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

api/modules: improve logic for builtin modules

Browse Source
This commit is contained in:
Johannes Kirschbauer
2025-08-31 12:50:36 +02:00
parent 5137d19b0f
commit c447aec9d3
4 changed files with 146 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
pkgs/clan-cli/clan_lib/fixtures/flakes/lib_clan/flake.nix
View File

@@ -1,11 +1,11 @@
{
inputs.clan-core.url = "https://git.clan.lol/clan/clan-core/archive/main.tar.gz";
inputs.nixpkgs.follows = "clan-core/nixpkgs";
inputs.Y2xhbi1jaW9yZS1uZXZlci1kZXBlbmQtb24tbWU.url = "https://git.clan.lol/clan/clan-core/archive/main.tar.gz";
inputs.nixpkgs.follows = "Y2xhbi1jaW9yZS1uZXZlci1kZXBlbmQtb24tbWU/nixpkgs";
outputs =
{ self, clan-core, ... }:
{ self, Y2xhbi1jaW9yZS1uZXZlci1kZXBlbmQtb24tbWU, ... }:
let
clan = clan-core.lib.clan ({
clan = Y2xhbi1jaW9yZS1uZXZlci1kZXBlbmQtb24tbWU.lib.clan ({
inherit self;
imports = [
./clan.nix

168
pkgs/clan-cli/clan_lib/services/modules.py
View File

@@ -60,7 +60,7 @@ class ModuleManifest:
raise ValueError(msg)
@classmethod
def from_dict(cls, data: dict) -> "ModuleManifest":
def from_dict(cls, data: dict[str, Any]) -> "ModuleManifest":
"""Create an instance of this class from a dictionary.
Drops any keys that are not defined in the dataclass.
"""
@@ -147,23 +147,51 @@ def extract_frontmatter[T](
@dataclass
class ModuleInfo(TypedDict):
class ModuleInfo:
manifest: ModuleManifest
roles: dict[str, None]
class Module(TypedDict):
module: InventoryInstanceModule
@dataclass
class Module:
# To use this module specify: InventoryInstanceModule :: { input, name } (foreign key)
usage_ref: InventoryInstanceModule
info: ModuleInfo
native: bool # Equivalent to input == None
@dataclass
class ClanModules:
modules: list[Module]
core_input_name: str
@API.register
def list_service_modules(flake: Flake) -> list[Module]:
def list_service_modules(flake: Flake) -> ClanModules:
"""Show information about a module"""
modules = flake.select("clanInternals.inventoryClass.modulesPerSource")
# inputName.moduleName -> ModuleInfo
modules: dict[str, dict[str, Any]] = flake.select(
"clanInternals.inventoryClass.modulesPerSource"
)
if "clan-core" not in modules:
msg = "Cannot find 'clan-core' input in the flake. Make sure your clan-core input is named 'clan-core'"
# moduleName -> ModuleInfo
builtin_modules: dict[str, Any] = flake.select(
"clanInternals.inventoryClass.staticModules"
)
first_name, first_module = next(iter(builtin_modules.items()))
clan_input_name = None
for input_name, module_set in modules.items():
if first_name in module_set:
# Compare the manifest name
module_set[first_name]["manifest"]["name"] = first_module["manifest"][
"name"
]
clan_input_name = input_name
break
if clan_input_name is None:
msg = "Could not determine the clan-core input name"
raise ClanError(msg)
res: list[Module] = []
@@ -171,86 +199,63 @@ def list_service_modules(flake: Flake) -> list[Module]:
for module_name, module_info in module_set.items():
res.append(
Module(
module={"name": module_name, "input": input_name},
usage_ref={
"name": module_name,
"input": None if input_name == clan_input_name else input_name,
},
info=ModuleInfo(
manifest=ModuleManifest.from_dict(
module_info.get("manifest"),
),
roles=module_info.get("roles", {}),
manifest=ModuleManifest.from_dict(module_info["manifest"]),
),
native=(input_name == clan_input_name),
)
)
return res
return ClanModules(res, clan_input_name)
@API.register
def get_service_module(
def resolve_service_module_ref(
flake: Flake,
module_ref: InventoryInstanceModuleType,
) -> ModuleInfo:
"""Returns the module information for a given module reference
:param module_ref: The module reference to get the information for
:return: Dict of module information
:raises ClanError: If the module_ref is invalid or missing required fields
"""
input_name, module_name = check_service_module_ref(flake, module_ref)
avilable_modules = list_service_modules(flake)
module_set: list[Module] = [
m for m in avilable_modules if m["module"].get("input", None) == input_name
]
if not module_set:
msg = f"Module set for input '{input_name}' not found"
raise ClanError(msg)
module = next((m for m in module_set if m["module"]["name"] == module_name), None)
if module is None:
msg = f"Module '{module_name}' not found in input '{input_name}'"
raise ClanError(msg)
return module["info"]
def check_service_module_ref(
flake: Flake,
module_ref: InventoryInstanceModuleType,
) -> tuple[str, str]:
) -> Module:
"""Checks if the module reference is valid
:param module_ref: The module reference to check
:raises ClanError: If the module_ref is invalid or missing required fields
"""
avilable_modules = list_service_modules(flake)
service_modules = list_service_modules(flake)
avilable_modules = service_modules.modules
input_ref = module_ref.get("input", None)
if input_ref is None:
msg = "Setting module_ref.input is currently required"
raise ClanError(msg)
if input_ref is None or input_ref == service_modules.core_input_name:
# Take only the native modules
module_set = [m for m in avilable_modules if m.native]
else:
# Match the input ref
module_set = [
m for m in avilable_modules if m["module"].get("input", None) == input_ref
m for m in avilable_modules if m.usage_ref.get("input", None) == input_ref
]
if module_set is None:
inputs = {m["module"].get("input") for m in avilable_modules}
if not module_set:
inputs = {m.usage_ref.get("input") for m in avilable_modules}
msg = f"module set for input '{input_ref}' not found"
msg += f"\nAvilable input_refs: {inputs}"
msg += "\nOmit the input field to use the built-in modules\n"
msg += "\n".join([m.usage_ref["name"] for m in avilable_modules if m.native])
raise ClanError(msg)
module_name = module_ref.get("name")
if not module_name:
msg = "Module name is required in module_ref"
raise ClanError(msg)
module = next((m for m in module_set if m["module"]["name"] == module_name), None)
module = next((m for m in module_set if m.usage_ref["name"] == module_name), None)
if module is None:
msg = f"module with name '{module_name}' not found"
raise ClanError(msg)
return (input_ref, module_name)
return module
@API.register
@@ -264,7 +269,16 @@ def get_service_module_schema(
:return: Dict of schemas for the service module roles
:raises ClanError: If the module_ref is invalid or missing required fields
"""
input_name, module_name = check_service_module_ref(flake, module_ref)
input_name, module_name = module_ref.get("input"), module_ref["name"]
module = resolve_service_module_ref(flake, module_ref)
if module is None:
msg = f"Module '{module_name}' not found in input '{input_name}'"
raise ClanError(msg)
if input_name is None:
msg = "Not implemented for: input_name is None"
raise ClanError(msg)
return flake.select(
f"clanInternals.inventoryClass.moduleSchemas.{input_name}.{module_name}",
@@ -278,7 +292,8 @@ def create_service_instance(
roles: InventoryInstanceRolesType,
) -> None:
"""Show information about a module"""
input_name, module_name = check_service_module_ref(flake, module_ref)
input_name, module_name = module_ref.get("input"), module_ref["name"]
module = resolve_service_module_ref(flake, module_ref)
inventory_store = InventoryStore(flake)
@@ -299,10 +314,10 @@ def create_service_instance(
all_machines = inventory.get("machines", {})
available_machine_refs = set(all_machines.keys())
schema = get_service_module_schema(flake, module_ref)
allowed_roles = module.info.roles
for role_name, role_members in roles.items():
if role_name not in schema:
msg = f"Role '{role_name}' is not defined in the module schema"
if role_name not in allowed_roles:
msg = f"Role '{role_name}' is not defined in the module"
raise ClanError(msg)
machine_refs = role_members.get("machines")
@@ -319,7 +334,15 @@ def create_service_instance(
# settings = role_members.get("settings", {})
# Create a new instance with the given roles
if not input_name:
new_instance: InventoryInstance = {
"module": {
"name": module_name,
},
"roles": roles,
}
else:
new_instance = {
"module": {
"name": module_name,
"input": input_name,
@@ -335,8 +358,10 @@ def create_service_instance(
)
class InventoryInstanceInfo(TypedDict):
module: Module
@dataclass
class InventoryInstanceInfo:
resolved: Module
module: InventoryInstanceModule
roles: InventoryInstanceRolesType
@@ -345,24 +370,19 @@ def list_service_instances(flake: Flake) -> dict[str, InventoryInstanceInfo]:
"""Returns all currently present service instances including their full configuration"""
inventory_store = InventoryStore(flake)
inventory = inventory_store.read()
service_modules = {
(mod["module"]["name"], mod["module"].get("input", "clan-core")): mod
for mod in list_service_modules(flake)
}
instances = inventory.get("instances", {})
res: dict[str, InventoryInstanceInfo] = {}
for instance_name, instance in instances.items():
module_key = (
instance.get("module", {})["name"],
instance.get("module", {}).get("input")
or "clan-core", # Replace None (or falsey) with "clan-core"
)
module = service_modules.get(module_key)
persisted_ref = instance.get("module", {"name": instance_name})
module = resolve_service_module_ref(flake, persisted_ref)
if module is None:
msg = f"Module '{module_key}' for instance '{instance_name}' not found"
msg = f"Module for instance '{instance_name}' not found"
raise ClanError(msg)
res[instance_name] = InventoryInstanceInfo(
module=module,
resolved=module,
module=persisted_ref,
roles=instance.get("roles", {}),
)
return res

50
pkgs/clan-cli/clan_lib/services/modules_test.py
View File

@@ -3,7 +3,7 @@ from collections.abc import Callable
import pytest
from clan_cli.tests.fixtures_flakes import nested_dict
from clan_lib.flake.flake import Flake
from clan_lib.services.modules import list_service_instances
from clan_lib.services.modules import list_service_instances, list_service_modules
@pytest.mark.with_core
@@ -11,25 +11,43 @@ def test_list_service_instances(
clan_flake: Callable[..., Flake],
) -> None:
config = nested_dict()
config["inventory"]["machines"]["alice"] = {}
config["inventory"]["machines"]["bob"] = {}
# implicit module selection (defaults to clan-core/admin)
config["inventory"]["instances"]["admin"]["roles"]["default"]["tags"]["all"] = {}
# explicit module selection
config["inventory"]["instances"]["my-sshd"]["module"]["input"] = "clan-core"
config["inventory"]["instances"]["my-sshd"]["module"]["name"] = "sshd"
# We use this random string in test to avoid code dependencies on the input name
config["inventory"]["instances"]["foo"]["module"]["input"] = (
"Y2xhbi1jaW9yZS1uZXZlci1kZXBlbmQtb24tbWU"
)
config["inventory"]["instances"]["foo"]["module"]["name"] = "sshd"
# input = null
config["inventory"]["instances"]["my-sshd-2"]["module"]["input"] = None
config["inventory"]["instances"]["my-sshd-2"]["module"]["name"] = "sshd"
config["inventory"]["instances"]["bar"]["module"]["input"] = None
config["inventory"]["instances"]["bar"]["module"]["name"] = "sshd"
# Omit input
config["inventory"]["instances"]["baz"]["module"]["name"] = "sshd"
# external input
flake = clan_flake(config)
service_modules = list_service_modules(flake)
assert len(service_modules.modules)
assert any(m.usage_ref["name"] == "sshd" for m in service_modules.modules)
instances = list_service_instances(flake)
assert list(instances.keys()) == ["admin", "my-sshd", "my-sshd-2"]
assert instances["admin"]["module"]["module"].get("input") == "clan-core"
assert instances["admin"]["module"]["module"].get("name") == "admin"
assert instances["my-sshd"]["module"]["module"].get("input") == "clan-core"
assert instances["my-sshd"]["module"]["module"].get("name") == "sshd"
assert instances["my-sshd-2"]["module"]["module"].get("input") == "clan-core"
assert instances["my-sshd-2"]["module"]["module"].get("name") == "sshd"
assert set(instances.keys()) == {"foo", "bar", "baz"}
# Reference to a built-in module
assert instances["foo"].resolved.usage_ref.get("input") is None
assert instances["foo"].resolved.usage_ref.get("name") == "sshd"
assert instances["foo"].resolved.info.manifest.name == "clan-core/sshd"
# Actual module
assert (
instances["foo"].module.get("input")
== "Y2xhbi1jaW9yZS1uZXZlci1kZXBlbmQtb24tbWU"
)
# Module exposes the input name?
assert instances["bar"].resolved.usage_ref.get("input") is None
assert instances["bar"].resolved.usage_ref.get("name") == "sshd"
assert instances["baz"].resolved.usage_ref.get("input") is None
assert instances["baz"].resolved.usage_ref.get("name") == "sshd"

8
pkgs/clan-cli/clan_lib/tests/test_create.py
View File

@@ -214,10 +214,12 @@ def test_clan_create_api(
store = InventoryStore(clan_dir_flake)
inventory = store.read()
modules = list_service_modules(clan_dir_flake)
service_modules = list_service_modules(clan_dir_flake)
admin_module = next(m for m in modules if m["module"]["name"] == "admin")
assert admin_module["info"]["manifest"].name == "clan-core/admin"
admin_module = next(
m for m in service_modules.modules if m.usage_ref.get("name") == "admin"
)
assert admin_module.info.manifest.name == "clan-core/admin"
set_value_by_path(inventory, "instances", inventory_conf.instances)
store.write(