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

ARG002/ARG005: fix

Browse Source
This commit is contained in:
Jörg Thalheim
2025-08-20 21:05:11 +02:00
parent 5233eb7fdb
commit 4986fe30c3
31 changed files with 67 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

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)

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"

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)

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/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

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

@@ -32,6 +32,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,
@@ -51,6 +52,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() == []:
@@ -103,6 +105,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/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()
@@ -148,6 +151,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)
@@ -259,6 +263,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)
@@ -348,6 +353,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)
@@ -367,6 +373,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()