refactor: replace eval_nix/build_nix with machine.select()
- Remove nix(), eval_nix(), and build_nix() methods from Machine class - Add select() method that handles machine-specific attribute prefixes - Update all usages to use machine.select() directly - Handle Path conversion and tmp_store logic at call sites - This simplifies the Machine API and prepares for deployment.json removal
This commit is contained in:
@@ -4,7 +4,7 @@ from clan_lib.machines.machines import Machine
|
||||
|
||||
def create_backup(machine: Machine, provider: str | None = None) -> None:
|
||||
machine.info(f"creating backup for {machine.name}")
|
||||
backup_scripts = machine.eval_nix("config.clan.core.backups")
|
||||
backup_scripts = machine.select("config.clan.core.backups")
|
||||
host = machine.target_host()
|
||||
if provider is None:
|
||||
if not backup_scripts["providers"]:
|
||||
|
||||
@@ -15,7 +15,7 @@ class Backup:
|
||||
|
||||
def list_provider(machine: Machine, host: Remote, provider: str) -> list[Backup]:
|
||||
results = []
|
||||
backup_metadata = machine.eval_nix("config.clan.core.backups")
|
||||
backup_metadata = machine.select("config.clan.core.backups")
|
||||
list_command = backup_metadata["providers"][provider]["list"]
|
||||
proc = host.run(
|
||||
[list_command],
|
||||
@@ -41,7 +41,7 @@ def list_provider(machine: Machine, host: Remote, provider: str) -> list[Backup]
|
||||
|
||||
|
||||
def list_backups(machine: Machine, provider: str | None = None) -> list[Backup]:
|
||||
backup_metadata = machine.eval_nix("config.clan.core.backups")
|
||||
backup_metadata = machine.select("config.clan.core.backups")
|
||||
results = []
|
||||
with machine.target_host().ssh_control_master() as host:
|
||||
if provider is None:
|
||||
|
||||
@@ -7,8 +7,8 @@ from clan_lib.ssh.remote import Remote
|
||||
def restore_service(
|
||||
machine: Machine, host: Remote, name: str, provider: str, service: str
|
||||
) -> None:
|
||||
backup_metadata = machine.eval_nix("config.clan.core.backups")
|
||||
backup_folders = machine.eval_nix("config.clan.core.state")
|
||||
backup_metadata = machine.select("config.clan.core.backups")
|
||||
backup_folders = machine.select("config.clan.core.state")
|
||||
|
||||
if service not in backup_folders:
|
||||
msg = f"Service {service} not found in configuration. Available services are: {', '.join(backup_folders.keys())}"
|
||||
@@ -60,7 +60,7 @@ def restore_backup(
|
||||
errors = []
|
||||
with machine.target_host().ssh_control_master() as host:
|
||||
if service is None:
|
||||
backup_folders = machine.eval_nix("config.clan.core.state")
|
||||
backup_folders = machine.select("config.clan.core.state")
|
||||
for _service in backup_folders:
|
||||
try:
|
||||
restore_service(machine, host, name, provider, _service)
|
||||
|
||||
@@ -81,9 +81,10 @@ class Machine:
|
||||
|
||||
@property
|
||||
def deployment(self) -> dict:
|
||||
deployment = json.loads(
|
||||
self.build_nix("config.system.clan.deployment.file").read_text()
|
||||
)
|
||||
output = Path(self.select("config.system.clan.deployment.file"))
|
||||
if tmp_store := nix_test_store():
|
||||
output = tmp_store.joinpath(*output.parts[1:])
|
||||
deployment = json.loads(output.read_text())
|
||||
return deployment
|
||||
|
||||
@cached_property
|
||||
@@ -159,13 +160,13 @@ class Machine:
|
||||
|
||||
return None
|
||||
|
||||
def nix(
|
||||
def select(
|
||||
self,
|
||||
attr: str,
|
||||
) -> Any:
|
||||
"""
|
||||
Build the machine and return the path to the result
|
||||
accepts a secret store and a facts store # TODO
|
||||
Select a nix attribute of the machine
|
||||
@attr: the attribute to get
|
||||
"""
|
||||
|
||||
config = nix_config()
|
||||
@@ -175,36 +176,6 @@ class Machine:
|
||||
f'clanInternals.machines."{system}"."{self.name}".{attr}'
|
||||
)
|
||||
|
||||
def eval_nix(self, attr: str, extra_config: None | dict = None) -> Any:
|
||||
"""
|
||||
eval a nix attribute of the machine
|
||||
@attr: the attribute to get
|
||||
"""
|
||||
|
||||
if extra_config:
|
||||
log.warning("extra_config in eval_nix is no longer supported")
|
||||
|
||||
return self.nix(attr)
|
||||
|
||||
def build_nix(self, attr: str, extra_config: None | dict = None) -> Path:
|
||||
"""
|
||||
build a nix attribute of the machine
|
||||
@attr: the attribute to get
|
||||
"""
|
||||
|
||||
if extra_config:
|
||||
log.warning("extra_config in build_nix is no longer supported")
|
||||
|
||||
output = self.nix(attr)
|
||||
output = Path(output)
|
||||
if tmp_store := nix_test_store():
|
||||
output = tmp_store.joinpath(*output.parts[1:])
|
||||
assert output.exists(), f"The output {output} doesn't exist"
|
||||
if isinstance(output, Path):
|
||||
return output
|
||||
msg = "build_nix returned not a Path"
|
||||
raise ClanError(msg)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RemoteSource:
|
||||
@@ -229,7 +200,7 @@ def get_host(
|
||||
machine.debug(
|
||||
f"'{field}' is not set in inventory, falling back to slower Nix config, set it either through the Nix or json interface to improve performance"
|
||||
)
|
||||
host_str = machine.eval_nix(f'config.clan.core.networking."{field}"')
|
||||
host_str = machine.select(f'config.clan.core.networking."{field}"')
|
||||
source = "nix_machine"
|
||||
|
||||
if not host_str:
|
||||
|
||||
@@ -282,5 +282,5 @@ def test_clan_create_api(
|
||||
clan_dir_flake.invalidate_cache()
|
||||
|
||||
with pytest.raises(ClanError) as exc_info:
|
||||
machine.build_nix("config.system.build.toplevel")
|
||||
Path(machine.select("config.system.build.toplevel"))
|
||||
assert "nixos-system-test-clan" in str(exc_info.value)
|
||||
|
||||
Reference in New Issue
Block a user