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.cmd import run_no_stdout
|
||||||
from clan_cli.errors import ClanCmdError, ClanError
|
from clan_cli.errors import ClanCmdError, ClanError
|
||||||
|
from clan_cli.inventory import Inventory, Service
|
||||||
from clan_cli.nix import nix_eval
|
from clan_cli.nix import nix_eval
|
||||||
|
|
||||||
from . import API
|
from . import API
|
||||||
@ -147,3 +148,26 @@ def get_module_info(
|
|||||||
roles=get_roles(module_path),
|
roles=get_roles(module_path),
|
||||||
readme=readme_content,
|
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
|
import json
|
||||||
from dataclasses import asdict, dataclass, is_dataclass
|
from dataclasses import asdict, dataclass, field, is_dataclass
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any, Literal
|
from typing import Any, Literal
|
||||||
|
|
||||||
from clan_cli.errors import ClanError
|
from clan_cli.errors import ClanError
|
||||||
|
from clan_cli.git import commit_file
|
||||||
|
|
||||||
|
|
||||||
def sanitize_string(s: str) -> str:
|
def sanitize_string(s: str) -> str:
|
||||||
@ -51,7 +52,7 @@ class Machine:
|
|||||||
system: Literal["x86_64-linux"] | str | None = None
|
system: Literal["x86_64-linux"] | str | None = None
|
||||||
description: str | None = None
|
description: str | None = None
|
||||||
icon: str | None = None
|
icon: str | None = None
|
||||||
tags: list[str] | None = None
|
tags: list[str] = field(default_factory=list)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def from_dict(d: dict[str, Any]) -> "Machine":
|
def from_dict(d: dict[str, Any]) -> "Machine":
|
||||||
@ -72,25 +73,29 @@ class ServiceMeta:
|
|||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class Role:
|
class Role:
|
||||||
machines: list[str] | None = None
|
machines: list[str] = field(default_factory=list)
|
||||||
tags: list[str] | None = None
|
tags: list[str] = field(default_factory=list)
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class Service:
|
class Service:
|
||||||
meta: ServiceMeta
|
meta: ServiceMeta
|
||||||
roles: dict[str, Role]
|
roles: dict[str, Role]
|
||||||
machines: dict[str, MachineServiceConfig] | None = None
|
machines: dict[str, MachineServiceConfig] = field(default_factory=dict)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def from_dict(d: dict[str, Any]) -> "Service":
|
def from_dict(d: dict[str, Any]) -> "Service":
|
||||||
return Service(
|
return Service(
|
||||||
meta=ServiceMeta(**d.get("meta", {})),
|
meta=ServiceMeta(**d.get("meta", {})),
|
||||||
roles={name: Role(**role) for name, role in d.get("roles", {}).items()},
|
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()
|
name: MachineServiceConfig(**machine)
|
||||||
},
|
for name, machine in d.get("machines", {}).items()
|
||||||
|
}
|
||||||
|
if d.get("machines")
|
||||||
|
else {}
|
||||||
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@ -133,7 +138,10 @@ class Inventory:
|
|||||||
|
|
||||||
return 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)
|
inventory_file = Inventory.get_path(flake_dir)
|
||||||
|
|
||||||
with open(inventory_file, "w") as f:
|
with open(inventory_file, "w") as f:
|
||||||
json.dump(dataclass_to_dict(self), f, indent=2)
|
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 = Inventory.load_file(flake_dir)
|
||||||
inventory.machines.update({machine.name: machine})
|
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))
|
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:
|
if machine is None:
|
||||||
raise ClanError(f"Machine {name} does not exist")
|
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)
|
folder = specific_machine_dir(Path(base_dir), name)
|
||||||
if folder.exists():
|
if folder.exists():
|
||||||
|
@ -1,7 +1,20 @@
|
|||||||
|
import json
|
||||||
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from fixtures_flakes import FlakeForTest
|
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
|
@pytest.mark.with_core
|
||||||
@ -13,3 +26,56 @@ def test_list_modules(test_flake_with_core: FlakeForTest) -> None:
|
|||||||
# Random test for those two modules
|
# Random test for those two modules
|
||||||
assert "borgbackup" in modules_info.keys()
|
assert "borgbackup" in modules_info.keys()
|
||||||
assert "syncthing" 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