Code cleanup

- Rename variables
- Change print messages
- Use sets where more appropriate
- Fix bug with removing build dependencies from chroots
This commit is contained in:
Kivi Kaitaniemi
2024-05-02 01:03:38 +03:00
parent 114c173c25
commit 5696ee647c
5 changed files with 105 additions and 68 deletions
+19 -10
View File
@@ -10,7 +10,8 @@ import traceback
import decman
import decman.error as err
import decman.lib as l
import decman.lib.aur as aur
import decman.config as conf
from decman.lib import fpm
def main():
@@ -94,9 +95,9 @@ def _set_up(store: l.Store):
if source is None:
l.print_error(
"Source was not specified. Please specify a source at least once with the --source argument."
"Source was not specified. Please specify a source with the '--source' argument."
)
l.print_info("Decman will remember the previous source.")
l.print_info("Decman will remember the previously specified source.")
sys.exit(1)
if source_changed or not store.allow_running_source_without_prompt:
@@ -130,6 +131,9 @@ def _set_up(store: l.Store):
class Core:
"""
Contains the main logic of decman.
"""
def __init__(self, store: l.Store, opts):
self.only_print, self.update_packages, self.update_files, self.update_units, self.upgrade_devel, self.force_build = opts
@@ -138,16 +142,19 @@ class Core:
self.source = _resolve_source()
self.pacman = l.Pacman()
self.systemctl = l.Systemd(store)
self.fpkg_search = aur.ExtendedPackageSearch(self.pacman)
self.fpkg_search = fpm.ExtendedPackageSearch(self.pacman)
for upkg in self.source.user_packages:
for upkg in self.source.all_user_pkgs():
self.fpkg_search.add_user_pkg(
aur.PackageInfo.from_user_package(upkg, self.pacman))
fpm.PackageInfo.from_user_package(upkg, self.pacman))
self.fpm = aur.ForeignPackageManager(store, self.pacman,
self.fpm = fpm.ForeignPackageManager(store, self.pacman,
self.fpkg_search)
def run(self):
"""
Run the main logic of decman.
"""
if self.update_units:
self._disable_units()
@@ -187,6 +194,7 @@ class Core:
l.print_summary("Upgrading packages.")
if not self.only_print:
self.pacman.upgrade()
if conf.enable_fpm:
self.fpm.upgrade(self.upgrade_devel, self.force_build,
self.source.ignored_packages)
@@ -202,13 +210,14 @@ class Core:
if not self.only_print:
self.pacman.install(to_install_pacman)
if conf.enable_fpm:
self.fpm.install(to_install_fpm, force=self.force_build)
def _create_and_remove_files(self):
l.print_list_summary("Copying files:",
l.print_list_summary("Installing files:",
self.source.all_file_targets(),
elements_per_line=1)
l.print_list_summary("Copying directories:",
l.print_list_summary("Installing directories:",
self.source.all_directory_targets(),
elements_per_line=1)
@@ -259,7 +268,7 @@ class Core:
def _resolve_source() -> l.Source:
enabled_systemd_user_units = {}
for user, units in decman.enabled_systemd_user_units:
for user, units in decman.enabled_systemd_user_units.items():
enabled_systemd_user_units[user] = set(units)
return l.Source(
+2 -1
View File
@@ -17,7 +17,7 @@ Commands are stored as methods in the Commands-class.
The global variable 'commands' of this module is an instance of the Commands-class.
To change the defalts, create a new child class of the Commands-class and set the 'commands'
variable to an instance of your class. Look in the README for examples.
variable to an instance of your class. Look in the example directory for an example.
"""
import typing
@@ -203,3 +203,4 @@ makepkg_user: str = "nobody"
build_dir: str = "/tmp/decman/build"
pkg_cache_dir: str = "/var/cache/decman"
aur_rpc_timeout: typing.Optional[int] = 30
enable_fpm: bool = True
+60 -32
View File
@@ -3,6 +3,7 @@ Library module for decman.
"""
import pwd
import shutil
import subprocess
import json
import os
@@ -18,13 +19,15 @@ _CYAN_PREFIX = "\033[96m"
_GREEN_PREFIX = "\033[92m"
_GRAY_PREFIX = "\033[90m"
_RESET_SUFFIX = "\033[m"
_SPACING = " "
_CONTINUATION_PREFIX = f"{_DECMAN_MSG_TAG}{_SPACING} "
def print_continuation(msg: str):
"""
Prints a message without a prefix.
"""
print(f"{_DECMAN_MSG_TAG}\t {msg}")
print(f"{_CONTINUATION_PREFIX}{msg}")
def print_error(error_msg: str):
@@ -53,7 +56,9 @@ def print_summary(msg: str):
def print_list_summary(msg: str,
l: list[str],
elements_per_line: typing.Optional[int] = None):
elements_per_line: typing.Optional[int] = None,
max_line_width: typing.Optional[int] = None,
limit_to_term_size: bool = True):
"""
Prints a summary message to the user along with a list of elements.
@@ -65,14 +70,37 @@ def print_list_summary(msg: str,
l = l.copy()
print_summary(msg)
print_continuation("")
if elements_per_line is None:
print_continuation(" ".join(l))
else:
elements_per_line = len(l)
if max_line_width is None:
max_line_width = 2**32 # Big enough to basically be unlimited
if limit_to_term_size:
max_line_width = shutil.get_terminal_size().columns - len(
_SPACING) - len(_CONTINUATION_PREFIX)
lines = [f"{l.pop(0)}"]
index = 0
elements_in_current_line = 1
while l:
to_print = []
for _ in range(elements_per_line):
to_print.append(l.pop())
print_continuation(" ".join(to_print))
next_element = l.pop(0)
can_fit_elements = elements_in_current_line + 1 <= elements_per_line
can_fit_text = len(lines[index]) + len(next_element) <= max_line_width
if can_fit_text and can_fit_elements:
lines[index] += f" {next_element}"
elements_in_current_line += 1
else:
lines.append(f"{next_element}")
index += 1
elements_in_current_line = 1
for line in lines:
print_continuation(line)
print_continuation("")
@@ -499,50 +527,50 @@ class Source:
result.append((module.name, module.version))
return result
def all_user_pkgs(self) -> list[decman.UserPackage]:
def all_user_pkgs(self) -> set[decman.UserPackage]:
"""
Returns all active UserPackages.
"""
result = []
result.extend(self.user_packages)
result = set()
result.update(self.user_packages)
for module in self.modules:
if module.enabled:
result.extend(module.user_packages())
result.update(module.user_packages())
return result
def _all_pacman_pkgs(self) -> list[str]:
result = []
result.extend(self.pacman_packages)
def _all_pacman_pkgs(self) -> set[str]:
result = set()
result.update(self.pacman_packages)
for module in self.modules:
if module.enabled:
result.extend(module.pacman_packages())
result.update(module.pacman_packages())
return result
def _all_foreign_pkgs(self) -> list[str]:
result = []
result.extend(self.aur_packages)
result.extend(map(lambda p: p.pkgname, self.user_packages))
def _all_foreign_pkgs(self) -> set[str]:
result = set()
result.update(self.aur_packages)
result.update(map(lambda p: p.pkgname, self.user_packages))
for module in self.modules:
if module.enabled:
result.extend(module.aur_packages())
result.extend(map(lambda p: p.pkgname, module.user_packages()))
result.update(module.aur_packages())
result.update(map(lambda p: p.pkgname, module.user_packages()))
return result
def _all_pkgs(self) -> list[str]:
result = []
result.extend(self._all_pacman_pkgs())
result.extend(self._all_foreign_pkgs())
def _all_pkgs(self) -> set[str]:
result = set()
result.update(self._all_pacman_pkgs())
result.update(self._all_foreign_pkgs())
return result
def _all_units(self) -> list[str]:
result = []
result.extend(self.systemd_units)
def _all_units(self) -> set[str]:
result = set()
result.update(self.systemd_units)
for module in self.modules:
if module.enabled:
result.extend(module.systemd_units())
result.update(module.systemd_units())
return result
def _all_user_units(self) -> dict[str, list[str]]:
def _all_user_units(self) -> dict[str, set[str]]:
result = {}
result.update(self.systemd_user_units)
for module in self.modules:
@@ -569,7 +597,7 @@ class Pacman:
conf.commands.list_pkgs(),
check=True,
stdout=subprocess.PIPE,
).stdout.decode().split('\n')
).stdout.decode().strip().split('\n')
return packages
except subprocess.CalledProcessError as error:
raise err.UserFacingError(
+10 -11
View File
@@ -673,13 +673,6 @@ class ForeignPackageManager:
) -> ResolvedDependencies:
"""
Resolves foreign dependencies of foreign packages.
Returns a tuple of (foreign_packages, pacman_deps)
foreign_packages are in the order they should be built
(the 1st element should be built 1st)
pacman_deps are dependencies that are required by the foreign packages.
"""
l.print_summary("Resolving foreign package dependencies.")
@@ -809,6 +802,8 @@ class PackageBuilder:
self.chroot_dir = os.path.join(self.chroot_wd_dir, "root")
self.pkgbase_dir_map = {}
self.original_wd = ""
self._pkgs_in_chroot = set(PackageBuilder.always_included_packages)
self._pkgs_in_chroot.update(resolved_deps.pacman_deps)
def __enter__(self):
self.store_wd()
@@ -876,9 +871,8 @@ class PackageBuilder:
except KeyError:
pass
subprocess.run(conf.commands.make_chroot(
self.chroot_dir, PackageBuilder.always_included_packages +
list(self._resolved_deps.pacman_deps)),
subprocess.run(conf.commands.make_chroot(self.chroot_dir,
list(self._pkgs_in_chroot)),
env=mkarchroot_env_vars,
check=True,
capture_output=conf.suppress_command_output)
@@ -955,9 +949,14 @@ class PackageBuilder:
l.print_info("Removing build dependencies from chroot.")
# FIX: If installed packages are virtual packages, removing them wont succeed.
if len(chroot_new_pacman_pkgs) != 0:
to_remove = []
for p in chroot_new_pacman_pkgs:
if p not in self._pkgs_in_chroot:
to_remove.append(strip_dependency(p))
subprocess.run(conf.commands.remove_chroot_packages(
self.chroot_dir, chroot_new_pacman_pkgs),
self.chroot_dir, to_remove),
check=True,
capture_output=conf.suppress_command_output)
+6 -6
View File
@@ -3,31 +3,31 @@
import unittest
from decman.error import UserFacingError
from decman.lib import Pacman, Store
from decman.lib.aur import ForeignPackageManager, DepGraph, ForeignPackage, ExtendedPackageSearch
from decman.lib.fpm import ForeignPackageManager, DepGraph, ForeignPackage, ExtendedPackageSearch
class TestVersionComparisons(unittest.TestCase):
def setUp(self):
pacman = Pacman()
self.aur = ForeignPackageManager(Store(), pacman,
self.pm = ForeignPackageManager(Store(), pacman,
ExtendedPackageSearch(pacman))
def test_should_upgrade_package_returns_true_on_newer_version(self):
self.assertTrue(
self.aur.should_upgrade_package("test", "0.1.9", "0.2.0"))
self.pm.should_upgrade_package("test", "0.1.9", "0.2.0"))
def test_should_upgrade_package_returns_false_on_older_version(self):
self.assertFalse(
self.aur.should_upgrade_package("test", "0.1.9", "0.1.8"))
self.pm.should_upgrade_package("test", "0.1.9", "0.1.8"))
def test_should_upgrade_package_returns_false_on_same_version(self):
self.assertFalse(
self.aur.should_upgrade_package("test", "0.1.9", "0.1.9"))
self.pm.should_upgrade_package("test", "0.1.9", "0.1.9"))
def test_should_upgrade_package_returns_true_on_devel(self):
self.assertTrue(
self.aur.should_upgrade_package("test-git",
self.pm.should_upgrade_package("test-git",
"0",
"0",
upgrade_devel=True))