generate_test_vars: move into clan_cli

Reduces overhead by not having to manage a separate package for this small tool

pkgs/clan-cli/bin/clan-generate-test-vars

@@ -0,0 +1,12 @@
+#!/usr/bin/env python3
+import os
+import sys
+
+sys.path.insert(
+    0, os.path.join(os.path.dirname(os.path.dirname(os.path.realpath(__file__))))
+)
+
+from clan_cli.generate_test_vars.cli import main  # NOQA
+
+if __name__ == "__main__":
+    main()

pkgs/clan-cli/clan_cli/generate_test_vars/cli.py

@@ -0,0 +1,291 @@
+#! /usr/bin/env python3
+
+import argparse
+import json
+import logging
+import os
+import shutil
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+from tempfile import NamedTemporaryFile
+from typing import TYPE_CHECKING, Any, override
+
+from clan_cli.vars.generator import Generator
+from clan_cli.vars.prompt import PromptType
+from clan_lib.dirs import find_toplevel
+from clan_lib.errors import ClanError
+from clan_lib.flake.flake import Flake
+from clan_lib.machines.machines import Machine
+from clan_lib.nix import nix_config, nix_eval, nix_test_store
+from clan_lib.vars.generate import run_generators
+
+if TYPE_CHECKING:
+    from clan_lib.machines.actions import (
+        ListOptions,
+        MachineResponse,
+    )
+
+log = logging.getLogger(__name__)
+
+sops_priv_key = (
+    "AGE-SECRET-KEY-1PL0M9CWRCG3PZ9DXRTTLMCVD57U6JDFE8K7DNVQ35F4JENZ6G3MQ0RQLRV"
+)
+sops_pub_key = "age1qm0p4vf9jvcnn43s6l4prk8zn6cx0ep9gzvevxecv729xz540v8qa742eg"
+
+
+def get_machine_names(repo_root: Path, check_attr: str, system: str) -> list[str]:
+    """Get the machine names from the test flake"""
+    nix_options = []
+    if tmp_store := nix_test_store():
+        nix_options += ["--store", str(tmp_store)]
+    cmd = nix_eval(
+        [
+            f"path://{repo_root}#checks.{system}.{check_attr}.nodes",
+            "--apply",
+            "builtins.attrNames",
+            *nix_options,
+        ],
+    )
+    out = subprocess.run(cmd, check=True, text=True, stdout=subprocess.PIPE)
+    return json.loads(out.stdout.strip())
+
+
+class TestFlake(Flake):
+    """Flake class which is able to deal with not having an actual flake.
+    All nix build and eval calls will be forwarded to:
+      clan-core#checks.<system>.<test_name>
+    """
+
+    def __init__(
+        self, check_attr: str, test_dir: Path, *args: Any, **kwargs: Any
+    ) -> None:
+        """Initialize the TestFlake with the check attribute."""
+        super().__init__(*args, **kwargs)
+        self.check_attr = check_attr
+        self.test_dir = test_dir
+
+    @override
+    def precache(self, selectors: list[str]) -> None:
+        # Precaching is broken since 501d02056222216330b3820d1c252ffdc81b7daf
+        # TODO @DavHau pls fix!
+        pass
+
+    @property
+    def path(self) -> Path:
+        return self.test_dir
+
+    def select_machine(self, machine_name: str, selector: str) -> Any:
+        """Select a nix attribute for a specific machine.
+
+        Args:
+            machine_name: The name of the machine
+            selector: The attribute selector string relative to the machine config
+            apply: Optional function to apply to the result
+
+        """
+        config = nix_config()
+        system = config["system"]
+        test_system = system
+        if system.endswith("-darwin"):
+            test_system = system.rstrip("darwin") + "linux"
+
+        full_selector = f'checks."{test_system}".{self.check_attr}.machinesCross.{system}."{machine_name}".{selector}'
+        return self.select(full_selector)
+
+    # we don't want to evaluate all machines of the flake. Only the ones defined in the test
+    def set_machine_names(self, machine_names: list[str]) -> None:
+        """Set the machine names for this flake instance to fake the machines defined by the test"""
+        self._machine_names = machine_names
+
+    def list_machines(
+        self,
+        opts: "ListOptions | None" = None,  # noqa: ARG002
+    ) -> "dict[str, MachineResponse]":
+        """List machines of a clan"""
+        from clan_lib.machines.actions import (  # noqa: PLC0415
+            InventoryMachine,
+            MachineResponse,
+        )
+
+        res = {}
+        for name in self._machine_names:
+            res[name] = MachineResponse(data=InventoryMachine())
+        return res
+
+
+class TestMachine(Machine):
+    """Machine class which is able to deal with not having an actual flake.
+    All nix build and eval calls will be forwarded to:
+      clan-core#checks.<system>.<test_name>.nodes.<machine_name>.<attr>
+    """
+
+    @override
+    def __init__(
+        self,
+        name: str,
+        flake: Flake,
+        test_dir: Path,
+        check_attr: str,
+    ) -> None:
+        super().__init__(name, flake)
+        self.check_attr = check_attr
+        self.test_dir = test_dir
+
+    @property
+    def flake_dir(self) -> Path:
+        return self.test_dir
+
+    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
+        """
+        config = nix_config()
+        system = config["system"]
+        test_system = system
+        if system.endswith("-darwin"):
+            test_system = system.rstrip("darwin") + "linux"
+
+        return self.flake.select(
+            f'checks."{test_system}".{self.check_attr}.machinesCross.{system}.{self.name}.{attr}',
+        )
+
+
+@dataclass
+class Options:
+    repo_root: Path
+    test_dir: Path
+    check_attr: str
+    clean: bool
+
+
+def parse_args() -> Options:
+    parser = argparse.ArgumentParser(
+        description="""
+            Update the vars of a 'clanTest' integration test.
+            See 'clanLib.test.clanTest' for more information on how to create such a test.
+        """,
+    )
+    parser.add_argument(
+        "--repo-root",
+        type=Path,
+        help="""
+            Should be an absolute path to the repo root.
+            This path is used as root to evaluate and build attributes using the nix commands.
+            i.e. 'nix eval <repo_root>#checks ...'
+        """,
+        required=False,
+        default=os.environ.get("PRJ_ROOT", find_toplevel([".git"])),
+    )
+    parser.add_argument(
+        "--clean",
+        help="wipe vars and sops directories before generating new vars",
+        action="store_true",
+        default=False,
+    )
+
+    parser.add_argument(
+        "test_dir",
+        type=Path,
+        help="""
+            The folder of the test. Usually passed as 'directory' to clan in the test.
+            Must be relative to the repo_root.
+        """,
+    )
+    parser.add_argument(
+        "check_attr",
+        type=str,
+        help="The attribute name of the flake#checks to update",
+    )
+    args = parser.parse_args()
+    return Options(
+        repo_root=args.repo_root,
+        test_dir=args.test_dir,
+        check_attr=args.check_attr,
+        clean=args.clean,
+    )
+
+
+def main() -> None:
+    logging.basicConfig(level=logging.DEBUG)
+    os.environ["CLAN_NO_COMMIT"] = "1"
+    opts = parse_args()
+    test_dir = opts.test_dir
+
+    if opts.clean:
+        shutil.rmtree(test_dir / "vars", ignore_errors=True)
+        shutil.rmtree(test_dir / "sops", ignore_errors=True)
+
+    config = nix_config()
+    system = config["system"]
+    test_system = system
+    if system.endswith("-darwin"):
+        test_system = system.rstrip("darwin") + "linux"
+
+    flake = TestFlake(opts.check_attr, test_dir, str(opts.repo_root))
+    machine_names = get_machine_names(
+        opts.repo_root,
+        opts.check_attr,
+        test_system,
+    )
+
+    flake.set_machine_names(machine_names)
+
+    flake.precache(
+        [
+            f"checks.{test_system}.{opts.check_attr}.machinesCross.{system}.{{{','.join(machine_names)}}}.config.clan.core.vars.generators.*.validationHash",
+        ],
+    )
+
+    # This hack is necessary because the sops store uses flake.path to find the machine keys
+    # This hack does not work because flake.invalidate_cache resets _path
+    flake._path = opts.test_dir  # noqa: SLF001
+
+    machines = [
+        TestMachine(name, flake, test_dir, opts.check_attr) for name in machine_names
+    ]
+    user = "admin"
+    admin_key_path = Path(test_dir.resolve() / "sops" / "users" / user / "key.json")
+    admin_key_path.parent.mkdir(parents=True, exist_ok=True)
+    os.environ["SOPS_AGE_KEY_FILE"] = str(admin_key_path)
+    admin_key_path.write_text(
+        json.dumps(
+            {
+                "publickey": sops_pub_key,
+                "type": "age",
+            },
+            indent=2,
+        )
+        + "\n",
+    )
+
+    def mocked_prompts(
+        generator: Generator,
+    ) -> dict[str, str]:
+        prompt_values: dict[str, str] = {}
+        for prompt in generator.prompts:
+            var_id = f"{generator.name}/{prompt.name}"
+            if prompt.prompt_type == PromptType.HIDDEN:
+                prompt_values[prompt.name] = "fake_hidden_value"
+            elif prompt.prompt_type == PromptType.MULTILINE_HIDDEN:
+                prompt_values[prompt.name] = "fake\nmultiline\nhidden\nvalue"
+            elif prompt.prompt_type == PromptType.MULTILINE:
+                prompt_values[prompt.name] = "fake\nmultiline\nvalue"
+            elif prompt.prompt_type == PromptType.LINE:
+                prompt_values[prompt.name] = "fake_line_value"
+            else:
+                msg = f"Unknown prompt type {prompt.prompt_type} for prompt {var_id} in generator {generator.name}"
+                raise ClanError(msg)
+        return prompt_values
+
+    with NamedTemporaryFile("w") as f:
+        f.write("# created: 2023-07-17T10:51:45+02:00\n")
+        f.write(f"# public key: {sops_pub_key}\n")
+        f.write(sops_priv_key)
+        f.seek(0)
+        os.environ["SOPS_AGE_KEY_FILE"] = f.name
+        run_generators(list(machines), prompt_values=mocked_prompts)
+
+
+if __name__ == "__main__":
+    main()

pkgs/clan-cli/pyproject.toml

@@ -6,16 +6,19 @@ build-backend = "setuptools.build_meta"
 name = "clan"
 description = "clan cli tool"
 dynamic = ["version"]
-scripts = { clan = "clan_cli.cli:main" }
 license = { text = "MIT" }
 
+[project.scripts]
+clan = "clan_cli.cli:main"
+clan-generate-test-vars = "clan_cli.generate_test_vars.cli:main"
+
 [project.urls]
 Homepage = "https://clan.lol/"
 Documentation = "https://docs.clan.lol/"
 Repository = "https://git.clan.lol/clan/clan-core"
 
 [tool.setuptools.packages.find]
-include = ["clan_lib*", "clan_cli*"]
+include = ["clan_lib*", "clan_cli*", "generate_test_vars*"]
 exclude = ["clan_lib.nixpkgs*", "result"]
 
 [tool.setuptools.package-data]
@@ -55,4 +58,4 @@ warn_redundant_casts = true
 disallow_untyped_calls = true
 disallow_untyped_defs = true
 no_implicit_optional = true
-exclude = "clan_lib.nixpkgs"
+exclude = "clan_lib.nixpkgs"