From 6a2dfb8176abfef690d928625b01785b79011273 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rg=20Thalheim?= Date: Mon, 25 Aug 2025 12:26:00 +0200 Subject: [PATCH] S101: fix --- pkgs/clan-cli/clan_cli/secrets/groups.py | 4 +++- pkgs/clan-cli/clan_cli/vars/migration.py | 4 +++- pkgs/clan-cli/clan_lib/api/serde.py | 4 +++- pkgs/clan-cli/clan_lib/async_run/__init__.py | 4 +++- pkgs/clan-cli/clan_lib/log_manager/api.py | 12 +++++++++--- pkgs/clan-cli/clan_lib/ssh/sudo_askpass_proxy.py | 12 +++++++----- pkgs/clan-vm-manager/clan_vm_manager/app.py | 5 ++++- pkgs/clan-vm-manager/clan_vm_manager/views/list.py | 8 ++++++-- 8 files changed, 38 insertions(+), 15 deletions(-) diff --git a/pkgs/clan-cli/clan_cli/secrets/groups.py b/pkgs/clan-cli/clan_cli/secrets/groups.py index e245bbb51..6f04a0aba 100644 --- a/pkgs/clan-cli/clan_cli/secrets/groups.py +++ b/pkgs/clan-cli/clan_cli/secrets/groups.py @@ -262,7 +262,9 @@ def add_secret( def get_groups(flake_dir: Path, what: str, name: str) -> list[str]: """Returns the list of group names the given user or machine is part of.""" - assert what in {"users", "machines"} + if what not in {"users", "machines"}: + msg = f"Invalid 'what' parameter: {what}. Must be 'users' or 'machines'" + raise ClanError(msg) groups_dir = sops_groups_folder(flake_dir) if not groups_dir.exists(): diff --git a/pkgs/clan-cli/clan_cli/vars/migration.py b/pkgs/clan-cli/clan_cli/vars/migration.py index 5057d3cb5..7b1b40736 100644 --- a/pkgs/clan-cli/clan_cli/vars/migration.py +++ b/pkgs/clan-cli/clan_cli/vars/migration.py @@ -80,7 +80,9 @@ def migrate_files( files_to_commit = [] for file in generator.files: if _migration_file_exists(machine, generator, file.name): - assert generator.migrate_fact is not None + if generator.migrate_fact is None: + msg = f"Generator {generator.name} has no migrate_fact defined" + raise ClanError(msg) files_to_commit += _migrate_file( machine, generator, diff --git a/pkgs/clan-cli/clan_lib/api/serde.py b/pkgs/clan-cli/clan_lib/api/serde.py index 35b9cae3e..463ddc6e6 100644 --- a/pkgs/clan-cli/clan_lib/api/serde.py +++ b/pkgs/clan-cli/clan_lib/api/serde.py @@ -223,7 +223,9 @@ def construct_value( # If the field is another dataclass # Field_value must be a dictionary if is_dataclass(t) and isinstance(field_value, dict): - assert isinstance(t, type) + if not isinstance(t, type): + msg = f"Expected a type, got {t}" + raise ClanError(msg) return construct_dataclass(t, field_value) # If the field expects a path diff --git a/pkgs/clan-cli/clan_lib/async_run/__init__.py b/pkgs/clan-cli/clan_lib/async_run/__init__.py index 3642abc6c..515c59fc1 100644 --- a/pkgs/clan-cli/clan_lib/async_run/__init__.py +++ b/pkgs/clan-cli/clan_lib/async_run/__init__.py @@ -281,7 +281,9 @@ class AsyncRuntime: for name, task in self.tasks.items(): if task.finished and task.async_opts.check: - assert task.result is not None + if task.result is None: + msg = f"Task {name} finished but has no result" + raise ClanError(msg) error = task.result.error if error is not None: if log.isEnabledFor(logging.DEBUG): diff --git a/pkgs/clan-cli/clan_lib/log_manager/api.py b/pkgs/clan-cli/clan_lib/log_manager/api.py index eb2608f36..07dce6767 100644 --- a/pkgs/clan-cli/clan_lib/log_manager/api.py +++ b/pkgs/clan-cli/clan_lib/log_manager/api.py @@ -16,7 +16,9 @@ def list_log_days() -> list[str]: AssertionError: If LOG_MANAGER_INSTANCE is not initialized. """ - assert LOG_MANAGER_INSTANCE is not None + if LOG_MANAGER_INSTANCE is None: + msg = "LOG_MANAGER_INSTANCE is not initialized" + raise ClanError(msg) return [day.date_day for day in LOG_MANAGER_INSTANCE.list_log_days()] @@ -38,7 +40,9 @@ def list_log_groups( AssertionError: If LOG_MANAGER_INSTANCE is not initialized. """ - assert LOG_MANAGER_INSTANCE is not None + if LOG_MANAGER_INSTANCE is None: + msg = "LOG_MANAGER_INSTANCE is not initialized" + raise ClanError(msg) return LOG_MANAGER_INSTANCE.filter(selector, date_day=date_day) @@ -63,7 +67,9 @@ def get_log_file( AssertionError: If LOG_MANAGER_INSTANCE is not initialized. """ - assert LOG_MANAGER_INSTANCE is not None + if LOG_MANAGER_INSTANCE is None: + msg = "LOG_MANAGER_INSTANCE is not initialized" + raise ClanError(msg) log_file = LOG_MANAGER_INSTANCE.get_log_file( op_key=id_key, diff --git a/pkgs/clan-cli/clan_lib/ssh/sudo_askpass_proxy.py b/pkgs/clan-cli/clan_lib/ssh/sudo_askpass_proxy.py index d5cf3a9b5..b0f720158 100644 --- a/pkgs/clan-cli/clan_lib/ssh/sudo_askpass_proxy.py +++ b/pkgs/clan-cli/clan_lib/ssh/sudo_askpass_proxy.py @@ -74,7 +74,9 @@ class SudoAskpassProxy: def _process(self, ssh_process: subprocess.Popen) -> None: """Execute the remote command with password proxying""" # Monitor SSH output for password requests - assert ssh_process.stdout is not None, "SSH process stdout is None" + if ssh_process.stdout is None: + msg = "SSH process stdout is None" + raise ClanError(msg) try: for line in ssh_process.stdout: line = line.strip() @@ -137,10 +139,10 @@ class SudoAskpassProxy: pass # Unclear why we have to close this manually, but pytest reports unclosed fd - assert self.ssh_process.stdout is not None - self.ssh_process.stdout.close() - assert self.ssh_process.stdin is not None - self.ssh_process.stdin.close() + if self.ssh_process.stdout is not None: + self.ssh_process.stdout.close() + if self.ssh_process.stdin is not None: + self.ssh_process.stdin.close() self.ssh_process = None if self.thread: self.thread.join() diff --git a/pkgs/clan-vm-manager/clan_vm_manager/app.py b/pkgs/clan-vm-manager/clan_vm_manager/app.py index 4a3781420..3f19e8c15 100644 --- a/pkgs/clan-vm-manager/clan_vm_manager/app.py +++ b/pkgs/clan-vm-manager/clan_vm_manager/app.py @@ -11,6 +11,7 @@ gi.require_version("Gtk", "4.0") gi.require_version("Adw", "1") from clan_lib.custom_logger import setup_logging +from clan_lib.errors import ClanError from gi.repository import Adw, Gdk, Gio, Gtk from clan_vm_manager.components.interfaces import ClanConfig @@ -118,7 +119,9 @@ class MainApplication(Adw.Application): css_provider = Gtk.CssProvider() css_provider.load_from_path(str(resource_path)) display = Gdk.Display.get_default() - assert display is not None + if display is None: + msg = "Could not get default display" + raise ClanError(msg) Gtk.StyleContext.add_provider_for_display( display, css_provider, diff --git a/pkgs/clan-vm-manager/clan_vm_manager/views/list.py b/pkgs/clan-vm-manager/clan_vm_manager/views/list.py index e5ef3511d..97b40db4d 100644 --- a/pkgs/clan-vm-manager/clan_vm_manager/views/list.py +++ b/pkgs/clan-vm-manager/clan_vm_manager/views/list.py @@ -116,7 +116,9 @@ class ClanList(Gtk.Box): add_action = Gio.SimpleAction.new("add", GLib.VariantType.new("s")) add_action.connect("activate", self.on_add) app = Gio.Application.get_default() - assert app is not None + if app is None: + msg = "Could not get default application" + raise ClanError(msg) app.add_action(add_action) # menu_model = Gio.Menu() @@ -214,7 +216,9 @@ class ClanList(Gtk.Box): build_logs_action.set_enabled(False) app = Gio.Application.get_default() - assert app is not None + if app is None: + msg = "Could not get default application" + raise ClanError(msg) app.add_action(open_action) app.add_action(build_logs_action)