S101: fix

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():

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,

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

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):

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,

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()

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,

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)