clan-config: introduce --machine + add tests

pkgs/clan-cli/clan_cli/config/__init__.py

@@ -5,9 +5,11 @@ import os
 import subprocess
 import sys
 from pathlib import Path
-from typing import Any, Optional, Type, Union
+from typing import Any, Optional, Type
 
+from clan_cli.dirs import get_clan_flake_toplevel
 from clan_cli.errors import ClanError
+from clan_cli.nix import nix_eval
 
 script_dir = Path(__file__).parent
 
@@ -91,16 +93,61 @@ def cast(value: Any, type: Type, opt_description: str) -> Any:
         )
 
 
-def read_option(option: str) -> str:
+def options_for_machine(machine_name: str, flake: Optional[Path] = None) -> dict:
+    if flake is None:
+        flake = get_clan_flake_toplevel()
+    # use nix eval to lib.evalModules .#clanModules.machine-{machine_name}
+    proc = subprocess.run(
+        nix_eval(
+            flags=[
+                "--json",
+                "--show-trace",
+                "--impure",
+                "--expr",
+                f"""
+                let
+                    flake = builtins.getFlake (toString {flake});
+                    lib = flake.inputs.nixpkgs.lib;
+                    module = flake.nixosModules.machine-{machine_name};
+                    evaled = lib.evalModules {{
+                        modules = [module];
+                    }};
+
+                    # this is actually system independent as it uses toFile
+                    docs = flake.inputs.nixpkgs.legacyPackages.x86_64-linux.nixosOptionsDoc {{
+                        inherit (evaled) options;
+                    }};
+                    options = builtins.fromJSON (builtins.readFile docs.optionsJSON.options);
+            in
+                options
+                """,
+            ],
+        ),
+        capture_output=True,
+        text=True,
+    )
+    if proc.returncode != 0:
+        print(proc.stderr, file=sys.stderr)
+        raise Exception(
+            f"Failed to read options for machine {machine_name}:\n{proc.stderr}"
+        )
+    options = json.loads(proc.stdout)
+    return options
+
+
+def read_machine_option_value(machine_name: str, option: str) -> str:
     # use nix eval to read from .#nixosConfigurations.default.config.{option}
     # this will give us the evaluated config with the options attribute
     proc = subprocess.run(
-        [
+        nix_eval(
-            "nix",
+            flags=[
-            "eval",
+                "--json",
-            "--json",
+                "--show-trace",
-            f".#nixosConfigurations.default.config.{option}",
+                "--extra-experimental-features",
-        ],
+                "nix-command flakes",
+                f".#nixosConfigurations.{machine_name}.config.{option}",
+            ],
+        ),
         capture_output=True,
         text=True,
     )
@@ -119,18 +166,44 @@ def read_option(option: str) -> str:
     return out
 
 
-def process_args(
+def get_or_set_option(args: argparse.Namespace) -> None:
+    if args.value == []:
+        print(read_machine_option_value(args.machine, args.option))
+    else:
+        # load options
+        print(args.options_file)
+        if args.options_file is None:
+            options = options_for_machine(machine_name=args.machine)
+        else:
+            with open(args.options_file) as f:
+                options = json.load(f)
+        # compute settings json file location
+        if args.settings_file is None:
+            flake = get_clan_flake_toplevel()
+            settings_file = flake / "machines" / f"{args.machine}.json"
+        else:
+            settings_file = args.settings_file
+        # set the option with the given value
+        set_option(
+            option=args.option,
+            value=args.value,
+            options=options,
+            settings_file=settings_file,
+            option_description=args.option,
+        )
+        if not args.quiet:
+            new_value = read_machine_option_value(args.machine, args.option)
+            print(f"New Value for {args.option}:")
+            print(new_value)
+
+
+def set_option(
     option: str,
     value: Any,
     options: dict,
     settings_file: Path,
-    quiet: bool = False,
     option_description: str = "",
 ) -> None:
-    if value == []:
-        print(read_option(option))
-        return
-
     option_path = option.split(".")
 
     # if the option cannot be found, then likely the type is attrs and we need to
@@ -140,12 +213,11 @@ def process_args(
             raise ClanError(f"Option {option_description} not found")
         option_parent = option_path[:-1]
         attr = option_path[-1]
-        return process_args(
+        return set_option(
             option=".".join(option_parent),
             value={attr: value},
             options=options,
             settings_file=settings_file,
-            quiet=quiet,
             option_description=option,
         )
 
@@ -170,45 +242,14 @@ def process_args(
         current_config = {}
     # merge and save the new config file
     new_config = merge(current_config, result)
+    settings_file.parent.mkdir(parents=True, exist_ok=True)
     with open(settings_file, "w") as f:
         json.dump(new_config, f, indent=2)
-    if not quiet:
-        new_value = read_option(option)
-        print(f"New Value for {option}:")
-        print(new_value)
-
-
-def register_parser(
-    parser: argparse.ArgumentParser,
-    options_file: Optional[Union[str, Path]] = os.environ.get("CLAN_OPTIONS_FILE"),
-) -> None:
-    if not options_file:
-        # use nix eval to evaluate .#clanOptions
-        # this will give us the evaluated config with the options attribute
-        proc = subprocess.run(
-            [
-                "nix",
-                "eval",
-                "--raw",
-                ".#clanOptions",
-            ],
-            check=True,
-            capture_output=True,
-            text=True,
-        )
-        file = proc.stdout.strip()
-        with open(file) as f:
-            options = json.load(f)
-    else:
-        with open(options_file) as f:
-            options = json.load(f)
-    return _register_parser(parser, options)
 
 
 # takes a (sub)parser and configures it
-def _register_parser(
+def register_parser(
     parser: Optional[argparse.ArgumentParser],
-    options: dict[str, Any],
 ) -> None:
     if parser is None:
         parser = argparse.ArgumentParser(
@@ -216,31 +257,35 @@ def _register_parser(
         )
 
     # inject callback function to process the input later
-    parser.set_defaults(
+    parser.set_defaults(func=get_or_set_option)
-        func=lambda args: process_args(
-            option=args.option,
-            value=args.value,
-            options=options,
-            quiet=args.quiet,
-            settings_file=args.settings_file,
-        )
-    )
 
-    # add --quiet option
+    # add --machine argument
     parser.add_argument(
-        "--quiet",
+        "--machine",
-        "-q",
+        "-m",
-        help="Suppress output",
+        help="Machine to configure",
-        action="store_true",
+        type=str,
+        default="default",
     )
 
-    # add argument to pass output file
+    # add --options-file argument
+    parser.add_argument(
+        "--options-file",
+        help="JSON file with options",
+        type=Path,
+    )
+
+    # add --settings-file argument
     parser.add_argument(
         "--settings-file",
-        "-o",
+        help="JSON file with settings",
-        help="Output file",
         type=Path,
-        default=Path("clan-settings.json"),
+    )
+    # add --quiet argument
+    parser.add_argument(
+        "--quiet",
+        help="Do not print the value",
+        action="store_true",
     )
 
     # add single positional argument for the option (e.g. "foo.bar")
@@ -248,7 +293,6 @@ def _register_parser(
         "option",
         help="Option to configure",
         type=str,
-        choices=AllContainer(list(options.keys())),
     )
 
     # add a single optional argument for the value
@@ -264,14 +308,8 @@ def main(argv: Optional[list[str]] = None) -> None:
     if argv is None:
         argv = sys.argv
     parser = argparse.ArgumentParser()
-    parser.add_argument(
+    register_parser(parser)
-        "schema",
+    parser.parse_args(argv[1:])
-        help="The schema to use for the configuration",
-        type=Path,
-    )
-    args = parser.parse_args(argv[1:2])
-    register_parser(parser, args.schema)
-    parser.parse_args(argv[2:])
 
 
 if __name__ == "__main__":

pkgs/clan-cli/clan_cli/config/machine.py

@@ -55,7 +55,7 @@ def schema_for_machine(machine_name: str, flake: Optional[Path] = None) -> dict:
             let
                 flake = builtins.getFlake (toString {flake});
                 lib = import {nixpkgs()}/lib;
-                module = builtins.trace (builtins.attrNames flake) flake.nixosModules.machine-{machine_name};
+                module = flake.nixosModules.machine-{machine_name};
                 evaled = lib.evalModules {{
                     modules = [module];
                 }};

pkgs/clan-cli/clan_cli/nix.py

@@ -1,8 +1,34 @@
 import os
+import tempfile
 
 from .dirs import flake_registry, unfree_nixpkgs
 
 
+def nix_eval(flags: list[str]) -> list[str]:
+    if os.environ.get("IN_NIX_SANDBOX"):
+        with tempfile.TemporaryDirectory() as nix_store:
+            return [
+                "nix",
+                "eval",
+                "--show-trace",
+                "--extra-experimental-features",
+                "nix-command flakes",
+                "--flake-registry",
+                str(flake_registry()),
+                # --store is required to prevent this error:
+                # error: cannot unlink '/nix/store/6xg259477c90a229xwmb53pdfkn6ig3g-default-builder.sh': Operation not permitted
+                "--store",
+                nix_store,
+            ] + flags
+    return [
+        "nix",
+        "eval",
+        "--show-trace",
+        "--extra-experimental-features",
+        "nix-command flakes",
+    ] + flags
+
+
 def nix_shell(packages: list[str], cmd: list[str]) -> list[str]:
     # we cannot use nix-shell inside the nix sandbox
     # in our tests we just make sure we have all the packages

pkgs/clan-cli/default.nix

@@ -77,7 +77,7 @@ python3.pkgs.buildPythonPackage {
   ];
   propagatedBuildInputs = dependencies;
 
-  passthru.tests.clan-pytest = runCommand "clan-tests"
+  passthru.tests.clan-pytest = runCommand "clan-pytest"
     {
       nativeBuildInputs = [ age zerotierone bubblewrap sops nix openssh rsync stdenv.cc ];
     } ''

pkgs/clan-cli/tests/machine_flake/flake.nix

@@ -4,7 +4,14 @@
     nixpkgs.url = "__NIXPKGS__";
   };
 
-  outputs = _inputs: {
+  outputs = inputs: {
     nixosModules.machine-machine1 = ./nixosModules/machine1.nix;
+    nixosConfigurations.machine1 = inputs.nixpkgs.lib.nixosSystem {
+      modules = [
+        inputs.self.nixosModules.machine-machine1
+        (builtins.fromJSON (builtins.readFile ./machines/machine1.json))
+        { nixpkgs.hostPlatform = "x86_64-linux"; }
+      ];
+    };
   };
 }

pkgs/clan-cli/tests/test_config.py

@@ -28,20 +28,44 @@ example_options = f"{Path(config.__file__).parent}/jsonschema/options.json"
 def test_set_some_option(
     args: list[str],
     expected: dict[str, Any],
-    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    monkeypatch.setenv("CLAN_OPTIONS_FILE", example_options)
-
     # create temporary file for out_file
     with tempfile.NamedTemporaryFile() as out_file:
         with open(out_file.name, "w") as f:
             json.dump({}, f)
         cli = Cli()
-        cli.run(["config", "--quiet", "--settings-file", out_file.name] + args)
+        cli.run(
+            [
+                "config",
+                "--quiet",
+                "--options-file",
+                example_options,
+                "--settings-file",
+                out_file.name,
+            ]
+            + args
+        )
         json_out = json.loads(open(out_file.name).read())
         assert json_out == expected
 
 
+def test_configure_machine(
+    machine_flake: Path,
+    temporary_dir: Path,
+    capsys: pytest.CaptureFixture,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("HOME", str(temporary_dir))
+    cli = Cli()
+    cli.run(["config", "-m", "machine1", "clan.jitsi.enable", "true"])
+    # clear the output buffer
+    capsys.readouterr()
+    # read a option value
+    cli.run(["config", "-m", "machine1", "clan.jitsi.enable"])
+    # read the output
+    assert capsys.readouterr().out == "true\n"
+
+
 def test_walk_jsonschema_all_types() -> None:
     schema = dict(
         type="object",