vars: fix listing vars + add test

This commit is contained in:
DavHau
2024-09-01 16:10:00 +02:00
parent 8bf4fddbdc
commit 8ad90aa44f
5 changed files with 35 additions and 4 deletions

View File

@@ -33,10 +33,18 @@ class StoreBase(ABC):
def get(self, service: str, name: str, shared: bool = False) -> bytes: def get(self, service: str, name: str, shared: bool = False) -> bytes:
pass pass
@property
@abstractmethod
def is_secret_store(self) -> bool:
pass
def get_all(self) -> list[Var]: def get_all(self) -> list[Var]:
all_vars = [] all_vars = []
for gen_name, generator in self.machine.vars_generators.items(): for gen_name, generator in self.machine.vars_generators.items():
for f_name, file in generator["files"].items(): for f_name, file in generator["files"].items():
# only handle vars compatible to this store
if self.is_secret_store != file["secret"]:
continue
all_vars.append( all_vars.append(
Var( Var(
store=self, store=self,

View File

@@ -18,11 +18,17 @@ def get_all_vars(machine: Machine) -> list[Var]:
return public_vars_store.get_all() + secret_vars_store.get_all() return public_vars_store.get_all() + secret_vars_store.get_all()
def stringify_vars(_vars: list[Var]) -> str:
return "\n".join([str(var) for var in _vars])
def stringify_all_vars(machine: Machine) -> str:
return stringify_vars(get_all_vars(machine))
def get_command(args: argparse.Namespace) -> None: def get_command(args: argparse.Namespace) -> None:
machine = Machine(name=args.machine, flake=args.flake) machine = Machine(name=args.machine, flake=args.flake)
print(stringify_all_vars(machine))
for var in get_all_vars(machine):
print(var)
def register_list_parser(parser: argparse.ArgumentParser) -> None: def register_list_parser(parser: argparse.ArgumentParser) -> None:

View File

@@ -5,6 +5,10 @@ from clan_cli.vars._types import StoreBase
class FactStoreBase(StoreBase): class FactStoreBase(StoreBase):
@property
def is_secret_store(self) -> bool:
return False
@abstractmethod @abstractmethod
def set( def set(
self, service: str, name: str, value: bytes, shared: bool = False self, service: str, name: str, value: bytes, shared: bool = False

View File

@@ -17,6 +17,10 @@ class SecretStoreBase(StoreBase):
) -> Path | None: ) -> Path | None:
pass pass
@property
def is_secret_store(self) -> bool:
return True
def update_check(self) -> bool: def update_check(self) -> bool:
return False return False

View File

@@ -13,6 +13,7 @@ from root import CLAN_CORE
from clan_cli.clan_uri import FlakeId from clan_cli.clan_uri import FlakeId
from clan_cli.machines.machines import Machine from clan_cli.machines.machines import Machine
from clan_cli.nix import nix_shell from clan_cli.nix import nix_shell
from clan_cli.vars.list import stringify_all_vars
from clan_cli.vars.public_modules import in_repo from clan_cli.vars.public_modules import in_repo
from clan_cli.vars.secret_modules import password_store, sops from clan_cli.vars.secret_modules import password_store, sops
@@ -66,7 +67,6 @@ def test_dependencies_as_files() -> None:
def test_generate_public_var( def test_generate_public_var(
monkeypatch: pytest.MonkeyPatch, monkeypatch: pytest.MonkeyPatch,
temporary_home: Path, temporary_home: Path,
# age_keys: list["KeyPair"],
) -> None: ) -> None:
config = nested_dict() config = nested_dict()
my_generator = config["clan"]["core"]["vars"]["generators"]["my_generator"] my_generator = config["clan"]["core"]["vars"]["generators"]["my_generator"]
@@ -84,6 +84,9 @@ def test_generate_public_var(
) )
assert store.exists("my_generator", "my_value") assert store.exists("my_generator", "my_value")
assert store.get("my_generator", "my_value").decode() == "hello\n" assert store.get("my_generator", "my_value").decode() == "hello\n"
machine = Machine(name="my_machine", flake=FlakeId(str(flake.path)))
vars_text = stringify_all_vars(machine)
assert "my_generator/my_value: hello" in vars_text
@pytest.mark.impure @pytest.mark.impure
@@ -113,6 +116,9 @@ def test_generate_secret_var_sops(
) )
assert sops_store.exists("my_generator", "my_secret") assert sops_store.exists("my_generator", "my_secret")
assert sops_store.get("my_generator", "my_secret").decode() == "hello\n" assert sops_store.get("my_generator", "my_secret").decode() == "hello\n"
machine = Machine(name="my_machine", flake=FlakeId(str(flake.path)))
vars_text = stringify_all_vars(machine)
assert "my_generator/my_secret" in vars_text
@pytest.mark.impure @pytest.mark.impure
@@ -194,6 +200,9 @@ def test_generate_secret_var_password_store(
) )
assert store.exists("my_generator", "my_secret") assert store.exists("my_generator", "my_secret")
assert store.get("my_generator", "my_secret").decode() == "hello\n" assert store.get("my_generator", "my_secret").decode() == "hello\n"
machine = Machine(name="my_machine", flake=FlakeId(str(flake.path)))
vars_text = stringify_all_vars(machine)
assert "my_generator/my_secret" in vars_text
@pytest.mark.impure @pytest.mark.impure