Merge pull request 'Templates: remove InputPrio and related classes' (#4221) from clan-templates into main
Reviewed-on: https://git.clan.lol/clan/clan-core/pulls/4221
This commit is contained in:
@@ -1,62 +1,15 @@
|
||||
# mypy: disable-error-code="var-annotated"
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
|
||||
from clan_cli.tests.fixtures_flakes import FlakeForTest
|
||||
from clan_lib.cmd import run
|
||||
from clan_lib.flake import Flake
|
||||
from clan_lib.git import commit_file
|
||||
from clan_lib.locked_open import locked_open
|
||||
from clan_lib.nix import nix_command
|
||||
from clan_lib.templates import (
|
||||
ClanExports,
|
||||
InputName,
|
||||
TemplateName,
|
||||
get_clan_nix_attrset,
|
||||
get_template,
|
||||
)
|
||||
|
||||
from clan_lib.templates import list_templates
|
||||
from clan_lib.templates.filesystem import copy_from_nixstore
|
||||
|
||||
|
||||
# Function to write clan attributes to a file
|
||||
def write_clan_attr(clan_attrset: dict[str, Any], flake: FlakeForTest) -> None:
|
||||
file = flake.path / "clan_attrs.json"
|
||||
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_with_core: FlakeForTest,
|
||||
injected: dict[str, Any],
|
||||
expected_self: dict[str, Any],
|
||||
test_number: int,
|
||||
) -> 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)
|
||||
|
||||
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
|
||||
def test_copy_from_nixstore_symlink(
|
||||
monkeypatch: pytest.MonkeyPatch, temporary_home: Path
|
||||
@@ -85,166 +38,37 @@ def test_clan_core_templates(
|
||||
temporary_home: Path,
|
||||
) -> 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"
|
||||
templates = list_templates(clan_dir)
|
||||
|
||||
assert list(templates.builtins.get("clan", {}).keys()) == [
|
||||
"default",
|
||||
"flake-parts",
|
||||
"minimal",
|
||||
"minimal-flake-parts",
|
||||
]
|
||||
clan_core_template_keys = list(clan_core_templates.keys())
|
||||
|
||||
expected_templates = ["default", "flake-parts", "minimal", "minimal-flake-parts"]
|
||||
assert clan_core_template_keys == expected_templates
|
||||
# clan.default
|
||||
default_template = templates.builtins.get("clan", {}).get("default")
|
||||
assert default_template is not None
|
||||
|
||||
default_template = get_template(
|
||||
TemplateName("default"),
|
||||
"clan",
|
||||
input_prio=None,
|
||||
clan_dir=clan_dir,
|
||||
)
|
||||
template_path = default_template.get("path", None)
|
||||
assert template_path is not None
|
||||
|
||||
new_clan = temporary_home / "new_clan"
|
||||
|
||||
copy_from_nixstore(
|
||||
Path(default_template.src["path"]),
|
||||
Path(template_path),
|
||||
new_clan,
|
||||
)
|
||||
assert (new_clan / "flake.nix").exists()
|
||||
assert (new_clan / "machines").is_dir()
|
||||
assert (new_clan / "machines" / "jon").is_dir()
|
||||
config_nix_p = new_clan / "machines" / "jon" / "configuration.nix"
|
||||
assert (config_nix_p).is_file()
|
||||
|
||||
# Test if we can write to the configuration.nix file
|
||||
with config_nix_p.open("r+") as f:
|
||||
flake_nix = new_clan / "flake.nix"
|
||||
assert (flake_nix).exists()
|
||||
assert (flake_nix).is_file()
|
||||
|
||||
assert (new_clan / "machines").is_dir()
|
||||
|
||||
# Test if we can write to the flake.nix file
|
||||
with flake_nix.open("r+") as f:
|
||||
data = f.read()
|
||||
f.write(data)
|
||||
|
||||
|
||||
# Test Case 1: Minimal input with empty templates
|
||||
@pytest.mark.with_core
|
||||
def test_clan_get_nix_attrset_case_1(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
temporary_home: Path,
|
||||
test_flake_with_core: FlakeForTest,
|
||||
) -> None:
|
||||
test_number = 1
|
||||
injected = {"templates": {"disko": {}, "machine": {}}}
|
||||
expected = {
|
||||
"inputs": {},
|
||||
"self": {"templates": {"disko": {}, "machine": {}, "clan": {}}},
|
||||
}
|
||||
nix_attr_tester(test_flake_with_core, injected, expected, test_number)
|
||||
|
||||
|
||||
# Test Case 2: Input with one template under 'clan'
|
||||
@pytest.mark.with_core
|
||||
def test_clan_get_nix_attrset_case_2(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
temporary_home: Path,
|
||||
test_flake_with_core: FlakeForTest,
|
||||
) -> None:
|
||||
test_number = 2
|
||||
injected = {
|
||||
"templates": {
|
||||
"clan": {
|
||||
"example_template": {
|
||||
"description": "An example clan template.",
|
||||
"path": "/example/path",
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
expected = {
|
||||
"inputs": {},
|
||||
"self": {
|
||||
"templates": {
|
||||
"clan": {
|
||||
"example_template": {
|
||||
"description": "An example clan template.",
|
||||
"path": "/example/path",
|
||||
}
|
||||
},
|
||||
"disko": {},
|
||||
"machine": {},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
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.with_core
|
||||
def test_clan_get_nix_attrset_case_3(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
temporary_home: Path,
|
||||
test_flake_with_core: FlakeForTest,
|
||||
) -> None:
|
||||
test_number = 3
|
||||
injected = {
|
||||
"templates": {
|
||||
"clan": {
|
||||
"clan_template": {
|
||||
"description": "A clan template.",
|
||||
"path": "/clan/path",
|
||||
}
|
||||
},
|
||||
"disko": {
|
||||
"disko_template": {
|
||||
"description": "A disko template.",
|
||||
"path": "/disko/path",
|
||||
}
|
||||
},
|
||||
"machine": {
|
||||
"machine_template": {
|
||||
"description": "A machine template.",
|
||||
"path": "/machine/path",
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
expected = {
|
||||
"inputs": {},
|
||||
"self": {
|
||||
"templates": {
|
||||
"clan": {
|
||||
"clan_template": {
|
||||
"description": "A clan template.",
|
||||
"path": "/clan/path",
|
||||
}
|
||||
},
|
||||
"disko": {
|
||||
"disko_template": {
|
||||
"description": "A disko template.",
|
||||
"path": "/disko/path",
|
||||
}
|
||||
},
|
||||
"machine": {
|
||||
"machine_template": {
|
||||
"description": "A machine template.",
|
||||
"path": "/machine/path",
|
||||
}
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
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.with_core
|
||||
def test_clan_get_nix_attrset_case_6(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
temporary_home: Path,
|
||||
test_flake_with_core: FlakeForTest,
|
||||
) -> None:
|
||||
test_number = 6
|
||||
injected = {}
|
||||
expected = {
|
||||
"inputs": {},
|
||||
"self": {"templates": {"disko": {}, "machine": {}, "clan": {}}},
|
||||
}
|
||||
nix_attr_tester(test_flake_with_core, injected, expected, test_number)
|
||||
|
||||
@@ -1,172 +1,12 @@
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Literal, NewType, TypedDict, cast
|
||||
|
||||
from clan_lib.dirs import clan_templates
|
||||
from clan_lib.errors import ClanCmdError, ClanError
|
||||
from clan_lib.flake import Flake
|
||||
from clan_lib.nix_models.clan import ClanTemplatesType
|
||||
from clan_lib.templates.filesystem import realize_nix_path
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
InputName = NewType("InputName", str)
|
||||
|
||||
|
||||
@dataclass
|
||||
class InputVariant:
|
||||
input_name: InputName | None
|
||||
|
||||
def is_self(self) -> bool:
|
||||
return self.input_name is None
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.input_name or "self"
|
||||
|
||||
|
||||
TemplateName = NewType("TemplateName", str)
|
||||
TemplateType = Literal["clan", "disko", "machine"]
|
||||
|
||||
|
||||
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, TemplatePath]
|
||||
clan: dict[TemplateName, TemplatePath]
|
||||
machine: dict[TemplateName, TemplatePath]
|
||||
|
||||
|
||||
class ClanAttrset(TypedDict):
|
||||
templates: TemplateTypeDict
|
||||
|
||||
|
||||
class ClanExports(TypedDict):
|
||||
inputs: dict[InputName, ClanAttrset]
|
||||
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", {}),
|
||||
}
|
||||
|
||||
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.
|
||||
"""
|
||||
if not clan_dir:
|
||||
clan_dir = Flake(str(clan_templates()))
|
||||
|
||||
log.debug(f"Evaluating flake {clan_dir} for Clan attrsets")
|
||||
|
||||
raw_clan_exports: dict[str, Any] = {"self": {"clan": {}}, "inputs": {"clan": {}}}
|
||||
|
||||
maybe_templates = clan_dir.select("?clan.?templates")
|
||||
if "clan" in maybe_templates:
|
||||
raw_clan_exports["self"] = maybe_templates["clan"]
|
||||
else:
|
||||
log.info("Current flake does not export the 'clan' attribute")
|
||||
|
||||
# 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:
|
||||
# FIXME: We expect here that if the input exports the clan attribute it also has clan.templates
|
||||
# this is not always the case if we just want to export clan.modules for example
|
||||
# However, there is no way to fix this, as clan.select does not support two optional selectors
|
||||
# and we cannot eval the clan attribute as clan.modules can be non JSON serializable because
|
||||
# of import statements.
|
||||
# This needs to be fixed in clan.select
|
||||
# For now always define clan.templates or no clan attribute at all
|
||||
temp = clan_dir.select("inputs.*.?clan.templates")
|
||||
|
||||
# FIXME: We need this because clan.select removes the templates attribute
|
||||
# but not the clan and other attributes leading up to templates
|
||||
for input_name, attrset in temp.items():
|
||||
if "clan" in attrset:
|
||||
raw_clan_exports["inputs"][input_name] = {
|
||||
"clan": {"templates": {**attrset["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
|
||||
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
|
||||
|
||||
@staticmethod
|
||||
def self_only() -> "InputPrio":
|
||||
# Only consider "self" (no external inputs)
|
||||
return InputPrio(prioritize_self=True, input_names=())
|
||||
|
||||
@staticmethod
|
||||
def try_inputs(input_names: tuple[str, ...]) -> "InputPrio":
|
||||
# Only consider the specified external inputs
|
||||
return InputPrio(prioritize_self=False, input_names=input_names)
|
||||
|
||||
@staticmethod
|
||||
def try_self_then_inputs(input_names: tuple[str, ...]) -> "InputPrio":
|
||||
# Consider "self" first, then the specified external inputs
|
||||
return InputPrio(prioritize_self=True, input_names=input_names)
|
||||
|
||||
|
||||
@dataclass
|
||||
class TemplateList:
|
||||
builtins: ClanTemplatesType
|
||||
@@ -181,76 +21,3 @@ def list_templates(flake: Flake) -> TemplateList:
|
||||
builtin_templates = flake.select("clanInternals.templates")
|
||||
|
||||
return TemplateList(builtin_templates, custom_templates)
|
||||
|
||||
|
||||
def get_template(
|
||||
template_name: TemplateName,
|
||||
template_type: TemplateType,
|
||||
*,
|
||||
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_dir = Flake(str(clan_templates()))
|
||||
|
||||
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
|
||||
if input_prio is None:
|
||||
input_prio = InputPrio.try_self_then_inputs(("clan-core",))
|
||||
|
||||
# Helper function to search for a specific template within a dictionary of templates
|
||||
def find_template(
|
||||
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: TemplatePath | None = None
|
||||
input_name: InputName | None = None
|
||||
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, 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"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,
|
||||
clan_exports["inputs"][input_name]["templates"][template_type],
|
||||
)
|
||||
if template:
|
||||
log.debug(f"Found template '{template_name}' in input '{input_name}'")
|
||||
break
|
||||
|
||||
# Step 3: Raise an error if the template wasn't found
|
||||
if not template:
|
||||
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
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user