yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

PLC0415: fix

Browse Source
This commit is contained in:
Jörg Thalheim
2025-08-26 14:10:30 +02:00
parent 9c9adc6e16
commit b2a54f5b0d
36 changed files with 146 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pkgs/clan-app/clan_app/api/api_bridge.py
View File

@@ -43,7 +43,7 @@ class ApiBridge(ABC):
def process_request(self, request: BackendRequest) -> None:
"""Process an API request through the middleware chain."""
from .middleware import MiddlewareContext
from .middleware import MiddlewareContext # noqa: PLC0415
with ExitStack() as stack:
context = MiddlewareContext(

3
pkgs/clan-app/clan_app/app.py
View File

@@ -17,6 +17,7 @@ from clan_app.api.middleware import (
LoggingMiddleware,
MethodExecutionMiddleware,
)
from clan_app.deps.http.http_server import HttpApiServer
from clan_app.deps.webview.webview import Size, SizeHint, Webview
log = logging.getLogger(__name__)
@@ -65,8 +66,6 @@ def app_run(app_opts: ClanAppOptions) -> int:
# Start HTTP API server if requested
http_server = None
if app_opts.http_api:
from clan_app.deps.http.http_server import HttpApiServer
openapi_file = os.getenv("OPENAPI_FILE", None)
swagger_dist = os.getenv("SWAGGER_UI_DIST", None)

9
pkgs/clan-app/clan_app/deps/webview/webview.py
View File

@@ -12,12 +12,11 @@ from clan_lib.api import MethodRegistry, message_queue
from clan_lib.api.tasks import WebThread
from ._webview_ffi import _encode_c_string, _webview_lib
from .webview_bridge import WebviewBridge
if TYPE_CHECKING:
from clan_app.api.middleware import Middleware
from .webview_bridge import WebviewBridge
log = logging.getLogger(__name__)
@@ -49,7 +48,7 @@ class Webview:
shared_threads: dict[str, WebThread] | None = None
# initialized later
_bridge: "WebviewBridge | None" = None
_bridge: WebviewBridge | None = None
_handle: Any | None = None
_callbacks: dict[str, Callable[..., Any]] = field(default_factory=dict)
_middleware: list["Middleware"] = field(default_factory=list)
@@ -132,10 +131,8 @@ class Webview:
self._middleware.append(middleware)
def create_bridge(self) -> "WebviewBridge":
def create_bridge(self) -> WebviewBridge:
"""Create and initialize the WebviewBridge with current middleware."""
from .webview_bridge import WebviewBridge
# Use shared_threads if provided, otherwise let WebviewBridge use its default
if self.shared_threads is not None:
bridge = WebviewBridge(

5
pkgs/clan-app/clan_app/deps/webview/webview_bridge.py
View File

@@ -8,8 +8,6 @@ from clan_lib.api.tasks import WebThread
from clan_app.api.api_bridge import ApiBridge, BackendRequest, BackendResponse
from .webview import FuncStatus
if TYPE_CHECKING:
from .webview import Webview
@@ -32,6 +30,9 @@ class WebviewBridge(ApiBridge):
)
log.debug(f"Sending response: {serialized}")
# Import FuncStatus locally to avoid circular import
from .webview import FuncStatus # noqa: PLC0415
self.webview.return_(response._op_key, FuncStatus.SUCCESS, serialized) # noqa: SLF001
def handle_webview_call(

5
pkgs/clan-cli/clan_cli/tests/test_machines_cli.py
View File

@@ -8,6 +8,7 @@ from clan_cli.tests import fixtures_flakes
from clan_cli.tests.age_keys import SopsSetup, assert_secrets_file_recipients
from clan_cli.tests.helpers import cli
from clan_cli.tests.stdout import CaptureOutput
from clan_lib.errors import ClanError
from clan_lib.flake import Flake
from clan_lib.persist.inventory_store import InventoryStore
@@ -93,8 +94,6 @@ def test_machines_update_nonexistent_machine(
test_flake_with_core: fixtures_flakes.FlakeForTest,
) -> None:
"""Test that update command gives helpful error messages for non-existent machines."""
from clan_lib.errors import ClanError
with pytest.raises(ClanError) as exc_info:
cli.run(
[
@@ -118,8 +117,6 @@ def test_machines_update_typo_in_machine_name(
test_flake_with_core: fixtures_flakes.FlakeForTest,
) -> None:
"""Test that update command suggests similar machine names for typos."""
from clan_lib.errors import ClanError
with pytest.raises(ClanError) as exc_info:
cli.run(
[

11
pkgs/clan-cli/clan_cli/tests/test_vars.py
View File

@@ -1,6 +1,7 @@
import json
import logging
import shutil
import subprocess
from pathlib import Path
import pytest
@@ -11,6 +12,7 @@ from clan_cli.vars.check import check_vars
from clan_cli.vars.generator import (
Generator,
GeneratorKey,
dependencies_as_dir,
)
from clan_cli.vars.get import get_machine_var
from clan_cli.vars.graph import all_missing_closure, requested_closure
@@ -21,7 +23,7 @@ from clan_cli.vars.set import set_var
from clan_lib.errors import ClanError
from clan_lib.flake import Flake
from clan_lib.machines.machines import Machine
from clan_lib.nix import nix_eval, run
from clan_lib.nix import nix_config, nix_eval, run
from clan_lib.vars.generate import (
get_generators,
run_generators,
@@ -29,8 +31,6 @@ from clan_lib.vars.generate import (
def test_dependencies_as_files(temp_dir: Path) -> None:
from clan_cli.vars.generator import dependencies_as_dir
decrypted_dependencies = {
"gen_1": {
"var_1a": b"var_1a",
@@ -506,7 +506,6 @@ def test_generate_secret_var_password_store(
monkeypatch.setenv("PASSWORD_STORE_DIR", str(password_store_dir))
# Initialize password store as a git repository
import subprocess
subprocess.run(["git", "init"], cwd=password_store_dir, check=True)
subprocess.run(
@@ -613,8 +612,6 @@ def test_generate_secret_for_multiple_machines(
) -> None:
flake = flake_with_sops
from clan_lib.nix import nix_config
local_system = nix_config()["system"]
machine1_config = flake.machines["machine1"]
@@ -1101,8 +1098,6 @@ def test_create_sops_age_secrets(
# check private key exists
assert (flake.temporary_home / ".config" / "sops" / "age" / "keys.txt").is_file()
# it should still work, even if the keys already exist
import shutil
shutil.rmtree(flake.path / "sops" / "users" / "user")
cli.run(["vars", "keygen", "--flake", str(flake.path), "--user", "user"])
# check public key exists

4
pkgs/clan-cli/clan_cli/vars/_types.py
View File

@@ -142,8 +142,6 @@ class StoreBase(ABC):
value: bytes,
is_migration: bool = False,
) -> list[Path]:
from clan_lib.machines.machines import Machine
changed_files: list[Path] = []
# if generator was switched from shared to per-machine or vice versa,
@@ -169,6 +167,8 @@ class StoreBase(ABC):
if generator.machine is None:
log_info = log.info
else:
from clan_lib.machines.machines import Machine # noqa: PLC0415
machine = Machine(name=generator.machine, flake=self.flake)
log_info = machine.info
if self.is_secret_store:

3
pkgs/clan-cli/clan_cli/vars/check.py
View File

@@ -32,13 +32,14 @@ def vars_status(
flake: Flake,
generator_name: None | str = None,
) -> VarStatus:
from clan_cli.vars.generator import Generator # noqa: PLC0415
machine = Machine(name=machine_name, flake=flake)
missing_secret_vars = []
missing_public_vars = []
# signals if a var needs to be updated (eg. needs re-encryption due to new users added)
unfixed_secret_vars = []
invalid_generators = []
from clan_cli.vars.generator import Generator
generators = Generator.get_machine_generators([machine.name], machine.flake)
if generator_name:

3
pkgs/clan-cli/clan_cli/vars/fix.py
View File

@@ -2,6 +2,7 @@ import argparse
import logging
from clan_cli.completions import add_dynamic_completer, complete_machines
from clan_cli.vars.generator import Generator
from clan_lib.errors import ClanError
from clan_lib.flake import require_flake
from clan_lib.machines.machines import Machine
@@ -10,8 +11,6 @@ log = logging.getLogger(__name__)
def fix_vars(machine: Machine, generator_name: None | str = None) -> None:
from clan_cli.vars.generator import Generator
generators = Generator.get_machine_generators([machine.name], machine.flake)
if generator_name:
for generator in generators:

45
pkgs/clan-cli/clan_cli/vars/generator.py
View File

@@ -1,23 +1,31 @@
import logging
import os
import shutil
import sys
from contextlib import ExitStack
from dataclasses import dataclass, field
from functools import cached_property
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import TYPE_CHECKING
from clan_lib import bwrap
from clan_lib.cmd import RunOpts, run
from clan_lib.errors import ClanError
from clan_lib.nix import nix_test_store
from clan_lib.git import commit_files
from clan_lib.nix import nix_config, nix_shell, nix_test_store
from .check import check_vars
from .prompt import Prompt
from .prompt import Prompt, ask
from .var import Var
if TYPE_CHECKING:
from clan_lib.flake import Flake
from clan_lib.machines.machines import Machine
if TYPE_CHECKING:
from ._types import StoreBase
from clan_lib.machines.machines import Machine
log = logging.getLogger(__name__)
@@ -91,8 +99,6 @@ class Generator:
list[Generator]: A list of (unsorted) generators for the machine.
"""
from clan_lib.nix import nix_config
config = nix_config()
system = config["system"]
@@ -125,8 +131,6 @@ class Generator:
files_selector,
)
from clan_lib.machines.machines import Machine
machine = Machine(name=machine_name, flake=flake)
pub_store = machine.public_vars_store
sec_store = machine.secret_vars_store
@@ -207,8 +211,6 @@ class Generator:
if self._flake is None:
msg = "Flake cannot be None"
raise ClanError(msg)
from clan_lib.machines.machines import Machine
machine = Machine(name=self.machine, flake=self._flake)
output = Path(
machine.select(
@@ -226,8 +228,6 @@ class Generator:
if self._flake is None:
msg = "Flake cannot be None"
raise ClanError(msg)
from clan_lib.machines.machines import Machine
machine = Machine(name=self.machine, flake=self._flake)
return machine.select(
f'config.clan.core.vars.generators."{self.name}".validationHash',
@@ -250,8 +250,6 @@ class Generator:
Dictionary mapping generator names to their variable values
"""
from clan_lib.errors import ClanError
generators = self.get_machine_generators([machine.name], machine.flake)
result: dict[str, dict[str, bytes]] = {}
@@ -297,8 +295,6 @@ class Generator:
Dictionary mapping prompt names to their values
"""
from .prompt import ask
prompt_values: dict[str, str] = {}
for prompt in self.prompts:
var_id = f"{self.name}/{prompt.name}"
@@ -323,17 +319,6 @@ class Generator:
no_sandbox: Whether to disable sandboxing when executing the generator
"""
import os
import sys
from contextlib import ExitStack
from pathlib import Path
from tempfile import TemporaryDirectory
from clan_lib import bwrap
from clan_lib.cmd import RunOpts, run
from clan_lib.errors import ClanError
from clan_lib.git import commit_files
if prompt_values is None:
prompt_values = self.ask_prompts()
@@ -353,10 +338,6 @@ class Generator:
def bubblewrap_cmd(generator: str, tmpdir: Path) -> list[str]:
"""Helper function to create bubblewrap command."""
import shutil
from clan_lib.nix import nix_shell, nix_test_store
test_store = nix_test_store()
real_bash_path = Path("bash")
if os.environ.get("IN_NIX_SANDBOX"):
@@ -414,7 +395,7 @@ class Generator:
if sys.platform == "linux" and bwrap.bubblewrap_works():
cmd = bubblewrap_cmd(str(final_script), tmpdir)
elif sys.platform == "darwin":
from clan_lib.sandbox_exec import sandbox_exec_cmd
from clan_lib.sandbox_exec import sandbox_exec_cmd # noqa: PLC0415
cmd = stack.enter_context(sandbox_exec_cmd(str(final_script), tmpdir))
else:

7
pkgs/clan-cli/clan_cli/vars/secret_modules/password_store.py
View File

@@ -8,6 +8,7 @@ from tempfile import TemporaryDirectory
from clan_cli.vars._types import StoreBase
from clan_cli.vars.generator import Generator, Var
from clan_lib.cmd import Log, RunOpts
from clan_lib.flake import Flake
from clan_lib.ssh.host import Host
from clan_lib.ssh.upload import upload
@@ -162,8 +163,6 @@ class SecretStore(StoreBase):
if not git_hash:
return b""
from clan_cli.vars.generator import Generator
generators = Generator.get_machine_generators([machine], self.flake)
manifest = [
f"{generator.name}/{file.name}".encode()
@@ -179,8 +178,6 @@ class SecretStore(StoreBase):
if not local_hash:
return True
from clan_lib.cmd import Log, RunOpts
remote_hash = host.run(
[
"cat",
@@ -195,8 +192,6 @@ class SecretStore(StoreBase):
return local_hash != remote_hash.encode()
def populate_dir(self, machine: str, output_dir: Path, phases: list[str]) -> None:
from clan_cli.vars.generator import Generator
vars_generators = Generator.get_machine_generators([machine], self.flake)
if "users" in phases:
with tarfile.open(

12
pkgs/clan-cli/clan_cli/vars/secret_modules/sops.py
View File

@@ -54,7 +54,7 @@ class SecretStore(StoreBase):
def ensure_machine_key(self, machine: str) -> None:
"""Ensure machine has sops keys initialized."""
# no need to generate keys if we don't manage secrets
from clan_cli.vars.generator import Generator
from clan_cli.vars.generator import Generator # noqa: PLC0415
vars_generators = Generator.get_machine_generators([machine], self.flake)
if not vars_generators:
@@ -141,7 +141,7 @@ class SecretStore(StoreBase):
"""
if generators is None:
from clan_cli.vars.generator import Generator
from clan_cli.vars.generator import Generator # noqa: PLC0415
generators = Generator.get_machine_generators([machine], self.flake)
file_found = False
@@ -219,7 +219,7 @@ class SecretStore(StoreBase):
return [store_folder]
def populate_dir(self, machine: str, output_dir: Path, phases: list[str]) -> None:
from clan_cli.vars.generator import Generator
from clan_cli.vars.generator import Generator # noqa: PLC0415
vars_generators = Generator.get_machine_generators([machine], self.flake)
if "users" in phases or "services" in phases:
@@ -291,7 +291,7 @@ class SecretStore(StoreBase):
)
def collect_keys_for_secret(self, machine: str, path: Path) -> set[sops.SopsKey]:
from clan_cli.secrets.secrets import (
from clan_cli.secrets.secrets import ( # noqa: PLC0415
collect_keys_for_path,
collect_keys_for_type,
)
@@ -352,10 +352,10 @@ class SecretStore(StoreBase):
ClanError: If the specified file_name is not found
"""
from clan_cli.secrets.secrets import update_keys
from clan_cli.secrets.secrets import update_keys # noqa: PLC0415
if generators is None:
from clan_cli.vars.generator import Generator
from clan_cli.vars.generator import Generator # noqa: PLC0415
generators = Generator.get_machine_generators([machine], self.flake)
file_found = False

11
pkgs/clan-cli/clan_lib/api/__init__.py
View File

@@ -20,6 +20,7 @@ from clan_lib.async_run import get_current_thread_opkey
from clan_lib.errors import ClanError
from .serde import dataclass_to_dict, from_dict, sanitize_string
from .type_to_jsonschema import JSchemaTypeError, type_to_dict
log = logging.getLogger(__name__)
@@ -201,10 +202,6 @@ API.register(get_system_file)
return fn
def to_json_schema(self) -> dict[str, Any]:
from typing import get_type_hints
from .type_to_jsonschema import JSchemaTypeError, type_to_dict
api_schema: dict[str, Any] = {
"$comment": "An object containing API methods. ",
"type": "object",
@@ -268,8 +265,6 @@ API.register(get_system_file)
return api_schema
def get_method_argtype(self, method_name: str, arg_name: str) -> Any:
from inspect import signature
func = self._registry.get(method_name, None)
if not func:
msg = f"API Method {method_name} not found in registry. Available methods: {list(self._registry.keys())}"
@@ -313,9 +308,9 @@ def load_in_all_api_functions() -> None:
We have to make sure python loads every wrapped function at least once.
This is done by importing all modules from the clan_lib and clan_cli packages.
"""
import clan_cli
import clan_cli # noqa: PLC0415 # Avoid circular imports - many modules import from clan_lib.api
import clan_lib
import clan_lib # noqa: PLC0415 # Avoid circular imports - many modules import from clan_lib.api
import_all_modules_from_package(clan_lib)
import_all_modules_from_package(clan_cli)

7
pkgs/clan-cli/clan_lib/api/directory.py
View File

@@ -4,8 +4,7 @@ from pathlib import Path
from typing import Any, Literal
from clan_lib.cmd import RunOpts, run
from clan_lib.flake import Flake
from clan_lib.nix import nix_shell
from clan_lib.flake.flake import Flake
from . import API
@@ -89,6 +88,8 @@ def list_system_storage_devices() -> Blockdevices:
A list of detected block devices with metadata like size, path, type, etc.
"""
from clan_lib.nix import nix_shell # noqa: PLC0415
cmd = nix_shell(
["util-linux"],
[
@@ -123,7 +124,7 @@ def get_clan_directory_relative(flake: Flake) -> str:
ClanError: If the flake evaluation fails or directories cannot be found
"""
from clan_lib.dirs import get_clan_directories
from clan_lib.dirs import get_clan_directories # noqa: PLC0415
_, relative_dir = get_clan_directories(flake)
return relative_dir

3
pkgs/clan-cli/clan_lib/api/serde_deserialize_test.py
View File

@@ -1,4 +1,5 @@
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
from typing import Any, Literal, TypedDict
@@ -334,8 +335,6 @@ def test_literal_field() -> None:
def test_enum_roundtrip() -> None:
from enum import Enum
class MyEnum(Enum):
FOO = "abc"
BAR = 2

3
pkgs/clan-cli/clan_lib/api/serde_serialize_test.py
View File

@@ -1,4 +1,5 @@
from dataclasses import dataclass, field
from enum import Enum
# Functions to test
from clan_lib.api import (
@@ -124,8 +125,6 @@ def test_filters_null_fields() -> None:
def test_custom_enum() -> None:
from enum import Enum
class CustomEnum(Enum):
FOO = "foo"
BAR = "bar"

13
pkgs/clan-cli/clan_lib/api/type_to_jsonschema_test.py
View File

@@ -1,5 +1,6 @@
from dataclasses import dataclass, field
from typing import Any, NotRequired, Required
from enum import Enum
from typing import Any, Generic, NotRequired, Required, TypedDict, TypeVar
import pytest
@@ -27,8 +28,6 @@ def test_simple_primitives() -> None:
def test_enum_type() -> None:
from enum import Enum
class Color(Enum):
RED = "red"
GREEN = "green"
@@ -224,8 +223,6 @@ def test_nested_open_dicts() -> None:
def test_type_variables() -> None:
from typing import Generic, TypeVar
T = TypeVar("T")
@dataclass
@@ -254,8 +251,6 @@ def test_type_variables() -> None:
def test_type_variable_nested_scopes() -> None:
# Define two type variables with the same name "T" but in different scopes
from typing import Generic, TypeVar
T = TypeVar("T")
@dataclass
@@ -284,8 +279,6 @@ def test_type_variable_nested_scopes() -> None:
def test_total_typed_dict() -> None:
from typing import TypedDict
class ExampleTypedDict(TypedDict):
name: str
value: NotRequired[int]
@@ -314,8 +307,6 @@ def test_total_typed_dict() -> None:
def test_open_typed_dict() -> None:
from typing import TypedDict
class ExampleTypedDict(TypedDict, total=False):
name: Required[str]
value: int

9
pkgs/clan-cli/clan_lib/dirs/__init__.py
View File

@@ -1,3 +1,4 @@
import json
import logging
import os
import sys
@@ -6,7 +7,9 @@ from enum import Enum
from pathlib import Path
from typing import TYPE_CHECKING, Protocol
from clan_lib.cmd import run
from clan_lib.errors import ClanError
from clan_lib.nix import nix_eval
if TYPE_CHECKING:
from clan_lib.flake import Flake
@@ -198,12 +201,6 @@ def get_clan_directories(flake: "Flake") -> tuple[str, str]:
ClanError: If the flake evaluation fails or directories cannot be found
"""
import json
from pathlib import Path
from clan_lib.cmd import run
from clan_lib.nix import nix_eval
# Get the source directory from nix store
root_directory = flake.select("sourceInfo")

29
pkgs/clan-cli/clan_lib/flake/flake.py
View File

@@ -11,7 +11,16 @@ from pathlib import Path
from tempfile import NamedTemporaryFile
from typing import Any
from clan_lib.cmd import Log, RunOpts, run
from clan_lib.dirs import select_source, user_cache_dir
from clan_lib.errors import ClanCmdError, ClanError
from clan_lib.nix import (
nix_build,
nix_command,
nix_config,
nix_metadata,
nix_test_store,
)
log = logging.getLogger(__name__)
@@ -851,11 +860,6 @@ class Flake:
def prefetch(self) -> None:
"""Loads the flake into the store and populates self.store_path and self.hash such that the flake can evaluate locally and offline"""
from clan_lib.cmd import RunOpts, run
from clan_lib.nix import (
nix_command,
)
if self.nix_options is None:
self.nix_options = []
@@ -895,11 +899,6 @@ class Flake:
This method is used to refresh the cache by reloading it from the flake.
"""
from clan_lib.dirs import user_cache_dir
from clan_lib.nix import (
nix_metadata,
)
self.prefetch()
self._cache = FlakeCache()
@@ -951,14 +950,6 @@ class Flake:
AssertionError: If the cache or flake cache path is not properly initialized.
"""
from clan_lib.cmd import Log, RunOpts, run
from clan_lib.dirs import select_source
from clan_lib.nix import (
nix_build,
nix_config,
nix_test_store,
)
if self._cache is None:
self.invalidate_cache()
if self._cache is None:
@@ -1125,8 +1116,6 @@ class Flake:
apply: Optional function to apply to the result
"""
from clan_lib.nix import nix_config
config = nix_config()
system = config["system"]

5
pkgs/clan-cli/clan_lib/flake/flake_cache_test.py
View File

@@ -360,11 +360,6 @@ def test_store_path_with_line_numbers_not_wrapped() -> None:
def test_store_reference_helpers() -> None:
"""Test the store reference helper functions."""
from clan_lib.flake.flake import (
find_store_references,
is_pure_store_path,
)
# Test find_store_references
assert find_store_references("/nix/store/abc123-pkg") == ["/nix/store/abc123-pkg"]
assert find_store_references("/nix/store/abc123-file.nix:42") == [

3
pkgs/clan-cli/clan_lib/flash/flash.py
View File

@@ -7,6 +7,7 @@ from tempfile import TemporaryDirectory
from typing import Any, Literal
from clan_cli.facts.generate import generate_facts
from clan_cli.vars.generator import Generator
from clan_cli.vars.upload import populate_secret_vars
from clan_lib.api import API
@@ -116,8 +117,6 @@ def run_machine_flash(
"users": {"root": {"openssh": {"authorizedKeys": {"keys": root_keys}}}},
}
from clan_cli.vars.generator import Generator
for generator in Generator.get_machine_generators(
[machine.name], machine.flake
):

7
pkgs/clan-cli/clan_lib/import_utils/import_utils_test.py
View File

@@ -1,3 +1,5 @@
import importlib.util
import sys
import tempfile
from pathlib import Path
from textwrap import dedent
@@ -52,8 +54,6 @@ def test_import_with_source(tmp_path: Path) -> None:
)
# Add the temp directory to sys.path
import sys
sys.path.insert(0, str(tmp_path))
try:
@@ -130,9 +130,6 @@ def test_import_with_source_with_args() -> None:
temp_file = Path(f.name)
# Import module dynamically
import importlib.util
import sys
spec = importlib.util.spec_from_file_location("temp_module", temp_file)
assert spec is not None
assert spec.loader is not None

5
pkgs/clan-cli/clan_lib/locked_open/__init__.py
View File

@@ -5,7 +5,6 @@ from contextlib import contextmanager
from pathlib import Path
from typing import Any
from clan_lib.dirs import user_history_file
from clan_lib.jsonrpc import ClanJSONEncoder
@@ -19,11 +18,15 @@ def locked_open(filename: Path, mode: str = "r") -> Generator:
def write_history_file(data: Any) -> None:
from clan_lib.dirs import user_history_file # noqa: PLC0415
with locked_open(user_history_file(), "w+") as f:
f.write(json.dumps(data, cls=ClanJSONEncoder, indent=4))
def read_history_file() -> list[dict]:
from clan_lib.dirs import user_history_file # noqa: PLC0415
with locked_open(user_history_file(), "r") as f:
content: str = f.read()
parsed: list[dict] = json.loads(content)

16
pkgs/clan-cli/clan_lib/log_manager/test_log_manager.py
View File

@@ -4,11 +4,13 @@ Tests are based on actual usage patterns from example_usage.py and api.py.
"""
import datetime
import tempfile
from pathlib import Path
import pytest
from clan_lib.log_manager import (
LogFile,
LogGroupConfig,
LogManager,
is_correct_day_format,
@@ -472,8 +474,6 @@ class TestLogFileSorting:
configured_log_manager: LogManager,
) -> None:
"""Test that LogFiles are sorted by datetime (newest first)."""
from clan_lib.log_manager import LogFile
# Create LogFiles with different times (same date)
newer_file = LogFile(
op_key="test_op",
@@ -508,8 +508,6 @@ class TestLogFileSorting:
configured_log_manager: LogManager,
) -> None:
"""Test that LogFiles are sorted by date (newer dates first)."""
from clan_lib.log_manager import LogFile
# Create LogFiles with different dates
newer_date_file = LogFile(
op_key="test_op",
@@ -543,8 +541,6 @@ class TestLogFileSorting:
configured_log_manager: LogManager,
) -> None:
"""Test that LogFiles with same datetime are sorted by group name (alphabetical)."""
from clan_lib.log_manager import LogFile
# Create LogFiles with same datetime but different groups
group_a_file = LogFile(
op_key="test_op",
@@ -579,8 +575,6 @@ class TestLogFileSorting:
configured_log_manager: LogManager,
) -> None:
"""Test that LogFiles with same datetime and group are sorted by func_name (alphabetical)."""
from clan_lib.log_manager import LogFile
# Create LogFiles with same datetime and group but different func_names
func_a_file = LogFile(
op_key="test_op",
@@ -614,8 +608,6 @@ class TestLogFileSorting:
configured_log_manager: LogManager,
) -> None:
"""Test that LogFiles with same datetime, group, and func_name are sorted by op_key (alphabetical)."""
from clan_lib.log_manager import LogFile
# Create LogFiles identical except for op_key
op_a_file = LogFile(
op_key="op_a", # Should sort first alphabetically
@@ -649,8 +641,6 @@ class TestLogFileSorting:
configured_log_manager: LogManager,
) -> None:
"""Test complex sorting with multiple LogFiles demonstrating full sort order."""
from clan_lib.log_manager import LogFile
# Create multiple files with different characteristics
files = [
# Oldest datetime, should be last
@@ -813,8 +803,6 @@ class TestLogFileSorting:
"""Test that list_log_days returns days sorted newest first."""
del configured_log_manager # Unused but kept for API compatibility
# Create log files on different days by manipulating the date
import tempfile
# Create files with different dates manually to test sorting
with tempfile.TemporaryDirectory() as tmp_dir:
base_dir = Path(tmp_dir)

4
pkgs/clan-cli/clan_lib/machines/machines.py
View File

@@ -33,7 +33,7 @@ class Machine:
def get_inv_machine(self) -> "InventoryMachine":
# Import on demand to avoid circular imports
from clan_lib.machines.actions import get_machine
from clan_lib.machines.actions import get_machine # noqa: PLC0415
return get_machine(self.flake, self.name)
@@ -121,7 +121,7 @@ class Machine:
return self.flake.path
def target_host(self) -> Remote:
from clan_lib.network.network import get_best_remote
from clan_lib.network.network import get_best_remote # noqa: PLC0415
with get_best_remote(self) as remote:
return remote

2
pkgs/clan-cli/clan_lib/machines/suggestions.py
View File

@@ -42,7 +42,7 @@ def _suggest_similar_names(
def get_available_machines(flake: Flake) -> list[str]:
from clan_lib.machines.list import list_machines
from clan_lib.machines.list import list_machines # noqa: PLC0415
machines = list_machines(flake)
return list(machines.keys())

2
pkgs/clan-cli/clan_lib/network/network.py
View File

@@ -34,7 +34,7 @@ class Peer:
_var: dict[str, str] = self._host["var"]
machine_name = _var["machine"]
generator = _var["generator"]
from clan_lib.machines.machines import Machine
from clan_lib.machines.machines import Machine # noqa: PLC0415
machine = Machine(name=machine_name, flake=self.flake)
var = get_machine_var(

2
pkgs/clan-cli/clan_lib/network/tor/__init__.py
View File

@@ -47,8 +47,6 @@ class NetworkTechnology(NetworkTechnologyBase):
yield network
def remote(self, peer: Peer) -> "Remote":
from clan_lib.ssh.remote import Remote
return Remote(
address=peer.host,
command_prefix=peer.name,

3
pkgs/clan-cli/clan_lib/network/tor/lib.py
View File

@@ -1,6 +1,7 @@
#!/usr/bin/env python3
import logging
import socket
import time
from collections.abc import Iterator
from contextlib import contextmanager
@@ -70,8 +71,6 @@ class TorCheck:
def tor_online_test(proxy_port: int) -> None:
"""Tests if Tor is online by checking if we can establish a SOCKS5 connection."""
import socket
# Try to establish a SOCKS5 handshake with the Tor proxy
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(2.0) # Short timeout for local connection

6
pkgs/clan-cli/clan_lib/nix/__init__.py
View File

@@ -8,7 +8,6 @@ from pathlib import Path
from typing import Any
from clan_lib.cmd import run
from clan_lib.dirs import nixpkgs_flake, nixpkgs_source
from clan_lib.errors import ClanError
from clan_lib.locked_open import locked_open
@@ -89,6 +88,8 @@ def nix_eval(flags: list[str]) -> list[str]:
],
)
if os.environ.get("IN_NIX_SANDBOX"):
from clan_lib.dirs import nixpkgs_source # noqa: PLC0415
return [
*default_flags,
"--override-input",
@@ -168,6 +169,9 @@ def nix_shell(packages: list[str], cmd: list[str]) -> list[str]:
] + [package for package in packages if "#" in package]
if not missing_packages:
return cmd
from clan_lib.dirs import nixpkgs_flake # noqa: PLC0415
return [
*nix_command(["shell", "--inputs-from", f"{nixpkgs_flake()!s}"]),
*missing_packages,

5
pkgs/clan-cli/clan_lib/ssh/remote.py
View File

@@ -459,12 +459,12 @@ class Remote:
self,
opts: "ConnectionOptions | None" = None,
) -> None:
from clan_lib.network.check import check_machine_ssh_reachable
from clan_lib.network.check import check_machine_ssh_reachable # noqa: PLC0415
return check_machine_ssh_reachable(self, opts)
def check_machine_ssh_login(self) -> None:
from clan_lib.network.check import check_machine_ssh_login
from clan_lib.network.check import check_machine_ssh_login # noqa: PLC0415
return check_machine_ssh_login(self)
@@ -521,7 +521,6 @@ def _parse_ssh_uri(
raise ClanError(msg)
hostname = result.hostname
port = result.port
from clan_lib.ssh.remote import Remote
return Remote(
address=hostname,

3
pkgs/clan-cli/clan_lib/vars/generate.py
View File

@@ -1,6 +1,7 @@
import logging
from collections.abc import Callable
from clan_cli.vars import graph
from clan_cli.vars.generator import Generator, GeneratorKey
from clan_cli.vars.graph import minimal_closure, requested_closure
from clan_cli.vars.migration import check_can_migrate, migrate_files
@@ -31,8 +32,6 @@ def get_generators(
List of generators based on the specified selection and closure mode.
"""
from clan_cli.vars import graph
machine_names = [machine.name for machine in machines]
vars_generators = Generator.get_machine_generators(
machine_names,

2
pkgs/clan-vm-manager/clan_vm_manager/components/executor.py
View File

@@ -36,7 +36,7 @@ class MPProcess:
def _set_proc_name(name: str) -> None:
if sys.platform != "linux":
return
import ctypes
import ctypes # noqa: PLC0415
# Define the prctl function with the appropriate arguments and return type
libc = ctypes.CDLL("libc.so.6")

110
pkgs/clan-vm-manager/clan_vm_manager/components/trayicon.py
View File

@@ -33,6 +33,15 @@ from gi.repository import GdkPixbuf, Gio, GLib, Gtk
from clan_vm_manager.assets import loc
# Windows-specific imports
if sys.platform == "win32":
from ctypes import byref, sizeof, windll # type: ignore[attr-defined]
else:
# Create dummy objects for type checking on non-Windows systems
windll = None
byref = None # type: ignore[assignment]
sizeof = None # type: ignore[assignment]
# DUMMY IMPLEMENTATION
################################################
@@ -747,12 +756,12 @@ class Win32Implementation(BaseImplementation):
SM_CXSMICON = 49
if sys.platform == "win32":
from ctypes import Structure
from ctypes import Structure # noqa: PLC0415
class WNDCLASSW(Structure):
"""Windows class structure for window registration."""
from ctypes import CFUNCTYPE, wintypes
from ctypes import CFUNCTYPE, wintypes # noqa: PLC0415
LPFN_WND_PROC = CFUNCTYPE(
wintypes.INT,
@@ -777,7 +786,7 @@ class Win32Implementation(BaseImplementation):
class MENUITEMINFOW(Structure):
"""Windows menu item information structure."""
from ctypes import wintypes
from ctypes import wintypes # noqa: PLC0415
_fields_: ClassVar = [
("cb_size", wintypes.UINT),
@@ -797,7 +806,7 @@ class Win32Implementation(BaseImplementation):
class NOTIFYICONDATAW(Structure):
"""Windows notification icon data structure."""
from ctypes import wintypes
from ctypes import wintypes # noqa: PLC0415
_fields_: ClassVar = [
("cb_size", wintypes.DWORD),
@@ -818,8 +827,6 @@ class Win32Implementation(BaseImplementation):
]
def __init__(self, application: Gtk.Application) -> None:
from ctypes import windll # type: ignore[attr-defined]
super().__init__(application)
self._window_class: Any = None
@@ -827,42 +834,45 @@ class Win32Implementation(BaseImplementation):
self._notify_id = None
self._h_icon = None
self._menu = None
self._wm_taskbarcreated = windll.user32.RegisterWindowMessageW("TaskbarCreated")
if sys.platform == "win32":
self._wm_taskbarcreated = windll.user32.RegisterWindowMessageW(
"TaskbarCreated"
) # type: ignore[attr-defined]
self._register_class()
self._create_window()
self.update_icon()
def _register_class(self) -> None:
from ctypes import byref, windll # type: ignore[attr-defined]
if sys.platform != "win32":
return
self._window_class = self.WNDCLASSW( # type: ignore[attr-defined]
style=(self.CS_VREDRAW | self.CS_HREDRAW),
lpfn_wnd_proc=self.WNDCLASSW.LPFN_WND_PROC(self.on_process_window_message), # type: ignore[attr-defined]
h_cursor=windll.user32.LoadCursorW(0, self.IDC_ARROW),
h_cursor=windll.user32.LoadCursorW(0, self.IDC_ARROW), # type: ignore[attr-defined]
hbr_background=self.COLOR_WINDOW,
lpsz_class_name=self.WINDOW_CLASS_NAME,
)
windll.user32.RegisterClassW(byref(self._window_class))
windll.user32.RegisterClassW(byref(self._window_class)) # type: ignore[attr-defined]
def _unregister_class(self):
if self._window_class is None:
if self._window_class is None or sys.platform != "win32":
return
from ctypes import windll
windll.user32.UnregisterClassW(
windll.user32.UnregisterClassW( # type: ignore[attr-defined]
self.WINDOW_CLASS_NAME,
self._window_class.h_instance,
)
self._window_class = None
def _create_window(self) -> None:
from ctypes import windll # type: ignore[attr-defined]
if sys.platform != "win32":
return
style = self.WS_OVERLAPPED | self.WS_SYSMENU
self._h_wnd = windll.user32.CreateWindowExW(
self._h_wnd = windll.user32.CreateWindowExW( # type: ignore[attr-defined]
0,
self.WINDOW_CLASS_NAME,
self.WINDOW_CLASS_NAME,
@@ -877,15 +887,13 @@ class Win32Implementation(BaseImplementation):
None,
)
windll.user32.UpdateWindow(self._h_wnd)
windll.user32.UpdateWindow(self._h_wnd) # type: ignore[attr-defined]
def _destroy_window(self):
if self._h_wnd is None:
if self._h_wnd is None or sys.platform != "win32":
return
from ctypes import windll
windll.user32.DestroyWindow(self._h_wnd)
windll.user32.DestroyWindow(self._h_wnd) # type: ignore[attr-defined]
self._h_wnd = None
def _load_ico_buffer(self, icon_name, icon_size):
@@ -922,10 +930,11 @@ class Win32Implementation(BaseImplementation):
return ico_buffer
def _load_h_icon(self, icon_name):
from ctypes import windll
if sys.platform != "win32":
return None
# Attempt to load custom icons first
icon_size = windll.user32.GetSystemMetrics(self.SM_CXSMICON)
icon_size = windll.user32.GetSystemMetrics(self.SM_CXSMICON) # type: ignore[attr-defined]
ico_buffer = self._load_ico_buffer(
icon_name.replace(f"{pynicotine.__application_id__}-", "nplus-tray-"),
icon_size,
@@ -937,7 +946,7 @@ class Win32Implementation(BaseImplementation):
with tempfile.NamedTemporaryFile(delete=False) as file_handle:
file_handle.write(ico_buffer)
return windll.user32.LoadImageA(
return windll.user32.LoadImageA( # type: ignore[attr-defined]
0,
encode_path(file_handle.name),
self.IMAGE_ICON,
@@ -947,16 +956,15 @@ class Win32Implementation(BaseImplementation):
)
def _destroy_h_icon(self):
from ctypes import windll
if sys.platform != "win32" or not self._h_icon:
return
if self._h_icon:
windll.user32.DestroyIcon(self._h_icon)
windll.user32.DestroyIcon(self._h_icon) # type: ignore[attr-defined]
self._h_icon = None
def _update_notify_icon(self, title="", message="", icon_name=None):
# pylint: disable=attribute-defined-outside-init,no-member
if self._h_wnd is None:
if sys.platform != "win32" or self._h_wnd is None:
return
if icon_name:
@@ -967,8 +975,6 @@ class Win32Implementation(BaseImplementation):
# When disabled by user, temporarily show tray icon when displaying a notification
return
from ctypes import byref, sizeof, windll
action = self.NIM_MODIFY
if self._notify_id is None:
@@ -1004,23 +1010,24 @@ class Win32Implementation(BaseImplementation):
ellipsize=True,
)
windll.shell32.Shell_NotifyIconW(action, byref(self._notify_id))
windll.shell32.Shell_NotifyIconW(action, byref(self._notify_id)) # type: ignore[attr-defined]
def _remove_notify_icon(self):
from ctypes import byref, windll
if sys.platform != "win32":
return
if self._notify_id:
windll.shell32.Shell_NotifyIconW(self.NIM_DELETE, byref(self._notify_id))
windll.shell32.Shell_NotifyIconW(self.NIM_DELETE, byref(self._notify_id)) # type: ignore[attr-defined]
self._notify_id = None
if self._menu:
windll.user32.DestroyMenu(self._menu)
windll.user32.DestroyMenu(self._menu) # type: ignore[attr-defined]
self._menu = None
def _serialize_menu_item(self, item):
# pylint: disable=attribute-defined-outside-init,no-member
from ctypes import sizeof
if sys.platform != "win32":
return None
item_info = self.MENUITEMINFOW(cb_size=sizeof(self.MENUITEMINFOW))
w_id = item["id"]
@@ -1048,38 +1055,42 @@ class Win32Implementation(BaseImplementation):
return item_info
def _show_menu(self):
from ctypes import byref, windll, wintypes
if sys.platform != "win32":
return
from ctypes import wintypes # noqa: PLC0415
if self._menu is None:
self.update_menu()
pos = wintypes.POINT()
windll.user32.GetCursorPos(byref(pos))
windll.user32.GetCursorPos(byref(pos)) # type: ignore[attr-defined]
# PRB: Menus for Notification Icons Do Not Work Correctly
# https://web.archive.org/web/20121015064650/http://support.microsoft.com/kb/135788
windll.user32.SetForegroundWindow(self._h_wnd)
windll.user32.TrackPopupMenu(self._menu, 0, pos.x, pos.y, 0, self._h_wnd, None)
windll.user32.PostMessageW(self._h_wnd, self.WM_NULL, 0, 0)
windll.user32.SetForegroundWindow(self._h_wnd) # type: ignore[attr-defined]
windll.user32.TrackPopupMenu(self._menu, 0, pos.x, pos.y, 0, self._h_wnd, None) # type: ignore[attr-defined]
windll.user32.PostMessageW(self._h_wnd, self.WM_NULL, 0, 0) # type: ignore[attr-defined]
def update_menu(self):
from ctypes import byref, windll
if sys.platform != "win32":
return
if self._menu is None:
self._menu = windll.user32.CreatePopupMenu()
self._menu = windll.user32.CreatePopupMenu() # type: ignore[attr-defined]
for item in self.menu_items.values():
item_id = item["id"]
item_info = self._serialize_menu_item(item)
if not windll.user32.SetMenuItemInfoW(
if not windll.user32.SetMenuItemInfoW( # type: ignore[attr-defined]
self._menu,
item_id,
False,
byref(item_info),
):
windll.user32.InsertMenuItemW(
windll.user32.InsertMenuItemW( # type: ignore[attr-defined]
self._menu,
item_id,
False,
@@ -1093,7 +1104,10 @@ class Win32Implementation(BaseImplementation):
self._update_notify_icon(title=title, message=message)
def on_process_window_message(self, h_wnd, msg, w_param, l_param):
from ctypes import windll, wintypes
if sys.platform != "win32":
return 0
from ctypes import wintypes # noqa: PLC0415
if msg == self.WM_TRAYICON:
if l_param == self.WM_RBUTTONUP:
@@ -1124,7 +1138,7 @@ class Win32Implementation(BaseImplementation):
self._remove_notify_icon()
self._update_notify_icon()
return windll.user32.DefWindowProcW(
return windll.user32.DefWindowProcW( # type: ignore[attr-defined]
wintypes.HWND(h_wnd),
msg,
wintypes.WPARAM(w_param),

2
pkgs/generate-test-vars/generate_test_vars/cli.py
View File

@@ -65,8 +65,6 @@ class TestFlake(Flake):
apply: Optional function to apply to the result
"""
from clan_lib.nix import nix_config
config = nix_config()
system = config["system"]
test_system = system

2
pyproject.toml
View File

@@ -66,8 +66,6 @@ lint.ignore = [
"TRY301",
"FBT003",
"INP001",
# TODO: fix later
"PLC0415",
]
[tool.ruff.lint.per-file-ignores]