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

Merge pull request 'ruff-7-misc' (#4939) from ruff-7-misc into main

Browse Source 
Reviewed-on: https://git.clan.lol/clan/clan-core/pulls/4939
This commit is contained in:
Mic92
2025-08-26 10:43:12 +00:00
parent 0ad40a0233 f634b8f1fb
commit a5474bc25f
29 changed files with 123 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
pkgs/clan-cli/clan_cli/facts/check.py
View File

@@ -13,24 +13,24 @@ def check_secrets(machine: Machine, service: None | str = None) -> bool:
missing_secret_facts = []
missing_public_facts = []
services = [service] if service else list(machine.facts_data.keys())
for service in services:
for secret_fact in machine.facts_data[service]["secret"]:
for svc in services:
for secret_fact in machine.facts_data[svc]["secret"]:
if isinstance(secret_fact, str):
secret_name = secret_fact
else:
secret_name = secret_fact["name"]
if not machine.secret_facts_store.exists(service, secret_name):
if not machine.secret_facts_store.exists(svc, secret_name):
machine.info(
f"Secret fact '{secret_fact}' for service '{service}' is missing.",
f"Secret fact '{secret_fact}' for service '{svc}' is missing.",
)
missing_secret_facts.append((service, secret_name))
missing_secret_facts.append((svc, secret_name))
for public_fact in machine.facts_data[service]["public"]:
if not machine.public_facts_store.exists(service, public_fact):
for public_fact in machine.facts_data[svc]["public"]:
if not machine.public_facts_store.exists(svc, public_fact):
machine.info(
f"Public fact '{public_fact}' for service '{service}' is missing.",
f"Public fact '{public_fact}' for service '{svc}' is missing.",
)
missing_public_facts.append((service, public_fact))
missing_public_facts.append((svc, public_fact))
machine.debug(f"missing_secret_facts: {missing_secret_facts}")
machine.debug(f"missing_public_facts: {missing_public_facts}")

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

@@ -178,10 +178,10 @@ def _generate_facts_for_machine(
else:
machine_service_facts = machine.facts_data
for service in machine_service_facts:
for svc in machine_service_facts:
machine_updated |= generate_service_facts(
machine=machine,
service=service,
service=svc,
regenerate=regenerate,
secret_facts_store=machine.secret_facts_store,
public_facts_store=machine.public_facts_store,

3
pkgs/clan-cli/clan_cli/network/list.py
View File

@@ -1,6 +1,7 @@
import argparse
import logging
from clan_lib.errors import ClanError
from clan_lib.flake import require_flake
from clan_lib.network.network import networks_from_flake
@@ -54,7 +55,7 @@ def list_command(args: argparse.Namespace) -> None:
try:
is_running = network.is_running()
running_status = "Yes" if is_running else "No"
except Exception:
except ClanError:
running_status = "Error"
print(

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

@@ -83,7 +83,7 @@ class KeyType(enum.Enum):
except FileNotFoundError:
return
except Exception as ex:
except OSError as ex:
log.warning(f"Could not read age keys from {key_path}", exc_info=ex)
if keys := os.environ.get("SOPS_AGE_KEY"):

8
pkgs/clan-cli/clan_cli/state/list.py
View File

@@ -52,12 +52,12 @@ def list_state_folders(machine: Machine, service: None | str = None) -> None:
description=f"The service: {service} needs to be configured for the machine.",
)
for service in state:
if not service:
for svc in state:
if not svc:
continue
print(f"· service: {service}")
service_cfg = state.get(service)
print(f"· service: {svc}")
service_cfg = state.get(svc)
if not service_cfg:
continue # or handle missing config

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

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
import contextlib
import socket
from collections.abc import Callable

2
pkgs/clan-cli/clan_cli/vms/virtiofsd.py
View File

@@ -34,7 +34,7 @@ def start_virtiofsd(socket_path: Path) -> Iterator[None]:
str(store),
],
)
log.debug("$ {}".format(" ".join(virtiofsd)))
log.debug("$ %s", " ".join(virtiofsd))
with subprocess.Popen(virtiofsd) as proc:
try:
while not socket_path.exists():

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

@@ -155,7 +155,7 @@ class AsyncThread[**P, R](threading.Thread):
set_should_cancel(lambda: self.stop_event.is_set())
# Arguments for ParamSpec "P@AsyncThread" are missing
self.result = AsyncResult(_result=self.function(*self.args, **self.kwargs))
except Exception as ex:
except Exception as ex: # noqa: BLE001
self.result = AsyncResult(_result=ex)
finally:
self.finished = True

4
pkgs/clan-cli/clan_lib/backups/create.py
View File

@@ -11,9 +11,9 @@ def create_backup(machine: Machine, provider: str | None = None) -> None:
msg = "No providers specified"
raise ClanError(msg)
with host.host_connection() as ssh:
for provider in backup_scripts["providers"]:
for prov in backup_scripts["providers"]:
proc = ssh.run(
[backup_scripts["providers"][provider]["create"]],
[backup_scripts["providers"][prov]["create"]],
)
if proc.returncode != 0:
msg = "failed to start backup"

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

@@ -846,7 +846,7 @@ class Flake:
return
try:
self._cache.load_from_file(path)
except Exception as e:
except (OSError, json.JSONDecodeError, KeyError, ValueError) as e:
log.warning(f"Failed load eval cache: {e}. Continue without cache")
def prefetch(self) -> None:

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

@@ -14,6 +14,7 @@ def list_log_days() -> list[str]:
Raises:
ClanError: If LOG_MANAGER_INSTANCE is not initialized.
"""
if LOG_MANAGER_INSTANCE is None:
msg = "LOG_MANAGER_INSTANCE is not initialized"
@@ -63,6 +64,7 @@ def get_log_file(
Raises:
ClanError: If the log file is not found or LOG_MANAGER_INSTANCE is not initialized.
"""
if LOG_MANAGER_INSTANCE is None:
msg = "LOG_MANAGER_INSTANCE is not initialized"

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

@@ -234,7 +234,7 @@ def run_machine_update(
is_mobile = machine.select(
"config.system.clan.deployment.nixosMobileWorkaround",
)
except Exception:
except ClanError:
is_mobile = False
# if the machine is mobile, we retry to deploy with the mobile workaround method
if is_mobile:

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

@@ -161,12 +161,9 @@ def get_best_remote(machine: "Machine") -> Iterator["Remote"]:
if target_host:
log.debug(f"Using targetHost from inventory for {machine.name}: {target_host}")
# Create a direct network with just this machine
try:
remote = Remote.from_ssh_uri(machine_name=machine.name, address=target_host)
yield remote
return
except Exception as e:
log.debug(f"Inventory targetHost not reachable for {machine.name}: {e}")
remote = Remote.from_ssh_uri(machine_name=machine.name, address=target_host)
yield remote
return
# Step 2: Try existing networks by priority
try:
@@ -189,7 +186,7 @@ def get_best_remote(machine: "Machine") -> Iterator["Remote"]:
)
yield network.remote(machine.name)
return
except Exception as e:
except ClanError as e:
log.debug(f"Failed to reach {machine.name} via {network_name}: {e}")
else:
try:
@@ -202,34 +199,26 @@ def get_best_remote(machine: "Machine") -> Iterator["Remote"]:
)
yield connected_network.remote(machine.name)
return
except Exception as e:
except ClanError as e:
log.debug(
f"Failed to establish connection to {machine.name} via {network_name}: {e}",
)
except Exception as e:
except (ImportError, AttributeError, KeyError) as e:
log.debug(f"Failed to use networking modules to determine machines remote: {e}")
# Step 3: Try targetHost from machine nixos config
try:
target_host = machine.select('config.clan.core.networking."targetHost"')
if target_host:
log.debug(
f"Using targetHost from machine config for {machine.name}: {target_host}",
)
# Check if reachable
try:
remote = Remote.from_ssh_uri(
machine_name=machine.name,
address=target_host,
)
yield remote
return
except Exception as e:
log.debug(
f"Machine config targetHost not reachable for {machine.name}: {e}",
)
except Exception as e:
log.debug(f"Could not get targetHost from machine config: {e}")
target_host = machine.select('config.clan.core.networking."targetHost"')
if target_host:
log.debug(
f"Using targetHost from machine config for {machine.name}: {target_host}",
)
# Check if reachable
remote = Remote.from_ssh_uri(
machine_name=machine.name,
address=target_host,
)
yield remote
return
# No connection method found
msg = f"Could not find any way to connect to machine '{machine.name}'. No targetHost configured and machine not reachable via any network."
@@ -249,12 +238,7 @@ def get_network_overview(networks: dict[str, Network]) -> dict:
else:
with module.connection(network) as conn:
for peer_name in conn.peers:
try:
result[network_name]["peers"][peer_name] = conn.ping(
peer_name,
)
except ClanError:
log.warning(
f"getting host for machine: {peer_name} in network: {network_name} failed",
)
result[network_name]["peers"][peer_name] = conn.ping(
peer_name,
)
return result

10
pkgs/clan-cli/clan_lib/network/qr_code.py
View File

@@ -29,6 +29,7 @@ class QRCodeData:
@contextmanager
def get_best_remote(self) -> Iterator[Remote]:
errors = []
for address in self.addresses:
try:
log.debug(f"Establishing connection via {address}")
@@ -39,8 +40,13 @@ class QRCodeData:
if ping_time is not None:
log.info(f"reachable via {address} after connection")
yield address.remote
except Exception as e:
log.debug(f"Failed to establish connection via {address}: {e}")
except ClanError as e:
errors.append((address, e))
if not errors:
msg = "No reachable remote found in QR code data: " + ", ".join(
f"{addr.remote} ({err})" for addr, err in errors
)
raise ClanError(msg)
def read_qr_json(qr_data: dict[str, Any], flake: Flake) -> QRCodeData:

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

@@ -43,18 +43,22 @@ def spawn_tor() -> Iterator[None]:
cmd_args = ["tor", "--HardwareAccel", "1"]
packages = ["tor"]
cmd = nix_shell(packages, cmd_args)
process = Popen(cmd)
try:
while not is_tor_running():
log.debug("Waiting for Tor to start...")
time.sleep(0.2)
log.info("Tor is now running")
yield
finally:
log.info("Terminating Tor process...")
process.terminate()
process.wait()
log.info("Tor process terminated")
process = Popen(cmd)
try:
while not is_tor_running():
log.debug("Waiting for Tor to start...")
time.sleep(0.2)
log.info("Tor is now running")
yield
finally:
log.info("Terminating Tor process...")
process.terminate()
process.wait()
log.info("Tor process terminated")
except OSError as e:
msg = f"Failed to spawn tor process with command: {cmd}"
raise ClanError(msg) from e
@dataclass(frozen=True)

4
pkgs/clan-cli/clan_lib/ssh/remote_test.py
View File

@@ -239,7 +239,7 @@ def test_run_exception(hosts: list[Remote], runtime: AsyncRuntime) -> None:
runtime.async_run(None, host.run_local, ["exit 1"], RunOpts(shell=True)) # noqa: S604
runtime.join_all()
runtime.check_all()
except Exception: # noqa: S110
except ClanError:
pass
else:
msg = "should have raised Exception"
@@ -255,7 +255,7 @@ def test_run_function_exception(hosts: list[Remote], runtime: AsyncRuntime) -> N
runtime.async_run(None, some_func, host)
runtime.join_all()
runtime.check_all()
except Exception: # noqa: S110
except ClanError:
pass
else:
msg = "should have raised Exception"

4
pkgs/clan-cli/clan_lib/ssh/sudo_askpass_proxy.py
View File

@@ -1,5 +1,3 @@
#!/usr/bin/env python3
import logging
import subprocess
import sys
@@ -90,7 +88,7 @@ class SudoAskpassProxy:
ssh_process.stdin.flush()
else:
print(stripped_line)
except Exception as e:
except (OSError, ClanError) as e:
logger.error(f"Error processing passwords requests output: {e}")
def run(self) -> str: