yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

clan-cli: Fix templates not downloading template, Make templates use Flake cache, Fix flake cache exception on conditional attribute, add more tests

Browse Source
This commit is contained in:
Qubasa
2025-03-17 14:31:53 +01:00
committed by Mic92
parent 33abb7ecd7
commit dc8bfab65d
13 changed files with 409 additions and 230 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
pkgs/clan-cli/clan_cli/clan/create.py
View File

@@ -9,7 +9,7 @@ from clan_cli.cmd import CmdOut, RunOpts, run
from clan_cli.errors import ClanError
from clan_cli.flake import Flake
from clan_cli.inventory import Inventory, init_inventory
from clan_cli.nix import nix_metadata, nix_shell
from clan_cli.nix import nix_command, nix_metadata, nix_shell
from clan_cli.templates import (
InputPrio,
TemplateName,
@@ -65,21 +65,15 @@ def create_clan(opts: CreateOptions) -> CreateClanResponse:
clan_dir=opts.src_flake,
)
log.info(f"Found template '{template.name}' in '{template.input_variant}'")
src = Path(template.src["path"])
if dest.exists():
dest /= src.name
dest /= template.name
if dest.exists():
msg = f"Destination directory {dest} already exists"
raise ClanError(msg)
if not src.exists():
msg = f"Template {template} does not exist"
raise ClanError(msg)
if not src.is_dir():
msg = f"Template {template} is not a directory"
raise ClanError(msg)
src = Path(template.src["path"])
copy_from_nixstore(src, dest)
@@ -108,9 +102,7 @@ def create_clan(opts: CreateOptions) -> CreateClanResponse:
)
if opts.update_clan:
flake_update = run(
nix_shell(["nixpkgs#nix"], ["nix", "flake", "update"]), RunOpts(cwd=dest)
)
flake_update = run(nix_command(["flake", "update"]), RunOpts(cwd=dest))
response.flake_update = flake_update
if opts.initial:
@@ -159,6 +151,13 @@ def register_create_parser(parser: argparse.ArgumentParser) -> None:
default=Path(),
)
parser.add_argument(
"--no-update",
help="Do not update the clan flake",
action="store_true",
default=False,
)
def create_flake_command(args: argparse.Namespace) -> None:
if len(args.input) == 0:
args.input = ["clan", "clan-core"]
@@ -175,6 +174,7 @@ def register_create_parser(parser: argparse.ArgumentParser) -> None:
template_name=args.template,
setup_git=not args.no_git,
src_flake=args.flake,
update_clan=not args.no_update,
)
)

68
pkgs/clan-cli/clan_cli/flake.py
View File

@@ -7,7 +7,7 @@ from hashlib import sha1
from pathlib import Path
from typing import Any, cast
from clan_cli.cmd import run
from clan_cli.cmd import Log, RunOpts, run
from clan_cli.dirs import user_cache_dir
from clan_cli.errors import ClanError
from clan_cli.nix import (
@@ -261,9 +261,10 @@ class FlakeCacheEntry:
return selectors == []
if isinstance(selector, AllSelector):
if isinstance(self.selector, AllSelector):
return all(
result = all(
self.value[sel].is_cached(selectors[1:]) for sel in self.value
)
return result
# TODO: check if we already have all the keys anyway?
return False
if (
@@ -273,11 +274,19 @@ class FlakeCacheEntry:
):
if not selector.issubset(self.selector):
return False
return all(self.value[sel].is_cached(selectors[1:]) for sel in selector)
result = all(
self.value[sel].is_cached(selectors[1:]) if sel in self.value else True
for sel in selector
)
return result
if isinstance(selector, str | int) and isinstance(self.value, dict):
if selector in self.value:
return self.value[selector].is_cached(selectors[1:])
result = self.value[selector].is_cached(selectors[1:])
return result
return False
return False
def select(self, selectors: list[Selector]) -> Any:
@@ -318,6 +327,7 @@ class FlakeCacheEntry:
return f"FlakeCache {self.value}"
@dataclass
class FlakeCache:
"""
an in-memory cache for flake outputs, uses a recursive FLakeCacheEntry structure
@@ -350,6 +360,7 @@ class FlakeCache:
def load_from_file(self, path: Path) -> None:
if path.exists():
with path.open("rb") as f:
log.debug(f"Loading cache from {path}")
self.cache = pickle.load(f)
@@ -361,11 +372,14 @@ class Flake:
"""
identifier: str
def __post_init__(self) -> None:
self._cache: FlakeCache | None = None
self._path: Path | None = None
self._is_local: bool | None = None
inputs_from: str | None = None
hash: str | None = None
flake_cache_path: Path | None = None
store_path: str | None = None
cache: FlakeCache | None = None
_cache: FlakeCache | None = None
_path: Path | None = None
_is_local: bool | None = None
@classmethod
def from_json(cls: type["Flake"], data: dict[str, Any]) -> "Flake":
@@ -400,24 +414,26 @@ class Flake:
"""
Run prefetch to flush the cache as well as initializing it.
"""
flake_prefetch = run(
nix_command(
[
"flake",
"prefetch",
"--json",
"--option",
"flake-registry",
"",
self.identifier,
]
)
)
cmd = [
"flake",
"prefetch",
"--json",
"--option",
"flake-registry",
"",
self.identifier,
]
if self.inputs_from:
cmd += ["--inputs-from", self.inputs_from]
flake_prefetch = run(nix_command(cmd))
flake_metadata = json.loads(flake_prefetch.stdout)
self.store_path = flake_metadata["storePath"]
self.hash = flake_metadata["hash"]
self._cache = FlakeCache()
assert self.hash is not None
hashed_hash = sha1(self.hash.encode()).hexdigest()
self.flake_cache_path = Path(user_cache_dir()) / "clan" / "flakes" / hashed_hash
if self.flake_cache_path.exists():
@@ -436,6 +452,7 @@ class Flake:
self._path = Path(flake_metadata["original"]["path"])
else:
self._is_local = False
assert self.store_path is not None
self._path = Path(self.store_path)
def get_from_nix(
@@ -464,7 +481,9 @@ class Flake:
nix_options.append("--impure")
build_output = Path(
run(nix_build(["--expr", nix_code, *nix_options])).stdout.strip()
run(
nix_build(["--expr", nix_code, *nix_options]), RunOpts(log=Log.NONE)
).stdout.strip()
)
if tmp_store:
@@ -473,6 +492,7 @@ class Flake:
if len(outputs) != len(selectors):
msg = f"flake_prepare_cache: Expected {len(outputs)} outputs, got {len(outputs)}"
raise ClanError(msg)
assert self.flake_cache_path is not None
self._cache.load_from_file(self.flake_cache_path)
for i, selector in enumerate(selectors):
self._cache.insert(outputs[i], selector)
@@ -486,8 +506,8 @@ class Flake:
if self._cache is None:
self.prefetch()
assert self._cache is not None
assert self.flake_cache_path is not None
self._cache.load_from_file(self.flake_cache_path)
if not self._cache.is_cached(selector):
log.debug(f"Cache miss for {selector}")
self.get_from_nix([selector], nix_options)

2
pkgs/clan-cli/clan_cli/select.py
View File

@@ -5,7 +5,7 @@ from clan_cli.flake import Flake
def select_command(args: argparse.Namespace) -> None:
flake = Flake(args.flake.path)
flake: Flake = args.flake
print(json.dumps(flake.select(args.selector), indent=4))

259
pkgs/clan-cli/clan_cli/templates.py
View File

@@ -1,15 +1,12 @@
import json
import logging
import shutil
import stat
import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Literal, NewType, TypedDict
from typing import Any, Literal, NewType, TypedDict, cast
from clan_cli.cmd import run
from clan_cli.errors import ClanError
from clan_cli.errors import ClanCmdError, ClanError
from clan_cli.flake import Flake
from clan_cli.nix import nix_eval
log = logging.getLogger(__name__)
@@ -35,18 +32,31 @@ ModuleName = NewType("ModuleName", str)
class ClanModule(TypedDict):
description: str
class InternalClanModule(ClanModule):
path: str
class Template(TypedDict):
description: str
class TemplatePath(Template):
path: str
@dataclass
class FoundTemplate:
input_variant: InputVariant
name: TemplateName
src: TemplatePath
class TemplateTypeDict(TypedDict):
disko: dict[TemplateName, Template] # Templates under "disko" type
clan: dict[TemplateName, Template] # Templates under "clan" type
machine: dict[TemplateName, Template] # Templates under "machine" type
disko: dict[TemplateName, TemplatePath]
clan: dict[TemplateName, TemplatePath]
machine: dict[TemplateName, TemplatePath]
class ClanAttrset(TypedDict):
@@ -59,48 +69,86 @@ class ClanExports(TypedDict):
self: ClanAttrset
def apply_fallback_structure(attrset: dict[str, Any]) -> ClanAttrset:
"""Ensure the attrset has all required fields with defaults when missing."""
# Deep copy not needed since we're constructing the dict from scratch
result: dict[str, Any] = {}
# Ensure templates field exists
if "templates" not in attrset:
result["templates"] = {"disko": {}, "clan": {}, "machine": {}}
else:
templates = attrset["templates"]
result["templates"] = {
"disko": templates.get("disko", {}),
"clan": templates.get("clan", {}),
"machine": templates.get("machine", {}),
}
# Ensure modules field exists
result["modules"] = attrset.get("modules", {})
return cast(ClanAttrset, result)
def get_clan_nix_attrset(clan_dir: Flake | None = None) -> ClanExports:
"""
Get the clan nix attrset from a flake, with fallback structure applied.
Path inside the attrsets have NOT YET been realized in the nix store.
"""
# Check if the clan directory is provided, otherwise use the environment variable
if not clan_dir:
# TODO: Quickfix, templates dir seems to be missing in CLAN_CORE_PATH??
clan_core_path = "git+https://git.clan.lol/clan/clan-core"
# clan_core_path = os.environ.get("CLAN_CORE_PATH")
# if not clan_core_path:
# msg = "Environment var CLAN_CORE_PATH is not set, this shouldn't happen"
# raise ClanError(msg)
clan_core_path = os.environ.get("CLAN_CORE_PATH")
if not clan_core_path:
msg = "Environment var CLAN_CORE_PATH is not set, this shouldn't happen"
raise ClanError(msg)
clan_dir = Flake(clan_core_path)
log.debug(f"Evaluating flake {clan_dir} for Clan attrsets")
# Nix evaluation script to compute find inputs that have a "clan" attribute
eval_script = f"""
let
self = builtins.getFlake "{clan_dir}";
lib = self.inputs.nixpkgs.lib;
inputsWithClan = lib.mapAttrs (
_name: value: value.clan
) (lib.filterAttrs(_name: value: value ? "clan") self.inputs);
in
{{ inputs = inputsWithClan; self = self.clan or {{}}; }}
"""
raw_clan_exports: dict[str, Any] = {"self": {"clan": {}}, "inputs": {"clan": {}}}
cmd = nix_eval(
[
"--json",
"--impure",
"--expr",
eval_script,
]
)
res = run(cmd).stdout
try:
raw_clan_exports["self"] = clan_dir.select("clan")
except ClanCmdError:
log.info("Current flake does not export the 'clan' attribute")
return json.loads(res)
# FIXME: flake.select destroys lazy evaluation
# this is why if one input has a template with a non existant path
# the whole evaluation will fail
try:
raw_clan_exports["inputs"] = clan_dir.select("inputs.*.{clan}")
except ClanCmdError as e:
msg = "Failed to evaluate flake inputs"
raise ClanError(msg) from e
inputs_with_fallback = {}
for input_name, attrset in raw_clan_exports["inputs"].items():
# FIXME: flake.select("inputs.*.{clan}") returns the wrong attrset
# depth when used with conditional fields
# this is why we have to do a attrset.get here
inputs_with_fallback[input_name] = apply_fallback_structure(
attrset.get("clan", {})
)
# Apply fallback structure to self
self_with_fallback = apply_fallback_structure(raw_clan_exports["self"])
# Construct the final result
clan_exports: ClanExports = {
"inputs": inputs_with_fallback,
"self": self_with_fallback,
}
return clan_exports
# Dataclass to manage input prioritization for templates
@dataclass
class InputPrio:
"""
Strategy for prioritizing inputs when searching for a template
"""
input_names: tuple[str, ...] # Tuple of input names (ordered priority list)
prioritize_self: bool = True # Whether to prioritize "self" first
@@ -120,51 +168,9 @@ class InputPrio:
return InputPrio(prioritize_self=True, input_names=input_names)
@dataclass
class FoundTemplate:
input_variant: InputVariant
name: TemplateName
src: Template
def copy_from_nixstore(src: Path, dest: Path) -> None:
if src.is_symlink():
target = src.readlink()
src.symlink_to(target)
return
if src.is_file():
shutil.copy(src, dest)
dest.chmod(stat.S_IWRITE | stat.S_IREAD | stat.S_IRGRP)
return
# Walk through the source directory
for root, dirs, files in src.walk(on_error=log.error):
relative_path = Path(root).relative_to(src)
dest_dir = dest / relative_path
dest_dir.mkdir(exist_ok=True)
log.debug(f"Creating directory '{dest_dir}'")
# Set permissions for directories
dest_dir.chmod(
stat.S_IWRITE | stat.S_IREAD | stat.S_IEXEC | stat.S_IRGRP | stat.S_IXGRP
)
for d in dirs:
(dest_dir / d).mkdir()
for file_name in files:
src_file = Path(root) / file_name
dest_file = dest_dir / file_name
if src_file.is_symlink():
target = src_file.readlink()
dest_file.symlink_to(target)
log.debug(f"Created symlink '{dest_file}' -> '{target}'")
else:
# Copy the file
shutil.copy(src_file, dest_file)
dest_file.chmod(stat.S_IWRITE | stat.S_IREAD | stat.S_IRGRP)
run(["cp", "-r", "--reflink=auto", str(src), str(dest)])
run(["chmod", "-R", "u+w", str(dest)])
@dataclass
@@ -176,33 +182,39 @@ class TemplateList:
def list_templates(
template_type: TemplateType, clan_dir: Flake | None = None
) -> TemplateList:
"""
List all templates of a specific type from a flake, without a path attribute.
As these paths are not yet downloaded into the nix store, and thus cannot be used directly.
"""
clan_exports = get_clan_nix_attrset(clan_dir)
result = TemplateList()
fallback: ClanAttrset = {
"templates": {"disko": {}, "clan": {}, "machine": {}},
"modules": {},
}
clan_templates = (
clan_exports["self"]
.get("templates", fallback["templates"])
.get(template_type, {})
)
result.self = clan_templates
for input_name, _attrset in clan_exports["inputs"].items():
clan_templates = (
clan_exports["inputs"]
.get(input_name, fallback)["templates"]
.get(template_type, {})
)
result.inputs[input_name] = {}
for template_name, template in clan_templates.items():
for template_name, template in clan_exports["self"]["templates"][
template_type
].items():
result.self[template_name] = template
for input_name, attrset in clan_exports["inputs"].items():
for template_name, template in attrset["templates"][template_type].items():
if input_name not in result.inputs:
result.inputs[input_name] = {}
result.inputs[input_name][template_name] = template
return result
# Function to retrieve a specific template from Clan exports
def realize_nix_path(clan_dir: Flake, nix_path: str) -> None:
"""
Downloads / realizes a nix path into the nix store
"""
if Path(nix_path).exists():
return
flake = Flake(identifier=nix_path, inputs_from=clan_dir.identifier)
flake.prefetch()
def get_template(
template_name: TemplateName,
template_type: TemplateType,
@@ -210,6 +222,19 @@ def get_template(
input_prio: InputPrio | None = None,
clan_dir: Flake | None = None,
) -> FoundTemplate:
"""
Find a specific template by name and type within a flake and then ensures it is realized in the nix store.
"""
if not clan_dir:
clan_core_path = os.environ.get("CLAN_CORE_PATH")
if not clan_core_path:
msg = "Environment var CLAN_CORE_PATH is not set, this shouldn't happen"
raise ClanError(msg)
clan_dir = Flake(clan_core_path)
log.info(f"Get template in {clan_dir}")
log.info(f"Searching for template '{template_name}' of type '{template_type}'")
# Set default priority strategy if none is provided
@@ -218,46 +243,50 @@ def get_template(
# Helper function to search for a specific template within a dictionary of templates
def find_template(
template_name: TemplateName, templates: dict[TemplateName, Template]
) -> Template | None:
template_name: TemplateName, templates: dict[TemplateName, TemplatePath]
) -> TemplatePath | None:
if template_name in templates:
return templates[template_name]
return None
# Initialize variables for the search results
template: Template | None = None
template: TemplatePath | None = None
input_name: InputName | None = None
template_list = list_templates(template_type, clan_dir)
clan_exports = get_clan_nix_attrset(clan_dir)
# Step 1: Check "self" first, if prioritize_self is enabled
if input_prio.prioritize_self:
log.info(f"Checking 'self' for template '{template_name}'")
template = find_template(template_name, template_list.self)
template = find_template(
template_name, clan_exports["self"]["templates"][template_type]
)
# Step 2: Otherwise, check the external inputs if no match is found
if not template and input_prio.input_names:
log.info(f"Checking external inputs for template '{template_name}'")
for input_name_str in input_prio.input_names:
input_name = InputName(input_name_str)
log.debug(f"Checking input '{input_name}' for template '{template_name}'")
log.debug(f"Searching in '{input_name}' for template '{template_name}'")
if input_name not in clan_exports["inputs"]:
log.debug(f"Skipping input '{input_name}', not found in '{clan_dir}'")
continue
template = find_template(
template_name, template_list.inputs.get(input_name, {})
template_name,
clan_exports["inputs"][input_name]["templates"][template_type],
)
if template:
log.debug(f"Found template '{template_name}' in input '{input_name}'")
break # Stop searching once the template is found
break
# Step 3: Raise an error if the template wasn't found
if not template:
source = (
f"inputs.{input_name}.clan.templates.{template_type}"
if input_name # Most recent "input_name"
else f"flake.clan.templates.{template_type}"
)
msg = f"Template '{template_name}' not in '{source}' in '{clan_dir}'"
msg = f"Template '{template_name}' could not be found in '{clan_dir}'"
raise ClanError(msg)
realize_nix_path(clan_dir, template["path"])
return FoundTemplate(
input_variant=InputVariant(input_name), src=template, name=template_name
)

14
pkgs/clan-cli/default.nix
View File

@@ -15,6 +15,7 @@
inventory-schema-abstract,
classgen,
pythonRuntime,
templateDerivation,
}:
let
pyDeps = ps: [
@@ -128,7 +129,16 @@ pythonRuntime.pkgs.buildPythonApplication {
(lib.mapAttrs' (n: lib.nameValuePair "clan-dep-${n}") testRuntimeDependenciesMap)
// {
clan-pytest-without-core =
runCommand "clan-pytest-without-core" { nativeBuildInputs = testDependencies; }
runCommand "clan-pytest-without-core"
{
nativeBuildInputs = testDependencies;
closureInfo = pkgs.closureInfo {
rootPaths = [
templateDerivation
];
};
}
''
set -u -o pipefail
cp -r ${source} ./src
@@ -160,6 +170,7 @@ pythonRuntime.pkgs.buildPythonApplication {
];
closureInfo = pkgs.closureInfo {
rootPaths = [
templateDerivation
pkgs.bash
pkgs.coreutils
pkgs.jq.dev
@@ -175,6 +186,7 @@ pythonRuntime.pkgs.buildPythonApplication {
cd ./src
export CLAN_CORE=${clan-core-path}
export CLAN_CORE_PATH=${clan-core-path}
export NIX_STATE_DIR=$TMPDIR/nix
export IN_NIX_SANDBOX=1
export PYTHONWARNINGS=error

46
pkgs/clan-cli/flake-module.nix
View File

@@ -19,6 +19,7 @@
"lib"
"nixosModules"
"flake.lock"
"templates"
];
};
flakeLock = lib.importJSON (clanCore + "/flake.lock");
@@ -48,7 +49,9 @@
};
clanCoreLock = flakeLockVendoredDeps flakeLock;
clanCoreLockFile = builtins.toFile "clan-core-flake.lock" (builtins.toJSON clanCoreLock);
clanCoreNode = {
inputs = lib.mapAttrs (name: _input: name) flakeInputs;
locked = {
lastModified = 1;
@@ -66,16 +69,33 @@
nodes = clanCoreLock.nodes // {
clan-core = clanCoreNode;
nixpkgs-lib = clanCoreLock.nodes.nixpkgs; # required by flake-parts
flake-parts = clanCoreLock.nodes.flake-parts;
root = clanCoreLock.nodes.root // {
inputs = clanCoreLock.nodes.root.inputs // {
clan-core = "clan-core";
nixpkgs = "nixpkgs";
nixpkgs-lib = "nixpkgs-lib";
clan = "clan-core";
flake-parts = "flake-parts";
};
};
};
};
templateLockFile = builtins.toFile "template-flake.lock" (builtins.toJSON templateLock);
# We need to add the paths of the templates to the nix store such that they are available
# only adding clanCoreWithVendoredDeps to the nix store is not enough
templateDerivation = pkgs.closureInfo {
rootPaths =
builtins.attrValues (self.lib.select "clan.templates.clan.*.path" self)
++ builtins.attrValues (self.lib.select "clan.templates.machine.*.path" self);
# FIXME: As the templates get modified in clanCoreWithVendoredDeps below, we need to add the modified version to the nix store too
# However it is not possible (or I don't know how) to add a nix path from a built derivation to the nix store
# rootPaths = [
# clanCoreWithVendoredDeps.clan.templates.clan.minimal.path
# ];
};
clanCoreWithVendoredDeps =
pkgs.runCommand "clan-core-with-vendored-deps"
{
@@ -96,15 +116,19 @@
cp ${clanCoreLockFile} $out/flake.lock
nix flake lock $out --extra-experimental-features 'nix-command flakes'
clanCoreHash=$(nix hash path ${clanCore} --extra-experimental-features 'nix-command')
for templateDir in $(find $out/templates -mindepth 1 -maxdepth 1 -type d); do
if ! [ -e "$templateDir/flake.nix" ]; then
continue
fi
cp ${templateLockFile} $templateDir/flake.lock
cat $templateDir/flake.lock | jq ".nodes.\"clan-core\".locked.narHash = \"$clanCoreHash\"" > $templateDir/flake.lock.final
mv $templateDir/flake.lock.final $templateDir/flake.lock
nix flake lock $templateDir --extra-experimental-features 'nix-command flakes'
done
## ==> We need this to make nix flake update work on the templates
## however then we have to re-add the clan templates to the nix store
## which is not possible (or I don't know how)
# for templateDir in $(find $out/templates/clan -mindepth 1 -maxdepth 1 -type d); do
# if ! [ -e "$templateDir/flake.nix" ]; then
# continue
# fi
# cp ${templateLockFile} $templateDir/flake.lock
# cat $templateDir/flake.lock | jq ".nodes.\"clan-core\".locked.narHash = \"$clanCoreHash\"" > $templateDir/flake.lock.final
# mv $templateDir/flake.lock.final $templateDir/flake.lock
# nix flake lock $templateDir --extra-experimental-features 'nix-command flakes'
# done
'';
in
{
@@ -117,6 +141,7 @@
inherit (inputs) nixpkgs;
inherit (self'.packages) classgen;
inherit (self'.legacyPackages.schemas) inventory-schema-abstract;
templateDerivation = templateDerivation;
pythonRuntime = pkgs.python3;
clan-core-path = clanCoreWithVendoredDeps;
includedRuntimeDeps = [
@@ -129,6 +154,7 @@
inherit (self'.packages) classgen;
inherit (self'.legacyPackages.schemas) inventory-schema-abstract;
clan-core-path = clanCoreWithVendoredDeps;
templateDerivation = templateDerivation;
pythonRuntime = pkgs.python3;
includedRuntimeDeps = lib.importJSON ./clan_cli/nix/allowed-programs.json;
};

124
pkgs/clan-cli/tests/test_clan_nix_attrset.py
View File

@@ -7,13 +7,16 @@ from typing import Any
import pytest
from clan_cli.cmd import run
from clan_cli.flake import Flake
from clan_cli.git import commit_file
from clan_cli.locked_open import locked_open
from clan_cli.nix import nix_command
from clan_cli.templates import (
ClanExports,
InputName,
TemplateName,
copy_from_nixstore,
get_clan_nix_attrset,
get_template,
list_templates,
)
from fixtures_flakes import FlakeForTest
@@ -25,18 +28,34 @@ def write_clan_attr(clan_attrset: dict[str, Any], flake: FlakeForTest) -> None:
with locked_open(file, "w") as cfile:
json.dump(clan_attrset, cfile, indent=2)
commit_file(file, flake.path, "Add clan attributes")
# Common function to test clan nix attrset
def nix_attr_tester(
test_flake: FlakeForTest,
test_flake_with_core: FlakeForTest,
injected: dict[str, Any],
expected: dict[str, Any],
expected_self: dict[str, Any],
test_number: int,
) -> None:
write_clan_attr(injected, test_flake)
nix_attrset = get_clan_nix_attrset(Flake(str(test_flake.path)))
) -> ClanExports:
write_clan_attr(injected, test_flake_with_core)
clan_dir = Flake(str(test_flake_with_core.path))
nix_attrset = get_clan_nix_attrset(clan_dir)
assert json.dumps(nix_attrset, indent=2) == json.dumps(expected, indent=2)
def recursive_sort(item: Any) -> Any:
if isinstance(item, dict):
return {k: recursive_sort(item[k]) for k in sorted(item)}
if isinstance(item, list):
return sorted(recursive_sort(elem) for elem in item)
return item
returned_sorted = recursive_sort(nix_attrset["self"])
expected_sorted = recursive_sort(expected_self["self"])
assert json.dumps(returned_sorted, indent=2) == json.dumps(
expected_sorted, indent=2
)
return nix_attrset
@pytest.mark.impure
@@ -68,6 +87,7 @@ def test_clan_core_templates(
) -> None:
clan_dir = Flake(str(test_flake_with_core.path))
nix_attrset = get_clan_nix_attrset(clan_dir)
clan_core_templates = nix_attrset["inputs"][InputName("clan-core")]["templates"][
"clan"
]
@@ -75,15 +95,21 @@ def test_clan_core_templates(
expected_templates = ["default", "flake-parts", "minimal", "minimal-flake-parts"]
assert clan_core_template_keys == expected_templates
vlist_temps = list_templates("clan", clan_dir)
list_template_keys = list(vlist_temps.inputs[InputName("clan-core")].keys())
assert list_template_keys == expected_templates
default_template = get_template(
TemplateName("default"),
"clan",
input_prio=None,
clan_dir=clan_dir,
)
new_clan = temporary_home / "new_clan"
copy_from_nixstore(
Path(
vlist_temps.inputs[InputName("clan-core")][TemplateName("default")]["path"]
),
Path(default_template.src["path"]),
new_clan,
)
assert (new_clan / "flake.nix").exists()
@@ -99,20 +125,27 @@ def test_clan_core_templates(
# Test Case 1: Minimal input with empty templates
@pytest.mark.impure
@pytest.mark.with_core
def test_clan_get_nix_attrset_case_1(
monkeypatch: pytest.MonkeyPatch, temporary_home: Path, test_flake: FlakeForTest
monkeypatch: pytest.MonkeyPatch,
temporary_home: Path,
test_flake_with_core: FlakeForTest,
) -> None:
test_number = 1
injected = {"templates": {"clan": {}}}
expected = {"inputs": {}, "self": {"templates": {"clan": {}}}}
nix_attr_tester(test_flake, injected, expected, test_number)
injected = {"templates": {"disko": {}, "machine": {}}}
expected = {
"inputs": {},
"self": {"templates": {"disko": {}, "machine": {}, "clan": {}}, "modules": {}},
}
nix_attr_tester(test_flake_with_core, injected, expected, test_number)
# Test Case 2: Input with one template under 'clan'
@pytest.mark.impure
@pytest.mark.with_core
def test_clan_get_nix_attrset_case_2(
monkeypatch: pytest.MonkeyPatch, temporary_home: Path, test_flake: FlakeForTest
monkeypatch: pytest.MonkeyPatch,
temporary_home: Path,
test_flake_with_core: FlakeForTest,
) -> None:
test_number = 2
injected = {
@@ -134,17 +167,27 @@ def test_clan_get_nix_attrset_case_2(
"description": "An example clan template.",
"path": "/example/path",
}
}
}
},
"disko": {},
"machine": {},
},
"modules": {},
},
}
nix_attr_tester(test_flake, injected, expected, test_number)
nix_attrset = nix_attr_tester(test_flake_with_core, injected, expected, test_number)
assert "default" in list(
nix_attrset["inputs"][InputName("clan-core")]["templates"]["clan"].keys()
)
# Test Case 3: Input with templates under multiple types
@pytest.mark.impure
@pytest.mark.with_core
def test_clan_get_nix_attrset_case_3(
monkeypatch: pytest.MonkeyPatch, temporary_home: Path, test_flake: FlakeForTest
monkeypatch: pytest.MonkeyPatch,
temporary_home: Path,
test_flake_with_core: FlakeForTest,
) -> None:
test_number = 3
injected = {
@@ -191,16 +234,19 @@ def test_clan_get_nix_attrset_case_3(
"path": "/machine/path",
}
},
}
},
"modules": {},
},
}
nix_attr_tester(test_flake, injected, expected, test_number)
nix_attr_tester(test_flake_with_core, injected, expected, test_number)
# Test Case 4: Input with modules only
@pytest.mark.impure
@pytest.mark.with_core
def test_clan_get_nix_attrset_case_4(
monkeypatch: pytest.MonkeyPatch, temporary_home: Path, test_flake: FlakeForTest
monkeypatch: pytest.MonkeyPatch,
temporary_home: Path,
test_flake_with_core: FlakeForTest,
) -> None:
test_number = 4
injected = {
@@ -216,15 +262,18 @@ def test_clan_get_nix_attrset_case_4(
"module1": {"description": "First module", "path": "/module1/path"},
"module2": {"description": "Second module", "path": "/module2/path"},
},
"templates": {"disko": {}, "machine": {}, "clan": {}},
},
}
nix_attr_tester(test_flake, injected, expected, test_number)
nix_attr_tester(test_flake_with_core, injected, expected, test_number)
# Test Case 5: Input with both templates and modules
@pytest.mark.impure
@pytest.mark.with_core
def test_clan_get_nix_attrset_case_5(
monkeypatch: pytest.MonkeyPatch, temporary_home: Path, test_flake: FlakeForTest
monkeypatch: pytest.MonkeyPatch,
temporary_home: Path,
test_flake_with_core: FlakeForTest,
) -> None:
test_number = 5
injected = {
@@ -252,19 +301,26 @@ def test_clan_get_nix_attrset_case_5(
"description": "A clan template.",
"path": "/clan/path",
}
}
},
"disko": {},
"machine": {},
},
},
}
nix_attr_tester(test_flake, injected, expected, test_number)
nix_attr_tester(test_flake_with_core, injected, expected, test_number)
# Test Case 6: Input with missing 'templates' and 'modules' (empty clan attrset)
@pytest.mark.impure
@pytest.mark.with_core
def test_clan_get_nix_attrset_case_6(
monkeypatch: pytest.MonkeyPatch, temporary_home: Path, test_flake: FlakeForTest
monkeypatch: pytest.MonkeyPatch,
temporary_home: Path,
test_flake_with_core: FlakeForTest,
) -> None:
test_number = 6
injected = {}
expected = {"inputs": {}, "self": {}}
nix_attr_tester(test_flake, injected, expected, test_number)
expected = {
"inputs": {},
"self": {"templates": {"disko": {}, "machine": {}, "clan": {}}, "modules": {}},
}
nix_attr_tester(test_flake_with_core, injected, expected, test_number)

44
pkgs/clan-cli/tests/test_create_flake.py
View File

@@ -1,15 +1,18 @@
import json
import subprocess
import logging
from pathlib import Path
import pytest
from clan_cli.cmd import run
from fixtures_flakes import substitute
from clan_cli.nix import nix_flake_show
from fixtures_flakes import FlakeForTest, substitute
from helpers import cli
from stdout import CaptureOutput
log = logging.getLogger(__name__)
@pytest.mark.impure
@pytest.mark.with_core
def test_create_flake(
monkeypatch: pytest.MonkeyPatch,
temporary_home: Path,
@@ -18,7 +21,7 @@ def test_create_flake(
) -> None:
flake_dir = temporary_home / "test-flake"
cli.run(["flakes", "create", str(flake_dir), "--template=default"])
cli.run(["flakes", "create", str(flake_dir), "--template=default", "--no-update"])
assert (flake_dir / ".clan-flake").exists()
# Replace the inputs.clan.url in the template flake.nix
@@ -29,6 +32,7 @@ def test_create_flake(
# Dont evaluate the inventory before the substitute call
monkeypatch.chdir(flake_dir)
cli.run(["machines", "create", "machine1"])
# create a hardware-configuration.nix that doesn't throw an eval error
@@ -41,11 +45,8 @@ def test_create_flake(
with capture_output as output:
cli.run(["machines", "list"])
assert "machine1" in output.out
flake_show = subprocess.run(
["nix", "flake", "show", "--json"],
check=True,
capture_output=True,
text=True,
flake_show = run(
nix_flake_show(str(flake_dir)),
)
flake_outputs = json.loads(flake_show.stdout)
try:
@@ -54,7 +55,7 @@ def test_create_flake(
pytest.fail("nixosConfigurations.machine1 not found in flake outputs")
@pytest.mark.impure
@pytest.mark.with_core
def test_create_flake_existing_git(
monkeypatch: pytest.MonkeyPatch,
temporary_home: Path,
@@ -65,7 +66,7 @@ def test_create_flake_existing_git(
run(["git", "init", str(temporary_home)])
cli.run(["flakes", "create", str(flake_dir), "--template=default"])
cli.run(["flakes", "create", str(flake_dir), "--template=default", "--no-update"])
assert (flake_dir / ".clan-flake").exists()
# Replace the inputs.clan.url in the template flake.nix
@@ -88,11 +89,8 @@ def test_create_flake_existing_git(
with capture_output as output:
cli.run(["machines", "list"])
assert "machine1" in output.out
flake_show = subprocess.run(
["nix", "flake", "show", "--json"],
check=True,
capture_output=True,
text=True,
flake_show = run(
nix_flake_show(str(flake_dir)),
)
flake_outputs = json.loads(flake_show.stdout)
try:
@@ -101,15 +99,17 @@ def test_create_flake_existing_git(
pytest.fail("nixosConfigurations.machine1 not found in flake outputs")
@pytest.mark.impure
@pytest.mark.with_core
def test_ui_template(
monkeypatch: pytest.MonkeyPatch,
temporary_home: Path,
test_flake_with_core: FlakeForTest,
clan_core: Path,
capture_output: CaptureOutput,
) -> None:
flake_dir = temporary_home / "test-flake"
cli.run(["flakes", "create", str(flake_dir), "--template=minimal"])
cli.run(["flakes", "create", str(flake_dir), "--template=minimal", "--no-update"])
# Replace the inputs.clan.url in the template flake.nix
substitute(
@@ -118,16 +118,14 @@ def test_ui_template(
)
monkeypatch.chdir(flake_dir)
cli.run(["machines", "create", "machine1"])
with capture_output as output:
cli.run(["machines", "list"])
assert "machine1" in output.out
flake_show = subprocess.run(
["nix", "flake", "show", "--json"],
check=True,
capture_output=True,
text=True,
flake_show = run(
nix_flake_show(str(flake_dir)),
)
flake_outputs = json.loads(flake_show.stdout)
try:

7
pkgs/clan-cli/tests/test_flake/flake.nix
View File

@@ -13,13 +13,14 @@
clan-core = fake-clan-core;
};
lib = inputs.nixpkgs.lib;
in
{
clan =
clan_attrs_json =
if lib.pathExists ./clan_attrs.json then
builtins.fromJSON (builtins.readFile ./clan_attrs.json)
else
{ };
in
{
clan = clan_attrs_json;
nixosConfigurations.machine1 = inputs.nixpkgs.lib.nixosSystem {
modules = [

28
pkgs/clan-cli/tests/test_flake_caching.py
View File

@@ -1,7 +1,11 @@
import logging
import pytest
from clan_cli.flake import Flake, FlakeCache, FlakeCacheEntry
from fixtures_flakes import ClanFlake
log = logging.getLogger(__name__)
def test_select() -> None:
testdict = {"x": {"y": [123, 345, 456], "z": "bla"}}
@@ -58,3 +62,27 @@ def test_cache_persistance(flake: ClanFlake) -> None:
assert flake2._cache.is_cached( # noqa: SLF001
"nixosConfigurations.*.config.networking.{hostName,hostId}"
)
@pytest.mark.with_core
def test_conditional_all_selector(flake: ClanFlake) -> None:
m1 = flake.machines["machine1"]
m1["nixpkgs"]["hostPlatform"] = "x86_64-linux"
flake.refresh()
flake1 = Flake(str(flake.path))
flake2 = Flake(str(flake.path))
flake1.prefetch()
flake2.prefetch()
assert isinstance(flake1._cache, FlakeCache) # noqa: SLF001
assert isinstance(flake2._cache, FlakeCache) # noqa: SLF001
log.info("First select")
res1 = flake1.select("inputs.*.{clan,missing}")
log.info("Second (cached) select")
res2 = flake1.select("inputs.*.{clan,missing}")
assert res1 == res2
assert res1["clan-core"].get("clan") is not None
flake2.prefetch()

13
pkgs/clan-cli/tests/test_flake_with_core/flake.nix
View File

@@ -7,8 +7,18 @@
inputs.nixpkgs.url = "__NIXPKGS__";
outputs =
{ self, clan-core, ... }:
{
self,
clan-core,
nixpkgs,
...
}:
let
clan_attrs_json =
if nixpkgs.lib.pathExists ./clan_attrs.json then
builtins.fromJSON (builtins.readFile ./clan_attrs.json)
else
{ };
clan = clan-core.lib.buildClan {
inherit self;
meta.name = "test_flake_with_core";
@@ -48,6 +58,7 @@
};
in
{
clan = clan_attrs_json;
inherit (clan) nixosConfigurations clanInternals;
};
}

2
templates/clan/flake-parts/flake.nix
View File

@@ -2,7 +2,7 @@
inputs.clan-core.url = "https://git.clan.lol/clan/clan-core/archive/main.tar.gz";
inputs.nixpkgs.follows = "clan-core/nixpkgs";
inputs.flake-parts.url = "github:hercules-ci/flake-parts";
inputs.flake-parts.inputs.nixpkgs-lib.follows = "nixpkgs";
inputs.flake-parts.inputs.nixpkgs-lib.follows = "clan-core/nixpkgs";
outputs =
inputs@{

8
templates/clan/minimal-flake-parts/flake.nix
View File

@@ -1,13 +1,11 @@
{
inputs = {
nixpkgs.url = "github:NixOS/nixpkgs/nixpkgs-unstable";
inputs = {
clan.url = "https://git.clan.lol/clan/clan-core/archive/main.tar.gz";
clan.inputs.nixpkgs.follows = "nixpkgs";
clan.inputs.flake-parts.follows = "flake-parts";
nixpkgs.follows = "clan/nixpkgs";
flake-parts.url = "github:hercules-ci/flake-parts";
flake-parts.inputs.nixpkgs-lib.follows = "nixpkgs";
flake-parts.inputs.nixpkgs-lib.follows = "clan/nixpkgs";
};
outputs =