JustAnyone/decman
1
0
Fork 0
mirror of https://github.com/kiviktnm/decman.git synced 2026-09-19 12:08:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve printed output

Browse Source
This commit is contained in:
Kivi Kaitaniemi
2025-12-17 21:54:53 +02:00
parent 757edfe7cc
commit 6a1f64c3dd
17 changed files with 204 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/README.md
+1 -1
View File
@@ -523,7 +523,7 @@ decman.sh(
### Errors
When your source needs to raise an error, decman provides `SourceError`s. These are the errors that should be raised when decman runs your `source.py` file.
When your source needs to raise an error, decman provides `SourceError`s. Running commands with `prg` and `sh` may raise `decman.core.error.CommandFailedError`s if `check` is set to `True`. These are the errors that should be raised when decman runs your `source.py` file.
```py
import decman
src/decman/__init__.py
+3
View File
@@ -119,6 +119,9 @@ def prg(
code, command_output = result
if code != 0:
output.print_warning(f"Command '{shlex.join(cmd)}' returned with an exit code {code}.")
if not pty:
for line in command_output.strip().split("\n"):
output.print_continuation(line.strip())
return command_output
src/decman/app.py
+33 -20
View File
@@ -81,11 +81,16 @@ def main():
output.print_traceback()
failed = True
except errors.CommandFailedError as error:
output.print_error(f"{error}")
output.print_error(str(error))
output.print_traceback()
failed = True
except ValueError as error:
output.print_error("ValueError raised from the source.")
output.print_error(str(error))
output.print_traceback()
failed = True
except errors.InvalidOnDisableError as error:
output.print_error(f"Invalid source. {error}")
output.print_error(str(error))
output.print_traceback()
failed = True
except Exception as error:
@@ -93,7 +98,9 @@ def main():
output.print_traceback()
failed = True
except OSError as error:
output.print_error(f"Failed to access decman store file '{_STORE_FILE}': {error.strerror}.")
output.print_error(
f"Failed to access decman store file '{_STORE_FILE}': {error.strerror or str(error)}."
)
output.print_error("This may cause already completed operations to run again.")
output.print_traceback()
finally:
@@ -186,7 +193,7 @@ def run_decman(store: _store.Store, args: argparse.Namespace) -> bool:
# Run main execution order
for step in execution_order:
output.print_debug(f"Running step '{step}'.")
output.print_info(f"Running step '{step}'.")
match step:
case "files":
if not file_manager.update_files(
@@ -201,8 +208,8 @@ def run_decman(store: _store.Store, args: argparse.Namespace) -> bool:
return False
else:
output.print_warning(
f"Plugin '{plugin_name}' configured in execution_order\
but not found in available plugins."
f"Plugin '{plugin_name}' configured in execution_order, "
"but not found in available plugins."
)
# On enable and on change should be ran last since they might depend on effects caused by
@@ -254,9 +261,9 @@ def _find_disabled_modules(store: _store.Store):
def _run_before_update(store: _store.Store, args: argparse.Namespace):
output.print_summary("Running 'before update' -hooks.")
output.print_summary("Running before_update -hooks.")
for module in decman.modules:
output.print_info(f"Running 'before update' for {module.name}.")
output.print_info(f"Running before_update for {module.name}.")
if not args.dry_run:
module.before_update(store)
@@ -265,12 +272,12 @@ def _run_on_disable(store: _store.Store, args: argparse.Namespace, disabled_modu
if not disabled_modules:
return
output.print_summary("Running 'on disable' -scripts.")
output.print_summary("Running on_disable -scripts.")
for disabled_module in disabled_modules:
on_disable_script = store["module_on_disable_scripts"].get(disabled_module, None)
if on_disable_script:
output.print_info(f"Running 'on disable' for {disabled_module}.")
output.print_info(f"Running on_disable for {disabled_module}.")
if not args.dry_run:
decman.prg([on_disable_script])
@@ -279,10 +286,13 @@ def _run_on_disable(store: _store.Store, args: argparse.Namespace, disabled_modu
def _run_on_enable(store: _store.Store, args: argparse.Namespace, new_modules: list[str]):
output.print_summary("Running 'on enable' -hooks.")
if not new_modules:
return
output.print_summary("Running on_enable -hooks.")
for module in decman.modules:
if module.name in new_modules:
output.print_info(f"Running 'on enable' for {module.name}.")
output.print_info(f"Running on_enable for {module.name}.")
if not args.dry_run:
module.on_enable(store)
@@ -295,30 +305,33 @@ def _run_on_enable(store: _store.Store, args: argparse.Namespace, new_modules: l
store["module_on_disable_scripts"][module.name] = script
except OSError as error:
output.print_error(
f"Failed to create 'on disable' script for module {module.name}:\
{error.strerror or str(error)}."
f"Failed to create on_disable script for module {module.name}: "
)
output.print_error(f"{error.strerror or str(error)}.")
output.print_traceback()
output.print_warning(
"This script will NOT be created when decman runs the next time."
)
output.print_warning(
"You should probably investigate the reason for the error. \
Try to fix it, and re-enable this module."
"You should investigate the reason for the error and try to fix it."
)
output.print_warning(
"Then disable and re-enable this module to create the script."
)
def _run_on_change(store: _store.Store, args: argparse.Namespace):
output.print_summary("Running 'on change' -hooks.")
output.print_summary("Running on_change -hooks.")
for module in decman.modules:
if module._changed:
output.print_info(f"Running 'on change' for {module.name}.")
output.print_info(f"Running on_change for {module.name}.")
if not args.dry_run:
module.on_change(store)
def _run_after_update(store: _store.Store, args: argparse.Namespace):
output.print_summary("Running 'after update' -hooks.")
output.print_summary("Running after_update -hooks.")
for module in decman.modules:
output.print_info(f"Running 'after update' for {module.name}.")
output.print_info(f"Running after_update for {module.name}.")
if not args.dry_run:
module.after_update(store)
src/decman/core/command.py
+22 -5
View File
@@ -6,6 +6,7 @@ import pwd
import select
import shlex
import shutil
import signal
import struct
import subprocess
import sys
@@ -151,6 +152,10 @@ def _build_env(
pass_environment: bool,
) -> dict[str, str]:
env = {}
output.print_debug(
f"Command environment is: user={user}, env_overrides={env_overrides},"
f"mimic_login={mimic_login}, pass_environment={pass_environment}"
)
if pass_environment:
env = os.environ.copy()
@@ -198,18 +203,30 @@ def _run_parent(master_fd: int, pid: int) -> tuple[int, str]:
old_tattr = termios.tcgetattr(stdin_fd)
tty.setraw(stdin_fd)
# Set PTY window size to match the current terminal size.
# We accept that resizing the real terminal causes issues here, it doesn't need to be handeled
# TODO: No actually, it's better to do resizing
rows, columns = shutil.get_terminal_size()
winsz = struct.pack("HHHH", rows, columns, 0, 0)
# Helper function to set PTY window size to the current terminal size
def resize_pty(*args):
try:
rows, cols = shutil.get_terminal_size()
winsz = struct.pack("HHHH", rows, cols, 0, 0)
fcntl.ioctl(master_fd, termios.TIOCSWINSZ, winsz)
except OSError:
# In case the child has exited before the signal handled was de-registered
pass
# Set PTY window size to match the current terminal size.
resize_pty()
# Handle terminal resizes automatically
old_winch = signal.getsignal(signal.SIGWINCH)
signal.signal(signal.SIGWINCH, resize_pty)
try:
output_bytes = _relay_pty(master_fd, stdin_fd, stdout_fd)
finally:
# Restore stdin termios attributes.
termios.tcsetattr(stdin_fd, termios.TCSADRAIN, old_tattr)
# Restore previous handler
signal.signal(signal.SIGWINCH, old_winch)
os.close(master_fd)
_, status = os.waitpid(pid, 0)
src/decman/core/error.py
+7 -4
View File
@@ -1,3 +1,6 @@
import shlex
class SourceError(Exception):
"""
Error raised manually from the user's source.
@@ -12,7 +15,7 @@ class FSInstallationFailedError(Exception):
def __init__(self, target: str, source: str, reason: str):
self.source = source
self.target = target
super().__init__(f"Failed to install file from {source} to {target}: {reason}")
super().__init__(f"Failed to install file from {source} to {target}: {reason}.")
class InvalidOnDisableError(Exception):
@@ -63,6 +66,6 @@ class CommandFailedError(Exception):
"""
def __init__(self, command: list[str], output: str) -> None:
self.command = command
self.output = output
super().__init__(f"Running a command '{' '.join(command)}' failed. Output: '{output}'.")
self.command = shlex.join(command)
self.output = output.strip()
super().__init__(f"Running a command '{self.command}' failed. Output:\n{self.output}")
src/decman/core/file_manager.py
+6 -3
View File
@@ -46,6 +46,8 @@ def update_files(
all_changed_files = []
store.ensure("all_files", [])
output.print_summary("Updating files.")
try:
output.print_debug("Applying common files.")
checked, changed = _install_files(files, dry_run=dry_run)
@@ -80,8 +82,8 @@ def update_files(
if len(module_changed_files) > 0:
output.print_debug(
f"Module '{mod.name}' set to changed due to modified \
files: {', '.join(module_changed_files)}"
f"Module '{mod.name}' set to changed due to modified "
f"files: '{"', '".join(module_changed_files)}'."
)
mod._changed = True
all_changed_files += module_changed_files
@@ -96,7 +98,6 @@ def update_files(
to_remove.append(file)
output.print_list("Updated files:", all_changed_files, elements_per_line=1)
output.print_list("Removing files:", to_remove, elements_per_line=1)
if not dry_run:
for file in to_remove:
@@ -106,6 +107,8 @@ def update_files(
output.print_warning(f"Failed to remove file: '{file}': {error.strerror}.")
store["all_files"] = all_checked_files
output.print_list("Removed files:", to_remove, elements_per_line=1)
return True
src/decman/core/fs.py
+7
View File
@@ -5,6 +5,7 @@ import typing
import decman.core.command as command
import decman.core.error as errors
import decman.core.output as output
class File:
@@ -133,6 +134,10 @@ class File:
parent_dir = os.path.dirname(dirct)
if not os.path.isdir(parent_dir):
create_missing_dirs(parent_dir, uid, gid)
output.print_debug(
f"While installing file '{target}' creating directory '{dirct}'."
)
os.mkdir(dirct)
if uid is not None:
@@ -143,6 +148,8 @@ class File:
create_missing_dirs(target_directory, self.uid, self.gid)
changed = self._write_content(target, variables, dry_run)
if changed:
output.print_debug(f"File '{target}' changed.")
if self.uid is not None and not dry_run:
assert self.gid is not None, "If uid is set, then gid is set."
src/decman/plugins/aur/__init__.py
+16 -8
View File
@@ -102,9 +102,15 @@ class AUR(plugins.Plugin):
if store["aur_packages_for_module"][mod.name] != aur_packages:
mod._changed = True
output.print_debug(
f"Module '{mod.name}' set to changed due to modified aur packages."
)
if store["custom_packages_for_module"][mod.name] != custom_package_strs:
mod._changed = True
output.print_debug(
f"Module '{mod.name}' set to changed due to modified custom packages."
)
self.packages |= aur_packages
self.custom_packages |= custom_packages
@@ -124,8 +130,10 @@ class AUR(plugins.Plugin):
try:
os.makedirs(pkg_cache_dir, exist_ok=True)
except OSError as error:
output.print_error("Failed to ensure AUR package cache directory exists.")
output.print_continuation(f"{error.strerror or error}")
output.print_error(
"Failed to ensure AUR package cache directory exists: "
f"{error.strerror or error}"
)
output.print_traceback()
return False
@@ -202,27 +210,27 @@ class AUR(plugins.Plugin):
fpm.install(list(to_install), force=force)
except AurRPCError as error:
output.print_error("Failed to fetch data from AUR RPC.")
output.print_continuation(f"{error}")
output.print_error(str(error))
output.print_traceback()
return False
except DependencyCycleError as error:
output.print_error("Foreign package dependency cycle detected.")
output.print_continuation(f"{error}")
output.print_error(str(error))
output.print_traceback()
return False
except PKGBUILDParseError as error:
output.print_error("Failed to parse a CustomPackage PKGBUILD.")
output.print_continuation(f"{error}")
output.print_error(str(error))
output.print_traceback()
return False
except ForeignPackageManagerError as error:
output.print_error("Foreign package manager failed.")
output.print_continuation(f"{error}")
output.print_error(str(error))
output.print_traceback()
return False
except errors.CommandFailedError as error:
output.print_error("Running a command failed.")
output.print_continuation(f"{error}")
output.print_error("Running a AUR command failed.")
output.print_error(str(error))
output.print_traceback()
return False
src/decman/plugins/aur/error.py
+3 -3
View File
@@ -11,9 +11,9 @@ class DependencyCycleError(Exception):
def __init__(self, package1: str, package2: str):
super().__init__(
f"Foreign package dependency cycle detected involving '{package1}' \
and '{package2}'. Foreign package dependencies are also required \
during package building and therefore dependency cycles cannot be handled."
f"Foreign package dependency cycle detected involving '{package1}' "
f"and '{package2}'. Foreign package dependencies are also required "
"during package building and therefore dependency cycles cannot be handled."
)
src/decman/plugins/aur/fpm.py
+19 -22
View File
@@ -54,8 +54,8 @@ def add_package_to_cache(store: _store.Store, package: str, version: str, path_t
for _, already_cached_path, __ in entries:
if already_cached_path == path_to_built_pkg:
output.print_debug(
f"Trying to cache {package} version {version}, but the version is already cached:\
{already_cached_path}"
f"Trying to cache {package} version {version}, but the version is already cached: "
f"{already_cached_path}"
)
return
entries.append(new_entry)
@@ -97,8 +97,8 @@ def clean_package_cache(store: _store.Store, package: str):
os.remove(oldest_path)
except OSError as e:
output.print_error(f"Failed to remove file '{oldest_path}' from the package cache.")
output.print_error(f"{e.strerror or e}")
output.print_continuation("You'll have to remove the file manually.")
output.print_error(e.strerror or str(e))
output.print_error("You'll have to remove the file manually.")
store["package_file_cache"][package] = entries
@@ -206,7 +206,7 @@ class ForeignPackageManager:
if ignored_pkgs is None:
ignored_pkgs = set()
output.print_summary("Determining foreign packages to upgrade.")
output.print_info("Determining foreign packages to upgrade.")
all_foreign_pkgs = self._pacman.get_versioned_foreign_packages()
all_explicit_foreign_pkgs = set(self._pacman.get_foreign_explicit())
@@ -258,25 +258,22 @@ class ForeignPackageManager:
output.print_list(
"The following foreign packages will be installed explicitly:",
list(resolved_dependencies.foreign_pkgs),
level=output.SUMMARY,
sorted(resolved_dependencies.foreign_pkgs),
)
output.print_list(
"The following foreign packages will be installed as dependencies:",
list(resolved_dependencies.foreign_dep_pkgs),
level=output.SUMMARY,
sorted(resolved_dependencies.foreign_dep_pkgs),
)
output.print_list(
"The following foreign packages will be built in order to install other packages.\
They will not be installed:",
list(resolved_dependencies.foreign_build_dep_pkgs),
level=output.SUMMARY,
"The following foreign packages will be built in order to install other packages. "
"They will not be installed:",
sorted(resolved_dependencies.foreign_build_dep_pkgs),
)
if not output.prompt_confirm("Proceed?", default=True):
raise ForeignPackageManagerError("Installing aborted.")
raise ForeignPackageManagerError("Installing aborted by the user.")
output.print_summary("Installing foreign package dependencies from pacman.")
self._pacman.install_dependencies(resolved_dependencies.pacman_deps)
@@ -438,8 +435,8 @@ class ForeignPackageManager:
should_upgrade = int(vercmp_output) < 0
output.print_debug(
f"Installed version is: {installed_version}. \
Available version is {fetched_version}. Should upgrade: {should_upgrade}"
f"Installed version is: {installed_version}. "
f"Available version is {fetched_version}. Should upgrade: {should_upgrade}."
)
return should_upgrade
except (ValueError, errors.CommandFailedError) as error:
@@ -538,8 +535,8 @@ class PackageBuilder:
)
assert pkgbase_info is not None, (
"All dependencies and packages should be resolved \
during the creation of ResolvedDependencies."
"All dependencies and packages should be resolved "
"during the creation of ResolvedDependencies."
)
output.print_debug(f"Git URL for '{pkgbase}' is '{pkgbase_info.git_url}'")
@@ -710,8 +707,8 @@ class PackageBuilder:
for foreign_pkg in chroot_foreign_pkgs:
entry = find_latest_cached_package(self._store, foreign_pkg)
assert entry is not None, (
"Build order determines that the dependencies are built \
before and thus are found in the cache."
"Build order determines that the dependencies are built "
"before and thus are found in the cache."
)
_, file = entry
@@ -739,8 +736,8 @@ before and thus are found in the cache."
if len(matches) != 1:
raise ForeignPackageManagerError(
f"Failed to build package '{pkgname}', because the pkg file cannot be determined.\
Possible files are: {matches}"
f"Failed to build package '{pkgname}', because the pkg file cannot be determined. "
f"Possible files are: {matches}"
)
return matches[0]
src/decman/plugins/aur/package.py
+2 -2
View File
@@ -422,8 +422,8 @@ class CustomPackage:
raise PKGBUILDParseError(
self.git_url,
self.pkgbuild_directory,
f"Package {self.pkgname} not found in SRCINFO.\
Packages present: {' '.join(found_pkgnames)}.",
f"Package {self.pkgname} not found in SRCINFO. "
f"Packages present: {' '.join(found_pkgnames)}.",
)
version_core = pkgver
src/decman/plugins/flatpak.py
+7 -1
View File
@@ -63,9 +63,15 @@ class Flatpak(plugins.Plugin):
if store["flatpaks_for_module"][mod.name] != packages:
mod._changed = True
output.print_debug(
f"Module '{mod.name}' set to changed due to modified system flatpaks."
)
if store["user_flatpaks_for_module"][mod.name] != user_packages:
mod._changed = True
output.print_debug(
f"Module '{mod.name}' set to changed due to modified user flatpaks."
)
self.packages |= packages
for user, flatpaks in user_packages.items():
@@ -86,7 +92,7 @@ class Flatpak(plugins.Plugin):
self.apply_packages(pm, user, packages, self.ignored_packages, dry_run)
except errors.CommandFailedError as error:
output.print_error("Running a flatpak command failed.")
output.print_continuation(f"{error}")
output.print_error(str(error))
output.print_traceback()
return False
return True
src/decman/plugins/pacman.py
+4 -1
View File
@@ -54,6 +54,9 @@ class Pacman(plugins.Plugin):
if store["packages_for_module"][mod.name] != packages:
mod._changed = True
output.print_debug(
f"Module '{mod.name}' set to changed due to modified pacman packages."
)
self.packages |= packages
@@ -107,7 +110,7 @@ class Pacman(plugins.Plugin):
pm.install(to_install)
except errors.CommandFailedError as error:
output.print_error("Running a pacman command failed.")
output.print_continuation(f"{error}")
output.print_error(str(error))
output.print_traceback()
return False
return True
src/decman/plugins/systemd.py
+40 -90
View File
@@ -1,6 +1,7 @@
import shutil
import decman.core.command as command
import decman.core.error as errors
import decman.core.module as module
import decman.core.output as output
import decman.core.store as _store
@@ -96,9 +97,15 @@ class Systemd(plugins.Plugin):
if store["systemd_units_for_module"][mod.name] != units:
mod._changed = True
output.print_debug(
f"Module '{mod.name}' set to changed due to modified systemd units."
)
if store["systemd_user_units_for_module"][mod.name] != user_units:
mod._changed = True
output.print_debug(
f"Module '{mod.name}' set to changed due to modified systemd user units."
)
self.enabled_units |= units
for user, u_units in user_units.items():
@@ -142,159 +149,102 @@ class Systemd(plugins.Plugin):
if unit not in self.enabled_user_units[user]:
user_units_to_disable[user].add(unit)
try:
output.print_info("Reloading systemd daemon.")
if not dry_run:
if not self.reload_daemon():
return False
self.reload_daemon()
output.print_info("Reloading systemd daemon for users.")
if not dry_run:
for user in user_units_to_enable.keys() | user_units_to_disable.keys():
if not self.reload_user_daemon(user):
return False
self.reload_user_daemon(user)
output.print_list("Enabling systemd units:", list(units_to_enable))
if not dry_run:
if not self.enable_units(store, units_to_enable):
return False
self.enable_units(store, units_to_enable)
output.print_list("Disabling systemd units:", list(units_to_disable))
if not dry_run:
if not self.disable_units(store, units_to_disable):
return False
self.disable_units(store, units_to_disable)
for user, units in user_units_to_enable.items():
output.print_list(f"Enabling systemd units for {user}:", list(units))
if not dry_run:
if not self.enable_user_units(store, units, user):
return False
self.enable_user_units(store, units, user)
for user, units in user_units_to_disable.items():
output.print_list(f"Disabling systemd units for {user}:", list(units))
if not dry_run:
if not self.disable_user_units(store, units, user):
self.disable_user_units(store, units, user)
except errors.CommandFailedError as error:
output.print_error("Running a systemd command failed.")
output.print_error(str(error))
output.print_traceback()
return False
return True
def enable_units(self, store: _store.Store, units: set[str]) -> bool:
def enable_units(self, store: _store.Store, units: set[str]):
"""
Enables the given units.
Returns ``True`` if the operation was successful.
"""
if not units:
return True
return
code, text = command.run(self.commands.enable_units(units))
output.print_command_output(text)
if code != 0:
output.print_error(f"Failed to enable systemd units '{' '.join(units)}'.")
output.print_error(f"Command exited with code: {code}")
output.print_error(f"{text}")
return False
cmd = self.commands.enable_units(units)
command.check_run_result(cmd, command.run(cmd))
store["systemd_units"] |= units
return True
def disable_units(self, store: _store.Store, units: set[str]) -> bool:
def disable_units(self, store: _store.Store, units: set[str]):
"""
Disables the given units.
Returns ``True`` if the operation was successful.
"""
if not units:
return True
return
code, text = command.run(self.commands.disable_units(units))
output.print_command_output(text)
if code != 0:
output.print_error(f"Failed to disable systemd units '{' '.join(units)}'.")
output.print_error(f"Command exited with code: {code}")
output.print_error(f"{text}")
return False
cmd = self.commands.disable_units(units)
command.check_run_result(cmd, command.run(cmd))
store["systemd_units"] -= units
return True
def enable_user_units(self, store: _store.Store, units: set[str], user: str) -> bool:
def enable_user_units(self, store: _store.Store, units: set[str], user: str):
"""
Enables the given units for the given user.
Returns ``True`` if the operation was successful.
"""
if not units:
return True
return
code, text = command.run(self.commands.enable_user_units(units, user))
output.print_command_output(text)
if code != 0:
output.print_error(
f"Failed to enable systemd units '{' '.join(units)}' for user {user}."
)
output.print_error(f"Command exited with code: {code}")
output.print_error(f"{text}")
return False
cmd = self.commands.enable_user_units(units, user)
command.check_run_result(cmd, command.run(cmd))
store["systemd_user_units"].setdefault(user, set())
store["systemd_user_units"][user] |= units
return True
def disable_user_units(self, store: _store.Store, units: set[str], user: str) -> bool:
def disable_user_units(self, store: _store.Store, units: set[str], user: str):
"""
Disables the given units for the given user.
Returns ``True`` if the operation was successful.
"""
if not units:
return True
return
code, text = command.run(self.commands.disable_user_units(units, user))
output.print_command_output(text)
if code != 0:
output.print_error(
f"Failed to disable systemd units '{' '.join(units)}' for user {user}."
)
output.print_error(f"Command exited with code: {code}")
output.print_error(f"{text}")
return False
cmd = self.commands.disable_user_units(units, user)
command.check_run_result(cmd, command.run(cmd))
store["systemd_user_units"].setdefault(user, set())
store["systemd_user_units"][user] -= units
return True
def reload_user_daemon(self, user: str) -> bool:
def reload_user_daemon(self, user: str):
"""
Reloads the user's systemd daemon.
Returns ``True`` if the operation was successful.
"""
code, text = command.run(self.commands.user_daemon_reload(user))
output.print_command_output(text)
if code != 0:
output.print_error(f"Failed to reload systemd daemon for {user}.")
output.print_error(f"Command exited with code: {code}")
output.print_error(f"{text}")
return False
return True
cmd = self.commands.user_daemon_reload(user)
_, text = command.check_run_result(cmd, command.run(cmd))
def reload_daemon(self) -> bool:
def reload_daemon(self):
"""
Reloads the systemd daemon.
Returns ``True`` if the operation was successful.
"""
code, text = command.run(self.commands.daemon_reload())
output.print_command_output(text)
if code != 0:
output.print_error("Failed to reload systemd daemon.")
output.print_error(f"Command exited with code: {code}")
output.print_error(f"{text}")
return False
return True
cmd = self.commands.daemon_reload()
command.check_run_result(cmd, command.run(cmd))
tests/test_decman_plugins_aur.py
+1 -5
View File
@@ -284,19 +284,15 @@ def test_apply_returns_false_on_aur_rpc_error(monkeypatch: pytest.MonkeyPatch) -
def fake_print_error(msg: str) -> None:
errors_logged.append(msg)
def fake_print_continuation(msg: str) -> None:
continuations.append(msg)
def fake_print_traceback() -> None:
traceback_called.append(True)
monkeypatch.setattr(aur_plugin.output, "print_error", fake_print_error)
monkeypatch.setattr(aur_plugin.output, "print_continuation", fake_print_continuation)
monkeypatch.setattr(aur_plugin.output, "print_traceback", fake_print_traceback)
ok = aur.apply(store, dry_run=False)
assert ok is False
assert any("AUR RPC" in msg or "fetch data from AUR RPC" in msg for msg in errors_logged)
assert any("RPC down" in msg for msg in continuations)
assert any("RPC down" in msg for msg in errors_logged)
assert traceback_called
tests/test_decman_plugins_pacman.py
+1 -5
View File
@@ -188,21 +188,17 @@ def test_apply_returns_false_on_command_failure(monkeypatch: pytest.MonkeyPatch)
def fake_print_error(msg: str) -> None:
errors_logged.append(msg)
def fake_print_continuation(msg: str) -> None:
continuations.append(msg)
def fake_print_traceback() -> None:
traceback_called.append(True)
monkeypatch.setattr(pacman_plugin.output, "print_error", fake_print_error)
monkeypatch.setattr(pacman_plugin.output, "print_continuation", fake_print_continuation)
monkeypatch.setattr(pacman_plugin.output, "print_traceback", fake_print_traceback)
ok = pacman.apply(store, dry_run=False)
assert ok is False
assert any("pacman command failed" in msg for msg in errors_logged)
assert any("boom" in msg for msg in continuations)
assert any("boom" in msg for msg in errors_logged)
assert traceback_called # at least once
tests/test_decman_plugins_systemd.py
+15 -27
View File
@@ -132,31 +132,25 @@ def test_apply_enables_and_disables_units_and_user_units(store):
def fake_reload_daemon():
calls.append(("reload_daemon",))
return True
def fake_reload_user_daemon(user):
calls.append(("reload_user_daemon", user))
return True
def fake_enable_units(store_arg, units_arg):
calls.append(("enable_units", frozenset(units_arg)))
store_arg["systemd_units"] |= units_arg
return True
def fake_disable_units(store_arg, units_arg):
calls.append(("disable_units", frozenset(units_arg)))
store_arg["systemd_units"] -= units_arg
return True
def fake_enable_user_units(store_arg, units_arg, user):
calls.append(("enable_user_units", user, frozenset(units_arg)))
store_arg["systemd_user_units"].setdefault(user, set()).update(units_arg)
return True
def fake_disable_user_units(store_arg, units_arg, user):
calls.append(("disable_user_units", user, frozenset(units_arg)))
store_arg["systemd_user_units"].setdefault(user, set()).difference_update(units_arg)
return True
# patch instance methods (no self parameter expected)
s.reload_daemon = fake_reload_daemon
@@ -167,7 +161,6 @@ def test_apply_enables_and_disables_units_and_user_units(store):
s.disable_user_units = fake_disable_user_units
result = s.apply(store, dry_run=False, params=None)
assert result is True
# reloads called once
assert ("reload_daemon",) in calls
@@ -223,8 +216,7 @@ def test_enable_units_success(monkeypatch, store, systemd):
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
result = systemd.enable_units(store, {"new.service"})
assert result is True
systemd.enable_units(store, {"new.service"})
assert store["systemd_units"] == {"old.service", "new.service"}
@@ -236,8 +228,8 @@ def test_enable_units_failure_does_not_update_store(monkeypatch, store, systemd)
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
result = systemd.enable_units(store, {"new.service"})
assert result is False
with pytest.raises(systemd_mod.errors.CommandFailedError):
systemd.enable_units(store, {"new.service"})
# unchanged
assert store["systemd_units"] == {"old.service"}
@@ -253,8 +245,7 @@ def test_disable_units_success(monkeypatch, store, systemd):
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
result = systemd.disable_units(store, {"new.service"})
assert result is True
systemd.disable_units(store, {"new.service"})
assert store["systemd_units"] == {"old.service"}
@@ -266,8 +257,8 @@ def test_disable_units_failure_does_not_update_store(monkeypatch, store, systemd
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
result = systemd.disable_units(store, {"new.service"})
assert result is False
with pytest.raises(systemd_mod.errors.CommandFailedError):
systemd.disable_units(store, {"new.service"})
assert store["systemd_units"] == {"old.service", "new.service"}
@@ -283,8 +274,7 @@ def test_enable_user_units_success(monkeypatch, store, systemd):
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
result = systemd.enable_user_units(store, {"newuser.service"}, "alice")
assert result is True
systemd.enable_user_units(store, {"newuser.service"}, "alice")
assert store["systemd_user_units"]["alice"] == {
"olduser.service",
"newuser.service",
@@ -299,8 +289,8 @@ def test_enable_user_units_failure_does_not_update_store(monkeypatch, store, sys
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
result = systemd.enable_user_units(store, {"newuser.service"}, "alice")
assert result is False
with pytest.raises(systemd_mod.errors.CommandFailedError):
systemd.enable_user_units(store, {"newuser.service"}, "alice")
assert store["systemd_user_units"]["alice"] == {"olduser.service"}
@@ -316,8 +306,7 @@ def test_disable_user_units_success(monkeypatch, store, systemd):
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
result = systemd.disable_user_units(store, {"newuser.service"}, "alice")
assert result is True
systemd.disable_user_units(store, {"newuser.service"}, "alice")
assert store["systemd_user_units"]["alice"] == {"olduser.service"}
@@ -329,8 +318,9 @@ def test_disable_user_units_failure_does_not_update_store(monkeypatch, store, sy
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
result = systemd.disable_user_units(store, {"newuser.service"}, "alice")
assert result is False
with pytest.raises(systemd_mod.errors.CommandFailedError):
systemd.disable_user_units(store, {"newuser.service"}, "alice")
assert store["systemd_user_units"]["alice"] == {
"olduser.service",
"newuser.service",
@@ -345,8 +335,7 @@ def test_reload_daemon_uses_command_run(monkeypatch, systemd):
return 0, "ok"
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
result = systemd.reload_daemon()
assert result is True
systemd.reload_daemon()
assert called["cmd"][:2] == ["systemctl", "daemon-reload"]
@@ -358,8 +347,7 @@ def test_reload_user_daemon_uses_command_run(monkeypatch, systemd):
return 0, "ok"
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
result = systemd.reload_user_daemon("alice")
assert result is True
systemd.reload_user_daemon("alice")
cmd = called["cmd"]
assert cmd[0] == "systemctl"
assert "--user" in cmd