Merge pull request 'make all same-module imports relative, the rest absolute' (#2013) from Mic92-type-checking into main

pkgs/clan-cli/clan_cli/__init__.py

@@ -5,9 +5,9 @@ from pathlib import Path
 from types import ModuleType
 
 # These imports are unused, but necessary for @API.register to run once.
-from clan_cli.api import admin, directory, disk, mdns_discovery, modules
-from clan_cli.arg_actions import AppendOptionAction
-from clan_cli.clan import show, update
+from .api import admin, directory, disk, mdns_discovery, modules
+from .arg_actions import AppendOptionAction
+from .clan import show, update
 
 # API endpoints that are not used in the cli.
 __all__ = ["directory", "mdns_discovery", "modules", "update", "disk", "admin"]
@@ -85,7 +85,7 @@ def register_common_flags(parser: argparse.ArgumentParser) -> None:
     has_subparsers = False
     for action in parser._actions:
         if isinstance(action, argparse._SubParsersAction):
-            for choice, child_parser in action.choices.items():
+            for _choice, child_parser in action.choices.items():
                 has_subparsers = True
                 register_common_flags(child_parser)
     if not has_subparsers:

pkgs/clan-cli/clan_cli/api/__init__.py

@@ -152,7 +152,7 @@ API.register(open_file)
     def to_json_schema(self) -> dict[str, Any]:
         from typing import get_type_hints
 
-        from clan_cli.api.util import type_to_dict
+        from .util import type_to_dict
 
         api_schema: dict[str, Any] = {
             "$comment": "An object containing API methods. ",

pkgs/clan-cli/clan_cli/api/admin.py

@@ -1,4 +1,3 @@
-from clan_cli.api import API
 from clan_cli.inventory import (
     AdminConfig,
     ServiceAdmin,
@@ -9,6 +8,8 @@ from clan_cli.inventory import (
     save_inventory,
 )
 
+from . import API
+
 
 @API.register
 def get_admin_service(base_url: str) -> ServiceAdmin | None:
@@ -26,12 +27,14 @@ def set_admin_service(
     base_url: str,
     allowed_keys: dict[str, str],
     instance_name: str = "admin",
-    extra_machines: list[str] = [],
+    extra_machines: list[str] | None = None,
 ) -> None:
     """
     Set the admin service of a clan
     Every machine is by default part of the admin service via the 'all' tag
     """
+    if extra_machines is None:
+        extra_machines = []
     inventory = load_inventory_eval(base_url)
 
     if not allowed_keys:

pkgs/clan-cli/clan_cli/api/cli.py

@@ -3,7 +3,7 @@
 import argparse
 import json
 
-from clan_cli.api import API
+from . import API
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description="Debug the API.")

pkgs/clan-cli/clan_cli/api/disk.py

@@ -1,4 +1,3 @@
-from clan_cli.api import API
 from clan_cli.inventory import (
     ServiceMeta,
     ServiceSingleDisk,
@@ -10,6 +9,8 @@ from clan_cli.inventory import (
     save_inventory,
 )
 
+from . import API
+
 
 def get_instance_name(machine_name: str) -> str:
     return f"{machine_name}-single-disk"

pkgs/clan-cli/clan_cli/api/modules.py

@@ -55,7 +55,7 @@ def extract_frontmatter(readme_content: str, err_scope: str) -> tuple[Frontmatte
                 f"Error parsing TOML frontmatter: {e}",
                 description=f"Invalid TOML frontmatter. {err_scope}",
                 location="extract_frontmatter",
-            )
+            ) from e
 
         return Frontmatter(**frontmatter_parsed), remaining_content
 
@@ -97,12 +97,12 @@ def get_modules(base_path: str) -> dict[str, str]:
     try:
         proc = run_no_stdout(cmd)
         res = proc.stdout.strip()
-    except ClanCmdError:
+    except ClanCmdError as e:
         raise ClanError(
             "clanInternals might not have clanModules attributes",
             location=f"list_modules {base_path}",
             description="Evaluation failed on clanInternals.clanModules attribute",
-        )
+        ) from e
     modules: dict[str, str] = json.loads(res)
     return modules
 

pkgs/clan-cli/clan_cli/api/serde.py

@@ -121,11 +121,13 @@ JsonValue = str | float | dict[str, Any] | list[Any] | None
 
 
 def construct_value(
-    t: type | UnionType, field_value: JsonValue, loc: list[str] = []
+    t: type | UnionType, field_value: JsonValue, loc: list[str] | None = None
 ) -> Any:
     """
     Construct a field value from a type hint and a field value.
     """
+    if loc is None:
+        loc = []
     if t is None and field_value:
         raise ClanError(f"Expected None but got: {field_value}", location=f"{loc}")
 
@@ -203,11 +205,15 @@ def construct_value(
         raise ClanError(f"Unhandled field type {t} with value {field_value}")
 
 
-def construct_dataclass(t: type[T], data: dict[str, Any], path: list[str] = []) -> T:
+def construct_dataclass(
+    t: type[T], data: dict[str, Any], path: list[str] | None = None
+) -> T:
     """
     type t MUST be a dataclass
     Dynamically instantiate a data class from a dictionary, handling nested data classes.
     """
+    if path is None:
+        path = []
     if not is_dataclass(t):
         raise ClanError(f"{t.__name__} is not a dataclass")
 
@@ -253,8 +259,10 @@ def construct_dataclass(t: type[T], data: dict[str, Any], path: list[str] = [])
 
 
 def from_dict(
-    t: type | UnionType, data: dict[str, Any] | Any, path: list[str] = []
+    t: type | UnionType, data: dict[str, Any] | Any, path: list[str] | None = None
 ) -> Any:
+    if path is None:
+        path = []
     if is_dataclass(t):
         if not isinstance(data, dict):
             raise ClanError(f"{data} is not a dict. Expected {t}")

pkgs/clan-cli/clan_cli/api/util.py

@@ -30,7 +30,7 @@ def inspect_dataclass_fields(t: type) -> dict[TypeVar, type]:
 
     type_params = origin.__parameters__
     # Create a map from type parameters to actual type arguments
-    type_map = dict(zip(type_params, type_args))
+    type_map = dict(zip(type_params, type_args, strict=False))
 
     return type_map
 
@@ -67,7 +67,11 @@ def apply_annotations(schema: dict[str, Any], annotations: list[Any]) -> dict[st
     return schema
 
 
-def type_to_dict(t: Any, scope: str = "", type_map: dict[TypeVar, type] = {}) -> dict:
+def type_to_dict(
+    t: Any, scope: str = "", type_map: dict[TypeVar, type] | None = None
+) -> dict:
+    if type_map is None:
+        type_map = {}
     if t is None:
         return {"type": "null"}
 

pkgs/clan-cli/clan_cli/backups/create.py

@@ -2,13 +2,13 @@ import argparse
 import json
 import logging
 
-from ..completions import (
+from clan_cli.completions import (
     add_dynamic_completer,
     complete_backup_providers_for_machine,
     complete_machines,
 )
-from ..errors import ClanError
-from ..machines.machines import Machine
+from clan_cli.errors import ClanError
+from clan_cli.machines.machines import Machine
 
 log = logging.getLogger(__name__)
 

pkgs/clan-cli/clan_cli/backups/list.py

@@ -3,13 +3,13 @@ import json
 import subprocess
 from dataclasses import dataclass
 
-from ..completions import (
+from clan_cli.completions import (
     add_dynamic_completer,
     complete_backup_providers_for_machine,
     complete_machines,
 )
-from ..errors import ClanError
-from ..machines.machines import Machine
+from clan_cli.errors import ClanError
+from clan_cli.machines.machines import Machine
 
 
 @dataclass

pkgs/clan-cli/clan_cli/backups/restore.py

@@ -2,13 +2,13 @@ import argparse
 import json
 import subprocess
 
-from ..completions import (
+from clan_cli.completions import (
     add_dynamic_completer,
     complete_backup_providers_for_machine,
     complete_machines,
 )
-from ..errors import ClanError
-from ..machines.machines import Machine
+from clan_cli.errors import ClanError
+from clan_cli.machines.machines import Machine
 
 
 def restore_service(machine: Machine, name: str, provider: str, service: str) -> None:

pkgs/clan-cli/clan_cli/clan/create.py

@@ -5,12 +5,11 @@ from dataclasses import dataclass
 from pathlib import Path
 
 from clan_cli.api import API
+from clan_cli.cmd import CmdOut, run
+from clan_cli.dirs import clan_templates
+from clan_cli.errors import ClanError
 from clan_cli.inventory import Inventory, init_inventory
-
-from ..cmd import CmdOut, run
-from ..dirs import clan_templates
-from ..errors import ClanError
-from ..nix import nix_command, nix_shell
+from clan_cli.nix import nix_command, nix_shell
 
 
 @dataclass

pkgs/clan-cli/clan_cli/clan/inspect.py

@@ -2,14 +2,20 @@ import argparse
 from dataclasses import dataclass
 from pathlib import Path
 
-from ..clan_uri import FlakeId
-from ..cmd import run
-from ..dirs import machine_gcroot
-from ..errors import ClanError
-from ..machines.list import list_nixos_machines
-from ..machines.machines import Machine
-from ..nix import nix_add_to_gcroots, nix_build, nix_config, nix_eval, nix_metadata
-from ..vms.inspect import VmConfig, inspect_vm
+from clan_cli.clan_uri import FlakeId
+from clan_cli.cmd import run
+from clan_cli.dirs import machine_gcroot
+from clan_cli.errors import ClanError
+from clan_cli.machines.list import list_nixos_machines
+from clan_cli.machines.machines import Machine
+from clan_cli.nix import (
+    nix_add_to_gcroots,
+    nix_build,
+    nix_config,
+    nix_eval,
+    nix_metadata,
+)
+from clan_cli.vms.inspect import VmConfig, inspect_vm
 
 
 @dataclass

pkgs/clan-cli/clan_cli/clan/show.py

@@ -5,11 +5,10 @@ from pathlib import Path
 from urllib.parse import urlparse
 
 from clan_cli.api import API
+from clan_cli.cmd import run_no_stdout
 from clan_cli.errors import ClanCmdError, ClanError
 from clan_cli.inventory import Meta
-
-from ..cmd import run_no_stdout
-from ..nix import nix_eval
+from clan_cli.nix import nix_eval
 
 log = logging.getLogger(__name__)
 
@@ -32,7 +31,7 @@ def show_clan_meta(uri: str | Path) -> Meta:
             "Evaluation failed on meta attribute",
             location=f"show_clan {uri}",
             description=str(e.cmd),
-        )
+        ) from e
 
     clan_meta = json.loads(res)
 

pkgs/clan-cli/clan_cli/cmd.py

@@ -29,28 +29,28 @@ def handle_output(process: subprocess.Popen, log: Log) -> tuple[str, str]:
     stderr_buf = b""
 
     while len(rlist) != 0:
-        r, _, _ = select.select(rlist, [], [], 0.1)
-        if len(r) == 0:  # timeout in select
+        readlist, _, _ = select.select(rlist, [], [], 0.1)
+        if len(readlist) == 0:  # timeout in select
             if process.poll() is None:
                 continue
             # Process has exited
             break
 
-        def handle_fd(fd: IO[Any] | None) -> bytes:
-            if fd and fd in r:
+        def handle_fd(fd: IO[Any] | None, readlist: list[IO[Any]]) -> bytes:
+            if fd and fd in readlist:
                 read = os.read(fd.fileno(), 4096)
                 if len(read) != 0:
                     return read
                 rlist.remove(fd)
             return b""
 
-        ret = handle_fd(process.stdout)
+        ret = handle_fd(process.stdout, readlist)
         if ret and log in [Log.STDOUT, Log.BOTH]:
             sys.stdout.buffer.write(ret)
             sys.stdout.flush()
 
         stdout_buf += ret
-        ret = handle_fd(process.stderr)
+        ret = handle_fd(process.stderr, readlist)
 
         if ret and log in [Log.STDERR, Log.BOTH]:
             sys.stderr.buffer.write(ret)
@@ -103,11 +103,13 @@ def run(
     *,
     input: bytes | None = None,  # noqa: A002
     env: dict[str, str] | None = None,
-    cwd: Path = Path.cwd(),
+    cwd: Path | None = None,
     log: Log = Log.STDERR,
     check: bool = True,
     error_msg: str | None = None,
 ) -> CmdOut:
+    if cwd is None:
+        cwd = Path.cwd()
     if input:
         glog.debug(
             f"""$: echo "{input.decode('utf-8', 'replace')}" | {shlex.join(cmd)} \nCaller: {get_caller()}"""
@@ -155,7 +157,7 @@ def run_no_stdout(
     cmd: list[str],
     *,
     env: dict[str, str] | None = None,
-    cwd: Path = Path.cwd(),
+    cwd: Path | None = None,
     log: Log = Log.STDERR,
     check: bool = True,
     error_msg: str | None = None,
@@ -164,6 +166,8 @@ def run_no_stdout(
     Like run, but automatically suppresses stdout, if not in DEBUG log level.
     If in DEBUG log level the stdout of commands will be shown.
     """
+    if cwd is None:
+        cwd = Path.cwd()
     if logging.getLogger(__name__.split(".")[0]).isEnabledFor(logging.DEBUG):
         return run(cmd, env=env, log=log, check=check, error_msg=error_msg)
     else:

pkgs/clan-cli/clan_cli/config/__init__.py

@@ -44,7 +44,9 @@ def map_type(nix_type: str) -> Any:
 
 
 # merge two dicts recursively
-def merge(a: dict, b: dict, path: list[str] = []) -> dict:
+def merge(a: dict, b: dict, path: list[str] | None = None) -> dict:
+    if path is None:
+        path = []
     for key in b:
         if key in a:
             if isinstance(a[key], dict) and isinstance(b[key], dict):
@@ -98,10 +100,10 @@ def cast(value: Any, input_type: Any, opt_description: str) -> Any:
             if len(value) > 1:
                 raise ClanError(f"Too many values for {opt_description}")
             return input_type(value[0])
-    except ValueError:
+    except ValueError as e:
         raise ClanError(
             f"Invalid type for option {opt_description} (expected {input_type.__name__})"
-        )
+        ) from e
 
 
 def options_for_machine(

pkgs/clan-cli/clan_cli/config/parsing.py

@@ -2,9 +2,9 @@ import json
 from pathlib import Path
 from typing import Any
 
-from ..cmd import run
-from ..errors import ClanError
-from ..nix import nix_eval
+from clan_cli.cmd import run
+from clan_cli.errors import ClanError
+from clan_cli.nix import nix_eval
 
 script_dir = Path(__file__).parent
 

pkgs/clan-cli/clan_cli/config/schema.py

@@ -18,8 +18,10 @@ def machine_schema(
     flake_dir: Path,
     config: dict[str, Any],
     clan_imports: list[str] | None = None,
-    option_path: list[str] = ["clan"],
+    option_path: list[str] | None = None,
 ) -> dict[str, Any]:
+    if option_path is None:
+        option_path = ["clan"]
     # use nix eval to lib.evalModules .#nixosConfigurations.<machine_name>.options.clan
     with NamedTemporaryFile(mode="w", dir=flake_dir) as clan_machine_settings_file:
         env = os.environ.copy()

pkgs/clan-cli/clan_cli/facts/__init__.py

@@ -1,7 +1,8 @@
 # !/usr/bin/env python3
 import argparse
 
-from ..hyperlink import help_hyperlink
+from clan_cli.hyperlink import help_hyperlink
+
 from .check import register_check_parser
 from .generate import register_generate_parser
 from .list import register_list_parser

pkgs/clan-cli/clan_cli/facts/check.py

@@ -2,8 +2,8 @@ import argparse
 import importlib
 import logging
 
-from ..completions import add_dynamic_completer, complete_machines
-from ..machines.machines import Machine
+from clan_cli.completions import add_dynamic_completer, complete_machines
+from clan_cli.machines.machines import Machine
 
 log = logging.getLogger(__name__)
 

pkgs/clan-cli/clan_cli/facts/generate.py

@@ -9,17 +9,17 @@ from pathlib import Path
 from tempfile import TemporaryDirectory
 
 from clan_cli.cmd import run
-
-from ..completions import (
+from clan_cli.completions import (
     add_dynamic_completer,
     complete_machines,
     complete_services_for_machine,
 )
-from ..errors import ClanError
-from ..git import commit_files
-from ..machines.inventory import get_all_machines, get_selected_machines
-from ..machines.machines import Machine
-from ..nix import nix_shell
+from clan_cli.errors import ClanError
+from clan_cli.git import commit_files
+from clan_cli.machines.inventory import get_all_machines, get_selected_machines
+from clan_cli.machines.machines import Machine
+from clan_cli.nix import nix_shell
+
 from .check import check_secrets
 from .public_modules import FactStoreBase
 from .secret_modules import SecretStoreBase

pkgs/clan-cli/clan_cli/facts/list.py

@@ -3,8 +3,8 @@ import importlib
 import json
 import logging
 
-from ..completions import add_dynamic_completer, complete_machines
-from ..machines.machines import Machine
+from clan_cli.completions import add_dynamic_completer, complete_machines
+from clan_cli.machines.machines import Machine
 
 log = logging.getLogger(__name__)
 

pkgs/clan-cli/clan_cli/facts/upload.py

@@ -4,10 +4,10 @@ import logging
 from pathlib import Path
 from tempfile import TemporaryDirectory
 
-from ..cmd import Log, run
-from ..completions import add_dynamic_completer, complete_machines
-from ..machines.machines import Machine
-from ..nix import nix_shell
+from clan_cli.cmd import Log, run
+from clan_cli.completions import add_dynamic_completer, complete_machines
+from clan_cli.machines.machines import Machine
+from clan_cli.nix import nix_shell
 
 log = logging.getLogger(__name__)
 

pkgs/clan-cli/clan_cli/flash.py

@@ -11,8 +11,7 @@ from pathlib import Path
 from tempfile import TemporaryDirectory
 from typing import Any
 
-from clan_cli.api import API
-
+from .api import API
 from .clan_uri import FlakeId
 from .cmd import Log, run
 from .completions import add_dynamic_completer, complete_machines
@@ -49,7 +48,7 @@ def list_possible_keymaps() -> list[str]:
 
     keymap_files = []
 
-    for root, _, files in os.walk(keymaps_dir):
+    for _root, _, files in os.walk(keymaps_dir):
         for file in files:
             if file.endswith(".map.gz"):
                 # Remove '.map.gz' ending
@@ -94,8 +93,10 @@ def flash_machine(
     dry_run: bool,
     write_efi_boot_entries: bool,
     debug: bool,
-    extra_args: list[str] = [],
+    extra_args: list[str] | None = None,
 ) -> None:
+    if extra_args is None:
+        extra_args = []
     system_config_nix: dict[str, Any] = {}
 
     if system_config.wifi_settings:
@@ -126,7 +127,9 @@ def flash_machine(
             try:
                 root_keys.append(key_path.read_text())
             except OSError as e:
-                raise ClanError(f"Cannot read SSH public key file: {key_path}: {e}")
+                raise ClanError(
+                    f"Cannot read SSH public key file: {key_path}: {e}"
+                ) from e
         system_config_nix["users"] = {
             "users": {"root": {"openssh": {"authorizedKeys": {"keys": root_keys}}}}
         }

pkgs/clan-cli/clan_cli/git.py

@@ -1,11 +1,9 @@
 from pathlib import Path
 
-# from clan_cli.dirs import find_git_repo_root
-from clan_cli.errors import ClanError
-from clan_cli.nix import run_cmd
-
 from .cmd import Log, run
+from .errors import ClanError
 from .locked_open import locked_open
+from .nix import run_cmd
 
 
 def commit_file(

pkgs/clan-cli/clan_cli/history/add.py

@@ -7,13 +7,12 @@ import logging
 from typing import Any
 
 from clan_cli.clan.inspect import FlakeConfig, inspect_flake
+from clan_cli.clan_uri import ClanURI
+from clan_cli.dirs import user_history_file
+from clan_cli.errors import ClanError
+from clan_cli.locked_open import read_history_file, write_history_file
 from clan_cli.machines.list import list_nixos_machines
 
-from ..clan_uri import ClanURI
-from ..dirs import user_history_file
-from ..errors import ClanError
-from ..locked_open import read_history_file, write_history_file
-
 log = logging.getLogger(__name__)
 
 

pkgs/clan-cli/clan_cli/history/update.py

@@ -3,11 +3,11 @@ import argparse
 import datetime
 
 from clan_cli.clan.inspect import inspect_flake
+from clan_cli.clan_uri import ClanURI
+from clan_cli.errors import ClanCmdError
+from clan_cli.locked_open import write_history_file
+from clan_cli.nix import nix_metadata
 
-from ..clan_uri import ClanURI
-from ..errors import ClanCmdError
-from ..locked_open import write_history_file
-from ..nix import nix_metadata
 from .add import HistoryEntry, list_history
 
 

pkgs/clan-cli/clan_cli/inventory/__init__.py

@@ -16,11 +16,11 @@ import json
 from pathlib import Path
 
 from clan_cli.api import API, dataclass_to_dict, from_dict
+from clan_cli.cmd import run_no_stdout
 from clan_cli.errors import ClanCmdError, ClanError
 from clan_cli.git import commit_file
+from clan_cli.nix import nix_eval
 
-from ..cmd import run_no_stdout
-from ..nix import nix_eval
 from .classes import (
     AdminConfig,
     Inventory,
@@ -114,7 +114,7 @@ def load_inventory_eval(flake_dir: str | Path) -> Inventory:
         inventory = from_dict(Inventory, data)
         return inventory
     except json.JSONDecodeError as e:
-        raise ClanError(f"Error decoding inventory from flake: {e}")
+        raise ClanError(f"Error decoding inventory from flake: {e}") from e
 
 
 def load_inventory_json(
@@ -134,7 +134,7 @@ def load_inventory_json(
                 inventory = from_dict(Inventory, res)
             except json.JSONDecodeError as e:
                 # Error decoding the inventory file
-                raise ClanError(f"Error decoding inventory file: {e}")
+                raise ClanError(f"Error decoding inventory file: {e}") from e
 
     if not inventory_file.exists():
         # Copy over the meta from the flake if the inventory is not initialized

pkgs/clan-cli/clan_cli/locked_open.py

@@ -5,9 +5,8 @@ from contextlib import contextmanager
 from pathlib import Path
 from typing import Any
 
-from clan_cli.jsonrpc import ClanJSONEncoder
-
 from .dirs import user_history_file
+from .jsonrpc import ClanJSONEncoder
 
 
 @contextmanager

pkgs/clan-cli/clan_cli/machines/create.py

@@ -2,10 +2,10 @@ import argparse
 import logging
 import re
 
-from ..api import API
-from ..clan_uri import FlakeId
-from ..errors import ClanError
-from ..inventory import (
+from clan_cli.api import API
+from clan_cli.clan_uri import FlakeId
+from clan_cli.errors import ClanError
+from clan_cli.inventory import (
     Machine,
     MachineDeploy,
     load_inventory_eval,

pkgs/clan-cli/clan_cli/machines/delete.py

@@ -1,12 +1,12 @@
 import argparse
 import shutil
 
-from ..api import API
-from ..clan_uri import FlakeId
-from ..completions import add_dynamic_completer, complete_machines
-from ..dirs import specific_machine_dir
-from ..errors import ClanError
-from ..inventory import load_inventory_json, save_inventory
+from clan_cli.api import API
+from clan_cli.clan_uri import FlakeId
+from clan_cli.completions import add_dynamic_completer, complete_machines
+from clan_cli.dirs import specific_machine_dir
+from clan_cli.errors import ClanError
+from clan_cli.inventory import load_inventory_json, save_inventory
 
 
 @API.register

pkgs/clan-cli/clan_cli/machines/facts.py

@@ -1,6 +1,6 @@
 from pathlib import Path
 
-from ..dirs import specific_machine_dir
+from clan_cli.dirs import specific_machine_dir
 
 
 def machine_has_fact(flake_dir: Path, machine: str, fact: str) -> bool:

pkgs/clan-cli/clan_cli/machines/hardware.py

@@ -7,13 +7,13 @@ from typing import Literal
 
 from clan_cli.api import API
 from clan_cli.clan_uri import FlakeId
+from clan_cli.cmd import run, run_no_stdout
+from clan_cli.completions import add_dynamic_completer, complete_machines
 from clan_cli.errors import ClanCmdError, ClanError
 from clan_cli.git import commit_file
+from clan_cli.machines.machines import Machine
+from clan_cli.nix import nix_config, nix_eval, nix_shell
 
-from ..cmd import run, run_no_stdout
-from ..completions import add_dynamic_completer, complete_machines
-from ..machines.machines import Machine
-from ..nix import nix_config, nix_eval, nix_shell
 from .types import machine_name_type
 
 log = logging.getLogger(__name__)
@@ -209,7 +209,7 @@ def generate_machine_hardware_info(
             "Invalid hardware-configuration.nix file",
             description="The hardware-configuration.nix file is invalid. Please check the file and try again.",
             location=f"{__name__} {hw_file}",
-        )
+        ) from e
 
     return HardwareReport(report_type)
 

pkgs/clan-cli/clan_cli/machines/install.py

@@ -8,14 +8,13 @@ from pathlib import Path
 from tempfile import TemporaryDirectory
 
 from clan_cli.api import API
-
-from ..clan_uri import FlakeId
-from ..cmd import Log, run
-from ..completions import add_dynamic_completer, complete_machines
-from ..facts.generate import generate_facts
-from ..machines.machines import Machine
-from ..nix import nix_shell
-from ..ssh.cli import is_ipv6, is_reachable, qrcode_scan
+from clan_cli.clan_uri import FlakeId
+from clan_cli.cmd import Log, run
+from clan_cli.completions import add_dynamic_completer, complete_machines
+from clan_cli.facts.generate import generate_facts
+from clan_cli.machines.machines import Machine
+from clan_cli.nix import nix_shell
+from clan_cli.ssh.cli import is_ipv6, is_reachable, qrcode_scan
 
 log = logging.getLogger(__name__)
 
@@ -30,8 +29,10 @@ def install_nixos(
     debug: bool = False,
     password: str | None = None,
     no_reboot: bool = False,
-    extra_args: list[str] = [],
+    extra_args: list[str] | None = None,
 ) -> None:
+    if extra_args is None:
+        extra_args = []
     secret_facts_module = importlib.import_module(machine.secret_facts_module)
     log.info(f"installing {machine.name}")
     secret_facts_store = secret_facts_module.SecretStore(machine=machine)

pkgs/clan-cli/clan_cli/machines/inventory.py

@@ -1,9 +1,10 @@
 import json
 from pathlib import Path
 
-from ..clan_uri import FlakeId
-from ..cmd import run
-from ..nix import nix_build, nix_config
+from clan_cli.clan_uri import FlakeId
+from clan_cli.cmd import run
+from clan_cli.nix import nix_build, nix_config
+
 from .machines import Machine
 
 

pkgs/clan-cli/clan_cli/machines/list.py

@@ -73,7 +73,7 @@ def list_nixos_machines(flake_url: str | Path) -> list[str]:
         data = json.loads(res)
         return data
     except json.JSONDecodeError as e:
-        raise ClanError(f"Error decoding machines from flake: {e}")
+        raise ClanError(f"Error decoding machines from flake: {e}") from e
 
 
 @dataclass

pkgs/clan-cli/clan_cli/machines/machine_group.py

@@ -1,7 +1,8 @@
 from collections.abc import Callable
 from typing import TypeVar
 
-from ..ssh import Host, HostGroup, HostResult
+from clan_cli.ssh import Host, HostGroup, HostResult
+
 from .machines import Machine
 
 T = TypeVar("T")

pkgs/clan-cli/clan_cli/machines/machines.py

@@ -6,11 +6,10 @@ from tempfile import NamedTemporaryFile
 from typing import Any
 
 from clan_cli.clan_uri import FlakeId
-
-from ..cmd import run_no_stdout
-from ..errors import ClanError
-from ..nix import nix_build, nix_config, nix_eval, nix_metadata
-from ..ssh import Host, parse_deployment_address
+from clan_cli.cmd import run_no_stdout
+from clan_cli.errors import ClanError
+from clan_cli.nix import nix_build, nix_config, nix_eval, nix_metadata
+from clan_cli.ssh import Host, parse_deployment_address
 
 log = logging.getLogger(__name__)
 
@@ -134,12 +133,14 @@ class Machine:
         attr: str,
         extra_config: None | dict = None,
         impure: bool = False,
-        nix_options: list[str] = [],
+        nix_options: list[str] | None = None,
     ) -> str | Path:
         """
         Build the machine and return the path to the result
         accepts a secret store and a facts store # TODO
         """
+        if nix_options is None:
+            nix_options = []
         config = nix_config()
         system = config["system"]
 
@@ -217,12 +218,14 @@ class Machine:
         refresh: bool = False,
         extra_config: None | dict = None,
         impure: bool = False,
-        nix_options: list[str] = [],
+        nix_options: list[str] | None = None,
     ) -> str:
         """
         eval a nix attribute of the machine
         @attr: the attribute to get
         """
+        if nix_options is None:
+            nix_options = []
         if attr in self._eval_cache and not refresh and extra_config is None:
             return self._eval_cache[attr]
 
@@ -239,13 +242,15 @@ class Machine:
         refresh: bool = False,
         extra_config: None | dict = None,
         impure: bool = False,
-        nix_options: list[str] = [],
+        nix_options: list[str] | None = None,
     ) -> Path:
         """
         build a nix attribute of the machine
         @attr: the attribute to get
         """
 
+        if nix_options is None:
+            nix_options = []
         if attr in self._build_cache and not refresh and extra_config is None:
             return self._build_cache[attr]
 

pkgs/clan-cli/clan_cli/machines/update.py

@@ -7,17 +7,17 @@ import sys
 
 from clan_cli.api import API
 from clan_cli.clan_uri import FlakeId
+from clan_cli.cmd import run
+from clan_cli.completions import add_dynamic_completer, complete_machines
+from clan_cli.errors import ClanError
+from clan_cli.facts.generate import generate_facts
+from clan_cli.facts.upload import upload_secrets
+from clan_cli.inventory import Machine as InventoryMachine
+from clan_cli.machines.machines import Machine
+from clan_cli.nix import nix_command, nix_metadata
+from clan_cli.ssh import HostKeyCheck
+from clan_cli.vars.generate import generate_vars
 
-from ..cmd import run
-from ..completions import add_dynamic_completer, complete_machines
-from ..errors import ClanError
-from ..facts.generate import generate_facts
-from ..facts.upload import upload_secrets
-from ..inventory import Machine as InventoryMachine
-from ..machines.machines import Machine
-from ..nix import nix_command, nix_metadata
-from ..ssh import HostKeyCheck
-from ..vars.generate import generate_vars
 from .inventory import get_all_machines, get_selected_machines
 from .machine_group import MachineGroup
 
@@ -82,7 +82,7 @@ def upload_sources(
     except (json.JSONDecodeError, OSError) as e:
         raise ClanError(
             f"failed to parse output of {shlex.join(cmd)}: {e}\nGot: {proc.stdout}"
-        )
+        ) from e
 
 
 @API.register
@@ -180,7 +180,7 @@ def update(args: argparse.Namespace) -> None:
                 if machine.deployment.get("requireExplicitUpdate", False):
                     continue
                 try:
-                    machine.build_host
+                    machine.build_host  # noqa: B018
                 except ClanError:  # check if we have a build host set
                     ignored_machines.append(machine)
                     continue

pkgs/clan-cli/clan_cli/nix/__init__.py

@@ -4,8 +4,8 @@ import tempfile
 from pathlib import Path
 from typing import Any
 
-from ..cmd import run, run_no_stdout
-from ..dirs import nixpkgs_flake, nixpkgs_source
+from clan_cli.cmd import run, run_no_stdout
+from clan_cli.dirs import nixpkgs_flake, nixpkgs_source
 
 
 def nix_command(flags: list[str]) -> list[str]:

pkgs/clan-cli/clan_cli/qemu/qmp.py

@@ -138,10 +138,10 @@ class QEMUMonitorProtocol:
                 self.__sock.settimeout(wait)
             try:
                 ret = self.__json_read(only_event=True)
-            except TimeoutError:
-                raise QMPTimeoutError("Timeout waiting for event")
-            except Exception:
-                raise QMPConnectError("Error while reading from socket")
+            except TimeoutError as e:
+                raise QMPTimeoutError("Timeout waiting for event") from e
+            except OSError as e:
+                raise QMPConnectError("Error while reading from socket") from e
             if ret is None:
                 raise QMPConnectError("Error while reading from socket")
             self.__sock.settimeout(None)

pkgs/clan-cli/clan_cli/secrets/folders.py

@@ -3,7 +3,7 @@ import shutil
 from collections.abc import Callable
 from pathlib import Path
 
-from ..errors import ClanError
+from clan_cli.errors import ClanError
 
 
 def get_sops_folder(flake_dir: Path) -> Path:
@@ -38,8 +38,8 @@ def remove_object(path: Path, name: str) -> list[Path]:
     try:
         shutil.rmtree(path / name)
         paths_to_commit.append(path / name)
-    except FileNotFoundError:
-        raise ClanError(f"{name} not found in {path}")
+    except FileNotFoundError as e:
+        raise ClanError(f"{name} not found in {path}") from e
     if not os.listdir(path):
         os.rmdir(path)
     return paths_to_commit

pkgs/clan-cli/clan_cli/secrets/groups.py

@@ -2,17 +2,17 @@ import argparse
 import os
 from pathlib import Path
 
-from clan_cli.git import commit_files
-
-from ..completions import (
+from clan_cli.completions import (
     add_dynamic_completer,
     complete_groups,
     complete_machines,
     complete_secrets,
     complete_users,
 )
-from ..errors import ClanError
-from ..machines.types import machine_name_type, validate_hostname
+from clan_cli.errors import ClanError
+from clan_cli.git import commit_files
+from clan_cli.machines.types import machine_name_type, validate_hostname
+
 from . import secrets
 from .folders import (
     sops_groups_folder,

pkgs/clan-cli/clan_cli/secrets/import_sops.py

@@ -3,15 +3,16 @@ import json
 import sys
 from pathlib import Path
 
-from ..cmd import run
-from ..completions import (
+from clan_cli.cmd import run
+from clan_cli.completions import (
     add_dynamic_completer,
     complete_groups,
     complete_machines,
     complete_users,
 )
-from ..errors import ClanError
-from ..nix import nix_shell
+from clan_cli.errors import ClanError
+from clan_cli.nix import nix_shell
+
 from .secrets import encrypt_secret, sops_secrets_folder
 
 

pkgs/clan-cli/clan_cli/secrets/key.py

@@ -2,9 +2,9 @@ import argparse
 import logging
 from pathlib import Path
 
+from clan_cli.errors import ClanError
 from clan_cli.git import commit_files
 
-from ..errors import ClanError
 from .secrets import update_secrets
 from .sops import default_sops_key_path, generate_private_key, get_public_key
 
@@ -22,10 +22,12 @@ def extract_public_key(filepath: Path) -> str:
                 if line.startswith("# public key:"):
                     # Extract and return the public key part after the prefix
                     return line.strip().split(": ")[1]
-    except FileNotFoundError:
-        raise ClanError(f"The file at {filepath} was not found.")
-    except Exception as e:
-        raise ClanError(f"An error occurred while extracting the public key: {e}")
+    except FileNotFoundError as e:
+        raise ClanError(f"The file at {filepath} was not found.") from e
+    except OSError as e:
+        raise ClanError(
+            f"An error occurred while extracting the public key: {e}"
+        ) from e
 
     raise ClanError(f"Could not find the public key in the file at {filepath}.")
 

pkgs/clan-cli/clan_cli/secrets/machines.py

@@ -1,10 +1,15 @@
 import argparse
 from pathlib import Path
 
-from ..completions import add_dynamic_completer, complete_machines, complete_secrets
-from ..errors import ClanError
-from ..git import commit_files
-from ..machines.types import machine_name_type, validate_hostname
+from clan_cli.completions import (
+    add_dynamic_completer,
+    complete_machines,
+    complete_secrets,
+)
+from clan_cli.errors import ClanError
+from clan_cli.git import commit_files
+from clan_cli.machines.types import machine_name_type, validate_hostname
+
 from . import secrets
 from .folders import (
     list_objects,

pkgs/clan-cli/clan_cli/secrets/secrets.py

@@ -8,17 +8,18 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import IO
 
-from .. import tty
-from ..clan_uri import FlakeId
-from ..completions import (
+from clan_cli import tty
+from clan_cli.clan_uri import FlakeId
+from clan_cli.completions import (
     add_dynamic_completer,
     complete_groups,
     complete_machines,
     complete_secrets,
     complete_users,
 )
-from ..errors import ClanError
-from ..git import commit_files
+from clan_cli.errors import ClanError
+from clan_cli.git import commit_files
+
 from .folders import (
     list_objects,
     sops_groups_folder,
@@ -84,11 +85,19 @@ def encrypt_secret(
     flake_dir: Path,
     secret_path: Path,
     value: IO[str] | str | bytes | None,
-    add_users: list[str] = [],
-    add_machines: list[str] = [],
-    add_groups: list[str] = [],
-    meta: dict = {},
+    add_users: list[str] | None = None,
+    add_machines: list[str] | None = None,
+    add_groups: list[str] | None = None,
+    meta: dict | None = None,
 ) -> None:
+    if meta is None:
+        meta = {}
+    if add_groups is None:
+        add_groups = []
+    if add_machines is None:
+        add_machines = []
+    if add_users is None:
+        add_users = []
     key = ensure_sops_key(flake_dir)
     recipient_keys = set([])
 

pkgs/clan-cli/clan_cli/secrets/sops.py

@@ -9,10 +9,11 @@ from pathlib import Path
 from tempfile import NamedTemporaryFile
 from typing import IO
 
-from ..cmd import Log, run
-from ..dirs import user_config_dir
-from ..errors import ClanError
-from ..nix import nix_shell
+from clan_cli.cmd import Log, run
+from clan_cli.dirs import user_config_dir
+from clan_cli.errors import ClanError
+from clan_cli.nix import nix_shell
+
 from .folders import sops_machines_folder, sops_users_folder
 
 
@@ -146,8 +147,10 @@ def encrypt_file(
     secret_path: Path,
     content: IO[str] | str | bytes | None,
     pubkeys: list[str],
-    meta: dict = {},
+    meta: dict | None = None,
 ) -> None:
+    if meta is None:
+        meta = {}
     folder = secret_path.parent
     folder.mkdir(parents=True, exist_ok=True)
 
@@ -224,10 +227,10 @@ def write_key(path: Path, publickey: str, overwrite: bool) -> None:
         if not overwrite:
             flags |= os.O_EXCL
         fd = os.open(path / "key.json", flags)
-    except FileExistsError:
+    except FileExistsError as e:
         raise ClanError(
             f"{path.name} already exists in {path}. Use --force to overwrite."
-        )
+        ) from e
     with os.fdopen(fd, "w") as f:
         json.dump({"publickey": publickey, "type": "age"}, f, indent=2)
 
@@ -237,7 +240,7 @@ def read_key(path: Path) -> str:
         try:
             key = json.load(f)
         except json.JSONDecodeError as e:
-            raise ClanError(f"Failed to decode {path.name}: {e}")
+            raise ClanError(f"Failed to decode {path.name}: {e}") from e
     if key["type"] != "age":
         raise ClanError(
             f"{path.name} is not an age key but {key['type']}. This is not supported"

pkgs/clan-cli/clan_cli/secrets/types.py

@@ -4,7 +4,8 @@ import re
 from collections.abc import Callable
 from pathlib import Path
 
-from ..errors import ClanError
+from clan_cli.errors import ClanError
+
 from .sops import get_public_key
 
 VALID_SECRET_NAME = re.compile(r"^[a-zA-Z0-9._-]+$")

pkgs/clan-cli/clan_cli/secrets/users.py

@@ -1,13 +1,10 @@
 import argparse
 from pathlib import Path
 
-from ..completions import (
-    add_dynamic_completer,
-    complete_secrets,
-    complete_users,
-)
-from ..errors import ClanError
-from ..git import commit_files
+from clan_cli.completions import add_dynamic_completer, complete_secrets, complete_users
+from clan_cli.errors import ClanError
+from clan_cli.git import commit_files
+
 from . import secrets
 from .folders import list_objects, remove_object, sops_secrets_folder, sops_users_folder
 from .secrets import update_secrets

pkgs/clan-cli/clan_cli/ssh/__init__.py

@@ -18,7 +18,7 @@ from shlex import quote
 from threading import Thread
 from typing import IO, Any, Generic, TypeVar
 
-from ..errors import ClanError
+from clan_cli.errors import ClanError
 
 # https://no-color.org
 DISABLE_COLOR = not sys.stderr.isatty() or os.environ.get("NO_COLOR", "") != ""
@@ -54,9 +54,9 @@ class CommandFormatter(logging.Formatter):
             prefix_color = ansi_color(self.hostname_colorcode(command_prefix))
             color_reset = "\x1b[0m"
 
-        setattr(record, "color", color)
-        setattr(record, "prefix_color", prefix_color)
-        setattr(record, "color_reset", color_reset)
+        record.color = color
+        record.prefix_color = prefix_color
+        record.color_reset = color_reset
 
         return super().format(record)
 
@@ -144,9 +144,9 @@ class Host:
         forward_agent: bool = False,
         command_prefix: str | None = None,
         host_key_check: HostKeyCheck = HostKeyCheck.STRICT,
-        meta: dict[str, Any] = {},
+        meta: dict[str, Any] | None = None,
         verbose_ssh: bool = False,
-        ssh_options: dict[str, str] = {},
+        ssh_options: dict[str, str] | None = None,
     ) -> None:
         """
         Creates a Host
@@ -158,6 +158,10 @@ class Host:
         @verbose_ssh: Enables verbose logging on ssh connections
         @meta: meta attributes associated with the host. Those can be accessed in custom functions passed to `run_function`
         """
+        if ssh_options is None:
+            ssh_options = {}
+        if meta is None:
+            meta = {}
         self.host = host
         self.user = user
         self.port = port
@@ -200,7 +204,9 @@ class Host:
         start = time.time()
         last_output = time.time()
         while len(rlist) != 0:
-            r, _, _ = select.select(rlist, [], [], min(timeout, NO_OUTPUT_TIMEOUT))
+            readlist, _, _ = select.select(
+                rlist, [], [], min(timeout, NO_OUTPUT_TIMEOUT)
+            )
 
             def print_from(
                 print_fd: IO[str], print_buf: str, is_err: bool = False
@@ -227,11 +233,11 @@ class Host:
                 last_output = time.time()
                 return (last_output, print_buf)
 
-            if print_std_fd in r and print_std_fd is not None:
+            if print_std_fd in readlist and print_std_fd is not None:
                 (last_output, print_std_buf) = print_from(
                     print_std_fd, print_std_buf, is_err=False
                 )
-            if print_err_fd in r and print_err_fd is not None:
+            if print_err_fd in readlist and print_err_fd is not None:
                 (last_output, print_err_buf) = print_from(
                     print_err_fd, print_err_buf, is_err=True
                 )
@@ -245,8 +251,8 @@ class Host:
                     extra=dict(command_prefix=self.command_prefix),
                 )
 
-            def handle_fd(fd: IO[Any] | None) -> str:
-                if fd and fd in r:
+            def handle_fd(fd: IO[Any] | None, readlist: list[IO[Any]]) -> str:
+                if fd and fd in readlist:
                     read = os.read(fd.fileno(), 4096)
                     if len(read) == 0:
                         rlist.remove(fd)
@@ -254,8 +260,8 @@ class Host:
                         return read.decode("utf-8")
                 return ""
 
-            stdout_buf += handle_fd(stdout)
-            stderr_buf += handle_fd(stderr)
+            stdout_buf += handle_fd(stdout, readlist)
+            stderr_buf += handle_fd(stderr, readlist)
 
             if now - last_output >= timeout:
                 break
@@ -268,11 +274,13 @@ class Host:
         shell: bool,
         stdout: FILE = None,
         stderr: FILE = None,
-        extra_env: dict[str, str] = {},
+        extra_env: dict[str, str] | None = None,
         cwd: None | str | Path = None,
         check: bool = True,
         timeout: float = math.inf,
     ) -> subprocess.CompletedProcess[str]:
+        if extra_env is None:
+            extra_env = {}
         with ExitStack() as stack:
             read_std_fd, write_std_fd = (None, None)
             read_err_fd, write_err_fd = (None, None)
@@ -354,7 +362,7 @@ class Host:
         cmd: str | list[str],
         stdout: FILE = None,
         stderr: FILE = None,
-        extra_env: dict[str, str] = {},
+        extra_env: dict[str, str] | None = None,
         cwd: None | str | Path = None,
         check: bool = True,
         timeout: float = math.inf,
@@ -371,6 +379,8 @@ class Host:
 
         @return subprocess.CompletedProcess result of the command
         """
+        if extra_env is None:
+            extra_env = {}
         shell = False
         if isinstance(cmd, str):
             cmd = [cmd]
@@ -397,7 +407,7 @@ class Host:
         stdout: FILE = None,
         stderr: FILE = None,
         become_root: bool = False,
-        extra_env: dict[str, str] = {},
+        extra_env: dict[str, str] | None = None,
         cwd: None | str | Path = None,
         check: bool = True,
         timeout: float = math.inf,
@@ -418,6 +428,8 @@ class Host:
 
         @return subprocess.CompletedProcess result of the ssh command
         """
+        if extra_env is None:
+            extra_env = {}
         sudo = ""
         if become_root and self.user != "root":
             sudo = "sudo -- "
@@ -548,13 +560,15 @@ class HostGroup:
         results: Results,
         stdout: FILE = None,
         stderr: FILE = None,
-        extra_env: dict[str, str] = {},
+        extra_env: dict[str, str] | None = None,
         cwd: None | str | Path = None,
         check: bool = True,
         verbose_ssh: bool = False,
         timeout: float = math.inf,
         tty: bool = False,
     ) -> None:
+        if extra_env is None:
+            extra_env = {}
         try:
             proc = host.run_local(
                 cmd,
@@ -577,13 +591,15 @@ class HostGroup:
         results: Results,
         stdout: FILE = None,
         stderr: FILE = None,
-        extra_env: dict[str, str] = {},
+        extra_env: dict[str, str] | None = None,
         cwd: None | str | Path = None,
         check: bool = True,
         verbose_ssh: bool = False,
         timeout: float = math.inf,
         tty: bool = False,
     ) -> None:
+        if extra_env is None:
+            extra_env = {}
         try:
             proc = host.run(
                 cmd,
@@ -622,13 +638,15 @@ class HostGroup:
         local: bool = False,
         stdout: FILE = None,
         stderr: FILE = None,
-        extra_env: dict[str, str] = {},
+        extra_env: dict[str, str] | None = None,
         cwd: None | str | Path = None,
         check: bool = True,
         timeout: float = math.inf,
         verbose_ssh: bool = False,
         tty: bool = False,
     ) -> Results:
+        if extra_env is None:
+            extra_env = {}
         results: Results = []
         threads = []
         for host in self.hosts:
@@ -665,7 +683,7 @@ class HostGroup:
         cmd: str | list[str],
         stdout: FILE = None,
         stderr: FILE = None,
-        extra_env: dict[str, str] = {},
+        extra_env: dict[str, str] | None = None,
         cwd: None | str | Path = None,
         check: bool = True,
         verbose_ssh: bool = False,
@@ -682,6 +700,8 @@ class HostGroup:
 
         @return a lists of tuples containing Host and the result of the command for this Host
         """
+        if extra_env is None:
+            extra_env = {}
         return self._run(
             cmd,
             stdout=stdout,
@@ -699,7 +719,7 @@ class HostGroup:
         cmd: str | list[str],
         stdout: FILE = None,
         stderr: FILE = None,
-        extra_env: dict[str, str] = {},
+        extra_env: dict[str, str] | None = None,
         cwd: None | str | Path = None,
         check: bool = True,
         timeout: float = math.inf,
@@ -715,6 +735,8 @@ class HostGroup:
 
         @return a lists of tuples containing Host and the result of the command for this Host
         """
+        if extra_env is None:
+            extra_env = {}
         return self._run(
             cmd,
             local=True,
@@ -761,8 +783,13 @@ class HostGroup:
 
 
 def parse_deployment_address(
-    machine_name: str, host: str, forward_agent: bool = True, meta: dict[str, Any] = {}
+    machine_name: str,
+    host: str,
+    forward_agent: bool = True,
+    meta: dict[str, Any] | None = None,
 ) -> Host:
+    if meta is None:
+        meta = {}
     parts = host.split("@")
     user: str | None = None
     if len(parts) > 1:

pkgs/clan-cli/clan_cli/ssh/cli.py

@@ -6,7 +6,7 @@ import socket
 import subprocess
 from pathlib import Path
 
-from ..nix import nix_shell
+from clan_cli.nix import nix_shell
 
 log = logging.getLogger(__name__)
 
@@ -15,9 +15,11 @@ def ssh(
     host: str,
     user: str = "root",
     password: str | None = None,
-    ssh_args: list[str] = [],
+    ssh_args: list[str] | None = None,
     torify: bool = False,
 ) -> None:
+    if ssh_args is None:
+        ssh_args = []
     packages = ["nixpkgs#openssh"]
     if torify:
         packages.append("nixpkgs#tor")

pkgs/clan-cli/clan_cli/state/list.py

@@ -3,15 +3,15 @@ import json
 import logging
 from pathlib import Path
 
-from ..cmd import run_no_stdout
-from ..completions import (
+from clan_cli.cmd import run_no_stdout
+from clan_cli.completions import (
     add_dynamic_completer,
     complete_machines,
     complete_state_services_for_machine,
 )
-from ..dirs import get_clan_flake_toplevel_or_env
-from ..errors import ClanCmdError, ClanError
-from ..nix import nix_eval
+from clan_cli.dirs import get_clan_flake_toplevel_or_env
+from clan_cli.errors import ClanCmdError, ClanError
+from clan_cli.nix import nix_eval
 
 log = logging.getLogger(__name__)
 
@@ -33,12 +33,12 @@ def list_state_folders(machine: str, service: None | str = None) -> None:
     try:
         proc = run_no_stdout(cmd)
         res = proc.stdout.strip()
-    except ClanCmdError:
+    except ClanCmdError as e:
         raise ClanError(
             "Clan might not have meta attributes",
             location=f"show_clan {uri}",
             description="Evaluation failed on clanInternals.meta attribute",
-        )
+        ) from e
 
     state = json.loads(res)
     if service:

pkgs/clan-cli/clan_cli/vars/__init__.py

@@ -1,7 +1,8 @@
 # !/usr/bin/env python3
 import argparse
 
-from ..hyperlink import help_hyperlink
+from clan_cli.hyperlink import help_hyperlink
+
 from .check import register_check_parser
 from .generate import register_generate_parser
 from .list import register_list_parser

pkgs/clan-cli/clan_cli/vars/check.py

@@ -2,8 +2,8 @@ import argparse
 import importlib
 import logging
 
-from ..completions import add_dynamic_completer, complete_machines
-from ..machines.machines import Machine
+from clan_cli.completions import add_dynamic_completer, complete_machines
+from clan_cli.machines.machines import Machine
 
 log = logging.getLogger(__name__)
 

pkgs/clan-cli/clan_cli/vars/generate.py

@@ -10,17 +10,17 @@ from tempfile import TemporaryDirectory
 from typing import Any
 
 from clan_cli.cmd import run
-
-from ..completions import (
+from clan_cli.completions import (
     add_dynamic_completer,
     complete_machines,
     complete_services_for_machine,
 )
-from ..errors import ClanError
-from ..git import commit_files
-from ..machines.inventory import get_all_machines, get_selected_machines
-from ..machines.machines import Machine
-from ..nix import nix_shell
+from clan_cli.errors import ClanError
+from clan_cli.git import commit_files
+from clan_cli.machines.inventory import get_all_machines, get_selected_machines
+from clan_cli.machines.machines import Machine
+from clan_cli.nix import nix_shell
+
 from .check import check_vars
 from .public_modules import FactStoreBase
 from .secret_modules import SecretStoreBase

pkgs/clan-cli/clan_cli/vars/list.py

@@ -2,8 +2,9 @@ import argparse
 import importlib
 import logging
 
-from ..completions import add_dynamic_completer, complete_machines
-from ..machines.machines import Machine
+from clan_cli.completions import add_dynamic_completer, complete_machines
+from clan_cli.machines.machines import Machine
+
 from ._types import Var
 
 log = logging.getLogger(__name__)

pkgs/clan-cli/clan_cli/vars/upload.py

@@ -4,10 +4,10 @@ import logging
 from pathlib import Path
 from tempfile import TemporaryDirectory
 
-from ..cmd import Log, run
-from ..completions import add_dynamic_completer, complete_machines
-from ..machines.machines import Machine
-from ..nix import nix_shell
+from clan_cli.cmd import Log, run
+from clan_cli.completions import add_dynamic_completer, complete_machines
+from clan_cli.machines.machines import Machine
+from clan_cli.nix import nix_shell
 
 log = logging.getLogger(__name__)
 

pkgs/clan-cli/clan_cli/vms/inspect.py

@@ -3,9 +3,9 @@ import json
 from dataclasses import dataclass
 from pathlib import Path
 
-from ..clan_uri import FlakeId
-from ..completions import add_dynamic_completer, complete_machines
-from ..machines.machines import Machine
+from clan_cli.clan_uri import FlakeId
+from clan_cli.completions import add_dynamic_completer, complete_machines
+from clan_cli.machines.machines import Machine
 
 
 @dataclass

pkgs/clan-cli/clan_cli/vms/qemu.py

@@ -5,9 +5,9 @@ from contextlib import contextmanager
 from dataclasses import dataclass
 from pathlib import Path
 
+from clan_cli.errors import ClanError
 from clan_cli.qemu.qmp import QEMUMonitorProtocol
 
-from ..errors import ClanError
 from .inspect import VmConfig
 
 
@@ -94,8 +94,10 @@ def qemu_command(
     virtiofsd_socket: Path,
     qmp_socket_file: Path,
     qga_socket_file: Path,
-    portmap: list[tuple[int, int]] = [],
+    portmap: list[tuple[int, int]] | None = None,
 ) -> QemuCommand:
+    if portmap is None:
+        portmap = []
     kernel_cmdline = [
         (Path(nixos_config["toplevel"]) / "kernel-params").read_text(),
         f'init={nixos_config["toplevel"]}/init',

pkgs/clan-cli/clan_cli/vms/run.py

@@ -7,13 +7,14 @@ from contextlib import ExitStack
 from pathlib import Path
 from tempfile import TemporaryDirectory
 
-from ..cmd import Log, run
-from ..completions import add_dynamic_completer, complete_machines
-from ..dirs import module_root, user_cache_dir, vm_state_dir
-from ..errors import ClanError
-from ..facts.generate import generate_facts
-from ..machines.machines import Machine
-from ..nix import nix_shell
+from clan_cli.cmd import Log, run
+from clan_cli.completions import add_dynamic_completer, complete_machines
+from clan_cli.dirs import module_root, user_cache_dir, vm_state_dir
+from clan_cli.errors import ClanError
+from clan_cli.facts.generate import generate_facts
+from clan_cli.machines.machines import Machine
+from clan_cli.nix import nix_shell
+
 from .inspect import VmConfig, inspect_vm
 from .qemu import qemu_command
 from .virtiofsd import start_virtiofsd
@@ -38,9 +39,11 @@ def facts_to_nixos_config(facts: dict[str, dict[str, bytes]]) -> dict:
 
 # TODO move this to the Machines class
 def build_vm(
-    machine: Machine, tmpdir: Path, nix_options: list[str] = []
+    machine: Machine, tmpdir: Path, nix_options: list[str] | None = None
 ) -> dict[str, str]:
     # TODO pass prompt here for the GTK gui
+    if nix_options is None:
+        nix_options = []
     secrets_dir = get_secrets(machine, tmpdir)
 
     public_facts_module = importlib.import_module(machine.public_facts_module)
@@ -57,7 +60,7 @@ def build_vm(
         vm_data["secrets_dir"] = str(secrets_dir)
         return vm_data
     except json.JSONDecodeError as e:
-        raise ClanError(f"Failed to parse vm config: {e}")
+        raise ClanError(f"Failed to parse vm config: {e}") from e
 
 
 def get_secrets(
@@ -107,9 +110,13 @@ def run_vm(
     *,
     cachedir: Path | None = None,
     socketdir: Path | None = None,
-    nix_options: list[str] = [],
-    portmap: list[tuple[int, int]] = [],
+    nix_options: list[str] | None = None,
+    portmap: list[tuple[int, int]] | None = None,
 ) -> None:
+    if portmap is None:
+        portmap = []
+    if nix_options is None:
+        nix_options = []
     with ExitStack() as stack:
         machine = Machine(name=vm.machine_name, flake=vm.flake_url)
         log.debug(f"Creating VM for {machine}")

pkgs/clan-cli/clan_cli/vms/virtiofsd.py

@@ -5,8 +5,8 @@ import time
 from collections.abc import Iterator
 from pathlib import Path
 
-from ..errors import ClanError
-from ..nix import nix_shell
+from clan_cli.errors import ClanError
+from clan_cli.nix import nix_shell
 
 
 @contextlib.contextmanager

pkgs/clan-cli/clan_cli/vms/waypipe.py

@@ -4,8 +4,8 @@ import subprocess
 import time
 from collections.abc import Iterator
 
-from ..errors import ClanError
-from ..nix import nix_shell
+from clan_cli.errors import ClanError
+from clan_cli.nix import nix_shell
 
 VMADDR_CID_HYPERVISOR = 2
 

pkgs/clan-cli/docs.py

@@ -156,7 +156,7 @@ def get_subcommands(
     parser: argparse.ArgumentParser,
     to: list[Category],
     level: int = 0,
-    prefix: list[str] = [],
+    prefix: list[str] | None = None,
 ) -> tuple[list[Option], list[Option], list[Subcommand]]:
     """
     Generate Markdown documentation for an argparse.ArgumentParser instance including its subcommands.
@@ -168,6 +168,8 @@ def get_subcommands(
 
     # Document each argument
     # --flake --option --debug, etc.
+    if prefix is None:
+        prefix = []
     flag_options: list[Option] = []
     positional_options: list[Option] = []
     subcommands: list[Subcommand] = []

pkgs/clan-cli/pyproject.toml

@@ -67,5 +67,16 @@ ignore_missing_imports = true
 [tool.ruff]
 target-version = "py311"
 line-length = 88
-lint.select = [ "E", "F", "I", "U", "N", "RUF", "ANN", "A" ]
+lint.select = [
+  "A",
+  "ANN",
+  "B",
+  "E",
+  "F",
+  "I",
+  "N",
+  "RUF",
+  "TID",
+  "U",
+]
 lint.ignore = ["E501", "E402", "E731", "ANN101", "ANN401", "A003"]

pkgs/clan-cli/tests/command.py

@@ -17,12 +17,14 @@ class Command:
     def run(
         self,
         command: list[str],
-        extra_env: dict[str, str] = {},
+        extra_env: dict[str, str] | None = None,
         stdin: _FILE = None,
         stdout: _FILE = None,
         stderr: _FILE = None,
         workdir: Path | None = None,
     ) -> subprocess.Popen[str]:
+        if extra_env is None:
+            extra_env = {}
         env = os.environ.copy()
         env.update(extra_env)
         # We start a new session here so that we can than more reliably kill all childs as well

pkgs/clan-cli/tests/fixtures_flakes.py

@@ -53,15 +53,10 @@ class FlakeForTest(NamedTuple):
 def generate_flake(
     temporary_home: Path,
     flake_template: Path,
-    substitutions: dict[str, str] = {
-        "__CHANGE_ME__": "_test_vm_persistence",
-        "git+https://git.clan.lol/clan/clan-core": "path://" + str(CLAN_CORE),
-        "https://git.clan.lol/clan/clan-core/archive/main.tar.gz": "path://"
-        + str(CLAN_CORE),
-    },
+    substitutions: dict[str, str] | None = None,
     # define the machines directly including their config
-    machine_configs: dict[str, dict] = {},
-    inventory: dict[str, dict] = {},
+    machine_configs: dict[str, dict] | None = None,
+    inventory: dict[str, dict] | None = None,
 ) -> FlakeForTest:
     """
     Creates a clan flake with the given name.
@@ -82,6 +77,17 @@ def generate_flake(
     """
 
     # copy the template to a new temporary location
+    if inventory is None:
+        inventory = {}
+    if machine_configs is None:
+        machine_configs = {}
+    if substitutions is None:
+        substitutions = {
+            "__CHANGE_ME__": "_test_vm_persistence",
+            "git+https://git.clan.lol/clan/clan-core": "path://" + str(CLAN_CORE),
+            "https://git.clan.lol/clan/clan-core/archive/main.tar.gz": "path://"
+            + str(CLAN_CORE),
+        }
     flake = temporary_home / "flake"
     shutil.copytree(flake_template, flake)
     sp.run(["chmod", "+w", "-R", str(flake)], check=True)
@@ -136,15 +142,19 @@ def create_flake(
     flake_template: str | Path,
     clan_core_flake: Path | None = None,
     # names referring to pre-defined machines from ../machines
-    machines: list[str] = [],
+    machines: list[str] | None = None,
     # alternatively specify the machines directly including their config
-    machine_configs: dict[str, dict] = {},
+    machine_configs: dict[str, dict] | None = None,
     remote: bool = False,
 ) -> Iterator[FlakeForTest]:
     """
     Creates a flake with the given name and machines.
     The machine names map to the machines in ./test_machines
     """
+    if machine_configs is None:
+        machine_configs = {}
+    if machines is None:
+        machines = []
     if isinstance(flake_template, Path):
         template_path = flake_template
     else:

pkgs/clan-cli/tests/test_api_dataclass_compat.py

@@ -11,7 +11,7 @@ from clan_cli.errors import ClanError
 
 
 def find_dataclasses_in_directory(
-    directory: Path, exclude_paths: list[str] = []
+    directory: Path, exclude_paths: list[str] | None = None
 ) -> list[tuple[str, str]]:
     """
     Find all dataclass classes in all Python files within a nested directory.
@@ -22,6 +22,8 @@ def find_dataclasses_in_directory(
     Returns:
         List[Tuple[str, str]]: A list of tuples containing the file path and the dataclass name.
     """
+    if exclude_paths is None:
+        exclude_paths = []
     dataclass_files = []
 
     excludes = [os.path.join(directory, d) for d in exclude_paths]
@@ -144,4 +146,4 @@ Help:
 --------------------------------------------------------------------------------
 """,
                 location=__file__,
-            )
+            ) from e

pkgs/clan-cli/tests/test_ssh_local.py

@@ -27,7 +27,7 @@ def test_timeout() -> None:
     except Exception:
         pass
     else:
-        assert False, "should have raised TimeoutExpired"
+        raise AssertionError("should have raised TimeoutExpired")
 
 
 def test_run_function() -> None:
@@ -45,7 +45,7 @@ def test_run_exception() -> None:
     except Exception:
         pass
     else:
-        assert False, "should have raised Exception"
+        raise AssertionError("should have raised Exception")
 
 
 def test_run_function_exception() -> None:
@@ -57,7 +57,7 @@ def test_run_function_exception() -> None:
     except Exception:
         pass
     else:
-        assert False, "should have raised Exception"
+        raise AssertionError("should have raised Exception")
 
 
 def test_run_local_non_shell() -> None:

pkgs/clan-cli/tests/test_ssh_remote.py

@@ -46,7 +46,7 @@ def test_timeout(host_group: HostGroup) -> None:
     except Exception:
         pass
     else:
-        assert False, "should have raised TimeoutExpired"
+        raise AssertionError("should have raised TimeoutExpired")
 
 
 def test_run_exception(host_group: HostGroup) -> None:
@@ -58,7 +58,7 @@ def test_run_exception(host_group: HostGroup) -> None:
     except Exception:
         pass
     else:
-        assert False, "should have raised Exception"
+        raise AssertionError("should have raised Exception")
 
 
 def test_run_function_exception(host_group: HostGroup) -> None:
@@ -70,4 +70,4 @@ def test_run_function_exception(host_group: HostGroup) -> None:
     except Exception:
         pass
     else:
-        assert False, "should have raised Exception"
+        raise AssertionError("should have raised Exception")