API: tests for module instance update #1692
@ -6,6 +6,7 @@ from pathlib import Path
|
||||
|
||||
from clan_cli.cmd import run_no_stdout
|
||||
from clan_cli.errors import ClanCmdError, ClanError
|
||||
from clan_cli.inventory import Inventory, Service
|
||||
from clan_cli.nix import nix_eval
|
||||
|
||||
from . import API
|
||||
@ -147,3 +148,26 @@ def get_module_info(
|
||||
roles=get_roles(module_path),
|
||||
readme=readme_content,
|
||||
)
|
||||
|
||||
|
||||
@API.register
|
||||
def update_module_instance(
|
||||
base_path: str, module_name: str, instance_name: str, instance_config: Service
|
||||
) -> Inventory:
|
||||
inventory = Inventory.load_file(base_path)
|
||||
|
||||
module_instances = inventory.services.get(module_name, {})
|
||||
module_instances[instance_name] = instance_config
|
||||
|
||||
inventory.services[module_name] = module_instances
|
||||
|
||||
inventory.persist(
|
||||
base_path, f"Updated module instance {module_name}/{instance_name}"
|
||||
)
|
||||
|
||||
return inventory
|
||||
|
||||
|
||||
@API.register
|
||||
def get_inventory(base_path: str) -> Inventory:
|
||||
return Inventory.load_file(base_path)
|
||||
|
@ -1,9 +1,10 @@
|
||||
import json
|
||||
from dataclasses import asdict, dataclass, is_dataclass
|
||||
from dataclasses import asdict, dataclass, field, is_dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Literal
|
||||
|
||||
from clan_cli.errors import ClanError
|
||||
from clan_cli.git import commit_file
|
||||
|
||||
|
||||
def sanitize_string(s: str) -> str:
|
||||
@ -51,7 +52,7 @@ class Machine:
|
||||
system: Literal["x86_64-linux"] | str | None = None
|
||||
description: str | None = None
|
||||
icon: str | None = None
|
||||
tags: list[str] | None = None
|
||||
tags: list[str] = field(default_factory=list)
|
||||
|
||||
@staticmethod
|
||||
def from_dict(d: dict[str, Any]) -> "Machine":
|
||||
@ -72,25 +73,29 @@ class ServiceMeta:
|
||||
|
||||
@dataclass
|
||||
class Role:
|
||||
machines: list[str] | None = None
|
||||
tags: list[str] | None = None
|
||||
machines: list[str] = field(default_factory=list)
|
||||
tags: list[str] = field(default_factory=list)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Service:
|
||||
meta: ServiceMeta
|
||||
roles: dict[str, Role]
|
||||
machines: dict[str, MachineServiceConfig] | None = None
|
||||
machines: dict[str, MachineServiceConfig] = field(default_factory=dict)
|
||||
|
||||
@staticmethod
|
||||
def from_dict(d: dict[str, Any]) -> "Service":
|
||||
return Service(
|
||||
meta=ServiceMeta(**d.get("meta", {})),
|
||||
roles={name: Role(**role) for name, role in d.get("roles", {}).items()},
|
||||
machines={
|
||||
machines=(
|
||||
{
|
||||
name: MachineServiceConfig(**machine)
|
||||
for name, machine in d.get("machines", {}).items()
|
||||
},
|
||||
}
|
||||
if d.get("machines")
|
||||
else {}
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@ -133,7 +138,10 @@ class Inventory:
|
||||
|
||||
return inventory
|
||||
|
||||
def persist(self, flake_dir: str | Path) -> None:
|
||||
def persist(self, flake_dir: str | Path, message: str) -> None:
|
||||
inventory_file = Inventory.get_path(flake_dir)
|
||||
|
||||
with open(inventory_file, "w") as f:
|
||||
json.dump(dataclass_to_dict(self), f, indent=2)
|
||||
|
||||
commit_file(inventory_file, Path(flake_dir), commit_message=message)
|
||||
|
@ -21,7 +21,7 @@ def create_machine(flake_dir: str | Path, machine: Machine) -> None:
|
||||
|
||||
inventory = Inventory.load_file(flake_dir)
|
||||
inventory.machines.update({machine.name: machine})
|
||||
inventory.persist(flake_dir)
|
||||
inventory.persist(flake_dir, f"Create machine {machine.name}")
|
||||
|
||||
commit_file(Inventory.get_path(flake_dir), Path(flake_dir))
|
||||
|
||||
|
@ -18,7 +18,7 @@ def delete_machine(base_dir: str | Path, name: str) -> None:
|
||||
if machine is None:
|
||||
raise ClanError(f"Machine {name} does not exist")
|
||||
|
||||
inventory.persist(base_dir)
|
||||
inventory.persist(base_dir, f"Delete machine {name}")
|
||||
|
||||
folder = specific_machine_dir(Path(base_dir), name)
|
||||
if folder.exists():
|
||||
|
@ -1,7 +1,20 @@
|
||||
import json
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import pytest
|
||||
from fixtures_flakes import FlakeForTest
|
||||
|
||||
from clan_cli.api.modules import list_modules
|
||||
from clan_cli.api.modules import list_modules, update_module_instance
|
||||
from clan_cli.inventory import Machine, Role, Service, ServiceMeta
|
||||
from clan_cli.machines.create import create_machine
|
||||
from clan_cli.nix import nix_eval, run_no_stdout
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from age_keys import KeyPair
|
||||
|
||||
from cli import Cli
|
||||
|
||||
from clan_cli.machines.facts import machine_get_fact
|
||||
|
||||
|
||||
@pytest.mark.with_core
|
||||
@ -13,3 +26,56 @@ def test_list_modules(test_flake_with_core: FlakeForTest) -> None:
|
||||
# Random test for those two modules
|
||||
assert "borgbackup" in modules_info.keys()
|
||||
assert "syncthing" in modules_info.keys()
|
||||
|
||||
|
||||
@pytest.mark.impure
|
||||
def test_add_module_to_inventory(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
test_flake_with_core: FlakeForTest,
|
||||
age_keys: list["KeyPair"],
|
||||
) -> None:
|
||||
base_path = test_flake_with_core.path
|
||||
monkeypatch.chdir(test_flake_with_core.path)
|
||||
monkeypatch.setenv("SOPS_AGE_KEY", age_keys[0].privkey)
|
||||
|
||||
cli = Cli()
|
||||
cli.run(
|
||||
[
|
||||
"secrets",
|
||||
"users",
|
||||
"add",
|
||||
"--flake",
|
||||
str(test_flake_with_core.path),
|
||||
"user1",
|
||||
age_keys[0].pubkey,
|
||||
]
|
||||
)
|
||||
create_machine(base_path, Machine(name="machine1", tags=[], system="x86_64-linux"))
|
||||
update_module_instance(
|
||||
base_path,
|
||||
"borgbackup",
|
||||
"borgbackup1",
|
||||
Service(
|
||||
meta=ServiceMeta(name="borgbackup"),
|
||||
roles={
|
||||
"client": Role(machines=["machine1"]),
|
||||
"server": Role(machines=["machine1"]),
|
||||
},
|
||||
),
|
||||
)
|
||||
|
||||
cmd = ["facts", "generate", "--flake", str(test_flake_with_core.path), "machine1"]
|
||||
cli.run(cmd)
|
||||
|
||||
ssh_key = machine_get_fact(base_path, "machine1", "borgbackup.ssh.pub")
|
||||
|
||||
cmd = nix_eval(
|
||||
[
|
||||
f"{base_path}#nixosConfigurations.machine1.config.services.borgbackup.repos",
|
||||
"--json",
|
||||
]
|
||||
)
|
||||
proc = run_no_stdout(cmd)
|
||||
res = json.loads(proc.stdout.strip())
|
||||
|
||||
assert res["machine1"]["authorizedKeys"] == [ssh_key]
|
||||
|
Loading…
Reference in New Issue
Block a user