clan-cli: test_flake_caching: add actual flake caching test
This commit is contained in:
@@ -7,7 +7,7 @@ from typing import Any
|
|||||||
|
|
||||||
from clan_cli.cmd import run
|
from clan_cli.cmd import run
|
||||||
from clan_cli.errors import ClanError
|
from clan_cli.errors import ClanError
|
||||||
from clan_cli.nix import nix_build, nix_config
|
from clan_cli.nix import nix_build, nix_command, nix_config, nix_test_store
|
||||||
|
|
||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
@@ -278,7 +278,19 @@ class Flake:
|
|||||||
cache: FlakeCache = field(default_factory=FlakeCache)
|
cache: FlakeCache = field(default_factory=FlakeCache)
|
||||||
|
|
||||||
def __post_init__(self) -> None:
|
def __post_init__(self) -> None:
|
||||||
flake_prefetch = run(["nix", "flake", "prefetch", "--json", self.identifier])
|
flake_prefetch = run(
|
||||||
|
nix_command(
|
||||||
|
[
|
||||||
|
"flake",
|
||||||
|
"prefetch",
|
||||||
|
"--json",
|
||||||
|
"--option",
|
||||||
|
"flake-registry",
|
||||||
|
"",
|
||||||
|
self.identifier,
|
||||||
|
]
|
||||||
|
)
|
||||||
|
)
|
||||||
flake_metadata = json.loads(flake_prefetch.stdout)
|
flake_metadata = json.loads(flake_prefetch.stdout)
|
||||||
self.store_path = flake_metadata["storePath"]
|
self.store_path = flake_metadata["storePath"]
|
||||||
self.hash = flake_metadata["hash"]
|
self.hash = flake_metadata["hash"]
|
||||||
@@ -292,8 +304,10 @@ class Flake:
|
|||||||
in
|
in
|
||||||
flake.inputs.nixpkgs.legacyPackages.{config["system"]}.writeText "clan-flake-select" (builtins.toJSON [ ({" ".join([f'flake.clanInternals.lib.select "{attr}" flake' for attr in selectors])}) ])
|
flake.inputs.nixpkgs.legacyPackages.{config["system"]}.writeText "clan-flake-select" (builtins.toJSON [ ({" ".join([f'flake.clanInternals.lib.select "{attr}" flake' for attr in selectors])}) ])
|
||||||
"""
|
"""
|
||||||
build_output = run(nix_build(["--expr", nix_code])).stdout.strip()
|
build_output = Path(run(nix_build(["--expr", nix_code])).stdout.strip())
|
||||||
outputs = json.loads(Path(build_output).read_text())
|
if tmp_store := nix_test_store():
|
||||||
|
build_output = tmp_store.joinpath(*build_output.parts[1:])
|
||||||
|
outputs = json.loads(build_output.read_text())
|
||||||
if len(outputs) != len(selectors):
|
if len(outputs) != len(selectors):
|
||||||
msg = f"flake_prepare_cache: Expected {len(outputs)} outputs, got {len(outputs)}"
|
msg = f"flake_prepare_cache: Expected {len(outputs)} outputs, got {len(outputs)}"
|
||||||
raise ClanError(msg)
|
raise ClanError(msg)
|
||||||
|
|||||||
@@ -1,15 +1,27 @@
|
|||||||
from clan_cli.flake import FlakeCacheEntry
|
import pytest
|
||||||
|
from clan_cli.flake import Flake, FlakeCacheEntry
|
||||||
from fixtures_flakes import ClanFlake
|
from fixtures_flakes import ClanFlake
|
||||||
|
|
||||||
|
|
||||||
def test_flake_caching(test_flake: ClanFlake) -> None:
|
@pytest.mark.with_core
|
||||||
|
def test_flake_caching(flake: ClanFlake) -> None:
|
||||||
testdict = {"x": {"y": [123, 345, 456], "z": "bla"}}
|
testdict = {"x": {"y": [123, 345, 456], "z": "bla"}}
|
||||||
test_cache = FlakeCacheEntry(testdict, [])
|
test_cache = FlakeCacheEntry(testdict, [])
|
||||||
assert test_cache["x"]["z"].value == "bla"
|
assert test_cache["x"]["z"].value == "bla"
|
||||||
assert test_cache.is_cached(["x", "z"])
|
assert test_cache.is_cached(["x", "z"])
|
||||||
assert test_cache.select(["x", "y", 0]) == 123
|
assert test_cache.select(["x", "y", 0]) == 123
|
||||||
assert not test_cache.is_cached(["x", "z", 1])
|
assert not test_cache.is_cached(["x", "z", 1])
|
||||||
# TODO check this, but test_flake is not a real clan flake (no clan-core, no clanInternals)
|
|
||||||
# cmd.run(["nix", "flake", "lock"], cmd.RunOpts(cwd=test_flake.path))
|
m1 = flake.machines["machine1"]
|
||||||
# flake = Flake(str(test_flake.path))
|
m1["nixpkgs"]["hostPlatform"] = "x86_64-linux"
|
||||||
# hostnames = flake.select("nixosConfigurations.*.config.networking.hostName")
|
flake.machines["machine2"] = m1.copy()
|
||||||
|
flake.machines["machine3"] = m1.copy()
|
||||||
|
flake.refresh()
|
||||||
|
|
||||||
|
flake_ = Flake(str(flake.path))
|
||||||
|
hostnames = flake_.select("nixosConfigurations.*.config.networking.hostName")
|
||||||
|
assert hostnames == {
|
||||||
|
"machine1": "machine1",
|
||||||
|
"machine2": "machine2",
|
||||||
|
"machine3": "machine3",
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user