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

Merge pull request 'ruff-3-arg-fixes' (#4934) from ruff-3-arg-fixes into main

Browse Source 
Reviewed-on: https://git.clan.lol/clan/clan-core/pulls/4934
This commit is contained in:
Mic92
2025-08-25 12:54:04 +00:00
parent de33a07875 2bebcab736
commit 05665b1c7e
52 changed files with 128 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
checks/flash/flake-module.nix
View File

@@ -55,7 +55,8 @@
closureInfo = pkgs.closureInfo { rootPaths = dependencies; };
in
{
checks = pkgs.lib.mkIf pkgs.stdenv.isLinux {
# Skip flash test on aarch64-linux for now as it's too slow
checks = lib.optionalAttrs (pkgs.stdenv.isLinux && pkgs.hostPlatform.system != "aarch64-linux") {
nixos-test-flash = self.clanLib.test.baseTest {
name = "flash";
nodes.target = {

8
lib/test/container-test-driver/test_driver/__init__.py
View File

@@ -150,7 +150,7 @@ class Error(Exception):
pass
def prepare_machine_root(machinename: str, root: Path) -> None:
def prepare_machine_root(root: Path) -> None:
root.mkdir(parents=True, exist_ok=True)
root.joinpath("etc").mkdir(parents=True, exist_ok=True)
root.joinpath(".env").write_text(
@@ -197,7 +197,7 @@ class Machine:
return self.get_systemd_process()
def start(self) -> None:
prepare_machine_root(self.name, self.rootdir)
prepare_machine_root(self.rootdir)
init_test_environment()
cmd = [
"systemd-nspawn",
@@ -294,8 +294,8 @@ class Machine:
def execute(
self,
command: str,
check_return: bool = True,
check_output: bool = True,
check_return: bool = True, # noqa: ARG002
check_output: bool = True, # noqa: ARG002
timeout: int | None = 900,
) -> subprocess.CompletedProcess:
"""Execute a shell command, returning a list `(status, stdout)`.

12
lib/test/container-test-driver/test_driver/logger.py
View File

@@ -41,15 +41,15 @@ class AbstractLogger(ABC):
pass
@abstractmethod
def info(self, *args: Any, **kwargs: Any) -> None: # type: ignore
def info(self, *args: Any, **kwargs: Any) -> None:
pass
@abstractmethod
def warning(self, *args: Any, **kwargs: Any) -> None: # type: ignore
def warning(self, *args: Any, **kwargs: Any) -> None:
pass
@abstractmethod
def error(self, *args: Any, **kwargs: Any) -> None: # type: ignore
def error(self, *args: Any, **kwargs: Any) -> None:
pass
@abstractmethod
@@ -78,6 +78,7 @@ class JunitXMLLogger(AbstractLogger):
atexit.register(self.close)
def log(self, message: str, attributes: dict[str, str] | None = None) -> None:
del attributes # Unused but kept for API compatibility
self.tests[self.currentSubtest].stdout += message + os.linesep
@contextmanager
@@ -86,6 +87,7 @@ class JunitXMLLogger(AbstractLogger):
name: str,
attributes: dict[str, str] | None = None,
) -> Iterator[None]:
del attributes # Unused but kept for API compatibility
old_test = self.currentSubtest
self.tests.setdefault(name, self.TestCaseState())
self.currentSubtest = name
@@ -100,16 +102,20 @@ class JunitXMLLogger(AbstractLogger):
message: str,
attributes: dict[str, str] | None = None,
) -> Iterator[None]:
del attributes # Unused but kept for API compatibility
self.log(message)
yield
def info(self, *args: Any, **kwargs: Any) -> None:
del kwargs # Unused but kept for API compatibility
self.tests[self.currentSubtest].stdout += args[0] + os.linesep
def warning(self, *args: Any, **kwargs: Any) -> None:
del kwargs # Unused but kept for API compatibility
self.tests[self.currentSubtest].stdout += args[0] + os.linesep
def error(self, *args: Any, **kwargs: Any) -> None:
del kwargs # Unused but kept for API compatibility
self.tests[self.currentSubtest].stderr += args[0] + os.linesep
self.tests[self.currentSubtest].failure = True

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

@@ -121,7 +121,7 @@ def app_run(app_opts: ClanAppOptions) -> int:
webview.add_middleware(LoggingMiddleware(log_manager=log_manager))
webview.add_middleware(MethodExecutionMiddleware(api=API))
webview.bind_jsonschema_api(API, log_manager=log_manager)
webview.bind_jsonschema_api(API)
webview.navigate(content_uri)
webview.run()

3
pkgs/clan-app/clan_app/deps/http/http_bridge.py
View File

@@ -313,7 +313,7 @@ class HttpBridge(ApiBridge, BaseHTTPRequestHandler):
)
return
self._process_api_request_in_thread(api_request, method_name)
self._process_api_request_in_thread(api_request)
def _parse_request_data(
self,
@@ -363,7 +363,6 @@ class HttpBridge(ApiBridge, BaseHTTPRequestHandler):
def _process_api_request_in_thread(
self,
api_request: BackendRequest,
method_name: str,
) -> None:
"""Process the API request in a separate thread."""
stop_event = threading.Event()

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

@@ -10,7 +10,6 @@ from typing import TYPE_CHECKING, Any
from clan_lib.api import MethodRegistry, message_queue
from clan_lib.api.tasks import WebThread
from clan_lib.log_manager import LogManager
from ._webview_ffi import _encode_c_string, _webview_lib
@@ -107,17 +106,16 @@ class Webview:
def api_wrapper(
self,
method_name: str,
wrap_method: Callable[..., Any],
op_key_bytes: bytes,
request_data: bytes,
arg: int,
) -> None:
"""Legacy API wrapper - delegates to the bridge."""
del arg # Unused but required for C callback signature
self.bridge.handle_webview_call(
method_name=method_name,
op_key_bytes=op_key_bytes,
request_data=request_data,
arg=arg,
)
@property
@@ -186,12 +184,11 @@ class Webview:
log.info("Shutting down webview...")
self.destroy()
def bind_jsonschema_api(self, api: MethodRegistry, log_manager: LogManager) -> None:
for name, method in api.functions.items():
def bind_jsonschema_api(self, api: MethodRegistry) -> None:
for name in api.functions:
wrapper = functools.partial(
self.api_wrapper,
name,
method,
)
c_callback = _webview_lib.binding_callback_t(wrapper)
@@ -208,7 +205,7 @@ class Webview:
)
def bind(self, name: str, callback: Callable[..., Any]) -> None:
def wrapper(seq: bytes, req: bytes, arg: int) -> None:
def wrapper(seq: bytes, req: bytes, _arg: int) -> None:
args = json.loads(req.decode())
try:
result = callback(*args)

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

@@ -39,7 +39,6 @@ class WebviewBridge(ApiBridge):
method_name: str,
op_key_bytes: bytes,
request_data: bytes,
arg: int,
) -> None:
"""Handle a call from webview's JavaScript bridge."""
try:

1
pkgs/clan-cli/clan_cli/arg_actions.py
View File

@@ -19,6 +19,7 @@ class AppendOptionAction(argparse.Action):
values: str | Sequence[str] | None,
option_string: str | None = None,
) -> None:
del parser, option_string # Unused but required by argparse API
lst = getattr(namespace, self.dest)
lst.append("--option")
if not values or not hasattr(values, "__getitem__"):

1
pkgs/clan-cli/clan_cli/clan/show_test.py
View File

@@ -23,7 +23,6 @@ def test_clan_show(
def test_clan_show_no_flake(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
capture_output: CaptureOutput,
) -> None:
monkeypatch.chdir(tmp_path)

48
pkgs/clan-cli/clan_cli/completions.py
View File

@@ -38,9 +38,9 @@ def clan_dir(flake: str | None) -> str | None:
def complete_machines(
prefix: str,
_prefix: str,
parsed_args: argparse.Namespace,
**kwargs: Any,
**_kwargs: Any,
) -> Iterable[str]:
"""Provides completion functionality for machine names configured in the clan."""
machines: list[str] = []
@@ -72,9 +72,9 @@ def complete_machines(
def complete_services_for_machine(
prefix: str,
_prefix: str,
parsed_args: argparse.Namespace,
**kwargs: Any,
**_kwargs: Any,
) -> Iterable[str]:
"""Provides completion functionality for machine facts generation services."""
services: list[str] = []
@@ -117,9 +117,9 @@ def complete_services_for_machine(
def complete_backup_providers_for_machine(
prefix: str,
_prefix: str,
parsed_args: argparse.Namespace,
**kwargs: Any,
**_kwargs: Any,
) -> Iterable[str]:
"""Provides completion functionality for machine backup providers."""
providers: list[str] = []
@@ -161,9 +161,9 @@ def complete_backup_providers_for_machine(
def complete_state_services_for_machine(
prefix: str,
_prefix: str,
parsed_args: argparse.Namespace,
**kwargs: Any,
**_kwargs: Any,
) -> Iterable[str]:
"""Provides completion functionality for machine state providers."""
providers: list[str] = []
@@ -205,9 +205,9 @@ def complete_state_services_for_machine(
def complete_secrets(
prefix: str,
_prefix: str,
parsed_args: argparse.Namespace,
**kwargs: Any,
**_kwargs: Any,
) -> Iterable[str]:
"""Provides completion functionality for clan secrets"""
from clan_lib.flake.flake import Flake
@@ -228,9 +228,9 @@ def complete_secrets(
def complete_users(
prefix: str,
_prefix: str,
parsed_args: argparse.Namespace,
**kwargs: Any,
**_kwargs: Any,
) -> Iterable[str]:
"""Provides completion functionality for clan users"""
from pathlib import Path
@@ -251,9 +251,9 @@ def complete_users(
def complete_groups(
prefix: str,
_prefix: str,
parsed_args: argparse.Namespace,
**kwargs: Any,
**_kwargs: Any,
) -> Iterable[str]:
"""Provides completion functionality for clan groups"""
from pathlib import Path
@@ -275,9 +275,9 @@ def complete_groups(
def complete_templates_disko(
prefix: str,
_prefix: str,
parsed_args: argparse.Namespace,
**kwargs: Any,
**_kwargs: Any,
) -> Iterable[str]:
"""Provides completion functionality for disko templates"""
from clan_lib.templates import list_templates
@@ -299,9 +299,9 @@ def complete_templates_disko(
def complete_templates_clan(
prefix: str,
_prefix: str,
parsed_args: argparse.Namespace,
**kwargs: Any,
**_kwargs: Any,
) -> Iterable[str]:
"""Provides completion functionality for clan templates"""
from clan_lib.templates import list_templates
@@ -323,9 +323,9 @@ def complete_templates_clan(
def complete_vars_for_machine(
prefix: str,
_prefix: str,
parsed_args: argparse.Namespace,
**kwargs: Any,
**_kwargs: Any,
) -> Iterable[str]:
"""Provides completion functionality for variable names for a specific machine.
Only completes vars that already exist in the vars directory on disk.
@@ -367,9 +367,9 @@ def complete_vars_for_machine(
def complete_target_host(
prefix: str,
_prefix: str,
parsed_args: argparse.Namespace,
**kwargs: Any,
**_kwargs: Any,
) -> Iterable[str]:
"""Provides completion functionality for target_host for a specific machine"""
target_hosts: list[str] = []
@@ -409,9 +409,9 @@ def complete_target_host(
def complete_tags(
prefix: str,
_prefix: str,
parsed_args: argparse.Namespace,
**kwargs: Any,
**_kwargs: Any,
) -> Iterable[str]:
"""Provides completion functionality for tags inside the inventory"""
tags: list[str] = []

2
pkgs/clan-cli/clan_cli/facts/generate.py
View File

@@ -151,7 +151,7 @@ def generate_service_facts(
return True
def prompt_func(service: str, text: str) -> str:
def prompt_func(_service: str, text: str) -> str:
print(f"{text}: ")
return read_multiline_input()

3
pkgs/clan-cli/clan_cli/facts/public_modules/in_repo.py
View File

@@ -12,6 +12,7 @@ class FactStore(FactStoreBase):
self.works_remotely = False
def set(self, service: str, name: str, value: bytes) -> Path | None:
del service # Unused but kept for API compatibility
if self.machine.flake.is_local:
fact_path = (
self.machine.flake.path
@@ -28,6 +29,7 @@ class FactStore(FactStoreBase):
raise ClanError(msg)
def exists(self, service: str, name: str) -> bool:
del service # Unused but kept for API compatibility
fact_path = (
self.machine.flake_dir / "machines" / self.machine.name / "facts" / name
)
@@ -35,6 +37,7 @@ class FactStore(FactStoreBase):
# get a single fact
def get(self, service: str, name: str) -> bytes:
del service # Unused but kept for API compatibility
fact_path = (
self.machine.flake_dir / "machines" / self.machine.name / "facts" / name
)

1
pkgs/clan-cli/clan_cli/facts/secret_modules/__init__.py
View File

@@ -34,6 +34,7 @@ class SecretStoreBase(ABC):
pass
def needs_upload(self, host: Host) -> bool:
del host # Unused but kept for API compatibility
return True
@abstractmethod

3
pkgs/clan-cli/clan_cli/facts/secret_modules/password_store.py
View File

@@ -22,6 +22,7 @@ class SecretStore(SecretStoreBase):
value: bytes,
groups: list[str],
) -> Path | None:
del service, groups # Unused but kept for API compatibility
subprocess.run(
nix_shell(
["pass"],
@@ -33,6 +34,7 @@ class SecretStore(SecretStoreBase):
return None # we manage the files outside of the git repo
def get(self, service: str, name: str) -> bytes:
del service # Unused but kept for API compatibility
return subprocess.run(
nix_shell(
["pass"],
@@ -43,6 +45,7 @@ class SecretStore(SecretStoreBase):
).stdout
def exists(self, service: str, name: str) -> bool:
del service # Unused but kept for API compatibility
password_store = os.environ.get(
"PASSWORD_STORE_DIR",
f"{os.environ['HOME']}/.password-store",

3
pkgs/clan-cli/clan_cli/facts/secret_modules/sops.py
View File

@@ -43,6 +43,7 @@ class SecretStore(SecretStoreBase):
value: bytes,
groups: list[str],
) -> Path | None:
del service # Unused but kept for API compatibility
path = (
sops_secrets_folder(self.machine.flake_dir) / f"{self.machine.name}-{name}"
)
@@ -57,12 +58,14 @@ class SecretStore(SecretStoreBase):
return path
def get(self, service: str, name: str) -> bytes:
del service # Unused but kept for API compatibility
return decrypt_secret(
sops_secrets_folder(self.machine.flake_dir) / f"{self.machine.name}-{name}",
age_plugins=load_age_plugins(self.machine.flake),
).encode("utf-8")
def exists(self, service: str, name: str) -> bool:
del service # Unused but kept for API compatibility
return has_secret(
sops_secrets_folder(self.machine.flake_dir) / f"{self.machine.name}-{name}",
)

1
pkgs/clan-cli/clan_cli/facts/secret_modules/vm.py
View File

@@ -21,6 +21,7 @@ class SecretStore(SecretStoreBase):
value: bytes,
groups: list[str],
) -> Path | None:
del groups # Unused but kept for API compatibility
secret_file = self.dir / service / name
secret_file.parent.mkdir(parents=True, exist_ok=True)
secret_file.write_bytes(value)

1
pkgs/clan-cli/clan_cli/flash/flash_cmd.py
View File

@@ -39,6 +39,7 @@ class AppendDiskAction(argparse.Action):
values: str | Sequence[str] | None, # Updated type hint
option_string: str | None = None,
) -> None:
del parser, option_string # Unused but required by argparse API
# Ensure 'values' is a sequence of two elements
if not (
isinstance(values, Sequence)

4
pkgs/clan-cli/clan_cli/flash/list_test.py
View File

@@ -8,7 +8,7 @@ from clan_cli.tests.stdout import CaptureOutput
@pytest.mark.with_core
def test_flash_list_languages(
temporary_home: Path,
temporary_home: Path, # noqa: ARG001
capture_output: CaptureOutput,
) -> None:
with capture_output as output:
@@ -21,7 +21,7 @@ def test_flash_list_languages(
@pytest.mark.with_core
def test_flash_list_keymaps(
temporary_home: Path,
temporary_home: Path, # noqa: ARG001
capture_output: CaptureOutput,
) -> None:
with capture_output as output:

6
pkgs/clan-cli/clan_cli/machines/list_test.py
View File

@@ -9,7 +9,7 @@ from clan_cli.tests.stdout import CaptureOutput
def list_basic(
temporary_home: Path,
temporary_home: Path, # noqa: ARG001
test_flake_with_core: fixtures_flakes.FlakeForTest,
capture_output: CaptureOutput,
) -> None:
@@ -49,7 +49,7 @@ def list_basic(
indirect=True,
)
def list_with_tags_single_tag(
temporary_home: Path,
temporary_home: Path, # noqa: ARG001
test_flake_with_core: fixtures_flakes.FlakeForTest,
capture_output: CaptureOutput,
) -> None:
@@ -100,7 +100,7 @@ def list_with_tags_single_tag(
indirect=True,
)
def list_with_tags_multiple_tags_intersection(
temporary_home: Path,
temporary_home: Path, # noqa: ARG001
test_flake_with_core: fixtures_flakes.FlakeForTest,
capture_output: CaptureOutput,
) -> None:

2
pkgs/clan-cli/clan_cli/secrets/key.py
View File

@@ -59,7 +59,7 @@ def generate_command(args: argparse.Namespace) -> None:
)
def show_command(args: argparse.Namespace) -> None:
def show_command(_args: argparse.Namespace) -> None:
keys = sops.maybe_get_admin_public_keys()
if not keys:
msg = "No public key found"

1
pkgs/clan-cli/clan_cli/templates/apply_disk.py
View File

@@ -27,6 +27,7 @@ class AppendSetAction(argparse.Action):
values: str | Sequence[str] | None,
option_string: str | None = None,
) -> None:
del parser, option_string # Unused but required by argparse API
lst = getattr(namespace, self.dest)
if not values or not hasattr(values, "__getitem__"):
msg = "values must be indexable"

2
pkgs/clan-cli/clan_cli/tests/fixtures_flakes.py
View File

@@ -505,7 +505,7 @@ def writable_clan_core(
@pytest.fixture
def vm_test_flake(
clan_core: Path,
clan_core: Path, # noqa: ARG001
tmp_path: Path,
) -> Path:
"""Creates a test flake that imports the VM test nixOS modules from clan-core."""

1
pkgs/clan-cli/clan_cli/tests/test_clan_nix_attrset.py
View File

@@ -10,7 +10,6 @@ from clan_lib.templates.filesystem import copy_from_nixstore
@pytest.mark.with_core
def test_clan_core_templates(
test_flake_with_core: FlakeForTest,
monkeypatch: pytest.MonkeyPatch,
temporary_home: Path,
) -> None:
clan_dir = Flake(str(test_flake_with_core.path))

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

@@ -67,8 +67,7 @@ def test_machine_subcommands(
@pytest.mark.with_core
def test_machines_update_with_tags(
test_flake_with_core: fixtures_flakes.FlakeForTest,
capture_output: CaptureOutput,
test_flake_with_core: fixtures_flakes.FlakeForTest, # noqa: ARG001
) -> None:
import argparse

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

@@ -94,6 +94,7 @@ class StoreBase(ABC):
str | None: An error message describing issues found, or None if everything is healthy
"""
del machine, generators, file_name # Unused but kept for API compatibility
return None
def fix(
@@ -116,7 +117,7 @@ class StoreBase(ABC):
None
"""
return
del machine, generators, file_name # Unused but kept for API compatibility
def backend_collision_error(self, folder: Path) -> None:
msg = (

2
pkgs/clan-cli/clan_cli/vars/secret_modules/fs.py
View File

@@ -41,6 +41,7 @@ class SecretStore(StoreBase):
return secret_file.read_bytes()
def populate_dir(self, machine: str, output_dir: Path, phases: list[str]) -> None:
del machine, phases # Unused but kept for API compatibility
if output_dir.exists():
shutil.rmtree(output_dir)
shutil.copytree(self.dir, output_dir)
@@ -53,6 +54,7 @@ class SecretStore(StoreBase):
return []
def delete_store(self, machine: str) -> list[Path]:
del machine # Unused but kept for API compatibility
if self.dir.exists():
shutil.rmtree(self.dir)
return []

1
pkgs/clan-cli/clan_cli/vars/secret_modules/vm.py
View File

@@ -66,6 +66,7 @@ class SecretStore(StoreBase):
return [vars_dir]
def populate_dir(self, machine: str, output_dir: Path, phases: list[str]) -> None:
del phases # Unused but kept for API compatibility
vars_dir = self.get_dir(machine)
if output_dir.exists():
shutil.rmtree(output_dir)

2
pkgs/clan-cli/clan_lib/api/mdns_discovery.py
View File

@@ -112,7 +112,7 @@ def list_system_services_mdns() -> DNSInfo:
return data
def mdns_command(args: argparse.Namespace) -> None:
def mdns_command(_args: argparse.Namespace) -> None:
dns_info = list_system_services_mdns()
for name, info in dns_info.services.items():
print(f"Hostname: {name} - ip: {info.ip}")

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

@@ -331,7 +331,7 @@ if __name__ == "__main__":
def concatenate(a: str, b: str) -> str:
time.sleep(1)
msg = "Hello World"
msg = a + b
raise ClanError(msg)
with runtime:

8
pkgs/clan-cli/clan_lib/fixtures/flakes/flakes.py
View File

@@ -39,10 +39,10 @@ def offline_template(tmp_path_factory: Any, offline_session_flake_hook: Any) ->
def patch_clan_template(monkeypatch: Any, offline_template: Path) -> None:
@contextmanager
def fake_clan_template(
flake: Flake,
template_ident: str,
_flake: Flake,
template_ident: str, # noqa: ARG001
dst_dir: Path,
post_process: Callable[[Path], None] | None = None,
post_process: Callable[[Path], None] | None = None, # noqa: ARG001
) -> Iterator[Path]:
# Just yield static destination directory without any processing
shutil.copytree(offline_template, dst_dir, dirs_exist_ok=True, symlinks=True)
@@ -54,7 +54,7 @@ def patch_clan_template(monkeypatch: Any, offline_template: Path) -> None:
@pytest.fixture()
def clan_flake(
tmp_path: Path,
patch_clan_template: Any,
patch_clan_template: Any, # noqa: ARG001
) -> Callable[[Clan | None, str | None], Flake]:
def factory(clan: Clan | None = None, raw: str | None = None) -> Flake:
# TODO: Make more options configurable

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

@@ -35,7 +35,7 @@ class ClassSource:
def import_with_source[T](
module_name: str,
class_name: str,
base_class: type[T],
_base_class: type[T],
*args: Any,
**kwargs: Any,
) -> T:

1
pkgs/clan-cli/clan_lib/log_manager/__init__.py
View File

@@ -575,6 +575,7 @@ class LogManager:
The LogFile if found, None otherwise.
"""
del group_path # Unused but kept for API compatibility
log_files: list[LogFile] = []
# Recursively search for log files

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

@@ -811,6 +811,7 @@ class TestLogFileSorting:
configured_log_manager: LogManager,
) -> None:
"""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

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

@@ -128,7 +128,7 @@ def get_machine_fields_schema(machine: Machine) -> dict[str, FieldSchema]:
"""
inventory_store = InventoryStore(machine.flake)
write_info = inventory_store.get_writeability_of(f"machines.{machine.name}")
write_info = inventory_store.get_writeability()
field_names = retrieve_typed_field_names(InventoryMachine)

7
pkgs/clan-cli/clan_lib/persist/inventory_store.py
View File

@@ -215,11 +215,10 @@ class InventoryStore:
return WriteInfo(writeables, data_eval, data_disk)
def get_writeability_of(self, path: str) -> Any:
"""Get the writeability of a path in the inventory
def get_writeability(self) -> Any:
"""Get the writeability of the inventory
:param path: The path to check
:return: A dictionary with the writeability of the path
:return: A dictionary with the writeability of all paths
"""
write_info = self._write_info()
return write_info.writeables

1
pkgs/clan-cli/clan_lib/persist/inventory_store_test.py
View File

@@ -28,6 +28,7 @@ class MockFlake:
selector: str,
nix_options: list[str] | None = None,
) -> Any:
del nix_options # Unused but kept for API compatibility
nixpkgs = os.environ.get("NIXPKGS")
select = os.environ.get("NIX_SELECT")
clan_core_path = os.environ.get("CLAN_CORE_PATH")

2
pkgs/clan-cli/clan_lib/ssh/create_test.py
View File

@@ -3,7 +3,7 @@ from pathlib import Path
from clan_lib.ssh.create import create_secret_key_nixos_anywhere
def test_clan_generate_sshkeys(temporary_home: Path) -> None:
def test_clan_generate_sshkeys(temporary_home: Path) -> None: # noqa: ARG001
keypair = create_secret_key_nixos_anywhere()
assert keypair.private.exists()

2
pkgs/clan-cli/clan_lib/ssh/localhost.py
View File

@@ -44,6 +44,7 @@ class LocalHost:
control_master: bool = True,
) -> CmdOut:
"""Run a command locally."""
del tty, verbose_ssh, control_master # Unused but kept for API compatibility
if opts is None:
opts = RunOpts()
@@ -99,6 +100,7 @@ class LocalHost:
control_master: bool = True,
) -> dict[str, str]:
"""LocalHost doesn't need SSH environment variables."""
del control_master # Unused but kept for API compatibility
if env is None:
env = {}
# Don't set NIX_SSHOPTS for localhost

2
pkgs/clan-cli/clan_lib/tests/test_create.py
View File

@@ -107,7 +107,7 @@ def fix_flake_inputs(clan_dir: Path, clan_core_dir: Path) -> None:
@pytest.mark.with_core
@pytest.mark.skipif(sys.platform == "darwin", reason="sshd fails to start on darwin")
def test_clan_create_api(
temporary_home: Path,
temporary_home: Path, # noqa: ARG001
test_lib_root: Path,
clan_core: Path,
hosts: list[Remote],

3
pkgs/clan-vm-manager/clan_vm_manager/app.py
View File

@@ -33,6 +33,7 @@ class MainApplication(Adw.Application):
}
def __init__(self, *args: Any, **kwargs: Any) -> None:
del args, kwargs # Unused but kept for API compatibility
super().__init__(
application_id="org.clan.vm-manager",
flags=Gio.ApplicationFlags.HANDLES_COMMAND_LINE,
@@ -52,6 +53,7 @@ class MainApplication(Adw.Application):
self.connect("shutdown", self.on_shutdown)
def on_shutdown(self, source: "MainApplication") -> None:
del source # Unused but kept for API compatibility
log.debug("Shutting down Adw.Application")
if self.get_windows() == []:
@@ -104,6 +106,7 @@ class MainApplication(Adw.Application):
log.info("Dummy menu entry called")
def on_activate(self, source: "MainApplication") -> None:
del source # Unused but kept for API compatibility
if not self.window:
self.init_style()
self.window = MainWindow(config=ClanConfig(initial_view="list"))

1
pkgs/clan-vm-manager/clan_vm_manager/components/list_splash.py
View File

@@ -70,5 +70,6 @@ class EmptySplash(Gtk.Box):
"""Callback for the join button
Extracts the text from the entry and calls the on_join callback
"""
del button # Unused but kept for API compatibility
log.info(f"Splash screen: Joining {entry.get_text()}")
self.on_join(entry.get_text())

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

@@ -90,7 +90,8 @@ LONG_PATH_PREFIX = "\\\\?\\"
# from pynicotine.gtkgui.widgets.theme import ICON_THEME
class IconTheme:
def lookup_icon(self, icon_name: str, **kwargs: Any) -> None:
return None
del icon_name
del kwargs
ICON_THEME = IconTheme()
@@ -177,7 +178,9 @@ class BaseImplementation:
self.application = application
self.menu_items: dict[int, Any] = {}
self.menu_item_id: int = 1
self.activate_callback: Callable = lambda a, b: self.update_window_visibility
self.activate_callback: Callable = (
lambda _a, _b: self.update_window_visibility()
)
self.is_visible: bool = True
self.create_menu()
@@ -275,9 +278,11 @@ class BaseImplementation:
pass
def set_download_status(self, status: str) -> None:
del status # Unused but kept for API compatibility
self.update_menu()
def set_upload_status(self, status) -> None:
del status # Unused but kept for API compatibility
self.update_menu()
def show_notification(self, title, message) -> None:

3
pkgs/clan-vm-manager/clan_vm_manager/components/vmobj.py
View File

@@ -49,6 +49,7 @@ class VMObject(GObject.Object):
data: HistoryEntry,
build_log_cb: Callable[[Gio.File], None],
) -> None:
del icon # Unused but kept for API compatibility
super().__init__()
# Store the data from the history entry
@@ -138,6 +139,7 @@ class VMObject(GObject.Object):
)
def _on_switch_toggle(self, switch: Gtk.Switch, user_state: bool) -> None:
del user_state # Unused but kept for API compatibility
if switch.get_active():
switch.set_state(False)
switch.set_sensitive(False)
@@ -265,6 +267,7 @@ class VMObject(GObject.Object):
other_file: Gio.File,
event_type: Gio.FileMonitorEvent,
) -> None:
del monitor, other_file # Unused but kept for API compatibility
if event_type == Gio.FileMonitorEvent.CHANGES_DONE_HINT:
# File was changed and the changes were written to disk
# wire up the callback for setting the logs

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

@@ -116,7 +116,7 @@ def add_history_command(args: argparse.Namespace) -> None:
add_history(args.uri)
def list_history_command(args: argparse.Namespace) -> None:
def list_history_command(_args: argparse.Namespace) -> None:
res: dict[str, list[HistoryEntry]] = {}
for history_entry in list_history():
url = str(history_entry.flake.flake_url)

2
pkgs/clan-vm-manager/clan_vm_manager/singletons/toast.py
View File

@@ -47,7 +47,7 @@ class ToastOverlay:
if key not in self.active_toasts:
self.active_toasts.add(key)
self.overlay.add_toast(toast)
toast.connect("dismissed", lambda toast: self.active_toasts.remove(key))
toast.connect("dismissed", lambda _toast: self.active_toasts.remove(key))
class ErrorToast:

1
pkgs/clan-vm-manager/clan_vm_manager/singletons/use_join.py
View File

@@ -75,6 +75,7 @@ class JoinList:
removed: int,
added: int,
) -> None:
del source, position, removed, added # Unused but kept for API compatibility
self.list_store.items_changed(
0,
self.list_store.get_n_items(),

1
pkgs/clan-vm-manager/clan_vm_manager/views/details.py
View File

@@ -59,6 +59,7 @@ class Details(Gtk.Box):
boxed_list: Gtk.ListBox,
item: PreferencesValue,
) -> Gtk.Widget:
del boxed_list # Unused but kept for API compatibility
cores: int | None = os.cpu_count()
fcores = float(cores) if cores else 1.0

7
pkgs/clan-vm-manager/clan_vm_manager/views/list.py
View File

@@ -61,6 +61,7 @@ class ClanList(Gtk.Box):
"""
def __init__(self, config: ClanConfig) -> None:
del config # Unused but kept for API compatibility
super().__init__(orientation=Gtk.Orientation.VERTICAL)
app = Gio.Application.get_default()
@@ -92,6 +93,7 @@ class ClanList(Gtk.Box):
self.splash = EmptySplash(on_join=lambda x: self.on_join_request(x, x))
def display_splash(self, source: GKVStore) -> None:
del source # Unused but kept for API compatibility
print("Displaying splash")
if (
ClanStore.use().clan_store.get_n_items() == 0
@@ -104,6 +106,7 @@ class ClanList(Gtk.Box):
boxed_list: Gtk.ListBox,
vm_store: VMStore,
) -> Gtk.Widget:
del boxed_list # Unused but kept for API compatibility
self.remove(self.splash)
vm = vm_store.first()
@@ -150,6 +153,7 @@ class ClanList(Gtk.Box):
return grp
def on_add(self, source: Any, parameter: Any) -> None:
del source # Unused but kept for API compatibility
target = parameter.get_string()
print("Adding new machine", target)
@@ -263,6 +267,7 @@ class ClanList(Gtk.Box):
return row
def on_edit(self, source: Any, parameter: Any) -> None:
del source # Unused but kept for API compatibility
target = parameter.get_string()
print("Editing settings for machine", target)
@@ -352,6 +357,7 @@ class ClanList(Gtk.Box):
return row
def on_join_request(self, source: Any, url: str) -> None:
del source # Unused but kept for API compatibility
log.debug("Join request: %s", url)
clan_uri = ClanURI.from_str(url)
JoinList.use().push(clan_uri, self.on_after_join)
@@ -371,6 +377,7 @@ class ClanList(Gtk.Box):
value.join()
def on_discard_clicked(self, value: JoinValue, source: Gtk.Widget) -> None:
del source # Unused but kept for API compatibility
JoinList.use().discard(value)
if JoinList.use().is_empty():
self.join_boxed_list.add_css_class("no-shadow")

1
pkgs/clan-vm-manager/clan_vm_manager/windows/main_window.py
View File

@@ -86,6 +86,7 @@ class MainWindow(Adw.ApplicationWindow):
ClanStore.use().kill_all()
def on_destroy(self, source: "Adw.ApplicationWindow") -> None:
del source # Unused but kept for API compatibility
log.info("====Destroying Adw.ApplicationWindow===")
ClanStore.use().kill_all()
self.tray_icon.destroy()

2
pkgs/clan-vm-manager/tests/test_cli.py
View File

@@ -2,7 +2,7 @@ import pytest
from cli import Cli
def test_help(capfd: pytest.CaptureFixture) -> None:
def test_help() -> None:
cli = Cli()
with pytest.raises(SystemExit):
cli.run(["clan-vm-manager", "--help"])

7
pkgs/classgen/main.py
View File

@@ -214,7 +214,6 @@ def field_def_from_default_value(
def get_field_def(
field_name: str,
field_meta: str | None,
field_types: list[str],
default: str | None = None,
default_factory: str | None = None,
@@ -332,11 +331,7 @@ def generate_dataclass(
msg = f"Python type not found for {prop} {prop_info}"
raise Error(msg)
field_meta = None
if field_name != prop:
field_meta = f"""{{"alias": "{prop}"}}"""
finalize_field = partial(get_field_def, field_name, field_meta)
finalize_field = partial(get_field_def, field_name)
# Sort and remove potential duplicates
field_types_sorted = sort_types(set(field_types))

13
pkgs/zerotier-members/zerotier-members.py
View File

@@ -74,9 +74,12 @@ def get_network_id() -> str:
def allow_member(args: argparse.Namespace) -> None:
if args.member_ip:
member_id = compute_member_id(args.member_ip)
member_id = compute_member_id(args.member_id_or_ip)
else:
member_id = args.member_id
if not args.member_id_or_ip:
msg = "Either --member-ip or member_id_or_ip must be provided"
raise ClanError(msg)
member_id = args.member_id_or_ip
network_id = get_network_id()
token = ZEROTIER_STATE_DIR.joinpath("authtoken.secret").read_text()
conn = http.client.HTTPConnection("localhost", 9993)
@@ -119,10 +122,12 @@ def main() -> None:
parser_allow = subparser.add_parser("allow", help="Allow a member to join")
parser_allow.add_argument(
"--member-ip",
help="Allow a member to join by their zerotier ipv6 address",
help="Interpret the positional argument as an IPv6 address instead of member ID",
action="store_true",
)
parser_allow.add_argument("member_id")
parser_allow.add_argument(
"member_id_or_ip", help="Member ID or IPv6 address (when --member-ip is used)"
)
parser_allow.set_defaults(func=allow_member)
parser_list = subparser.add_parser("list", help="List members")