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

Merge pull request 'clan_lib: persist, compute static data for simpler patch validation' (#5218) from update-service into main

Browse Source 
Reviewed-on: https://git.clan.lol/clan/clan-core/pulls/5218
This commit is contained in:
hsjobeki
2025-09-22 16:39:29 +00:00
parent 519ff4c0f3 9ea5156f32
commit 6b059ca71d
23 changed files with 1448 additions and 885 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
pkgs/clan-cli/clan_cli/machines/create.py
View File

@@ -11,7 +11,8 @@ from clan_lib.git import commit_file
from clan_lib.nix_models.clan import InventoryMachine
from clan_lib.nix_models.clan import InventoryMachineDeploy as MachineDeploy
from clan_lib.persist.inventory_store import InventoryStore
from clan_lib.persist.util import merge_objects, set_value_by_path
from clan_lib.persist.patch_engine import merge_objects
from clan_lib.persist.path_utils import set_value_by_path
from clan_lib.templates.handler import machine_template
from clan_cli.completions import add_dynamic_completer, complete_tags

3
pkgs/clan-cli/clan_lib/clan/create.py
View File

@@ -11,7 +11,8 @@ from clan_lib.flake import Flake
from clan_lib.nix import nix_command, nix_metadata, nix_shell
from clan_lib.nix_models.clan import InventoryMeta
from clan_lib.persist.inventory_store import InventoryStore
from clan_lib.persist.util import merge_objects, set_value_by_path
from clan_lib.persist.patch_engine import merge_objects
from clan_lib.persist.path_utils import set_value_by_path
from clan_lib.templates.handler import clan_template
from clan_lib.validator.hostname import hostname

5
pkgs/clan-cli/clan_lib/clan/get.py
View File

@@ -5,8 +5,9 @@ from clan_lib.errors import ClanError
from clan_lib.flake import Flake
from clan_lib.machines.actions import FieldSchema
from clan_lib.nix_models.clan import InventoryMeta
from clan_lib.persist.introspection import retrieve_typed_field_names
from clan_lib.persist.inventory_store import InventoryStore
from clan_lib.persist.util import is_writeable_key, retrieve_typed_field_names
from clan_lib.persist.write_rules import is_writeable_key
log = logging.getLogger(__name__)
@@ -50,7 +51,7 @@ def get_clan_details_schema(flake: Flake) -> dict[str, FieldSchema]:
"""
inventory_store = InventoryStore(flake)
write_info = inventory_store.get_writeability()
write_info = inventory_store.get_write_map()
field_names = retrieve_typed_field_names(InventoryMeta)

5
pkgs/clan-cli/clan_lib/clan/test_create.py
View File

@@ -124,10 +124,7 @@ def test_create_cannot_set_name(tmp_path: Path, offline_flake_hook: Any) -> None
with pytest.raises(ClanError) as exc_info:
create_clan(opts)
assert (
"Key 'meta.name' is not writeable. It seems its value is statically defined in nix."
in str(exc_info.value)
)
assert "Path 'meta.name' is readonly" in str(exc_info.value)
@pytest.mark.with_core

2
pkgs/clan-cli/clan_lib/clan/update.py
View File

@@ -4,7 +4,7 @@ from clan_lib.api import API
from clan_lib.flake import Flake
from clan_lib.nix_models.clan import InventoryMeta as Meta
from clan_lib.persist.inventory_store import InventorySnapshot, InventoryStore
from clan_lib.persist.util import set_value_by_path
from clan_lib.persist.path_utils import set_value_by_path
@dataclass

8
pkgs/clan-cli/clan_lib/machines/actions.py
View File

@@ -11,14 +11,14 @@ from clan_lib.nix_models.clan import (
InventoryMachine,
InventoryMachineTagsType,
)
from clan_lib.persist.introspection import retrieve_typed_field_names
from clan_lib.persist.inventory_store import InventoryStore
from clan_lib.persist.util import (
from clan_lib.persist.path_utils import (
get_value_by_path,
is_writeable_key,
list_difference,
retrieve_typed_field_names,
set_value_by_path,
)
from clan_lib.persist.write_rules import is_writeable_key
@dataclass
@@ -170,7 +170,7 @@ def get_machine_fields_schema(machine: Machine) -> dict[str, FieldSchema]:
"""
inventory_store = InventoryStore(machine.flake)
write_info = inventory_store.get_writeability()
write_info = inventory_store.get_write_map()
field_names = retrieve_typed_field_names(InventoryMachine)

14
pkgs/clan-cli/clan_lib/machines/actions_test.py
View File

@@ -16,7 +16,7 @@ from clan_lib.nix_models.clan import (
Unknown,
)
from clan_lib.persist.inventory_store import InventoryStore
from clan_lib.persist.util import get_value_by_path, set_value_by_path
from clan_lib.persist.path_utils import get_value_by_path, set_value_by_path
from .actions import (
MachineState,
@@ -164,10 +164,7 @@ def test_set_machine_fully_defined_in_nix(clan_flake: Callable[..., Flake]) -> N
with pytest.raises(ClanError) as exc_info:
set_machine(Machine("jon", flake), machine_jon)
assert (
"Key 'machines.jon.description' is not writeable. It seems its value is statically defined in nix."
in str(exc_info.value)
)
assert "Path 'machines.jon.description' is readonly" in str(exc_info.value)
# Assert _write should not be called
mock_write.assert_not_called()
@@ -213,8 +210,11 @@ def test_set_machine_manage_tags(clan_flake: Callable[..., Flake]) -> None:
with pytest.raises(ClanError) as exc_info:
set_jon(invalid_tags)
assert "Key 'machines.jon.tags' doesn't contain items ['nix1', 'nix2']" in str(
exc_info.value,
assert (
"Path 'machines.jon.tags' doesn't contain static items ['nix1', 'nix2']"
in str(
exc_info.value,
)
)

2
pkgs/clan-cli/clan_lib/machines/delete.py
View File

@@ -35,7 +35,7 @@ def delete_machine(machine: Machine) -> None:
inventory_store = InventoryStore(machine.flake)
try:
inventory_store.delete(
{f"machines.{machine.name}"},
{("machines", machine.name)},
)
except KeyError as exc:
# louis@(2025-03-09): test infrastructure does not seem to set the

2
pkgs/clan-cli/clan_lib/machines/install.py
View File

@@ -15,7 +15,7 @@ from clan_lib.cmd import Log, RunOpts, run
from clan_lib.machines.machines import Machine
from clan_lib.nix import nix_config, nix_shell
from clan_lib.persist.inventory_store import InventoryStore
from clan_lib.persist.util import set_value_by_path
from clan_lib.persist.path_utils import set_value_by_path
from clan_lib.ssh.create import create_secret_key_nixos_anywhere
from clan_lib.ssh.remote import Remote
from clan_lib.vars.generate import run_generators

36
pkgs/clan-cli/clan_lib/persist/introspection.py Normal file
View File

@@ -0,0 +1,36 @@
from typing import NotRequired, Required, get_args, get_origin, get_type_hints
def is_typeddict_class(obj: type) -> bool:
"""Safely checks if a class is a TypedDict."""
return (
isinstance(obj, type)
and hasattr(obj, "__annotations__")
and obj.__class__.__name__ == "_TypedDictMeta"
)
def retrieve_typed_field_names(obj: type, prefix: str = "") -> set[str]:
fields = set()
hints = get_type_hints(obj, include_extras=True)
for field, field_type in hints.items():
full_key = f"{prefix}.{field}" if prefix else field
origin = get_origin(field_type)
args = get_args(field_type)
# Unwrap Required/NotRequired
if origin in {NotRequired, Required}:
unwrapped_type = args[0]
origin = get_origin(unwrapped_type)
args = get_args(unwrapped_type)
else:
unwrapped_type = field_type
if is_typeddict_class(unwrapped_type):
fields |= retrieve_typed_field_names(unwrapped_type, prefix=full_key)
else:
fields.add(full_key)
return fields

34
pkgs/clan-cli/clan_lib/persist/inventory_store.py
View File

@@ -13,14 +13,14 @@ from clan_lib.nix_models.clan import (
InventoryMetaType,
InventoryTagsType,
)
from .util import (
calc_patches,
delete_by_path,
determine_writeability,
from clan_lib.persist.patch_engine import calc_patches
from clan_lib.persist.path_utils import (
PathTuple,
delete_by_path_tuple,
path_match,
set_value_by_path,
set_value_by_path_tuple,
)
from clan_lib.persist.write_rules import WriteMap, compute_write_map
def unwrap_known_unknown(value: Any) -> Any:
@@ -79,7 +79,7 @@ def sanitize(data: Any, whitelist_paths: list[str], current_path: list[str]) ->
@dataclass
class WriteInfo:
writeables: dict[str, set[str]]
writeables: WriteMap
data_eval: "InventorySnapshot"
data_disk: "InventorySnapshot"
@@ -194,7 +194,7 @@ class InventoryStore:
"""
return self._flake.select("clanInternals.inventoryClass.introspection")
def _write_info(self) -> WriteInfo:
def _write_map(self) -> WriteInfo:
"""Get the paths of the writeable keys in the inventory
Load the inventory and determine the writeable keys
@@ -205,20 +205,20 @@ class InventoryStore:
data_eval: InventorySnapshot = self._load_merged_inventory()
data_disk: InventorySnapshot = self._get_persisted()
writeables = determine_writeability(
write_map = compute_write_map(
current_priority,
dict(data_eval),
dict(data_disk),
)
return WriteInfo(writeables, data_eval, data_disk)
return WriteInfo(write_map, data_eval, data_disk)
def get_writeability(self) -> Any:
def get_write_map(self) -> Any:
"""Get the writeability of the inventory
:return: A dictionary with the writeability of all paths
"""
write_info = self._write_info()
write_info = self._write_map()
return write_info.writeables
def read(self) -> InventorySnapshot:
@@ -229,12 +229,12 @@ class InventoryStore:
"""
return self._load_merged_inventory()
def delete(self, delete_set: set[str], commit: bool = True) -> None:
def delete(self, delete_set: set[PathTuple], commit: bool = True) -> None:
"""Delete keys from the inventory"""
data_disk = dict(self._get_persisted())
for delete_path in delete_set:
delete_by_path(data_disk, delete_path)
delete_by_path_tuple(data_disk, delete_path)
with self.inventory_file.open("w") as f:
json.dump(data_disk, f, indent=2)
@@ -255,7 +255,7 @@ class InventoryStore:
"""Write the inventory to the flake directory
and commit it to git with the given message
"""
write_info = self._write_info()
write_info = self._write_map()
patchset, delete_set = calc_patches(
dict(write_info.data_disk),
dict(update),
@@ -265,10 +265,10 @@ class InventoryStore:
persisted = dict(write_info.data_disk)
for patch_path, data in patchset.items():
set_value_by_path(persisted, patch_path, data)
set_value_by_path_tuple(persisted, patch_path, data)
for delete_path in delete_set:
delete_by_path(persisted, delete_path)
delete_by_path_tuple(persisted, delete_path)
def post_write() -> None:
if commit:

17
pkgs/clan-cli/clan_lib/persist/inventory_store_test.py
View File

@@ -10,7 +10,7 @@ import pytest
from clan_lib.errors import ClanError
from clan_lib.nix import nix_eval
from clan_lib.persist.inventory_store import InventoryStore
from clan_lib.persist.util import delete_by_path, set_value_by_path
from clan_lib.persist.path_utils import delete_by_path, set_value_by_path
class MockFlake:
@@ -120,10 +120,7 @@ def test_simple_read_write(setup_test_files: Path) -> None:
invalid_data = {"protected": "foo"}
with pytest.raises(ClanError) as e:
store.write(invalid_data, "test", commit=False) # type: ignore[arg-type]
assert (
str(e.value)
== "Key 'protected' is not writeable. It seems its value is statically defined in nix."
)
assert "Path 'protected' is readonly" in str(e.value)
# Test the data is not touched
assert store.read() == data
@@ -255,10 +252,7 @@ def test_manipulate_list(setup_test_files: Path) -> None:
with pytest.raises(ClanError) as e:
store.write(data, "test", commit=False)
assert (
str(e.value)
== "Key 'empty' contains list duplicates: ['a'] - List values must be unique."
)
assert "Path 'empty' contains list duplicates: ['a']" in str(e.value)
assert store.read() == {"empty": [], "predefined": ["a", "b"]}
@@ -287,7 +281,4 @@ def test_static_list_items(setup_test_files: Path) -> None:
with pytest.raises(ClanError) as e:
store.write(data, "test", commit=False)
assert (
str(e.value)
== "Key 'predefined' doesn't contain items ['a'] - Deleting them is not possible, they are static values set via a .nix file"
)
assert "Path 'predefined' doesn't contain static items ['a']" in str(e.value)

233
pkgs/clan-cli/clan_lib/persist/patch_engine.py Normal file
View File

@@ -0,0 +1,233 @@
import json
from typing import Any, TypeVar, cast
from clan_lib.errors import ClanError
from clan_lib.persist.path_utils import (
PathTuple,
flatten_data_structured,
list_difference,
should_skip_path,
)
from clan_lib.persist.validate import (
validate_list_uniqueness,
validate_no_static_deletion,
validate_patch_conflicts,
validate_type_compatibility,
validate_writeability,
)
from clan_lib.persist.write_rules import WriteMap
def find_deleted_paths_structured(
all_values: dict[str, Any],
update: dict[str, Any],
parent_key: PathTuple = (),
) -> set[PathTuple]:
"""Find paths that are marked for deletion in the structured format."""
deleted_paths: set[PathTuple] = set()
for key, p_value in all_values.items():
current_path = (*parent_key, key) if parent_key else (key,)
if key not in update:
# Key doesn't exist at all -> entire branch deleted
deleted_paths.add(current_path)
else:
u_value = update[key]
# If persisted value is dict, check the update value
if isinstance(p_value, dict):
if isinstance(u_value, dict):
# If persisted dict is non-empty but updated dict is empty,
# that means everything under this branch is removed.
if p_value and not u_value:
# All children are removed
for child_key in p_value:
child_path = (*current_path, child_key)
deleted_paths.add(child_path)
else:
# Both are dicts, recurse deeper
deleted_paths |= find_deleted_paths_structured(
p_value,
u_value,
current_path,
)
else:
# deleted_paths.add(current_path)
# Persisted was a dict, update is not a dict
# This can happen if user sets the value to None, explicitly. # produce 'set path None' is not a deletion
pass
return deleted_paths
def calculate_static_data(
all_values: dict[str, Any], persisted: dict[str, Any]
) -> dict[PathTuple, Any]:
"""Calculate the static (read-only) data by finding what exists in all_values
but not in persisted data.
This gives us a clear view of what cannot be modified/deleted.
"""
all_flat = flatten_data_structured(all_values)
persisted_flat = flatten_data_structured(persisted)
static_flat = {}
for key, value in all_flat.items():
if key not in persisted_flat:
# This key exists only in static data
static_flat[key] = value
elif isinstance(value, list) and isinstance(persisted_flat[key], list):
# For lists, find items that are only in all_values (static items)
static_items = list_difference(value, persisted_flat[key])
if static_items:
static_flat[key] = static_items
return static_flat
def calc_patches(
persisted: dict[str, Any],
update: dict[str, Any],
all_values: dict[str, Any],
writeables: WriteMap,
) -> tuple[dict[PathTuple, Any], set[PathTuple]]:
"""Calculate the patches to apply to the inventory using structured paths.
Given its current state and the update to apply.
Calulates the necessary SET patches and DELETE paths.
While validating writeability rules.
Args:
persisted: The current mutable state of the inventory
update: The update to apply
all_values: All values in the inventory (static + mutable merged)
writeables: The writeable keys. Use 'determine_writeability'.
Example: {'writeable': {'foo', 'foo.bar'}, 'non_writeable': {'foo.nix'}}
Returns:
Tuple of (SET patches dict, DELETE paths set)
Raises:
ClanError: When validation fails or invalid operations are attempted
"""
# Calculate static data using structured paths
static_data = calculate_static_data(all_values, persisted)
# Flatten all data structures using structured paths
# persisted_flat = flatten_data_structured(persisted)
update_flat = flatten_data_structured(update)
all_values_flat = flatten_data_structured(all_values)
# Early validation: ensure we're not trying to modify static-only paths
# validate_no_static_modification(update_flat, static_data)
# Find paths marked for deletion
delete_paths = find_deleted_paths_structured(all_values, update)
# Validate deletions don't affect static data
# TODO: We currently cannot validate this properly.
# for delete_path in delete_paths:
# for static_path in static_data:
# if path_starts_with(static_path, delete_path):
# msg = f"Cannot delete path '{path_to_string(delete_path)}' - Readonly path '{path_to_string(static_path)}' is set via .nix file"
# raise ClanError(msg)
# Get all paths that might need processing
all_paths: set[PathTuple] = set(all_values_flat) | set(update_flat)
# Calculate patches
patches: dict[PathTuple, Any] = {}
for path in all_paths:
old_value = all_values_flat.get(path)
new_value = update_flat.get(path)
has_value = path in update_flat
# Skip if no change
if old_value == new_value:
continue
# Skip if path is marked for deletion or under a deletion path
if should_skip_path(path, delete_paths):
continue
# Skip deletions (they're handled by delete_paths)
if old_value is not None and not has_value:
continue
# Validate the change is allowed
validate_no_static_deletion(path, new_value, static_data)
validate_writeability(path, writeables)
validate_type_compatibility(path, old_value, new_value)
validate_list_uniqueness(path, new_value)
patch_value = new_value # Init
if isinstance(new_value, list):
patch_value = list_difference(new_value, static_data.get(path, []))
patches[path] = patch_value
validate_patch_conflicts(set(patches.keys()), delete_paths)
return patches, delete_paths
empty: list[str] = []
T = TypeVar("T")
def merge_objects(
curr: T,
update: T,
merge_lists: bool = True,
path: list[str] = empty,
) -> T:
"""Updates values in curr by values of update
The output contains values for all keys of curr and update together.
Lists are deduplicated and appended almost like in the nix module system.
Example:
merge_objects({"a": 1}, {"a": null }) -> {"a": null}
merge_objects({"a": null}, {"a": 1 }) -> {"a": 1}
"""
result = {}
msg = f"cannot update non-dictionary values: {curr} by {update}"
if not isinstance(update, dict):
raise ClanError(msg)
if not isinstance(curr, dict):
raise ClanError(msg)
all_keys = set(update.keys()).union(curr.keys())
for key in all_keys:
curr_val = curr.get(key)
update_val = update.get(key)
if isinstance(update_val, dict) and isinstance(curr_val, dict):
result[key] = merge_objects(
curr_val,
update_val,
merge_lists=merge_lists,
path=[*path, key],
)
elif isinstance(update_val, list) and isinstance(curr_val, list):
if merge_lists:
result[key] = list(dict.fromkeys(curr_val + update_val)) # type: ignore[assignment]
else:
result[key] = update_val # type: ignore[assignment]
elif (
update_val is not None
and curr_val is not None
and type(update_val) is not type(curr_val)
):
msg = f"Type mismatch for key '{key}'. Cannot update {type(curr_val)} with {type(update_val)}"
raise ClanError(msg, location=json.dumps([*path, key]))
elif key in update:
result[key] = update_val # type: ignore[assignment]
elif key in curr:
result[key] = curr_val # type: ignore[assignment]
return cast("T", result)

593
pkgs/clan-cli/clan_lib/persist/util_test.py → pkgs/clan-cli/clan_lib/persist/patch_engine_test.py
View File

@@ -1,229 +1,209 @@
# Functions to test
from copy import deepcopy
from typing import Any
import pytest
from clan_lib.errors import ClanError
from clan_lib.persist.util import (
from clan_lib.persist.patch_engine import (
calc_patches,
delete_by_path,
determine_writeability,
list_difference,
calculate_static_data,
merge_objects,
path_match,
)
from clan_lib.persist.path_utils import (
delete_by_path,
set_value_by_path,
set_value_by_path_tuple,
)
from clan_lib.persist.write_rules import compute_write_map
# --- calculate_static_data ---
@pytest.mark.parametrize(
("path", "whitelist", "expected"),
[
# Exact matches
(["a", "b", "c"], [["a", "b", "c"]], True),
(["a", "b"], [["a", "b"]], True),
([], [[]], True),
# Wildcard matches
(["a", "b", "c"], [["a", "*", "c"]], True),
(["a", "x", "c"], [["a", "*", "c"]], True),
(["a", "b", "c"], [["*", "b", "c"]], True),
(["a", "b", "c"], [["a", "b", "*"]], True),
(["a", "b", "c"], [["*", "*", "*"]], True),
# Multiple patterns - one matches
(["a", "b", "c"], [["x", "y", "z"], ["a", "*", "c"]], True),
(["x", "y", "z"], [["a", "*", "c"], ["x", "y", "z"]], True),
# Length mismatch
(["a", "b", "c"], [["a", "b"]], False),
(["a", "b"], [["a", "b", "c"]], False),
# Non-matching
(["a", "b", "c"], [["a", "b", "x"]], False),
(["a", "b", "c"], [["a", "x", "x"]], False),
(["a", "b", "c"], [["x", "x", "x"]], False),
# Empty whitelist
(["a"], [], False),
# Wildcards and exact mixed
(
["instances", "inst1", "roles", "roleA", "settings"],
[["instances", "*", "roles", "*", "settings"]],
True,
),
# Partial wildcard - length mismatch should fail
(
["instances", "inst1", "roles", "roleA"],
[["instances", "*", "roles", "*", "settings"]],
False,
),
# Empty path, no patterns
([], [], False),
],
)
def test_path_match(
path: list[str],
whitelist: list[list[str]],
expected: bool,
) -> None:
assert path_match(path, whitelist) == expected
# --------- Patching tests ---------
def test_patch_nested() -> None:
orig = {"a": 1, "b": {"a": 2.1, "b": 2.2}, "c": 3}
set_value_by_path(orig, "b.b", "foo")
# Should only update the nested value
assert orig == {"a": 1, "b": {"a": 2.1, "b": "foo"}, "c": 3}
def test_patch_nested_dict() -> None:
orig = {"a": 1, "b": {"a": 2.1, "b": 2.2}, "c": 3}
# This should update the whole "b" dict
# Which also removes all other keys
set_value_by_path(orig, "b", {"b": "foo"})
# Should only update the nested value
assert orig == {"a": 1, "b": {"b": "foo"}, "c": 3}
def test_create_missing_paths() -> None:
orig = {"a": 1}
set_value_by_path(orig, "b.c", "foo")
# Should only update the nested value
assert orig == {"a": 1, "b": {"c": "foo"}}
orig = {}
set_value_by_path(orig, "a.b.c", "foo")
assert orig == {"a": {"b": {"c": "foo"}}}
# --------- Write tests ---------
#
def test_write_simple() -> None:
prios = {
"foo": {
"__prio": 100, # <- writeable: "foo"
"bar": {"__prio": 1000}, # <- writeable: mkDefault "foo.bar"
def test_calculate_static_data_basic() -> None:
all_values = {
"name": "example",
"version": 1,
"settings": {
"optionA": True,
"optionB": False,
"listSetting": [1, 2, 3, 4],
},
"staticOnly": "staticValue",
}
persisted = {
"name": "example",
"version": 1,
"settings": {
"optionA": True,
"listSetting": [2, 3],
},
}
default: dict = {"foo": {}}
data: dict = {}
res = determine_writeability(prios, default, data)
expected_static = {
("settings", "optionB"): False,
("settings", "listSetting"): [1, 4],
("staticOnly",): "staticValue",
}
assert res == {"writeable": {"foo", "foo.bar"}, "non_writeable": set({})}
static_data = calculate_static_data(all_values, persisted)
assert static_data == expected_static
def test_write_inherited() -> None:
prios = {
"foo": {
"__prio": 100, # <- writeable: "foo"
"bar": {
# Inherits prio from parent <- writeable: "foo.bar"
"baz": {"__prio": 1000}, # <- writeable: "foo.bar.baz"
def test_calculate_static_data_no_static() -> None:
all_values = {
"name": "example",
"version": 1,
"settings": {
"optionA": True,
"listSetting": [1, 2, 3],
},
}
persisted = {
"name": "example",
"version": 1,
"settings": {
"optionA": True,
"listSetting": [1, 2, 3],
},
}
expected_static: dict = {}
static_data = calculate_static_data(all_values, persisted)
assert static_data == expected_static
# See: https://git.clan.lol/clan/clan-core/issues/5231
# Uncomment this test if the issue is resolved
# def test_calculate_static_data_no_static_sets() -> None:
# all_values = {
# "instance": {
# "hello": {
# "roles": {
# "default": {
# "machines": {
# "jon": {
# # Default
# "settings": {
# }
# }
# }
# }
# }
# }
# }
# }
# persisted = {
# "instance": {
# "hello": {
# "roles": {
# "default": {
# "machines": {
# "jon": {
# }
# }
# }
# }
# }
# }
# }
# expected_static: dict = {}
# static_data = calculate_static_data(all_values, persisted)
# assert static_data == expected_static
def test_calculate_static_data_all_static() -> None:
all_values = {
"name": "example",
"version": 1,
"settings": {
"optionA": True,
"listSetting": [1, 2, 3],
},
"staticOnly": "staticValue",
}
persisted: dict = {}
expected_static = {
("name",): "example",
("version",): 1,
("settings", "optionA"): True,
("settings", "listSetting"): [1, 2, 3],
("staticOnly",): "staticValue",
}
static_data = calculate_static_data(all_values, persisted)
assert static_data == expected_static
def test_calculate_static_data_empty_all_values() -> None:
# This should never happen in practice, but we test it for completeness.
# Maybe this should emit a warning in the future?
all_values: dict = {}
persisted = {
"name": "example",
"version": 1,
}
expected_static: dict = {}
static_data = calculate_static_data(all_values, persisted)
assert static_data == expected_static
def test_calculate_nested_dicts() -> None:
all_values = {
"level1": {
"level2": {
"staticKey": "staticValue",
"persistedKey": "persistedValue",
},
"anotherStatic": 42,
},
"topLevelStatic": True,
}
persisted = {
"level1": {
"level2": {
"persistedKey": "persistedValue",
},
},
}
data: dict = {}
res = determine_writeability(prios, {"foo": {"bar": {}}}, data)
assert res == {
"writeable": {"foo", "foo.bar", "foo.bar.baz"},
"non_writeable": set(),
expected_static = {
("level1", "level2", "staticKey"): "staticValue",
("level1", "anotherStatic"): 42,
("topLevelStatic",): True,
}
static_data = calculate_static_data(all_values, persisted)
assert static_data == expected_static
def test_non_write_inherited() -> None:
prios = {
"foo": {
"__prio": 50, # <- non writeable: mkForce "foo" = {...}
"bar": {
# Inherits prio from parent <- non writeable
"baz": {"__prio": 1000}, # <- non writeable: mkDefault "foo.bar.baz"
},
def test_dot_in_keys() -> None:
all_values = {
"key.foo": "staticValue",
"key": {
"foo": "anotherStaticValue",
},
}
persisted: dict = {}
data: dict = {}
res = determine_writeability(prios, {}, data)
assert res == {
"writeable": set(),
"non_writeable": {"foo", "foo.bar", "foo.bar.baz"},
expected_static = {
("key.foo",): "staticValue",
("key", "foo"): "anotherStaticValue",
}
static_data = calculate_static_data(all_values, persisted)
def test_write_list() -> None:
prios = {
"foo": {
"__prio": 100,
},
}
data: dict = {}
default: dict = {
"foo": [
"a",
"b",
], # <- writeable: because lists are merged. Filtering out nix-values comes later
}
res = determine_writeability(prios, default, data)
assert res == {
"writeable": {"foo"},
"non_writeable": set(),
}
assert static_data == expected_static
def test_write_because_written() -> None:
prios = {
"foo": {
"__prio": 100, # <- writeable: "foo"
"bar": {
# Inherits prio from parent <- writeable
"baz": {"__prio": 100}, # <- non writeable usually
"foobar": {"__prio": 100}, # <- non writeable
},
},
}
# Given the following data. {}
# Check that the non-writeable paths are correct.
res = determine_writeability(prios, {"foo": {"bar": {}}}, {})
assert res == {
"writeable": {"foo", "foo.bar"},
"non_writeable": {"foo.bar.baz", "foo.bar.foobar"},
}
data: dict = {
"foo": {
"bar": {
"baz": "foo", # <- written. Since we created the data, we know we can write to it
},
},
}
res = determine_writeability(prios, {}, data)
assert res == {
"writeable": {"foo", "foo.bar", "foo.bar.baz"},
"non_writeable": {"foo.bar.foobar"},
}
# --------- List unmerge tests ---------
def test_list_unmerge() -> None:
all_machines = ["machineA", "machineB"]
inventory = ["machineB"]
nix_machines = list_difference(all_machines, inventory)
assert nix_machines == ["machineA"]
# --------- Write tests ---------
# --------- calc_patches ---------
def test_update_simple() -> None:
@@ -239,25 +219,28 @@ def test_update_simple() -> None:
data_disk: dict = {}
writeables = determine_writeability(prios, data_eval, data_disk)
assert writeables == {"writeable": {"foo", "foo.bar"}, "non_writeable": {"foo.nix"}}
writeables = compute_write_map(prios, data_eval, data_disk)
assert writeables == {
"writeable": {("foo",), ("foo", "bar")},
"non_writeable": {("foo", "nix")},
}
update = {
"foo": {
"bar": "new value", # <- user sets this value
"nix": "this is set in nix", # <- user didnt touch this value
# If the user would have set this value, it would trigger an error
"nix": "this is set in nix", # <- user didnt touch this value
},
}
patchset, _ = calc_patches(
patchset, delete_set = calc_patches(
data_disk,
update,
all_values=data_eval,
writeables=writeables,
)
assert patchset == {"foo.bar": "new value"}
assert patchset == {("foo", "bar"): "new value"}
assert delete_set == set()
def test_update_add_empty_dict() -> None:
@@ -272,20 +255,21 @@ def test_update_add_empty_dict() -> None:
data_disk: dict = {}
writeables = determine_writeability(prios, data_eval, data_disk)
writeables = compute_write_map(prios, data_eval, data_disk)
update = deepcopy(data_eval)
set_value_by_path(update, "foo.mimi", {})
set_value_by_path_tuple(update, ("foo", "mimi"), {})
patchset, _ = calc_patches(
patchset, delete_set = calc_patches(
data_disk,
update,
all_values=data_eval,
writeables=writeables,
)
assert patchset == {"foo.mimi": {}} # this is what gets persisted
assert patchset == {("foo", "mimi"): {}} # this is what gets persisted
assert delete_set == set()
def test_update_many() -> None:
@@ -312,11 +296,16 @@ def test_update_many() -> None:
data_disk = {"foo": {"bar": "baz", "nested": {"x": "x"}}}
writeables = determine_writeability(prios, data_eval, data_disk)
writeables = compute_write_map(prios, data_eval, data_disk)
assert writeables == {
"writeable": {"foo.nested", "foo", "foo.bar", "foo.nested.x"},
"non_writeable": {"foo.nix", "foo.nested.y"},
"writeable": {
("foo",),
("foo", "bar"),
("foo", "nested"),
("foo", "nested", "x"),
},
"non_writeable": {("foo", "nix"), ("foo", "nested", "y")},
}
update = {
@@ -329,7 +318,7 @@ def test_update_many() -> None:
},
},
}
patchset, _ = calc_patches(
patchset, delete_set = calc_patches(
data_disk,
update,
all_values=data_eval,
@@ -337,9 +326,10 @@ def test_update_many() -> None:
)
assert patchset == {
"foo.bar": "new value for bar",
"foo.nested.x": "new value for x",
("foo", "bar"): "new value for bar",
("foo", "nested", "x"): "new value for x",
}
assert delete_set == set()
def test_update_parent_non_writeable() -> None:
@@ -362,9 +352,12 @@ def test_update_parent_non_writeable() -> None:
},
}
writeables = determine_writeability(prios, data_eval, data_disk)
writeables = compute_write_map(prios, data_eval, data_disk)
assert writeables == {"writeable": set(), "non_writeable": {"foo", "foo.bar"}}
assert writeables == {
"writeable": set(),
"non_writeable": {("foo",), ("foo", "bar")},
}
update = {
"foo": {
@@ -374,7 +367,34 @@ def test_update_parent_non_writeable() -> None:
with pytest.raises(ClanError) as error:
calc_patches(data_disk, update, all_values=data_eval, writeables=writeables)
assert "Key 'foo.bar' is not writeable." in str(error.value)
assert "Path 'foo.bar' is readonly." in str(error.value)
# TODO: Resolve the issue https://git.clan.lol/clan/clan-core/issues/5231
# def test_remove_non_writable_attrs() -> None:
# prios = {
# "foo": {
# "__prio": 100, # <- writeable: "foo"
# },
# }
# data_eval: dict = {"foo": {"bar": {}, "baz": {}}}
# data_disk: dict = {}
# writeables = compute_write_map(prios, data_eval, data_disk)
# update: dict = {
# "foo": {
# "bar": {}, # <- user leaves this value
# # User removed "baz"
# },
# }
# with pytest.raises(ClanError) as error:
# calc_patches(data_disk, update, all_values=data_eval, writeables=writeables)
# assert "Cannot delete path 'foo.baz'" in str(error.value)
def test_update_list() -> None:
@@ -391,9 +411,9 @@ def test_update_list() -> None:
data_disk = {"foo": ["B"]}
writeables = determine_writeability(prios, data_eval, data_disk)
writeables = compute_write_map(prios, data_eval, data_disk)
assert writeables == {"writeable": {"foo"}, "non_writeable": set()}
assert writeables == {"writeable": {("foo",)}, "non_writeable": set()}
# Add "C" to the list
update = {"foo": ["A", "B", "C"]} # User wants to add "C"
@@ -405,7 +425,7 @@ def test_update_list() -> None:
writeables=writeables,
)
assert patchset == {"foo": ["B", "C"]}
assert patchset == {("foo",): ["B", "C"]}
# "foo": ["A", "B"]
# Remove "B" from the list
@@ -419,7 +439,7 @@ def test_update_list() -> None:
writeables=writeables,
)
assert patchset == {"foo": []}
assert patchset == {("foo",): []}
def test_update_list_duplicates() -> None:
@@ -436,9 +456,9 @@ def test_update_list_duplicates() -> None:
data_disk = {"foo": ["B"]}
writeables = determine_writeability(prios, data_eval, data_disk)
writeables = compute_write_map(prios, data_eval, data_disk)
assert writeables == {"writeable": {"foo"}, "non_writeable": set()}
assert writeables == {"writeable": {("foo",)}, "non_writeable": set()}
# Add "A" to the list
update = {"foo": ["A", "B", "A"]} # User wants to add duplicate "A"
@@ -446,7 +466,7 @@ def test_update_list_duplicates() -> None:
with pytest.raises(ClanError) as error:
calc_patches(data_disk, update, all_values=data_eval, writeables=writeables)
assert "Key 'foo' contains list duplicates: ['A']" in str(error.value)
assert "Path 'foo' contains list duplicates: ['A']" in str(error.value)
def test_dont_persist_defaults() -> None:
@@ -460,8 +480,11 @@ def test_dont_persist_defaults() -> None:
"config": {"foo": "bar"},
}
data_disk: dict[str, Any] = {}
writeables = determine_writeability(prios, data_eval, data_disk)
assert writeables == {"writeable": {"config", "enabled"}, "non_writeable": set()}
writeables = compute_write_map(prios, data_eval, data_disk)
assert writeables == {
"writeable": {("config",), ("enabled",)},
"non_writeable": set(),
}
update = deepcopy(data_eval)
set_value_by_path(update, "config.foo", "foo")
@@ -472,10 +495,35 @@ def test_dont_persist_defaults() -> None:
all_values=data_eval,
writeables=writeables,
)
assert patchset == {"config.foo": "foo"}
assert patchset == {("config", "foo"): "foo"}
assert delete_set == set()
def test_set_null() -> None:
data_eval: dict = {
"foo": {},
"bar": {},
}
data_disk = data_eval
# User set Foo to null
# User deleted bar
update = {"foo": None}
patchset, delete_set = calc_patches(
data_disk,
update,
all_values=data_eval,
writeables=compute_write_map(
{"__prio": 100, "foo": {"__prio": 100}},
data_eval,
data_disk,
),
)
assert patchset == {("foo",): None}
assert delete_set == {("bar",)}
def test_machine_delete() -> None:
prios = {
"machines": {"__prio": 100},
@@ -489,8 +537,8 @@ def test_machine_delete() -> None:
}
data_disk = data_eval
writeables = determine_writeability(prios, data_eval, data_disk)
assert writeables == {"writeable": {"machines"}, "non_writeable": set()}
writeables = compute_write_map(prios, data_eval, data_disk)
assert writeables == {"writeable": {("machines",)}, "non_writeable": set()}
# Delete machine "bar" from the inventory
update = deepcopy(data_eval)
@@ -504,7 +552,7 @@ def test_machine_delete() -> None:
)
assert patchset == {}
assert delete_set == {"machines.bar"}
assert delete_set == {("machines", "bar")}
def test_update_mismatching_update_type() -> None:
@@ -518,9 +566,9 @@ def test_update_mismatching_update_type() -> None:
data_disk: dict = {}
writeables = determine_writeability(prios, data_eval, data_disk)
writeables = compute_write_map(prios, data_eval, data_disk)
assert writeables == {"writeable": {"foo"}, "non_writeable": set()}
assert writeables == {"writeable": {("foo",)}, "non_writeable": set()}
# set foo to an int but it is a list
update: dict = {"foo": 1}
@@ -529,7 +577,7 @@ def test_update_mismatching_update_type() -> None:
calc_patches(data_disk, update, all_values=data_eval, writeables=writeables)
assert (
"Type mismatch for key 'foo'. Cannot update <class 'list'> with <class 'int'>"
"Type mismatch for path 'foo'. Cannot update <class 'list'> with <class 'int'>"
in str(error.value)
)
@@ -545,9 +593,9 @@ def test_delete_key() -> None:
data_disk = data_eval
writeables = determine_writeability(prios, data_eval, data_disk)
writeables = compute_write_map(prios, data_eval, data_disk)
assert writeables == {"writeable": {"foo"}, "non_writeable": set()}
assert writeables == {"writeable": {("foo",)}, "non_writeable": set()}
# remove all keys from foo
update: dict = {"foo": {}}
@@ -559,8 +607,8 @@ def test_delete_key() -> None:
writeables=writeables,
)
assert patchset == {"foo": {}}
assert delete_set == {"foo.bar"}
assert patchset == {("foo",): {}}
assert delete_set == {("foo", "bar")}
def test_delete_key_intermediate() -> None:
@@ -584,9 +632,9 @@ def test_delete_key_intermediate() -> None:
data_disk = data_eval
writeables = determine_writeability(prios, data_eval, data_disk)
writeables = compute_write_map(prios, data_eval, data_disk)
assert writeables == {"writeable": {"foo"}, "non_writeable": set()}
assert writeables == {"writeable": {("foo",)}, "non_writeable": set()}
# remove all keys from foo
@@ -598,7 +646,7 @@ def test_delete_key_intermediate() -> None:
)
assert patchset == {}
assert delete_set == {"foo.bar"}
assert delete_set == {("foo", "bar")}
def test_delete_key_non_writeable() -> None:
@@ -618,85 +666,18 @@ def test_delete_key_non_writeable() -> None:
data_disk = data_eval
writeables = determine_writeability(prios, data_eval, data_disk)
writeables = compute_write_map(prios, data_eval, data_disk)
assert writeables == {"writeable": set(), "non_writeable": {"foo"}}
assert writeables == {"writeable": set(), "non_writeable": {("foo",)}}
# remove all keys from foo
with pytest.raises(ClanError) as error:
calc_patches(data_disk, update, all_values=data_eval, writeables=writeables)
assert "is not writeable" in str(error.value)
assert "Path 'foo' is readonly." in str(error.value)
def test_delete_atom() -> None:
data = {"foo": {"bar": 1}}
# Removes the key "foo.bar"
# Returns the deleted key-value pair { "bar": 1 }
entry = delete_by_path(data, "foo.bar")
assert entry == {"bar": 1}
assert data == {"foo": {}}
def test_delete_intermediate() -> None:
data = {"a": {"b": {"c": {"d": 42}}}}
# Removes "a.b.c.d"
entry = delete_by_path(data, "a.b.c")
assert entry == {"c": {"d": 42}}
# Check all intermediate dictionaries remain intact
assert data == {"a": {"b": {}}}
def test_delete_top_level() -> None:
data = {"x": 100, "y": 200}
# Deletes top-level key
entry = delete_by_path(data, "x")
assert entry == {"x": 100}
assert data == {"y": 200}
def test_delete_key_not_found() -> None:
data = {"foo": {"bar": 1}}
# Trying to delete a non-existing key "foo.baz" - should return empty dict
result = delete_by_path(data, "foo.baz")
assert result == {}
# Data should remain unchanged
assert data == {"foo": {"bar": 1}}
def test_delete_intermediate_not_dict() -> None:
data = {"foo": "not a dict"}
# Trying to go deeper into a non-dict value
with pytest.raises(KeyError) as excinfo:
delete_by_path(data, "foo.bar")
assert "not found or not a dictionary" in str(excinfo.value)
# Data should remain unchanged
assert data == {"foo": "not a dict"}
def test_delete_empty_path() -> None:
data = {"foo": {"bar": 1}}
# Attempting to delete with an empty path
with pytest.raises(KeyError) as excinfo:
delete_by_path(data, "")
# Depending on how you handle empty paths, you might raise an error or handle it differently.
# If you do raise an error, check the message.
assert "Cannot delete. Path is empty" in str(excinfo.value)
assert data == {"foo": {"bar": 1}}
def test_delete_non_existent_path_deep() -> None:
data = {"foo": {"bar": {"baz": 123}}}
# non-existent deep path - should return empty dict
result = delete_by_path(data, "foo.bar.qux")
assert result == {}
# Data remains unchanged
assert data == {"foo": {"bar": {"baz": 123}}}
### Merge Objects Tests ###
# --- test merge_objects ---
def test_merge_objects_empty() -> None:

215
pkgs/clan-cli/clan_lib/persist/path_utils.py Normal file
View File

@@ -0,0 +1,215 @@
from collections import Counter
from typing import Any, TypeVar, cast
PathTuple = tuple[str, ...]
def list_difference(all_items: list, filter_items: list) -> list:
"""Applys a filter to a list and returns the items in all_items that are not in filter_items
Example:
all_items = [1, 2, 3, 4]
filter_items = [3, 4]
list_difference(all_items, filter_items) == [1, 2]
"""
return [value for value in all_items if value not in filter_items]
def find_duplicates(string_list: list[str]) -> list[str]:
count = Counter(string_list)
return [item for item, freq in count.items() if freq > 1]
def path_to_string(path: PathTuple) -> str:
"""Convert tuple path to string for display/error messages."""
return ".".join(str(p) for p in path)
def path_starts_with(path: PathTuple, prefix: PathTuple) -> bool:
"""Check if path starts with prefix tuple."""
return len(path) >= len(prefix) and path[: len(prefix)] == prefix
type DictLike = dict[str, Any] | Any
def set_value_by_path(d: DictLike, path: str, content: Any) -> None:
"""Update the value at a specific dot-separated path in a nested dictionary.
If the value didn't exist before, it will be created recursively.
:param d: The dictionary to update.
:param path: The dot-separated path to the key (e.g., 'foo.bar').
:param content: The new value to set.
"""
keys = path.split(".")
current = d
for key in keys[:-1]:
current = current.setdefault(key, {})
current[keys[-1]] = content
def set_value_by_path_tuple(d: DictLike, path: PathTuple, content: Any) -> None:
"""Update the value at a specific path in a nested dictionary.
If the value didn't exist before, it will be created recursively.
:param d: The dictionary to update.
:param path: A tuple of strings representing the path to the value.
:param content: The new value to set.
"""
keys = path
current = d
for key in keys[:-1]:
current = current.setdefault(key, {})
current[keys[-1]] = content
def delete_by_path_tuple(d: dict[str, Any], path: PathTuple) -> Any:
"""Deletes the nested entry specified by a dot-separated path from the dictionary using pop().
:param data: The dictionary to modify.
:param path: A dot-separated string indicating the nested key to delete.
e.g., "foo.bar.baz" will attempt to delete data["foo"]["bar"]["baz"].
:raises KeyError: If any intermediate key is missing or not a dictionary,
or if the final key to delete is not found.
"""
if not path:
msg = "Cannot delete. Path is empty."
raise KeyError(msg)
keys = path
current = d
# Navigate to the parent dictionary of the final key
for key in keys[:-1]:
if key not in current or not isinstance(current[key], dict):
msg = f"Cannot delete. Key '{path_to_string(path)}' not found or not a dictionary '{d}'"
raise KeyError(msg)
current = current[key]
# Attempt to pop the final key
last_key = keys[-1]
try:
value = current.pop(last_key)
except KeyError:
# TODO(@hsjobeki): It should be save to raise an error here.
# Possibly data was already deleted
# msg = f"Canot delete. Path '{path}' not found in data '{d}'"
# raise KeyError(msg) from exc
return {}
else:
return {last_key: value}
def delete_by_path(d: dict[str, Any], path: str) -> Any:
"""Deletes the nested entry specified by a dot-separated path from the dictionary using pop().
:param data: The dictionary to modify.
:param path: A dot-separated string indicating the nested key to delete.
e.g., "foo.bar.baz" will attempt to delete data["foo"]["bar"]["baz"].
:raises KeyError: If any intermediate key is missing or not a dictionary,
or if the final key to delete is not found.
"""
if not path:
msg = "Cannot delete. Path is empty."
raise KeyError(msg)
keys = path.split(".")
return delete_by_path_tuple(d, tuple(keys))
V = TypeVar("V")
def get_value_by_path(
d: DictLike,
path: str,
fallback: V | None = None,
expected_type: type[V] | None = None, # noqa: ARG001
) -> V:
"""Get the value at a specific dot-separated path in a nested dictionary.
If the path does not exist, it returns fallback.
:param d: The dictionary to get from.
:param path: The dot-separated path to the key (e.g., 'foo.bar').
"""
keys = path.split(".")
current = d
for key in keys[:-1]:
current = current.setdefault(key, {})
if isinstance(current, dict):
return cast("V", current.get(keys[-1], fallback))
return cast("V", fallback)
def flatten_data_structured(
data: dict, parent_path: PathTuple = ()
) -> dict[PathTuple, Any]:
"""Flatten data using tuple keys instead of string concatenation.
This eliminates ambiguity between literal dots in keys vs nested structure.
Args:
data: The nested dictionary to flatten
parent_path: Current path as tuple (used for recursion)
Returns:
Dict with tuple keys representing the full path to each value
Example:
{"key.foo": "val1", "key": {"foo": "val2"}}
becomes:
{("key.foo",): "val1", ("key", "foo"): "val2"}
"""
flattened = {}
for key, value in data.items():
current_path = (*parent_path, key)
if isinstance(value, dict):
if value:
flattened.update(flatten_data_structured(value, current_path))
else:
flattened[current_path] = {}
else:
flattened[current_path] = value
return flattened
def should_skip_path(path: tuple, delete_paths: set[tuple]) -> bool:
"""Check if path should be skipped because it's under a deletion path."""
return any(path_starts_with(path, delete_path) for delete_path in delete_paths)
# TODO: use PathTuple
def path_match(path: list[str], whitelist_paths: list[list[str]]) -> bool:
"""Returns True if path matches any whitelist path with "*" wildcards.
I.e.:
whitelist_paths = [["a.b.*"]]
path = ["a", "b", "c"]
path_match(path, whitelist_paths) == True
whitelist_paths = ["a.b.c", "a.b.*"]
path = ["a", "b", "d"]
path_match(path, whitelist_paths) == False
"""
for wp in whitelist_paths:
if len(path) != len(wp):
continue
match = True
for p, w in zip(path, wp, strict=False):
if w not in ("*", p):
match = False
break
if match:
return True
return False

245
pkgs/clan-cli/clan_lib/persist/path_utils_test.py Normal file
View File

@@ -0,0 +1,245 @@
import pytest
from clan_lib.persist.path_utils import (
delete_by_path,
flatten_data_structured,
list_difference,
path_match,
set_value_by_path,
)
# --------- path_match ---------
@pytest.mark.parametrize(
("path", "whitelist", "expected"),
[
# Exact matches
(["a", "b", "c"], [["a", "b", "c"]], True),
(["a", "b"], [["a", "b"]], True),
([], [[]], True),
# Wildcard matches
(["a", "b", "c"], [["a", "*", "c"]], True),
(["a", "x", "c"], [["a", "*", "c"]], True),
(["a", "b", "c"], [["*", "b", "c"]], True),
(["a", "b", "c"], [["a", "b", "*"]], True),
(["a", "b", "c"], [["*", "*", "*"]], True),
# Multiple patterns - one matches
(["a", "b", "c"], [["x", "y", "z"], ["a", "*", "c"]], True),
(["x", "y", "z"], [["a", "*", "c"], ["x", "y", "z"]], True),
# Length mismatch
(["a", "b", "c"], [["a", "b"]], False),
(["a", "b"], [["a", "b", "c"]], False),
# Non-matching
(["a", "b", "c"], [["a", "b", "x"]], False),
(["a", "b", "c"], [["a", "x", "x"]], False),
(["a", "b", "c"], [["x", "x", "x"]], False),
# Empty whitelist
(["a"], [], False),
# Wildcards and exact mixed
(
["instances", "inst1", "roles", "roleA", "settings"],
[["instances", "*", "roles", "*", "settings"]],
True,
),
# Partial wildcard - length mismatch should fail
(
["instances", "inst1", "roles", "roleA"],
[["instances", "*", "roles", "*", "settings"]],
False,
),
# Empty path, no patterns
([], [], False),
],
)
def test_path_match(
path: list[str],
whitelist: list[list[str]],
expected: bool,
) -> None:
assert path_match(path, whitelist) == expected
# --------- set_value_by_path ---------
def test_patch_nested() -> None:
orig = {"a": 1, "b": {"a": 2.1, "b": 2.2}, "c": 3}
set_value_by_path(orig, "b.b", "foo")
# Should only update the nested value
assert orig == {"a": 1, "b": {"a": 2.1, "b": "foo"}, "c": 3}
def test_patch_nested_dict() -> None:
orig = {"a": 1, "b": {"a": 2.1, "b": 2.2}, "c": 3}
# This should update the whole "b" dict
# Which also removes all other keys
set_value_by_path(orig, "b", {"b": "foo"})
# Should only update the nested value
assert orig == {"a": 1, "b": {"b": "foo"}, "c": 3}
def test_create_missing_paths() -> None:
orig = {"a": 1}
set_value_by_path(orig, "b.c", "foo")
# Should only update the nested value
assert orig == {"a": 1, "b": {"c": "foo"}}
orig = {}
set_value_by_path(orig, "a.b.c", "foo")
assert orig == {"a": {"b": {"c": "foo"}}}
# ---- delete_by_path ----
def test_delete_atom() -> None:
data = {"foo": {"bar": 1}}
# Removes the key "foo.bar"
# Returns the deleted key-value pair { "bar": 1 }
entry = delete_by_path(data, "foo.bar")
assert entry == {"bar": 1}
assert data == {"foo": {}}
def test_delete_intermediate() -> None:
data = {"a": {"b": {"c": {"d": 42}}}}
# Removes "a.b.c.d"
entry = delete_by_path(data, "a.b.c")
assert entry == {"c": {"d": 42}}
# Check all intermediate dictionaries remain intact
assert data == {"a": {"b": {}}}
def test_delete_top_level() -> None:
data = {"x": 100, "y": 200}
# Deletes top-level key
entry = delete_by_path(data, "x")
assert entry == {"x": 100}
assert data == {"y": 200}
def test_delete_key_not_found() -> None:
data = {"foo": {"bar": 1}}
# Trying to delete a non-existing key "foo.baz" - should return empty dict
result = delete_by_path(data, "foo.baz")
assert result == {}
# Data should remain unchanged
assert data == {"foo": {"bar": 1}}
def test_delete_intermediate_not_dict() -> None:
data = {"foo": "not a dict"}
# Trying to go deeper into a non-dict value
with pytest.raises(KeyError) as excinfo:
delete_by_path(data, "foo.bar")
assert "not found or not a dictionary" in str(excinfo.value)
# Data should remain unchanged
assert data == {"foo": "not a dict"}
def test_delete_empty_path() -> None:
data = {"foo": {"bar": 1}}
# Attempting to delete with an empty path
with pytest.raises(KeyError) as excinfo:
delete_by_path(data, "")
# Depending on how you handle empty paths, you might raise an error or handle it differently.
# If you do raise an error, check the message.
assert "Cannot delete. Path is empty" in str(excinfo.value)
assert data == {"foo": {"bar": 1}}
def test_delete_non_existent_path_deep() -> None:
data = {"foo": {"bar": {"baz": 123}}}
# non-existent deep path - should return empty dict
result = delete_by_path(data, "foo.bar.qux")
assert result == {}
# Data remains unchanged
assert data == {"foo": {"bar": {"baz": 123}}}
# --- list_difference ---
def test_list_unmerge() -> None:
all_machines = ["machineA", "machineB"]
inventory = ["machineB"]
nix_machines = list_difference(all_machines, inventory)
assert nix_machines == ["machineA"]
# --- flatten_data_structured ---
def test_flatten_data_structured() -> None:
data = {
"name": "example",
"settings": {
"optionA": True,
"optionB": {
"subOption1": 10,
"subOption2": 20,
},
"emptyDict": {},
},
"listSetting": [1, 2, 3],
}
expected_flat = {
("name",): "example",
("settings", "optionA"): True,
("settings", "optionB", "subOption1"): 10,
("settings", "optionB", "subOption2"): 20,
("settings", "emptyDict"): {},
("listSetting",): [1, 2, 3],
}
flattened = flatten_data_structured(data)
assert flattened == expected_flat
def test_flatten_data_structured_empty() -> None:
data: dict = {}
expected_flat: dict = {}
flattened = flatten_data_structured(data)
assert flattened == expected_flat
def test_flatten_data_structured_nested_empty() -> None:
data: dict = {
"level1": {
"level2": {
"level3": {},
},
},
}
expected_flat: dict = {
("level1", "level2", "level3"): {},
}
flattened = flatten_data_structured(data)
assert flattened == expected_flat
def test_flatten_data_dot_in_keys() -> None:
data = {
"key.foo": "value1",
"key": {
"foo": "value2",
},
}
expected_flat = {
("key.foo",): "value1",
("key", "foo"): "value2",
}
flattened = flatten_data_structured(data)
assert flattened == expected_flat

524
pkgs/clan-cli/clan_lib/persist/util.py
View File

@@ -1,524 +0,0 @@
"""Utilities for working with nested dictionaries, particularly for
flattening, unmerging lists, finding duplicates, and calculating patches.
"""
import json
from collections import Counter
from typing import Any, TypeVar, cast
from clan_lib.errors import ClanError
T = TypeVar("T")
# Priority constants for configuration merging
WRITABLE_PRIORITY_THRESHOLD = 100 # Values below this are not writeable
empty: list[str] = []
def merge_objects(
curr: T,
update: T,
merge_lists: bool = True,
path: list[str] = empty,
) -> T:
"""Updates values in curr by values of update
The output contains values for all keys of curr and update together.
Lists are deduplicated and appended almost like in the nix module system.
Example:
merge_objects({"a": 1}, {"a": null }) -> {"a": null}
merge_objects({"a": null}, {"a": 1 }) -> {"a": 1}
"""
result = {}
msg = f"cannot update non-dictionary values: {curr} by {update}"
if not isinstance(update, dict):
raise ClanError(msg)
if not isinstance(curr, dict):
raise ClanError(msg)
all_keys = set(update.keys()).union(curr.keys())
for key in all_keys:
curr_val = curr.get(key)
update_val = update.get(key)
if isinstance(update_val, dict) and isinstance(curr_val, dict):
result[key] = merge_objects(
curr_val,
update_val,
merge_lists=merge_lists,
path=[*path, key],
)
elif isinstance(update_val, list) and isinstance(curr_val, list):
if merge_lists:
result[key] = list(dict.fromkeys(curr_val + update_val)) # type: ignore[assignment]
else:
result[key] = update_val # type: ignore[assignment]
elif (
update_val is not None
and curr_val is not None
and type(update_val) is not type(curr_val)
):
msg = f"Type mismatch for key '{key}'. Cannot update {type(curr_val)} with {type(update_val)}"
raise ClanError(msg, location=json.dumps([*path, key]))
elif key in update:
result[key] = update_val # type: ignore[assignment]
elif key in curr:
result[key] = curr_val # type: ignore[assignment]
return cast("T", result)
def path_match(path: list[str], whitelist_paths: list[list[str]]) -> bool:
"""Returns True if path matches any whitelist path with "*" wildcards.
I.e.:
whitelist_paths = [["a.b.*"]]
path = ["a", "b", "c"]
path_match(path, whitelist_paths) == True
whitelist_paths = ["a.b.c", "a.b.*"]
path = ["a", "b", "d"]
path_match(path, whitelist_paths) == False
"""
for wp in whitelist_paths:
if len(path) != len(wp):
continue
match = True
for p, w in zip(path, wp, strict=False):
if w not in ("*", p):
match = False
break
if match:
return True
return False
def flatten_data(data: dict, parent_key: str = "", separator: str = ".") -> dict:
"""Recursively flattens a nested dictionary structure where keys are joined by the separator.
Args:
data (dict): The nested dictionary structure.
parent_key (str): The current path to the nested dictionary (used for recursion).
separator (str): The string to use for joining keys.
Returns:
dict: A flattened dictionary with all values. Directly in the root.
"""
flattened = {}
for key, value in data.items():
new_key = f"{parent_key}{separator}{key}" if parent_key else key
if isinstance(value, dict):
# Recursively flatten the nested dictionary
if value:
flattened.update(flatten_data(value, new_key, separator))
else:
# If the value is an empty dictionary, add it to the flattened dict
flattened[new_key] = {}
else:
flattened[new_key] = value
return flattened
def list_difference(all_items: list, filter_items: list) -> list:
"""Unmerge the current list. Given a previous list.
Returns:
The other list.
"""
# Unmerge the lists
return [value for value in all_items if value not in filter_items]
def find_duplicates(string_list: list[str]) -> list[str]:
count = Counter(string_list)
return [item for item, freq in count.items() if freq > 1]
def find_deleted_paths(
curr: dict[str, Any],
update: dict[str, Any],
parent_key: str = "",
) -> set[str]:
"""Recursively find keys (at any nesting level) that exist in persisted but do not
exist in update. If a nested dictionary is completely removed, return that dictionary key.
:param persisted: The original (persisted) nested dictionary.
:param update: The updated nested dictionary (some keys might be removed).
:param parent_key: The dotted path to the current dictionary's location.
:return: A set of dotted paths indicating keys or entire nested paths that were deleted.
"""
deleted_paths = set()
# Iterate over keys in persisted
for key, p_value in curr.items():
current_path = f"{parent_key}.{key}" if parent_key else key
# Check if this key exists in update
if key not in update:
# Key doesn't exist at all -> entire branch deleted
deleted_paths.add(current_path)
else:
u_value = update[key]
# If persisted value is dict, check the update value
if isinstance(p_value, dict):
if isinstance(u_value, dict):
# If persisted dict is non-empty but updated dict is empty,
# that means everything under this branch is removed.
if p_value and not u_value:
# All children are removed
for child_key in p_value:
child_path = f"{current_path}.{child_key}"
deleted_paths.add(child_path)
else:
# Both are dicts, recurse deeper
deleted_paths |= find_deleted_paths(
p_value,
u_value,
current_path,
)
else:
# Persisted was a dict, update is not a dict -> entire branch changed
# Consider this as a full deletion of the persisted branch
deleted_paths.add(current_path)
return deleted_paths
def parent_is_dict(key: str, data: dict[str, Any]) -> bool:
parts = key.split(".")
while len(parts) > 1:
parts.pop()
parent_key = ".".join(parts)
if parent_key in data:
return isinstance(data[parent_key], dict)
return False
def is_writeable_key(
key: str,
writeables: dict[str, set[str]],
) -> bool:
"""Recursively check if a key is writeable.
key "machines.machine1.deploy.targetHost" is specified but writeability is only defined for "machines"
We pop the last key and check if the parent key is writeable/non-writeable.
"""
remaining = key.split(".")
while remaining:
if ".".join(remaining) in writeables["writeable"]:
return True
if ".".join(remaining) in writeables["non_writeable"]:
return False
remaining.pop()
msg = f"Cannot determine writeability for key '{key}'"
raise ClanError(msg, description="F001")
def calc_patches(
persisted: dict[str, Any],
update: dict[str, Any],
all_values: dict[str, Any],
writeables: dict[str, set[str]],
) -> tuple[dict[str, Any], set[str]]:
"""Calculate the patches to apply to the inventory.
Given its current state and the update to apply.
Filters out nix-values so it doesn't matter if the anyone sends them.
: param persisted: The current state of the inventory.
: param update: The update to apply.
: param writeable: The writeable keys. Use 'determine_writeability'.
Example: {'writeable': {'foo', 'foo.bar'}, 'non_writeable': {'foo.nix'}}
: param all_values: All values in the inventory retrieved from the flake evaluation.
Returns a tuple with the SET and DELETE patches.
"""
data_all = flatten_data(all_values)
data_all_updated = flatten_data(update)
data_dyn = flatten_data(persisted)
all_keys = set(data_all) | set(data_all_updated)
patchset: dict[str, Any] = {}
delete_set = find_deleted_paths(all_values, update, parent_key="")
for key in all_keys:
# Get the old and new values
old = data_all.get(key, None)
new = data_all_updated.get(key, None)
# Some kind of change
if old != new:
# If there is a change, check if the key is writeable
if not is_writeable_key(key, writeables):
msg = f"Key '{key}' is not writeable. It seems its value is statically defined in nix."
raise ClanError(msg)
if any(key.startswith(d) for d in delete_set):
# Skip this key if it or any of its parent paths are marked for deletion
continue
if old is not None and type(old) is not type(new):
if new is None:
# If this is a deleted key, they are handled by 'find_deleted_paths'
continue
msg = f"Type mismatch for key '{key}'. Cannot update {type(old)} with {type(new)}"
description = f"""
Previous_value is of type '{type(old)}' this operation would change it to '{type(new)}'.
Prev: {old}
->
After: {new}
"""
raise ClanError(msg, description=description)
if isinstance(new, list):
duplicates = find_duplicates(new)
if duplicates:
msg = f"Key '{key}' contains list duplicates: {duplicates} - List values must be unique."
raise ClanError(msg)
# List of current values
persisted_data = data_dyn.get(key, [])
# List including nix values
all_list = data_all.get(key, [])
nix_list = list_difference(all_list, persisted_data)
# every item in nix_list MUST be in new
nix_items_to_remove = list(
filter(lambda item: item not in new, nix_list),
)
if nix_items_to_remove:
msg = (
f"Key '{key}' doesn't contain items {nix_items_to_remove} - "
"Deleting them is not possible, they are static values set via a .nix file"
)
raise ClanError(msg)
if new != all_list:
patchset[key] = list_difference(new, nix_list)
else:
patchset[key] = new
# Ensure not inadvertently patching something already marked for deletion
conflicts = {key for d in delete_set for key in patchset if key.startswith(d)}
if conflicts:
conflict_list = ", ".join(sorted(conflicts))
msg = (
f"The following keys are marked for deletion but also have update values: {conflict_list}. "
"You cannot delete and patch the same key and its subkeys."
)
raise ClanError(msg)
return patchset, delete_set
def determine_writeability(
priorities: dict[str, Any],
defaults: dict[str, Any],
persisted: dict[str, Any],
parent_key: str = "",
parent_prio: int | None = None,
results: dict | None = None,
non_writeable: bool = False,
) -> dict[str, set[str]]:
if results is None:
results = {"writeable": set({}), "non_writeable": set({})}
for key, value in priorities.items():
if key == "__prio":
continue
full_key = f"{parent_key}.{key}" if parent_key else key
# Determine the priority for the current key
# Inherit from parent if no priority is defined
prio = value.get("__prio", None)
if prio is None:
prio = parent_prio
# If priority is less than 100, all children are not writeable
# If the parent passed "non_writeable" earlier, this makes all children not writeable
if (prio is not None and prio < WRITABLE_PRIORITY_THRESHOLD) or non_writeable:
results["non_writeable"].add(full_key)
if isinstance(value, dict):
determine_writeability(
value,
defaults,
{}, # Children won't be writeable, so correlation doesn't matter here
full_key,
prio, # Pass the same priority down
results,
# Recursively mark all children as non-writeable
non_writeable=True,
)
continue
# Check if the key is writeable otherwise
key_in_correlated = key in persisted
if prio is None:
msg = f"Priority for key '{full_key}' is not defined. Cannot determine if it is writeable."
raise ClanError(msg)
is_mergeable = False
if prio == WRITABLE_PRIORITY_THRESHOLD:
default = defaults.get(key)
if isinstance(default, dict):
is_mergeable = True
if isinstance(default, list):
is_mergeable = True
if key_in_correlated:
is_mergeable = True
is_writeable = prio > WRITABLE_PRIORITY_THRESHOLD or is_mergeable
# Append the result
if is_writeable:
results["writeable"].add(full_key)
else:
results["non_writeable"].add(full_key)
# Recursive
if isinstance(value, dict):
determine_writeability(
value,
defaults.get(key, {}),
persisted.get(key, {}),
full_key,
prio, # Pass down current priority
results,
)
return results
def delete_by_path(d: dict[str, Any], path: str) -> Any:
"""Deletes the nested entry specified by a dot-separated path from the dictionary using pop().
:param data: The dictionary to modify.
:param path: A dot-separated string indicating the nested key to delete.
e.g., "foo.bar.baz" will attempt to delete data["foo"]["bar"]["baz"].
:raises KeyError: If any intermediate key is missing or not a dictionary,
or if the final key to delete is not found.
"""
if not path:
msg = "Cannot delete. Path is empty."
raise KeyError(msg)
keys = path.split(".")
current = d
# Navigate to the parent dictionary of the final key
for key in keys[:-1]:
if key not in current or not isinstance(current[key], dict):
msg = f"Cannot delete. Key '{path}' not found or not a dictionary '{d}'"
raise KeyError(msg)
current = current[key]
# Attempt to pop the final key
last_key = keys[-1]
try:
value = current.pop(last_key)
except KeyError:
# Possibly data was already deleted
# msg = f"Canot delete. Path '{path}' not found in data '{d}'"
# raise KeyError(msg) from exc
return {}
else:
return {last_key: value}
type DictLike = dict[str, Any] | Any
V = TypeVar("V")
def get_value_by_path(
d: DictLike,
path: str,
fallback: V | None = None,
expected_type: type[V] | None = None, # noqa: ARG001
) -> V:
"""Get the value at a specific dot-separated path in a nested dictionary.
If the path does not exist, it returns fallback.
:param d: The dictionary to get from.
:param path: The dot-separated path to the key (e.g., 'foo.bar').
"""
keys = path.split(".")
current = d
for key in keys[:-1]:
current = current.setdefault(key, {})
if isinstance(current, dict):
return cast("V", current.get(keys[-1], fallback))
return cast("V", fallback)
def set_value_by_path(d: DictLike, path: str, content: Any) -> None:
"""Update the value at a specific dot-separated path in a nested dictionary.
If the value didn't exist before, it will be created recursively.
:param d: The dictionary to update.
:param path: The dot-separated path to the key (e.g., 'foo.bar').
:param content: The new value to set.
"""
keys = path.split(".")
current = d
for key in keys[:-1]:
current = current.setdefault(key, {})
current[keys[-1]] = content
from typing import NotRequired, Required, get_args, get_origin, get_type_hints
def is_typeddict_class(obj: type) -> bool:
"""Safely checks if a class is a TypedDict."""
return (
isinstance(obj, type)
and hasattr(obj, "__annotations__")
and obj.__class__.__name__ == "_TypedDictMeta"
)
def retrieve_typed_field_names(obj: type, prefix: str = "") -> set[str]:
fields = set()
hints = get_type_hints(obj, include_extras=True)
for field, field_type in hints.items():
full_key = f"{prefix}.{field}" if prefix else field
origin = get_origin(field_type)
args = get_args(field_type)
# Unwrap Required/NotRequired
if origin in {NotRequired, Required}:
unwrapped_type = args[0]
origin = get_origin(unwrapped_type)
args = get_args(unwrapped_type)
else:
unwrapped_type = field_type
if is_typeddict_class(unwrapped_type):
fields |= retrieve_typed_field_names(unwrapped_type, prefix=full_key)
else:
fields.add(full_key)
return fields

79
pkgs/clan-cli/clan_lib/persist/validate.py Normal file
View File

@@ -0,0 +1,79 @@
from typing import Any
from clan_lib.errors import ClanError
from clan_lib.persist.path_utils import (
PathTuple,
find_duplicates,
path_starts_with,
path_to_string,
)
from clan_lib.persist.write_rules import WriteMap, is_writeable_path
def validate_no_static_deletion(
path: PathTuple, new_value: Any, static_data: dict[PathTuple, Any]
) -> None:
"""Validate that we're not trying to delete static data."""
# Check if we're trying to delete a path that exists in static data
if path in static_data and new_value is None:
msg = f"Path '{path_to_string(path)}' is readonly - since its defined via a .nix file"
raise ClanError(msg)
# For lists, check if we're trying to remove static items
if isinstance(new_value, list) and path in static_data:
static_items = static_data[path]
if isinstance(static_items, list):
missing_static = [item for item in static_items if item not in new_value]
if missing_static:
msg = f"Path '{path_to_string(path)}' doesn't contain static items {missing_static} - They are readonly - since they are defined via a .nix file"
raise ClanError(msg)
def validate_writeability(path: PathTuple, writeables: WriteMap) -> None:
"""Validate that a path is writeable."""
if not is_writeable_path(path, writeables):
msg = f"Path '{path_to_string(path)}' is readonly. - It seems its value is statically defined in nix."
raise ClanError(msg)
def validate_type_compatibility(path: tuple, old_value: Any, new_value: Any) -> None:
"""Validate that type changes are allowed."""
if old_value is not None and type(old_value) is not type(new_value):
if new_value is None:
return # Deletion is handled separately
path_str = path_to_string(path)
msg = f"Type mismatch for path '{path_str}'. Cannot update {type(old_value)} with {type(new_value)}"
description = f"""
Previous value is of type '{type(old_value)}' this operation would change it to '{type(new_value)}'.
Prev: {old_value}
->
After: {new_value}
"""
raise ClanError(msg, description=description)
def validate_list_uniqueness(path: tuple, value: Any) -> None:
"""Validate that lists don't contain duplicates."""
if isinstance(value, list):
duplicates = find_duplicates(value)
if duplicates:
msg = f"Path '{path_to_string(path)}' contains list duplicates: {duplicates} - List values must be unique."
raise ClanError(msg)
def validate_patch_conflicts(
patches: set[PathTuple], delete_paths: set[PathTuple]
) -> None:
"""Ensure patches don't conflict with deletions."""
conflicts = {
path
for delete_path in delete_paths
for path in patches
if path_starts_with(path, delete_path)
}
if conflicts:
conflict_list = ", ".join(path_to_string(path) for path in sorted(conflicts))
msg = f"The following paths are marked for deletion but also have update values: {conflict_list}. - You cannot delete and patch the same path and its subpaths."
raise ClanError(msg)

189
pkgs/clan-cli/clan_lib/persist/write_rules.py Normal file
View File

@@ -0,0 +1,189 @@
from typing import Any, TypedDict
from clan_lib.errors import ClanError
from clan_lib.persist.path_utils import PathTuple, path_to_string
WRITABLE_PRIORITY_THRESHOLD = 100 # Values below this are not writeable
class WriteMap(TypedDict):
writeable: set[PathTuple]
non_writeable: set[PathTuple]
def is_writeable_path(
key: PathTuple,
writeables: WriteMap,
) -> bool:
"""Recursively check if a key is writeable.
If key "machines.machine1.deploy.targetHost" is specified but writeability is only
defined for "machines", we pop the last key and check if the parent key is writeable/non-writeable.
"""
remaining = key
while remaining:
current_path = remaining
if current_path in writeables["writeable"]:
return True
if current_path in writeables["non_writeable"]:
return False
remaining = remaining[:-1]
msg = f"Cannot determine writeability for key '{key}'"
raise ClanError(msg)
def is_writeable_key(
key: str,
writeables: WriteMap,
) -> bool:
"""Recursively check if a key is writeable.
Key is a dot-separated string, e.g. "machines.machine1.deploy.targetHost".
In case of ambiguity use is_writeable_path with tuple keys.
"""
items = key.split(".")
return is_writeable_path(tuple(items), writeables)
def get_priority(value: Any) -> int | None:
"""Extract priority from a value, handling both dict and non-dict cases."""
if isinstance(value, dict) and "__prio" in value:
return value["__prio"]
return None
def is_mergeable_type(value: Any) -> bool:
"""Check if a value type supports merging (dict or list)."""
return isinstance(value, (dict, list))
def should_inherit_non_writeable(
priority: int | None, parent_non_writeable: bool
) -> bool:
"""Determine if this node should be marked as non-writeable due to inheritance."""
if parent_non_writeable:
return True
return priority is not None and priority < WRITABLE_PRIORITY_THRESHOLD
def is_key_writeable(
priority: int | None, exists_in_persisted: bool, value_in_all: Any
) -> bool:
"""Determine if a key is writeable based on priority and mergeability rules.
Rules:
- Priority > 100: Always writeable
- Priority < 100: Never writeable
- Priority == 100: Writeable if mergeable (dict/list) OR exists in persisted
"""
if priority is None:
return False # No priority means not writeable
if priority > WRITABLE_PRIORITY_THRESHOLD:
return True
if priority < WRITABLE_PRIORITY_THRESHOLD:
return False
# priority == WRITABLE_PRIORITY_THRESHOLD (100)
return is_mergeable_type(value_in_all) or exists_in_persisted
def _determine_writeability_recursive(
priorities: dict[str, Any],
all_values: dict[str, Any],
persisted: dict[str, Any],
current_path: PathTuple = (),
inherited_priority: int | None = None,
parent_non_writeable: bool = False,
results: WriteMap | None = None,
) -> WriteMap:
"""Recursively determine writeability for all paths in the priority structure.
This is internal recursive function. Use 'determine_writeability' as entry point.
"""
if results is None:
results = WriteMap(writeable=set(), non_writeable=set())
for key, value in priorities.items():
# Skip metadata keys
if key == "__prio":
continue
path = (*current_path, key)
# Determine priority for this key
key_priority = get_priority(value)
effective_priority = (
key_priority if key_priority is not None else inherited_priority
)
# Check if this should be non-writeable due to inheritance
force_non_writeable = should_inherit_non_writeable(
effective_priority, parent_non_writeable
)
if force_non_writeable:
results["non_writeable"].add(path)
# All children are also non-writeable
if isinstance(value, dict):
_determine_writeability_recursive(
value,
all_values.get(key, {}),
{}, # Doesn't matter since all children will be non-writeable
path,
effective_priority,
parent_non_writeable=True,
results=results,
)
else:
# Determine writeability based on rules
if effective_priority is None:
msg = f"Priority for path '{path_to_string(path)}' is not defined. Cannot determine writeability."
raise ClanError(msg)
exists_in_persisted = key in persisted
value_in_all = all_values.get(key)
if is_key_writeable(effective_priority, exists_in_persisted, value_in_all):
results["writeable"].add(path)
else:
results["non_writeable"].add(path)
# Recurse into children
if isinstance(value, dict):
_determine_writeability_recursive(
value,
all_values.get(key, {}),
persisted.get(key, {}),
path,
effective_priority,
parent_non_writeable=False,
results=results,
)
return results
def compute_write_map(
priorities: dict[str, Any], all_values: dict[str, Any], persisted: dict[str, Any]
) -> WriteMap:
"""Determine writeability for all paths based on priorities and current data.
- Priority-based writeability: Values with priority < 100 are not writeable
- Inheritance: Children inherit parent priorities if not specified
- Special case at priority 100: Can be writeable if it's mergeable (dict/list) or exists in persisted data
Args:
priorities: The priority structure defining writeability rules. See: 'clanInternals.inventoryClass.introspection'
all_values: All values in the inventory, See: 'clanInternals.inventoryClass.allValues'
persisted: The current mutable state of the inventory, see: 'readFile inventory.json'
Returns:
Dict with sets of writeable and non-writeable paths using tuple keys
"""
return _determine_writeability_recursive(priorities, all_values, persisted)

118
pkgs/clan-cli/clan_lib/persist/write_rules_test.py Normal file
View File

@@ -0,0 +1,118 @@
from clan_lib.persist.write_rules import compute_write_map
def test_write_simple() -> None:
prios = {
"foo": {
"__prio": 100, # <- writeable: "foo"
"bar": {"__prio": 1000}, # <- writeable: mkDefault "foo.bar"
},
"foo.bar": {"__prio": 1000},
}
default: dict = {"foo": {}}
data: dict = {}
res = compute_write_map(prios, default, data)
assert res == {
"writeable": {("foo", "bar"), ("foo",), ("foo.bar",)},
"non_writeable": set(),
}
def test_write_inherited() -> None:
prios = {
"foo": {
"__prio": 100, # <- writeable: "foo"
"bar": {
# Inherits prio from parent <- writeable: "foo.bar"
"baz": {"__prio": 1000}, # <- writeable: "foo.bar.baz"
},
},
}
data: dict = {}
res = compute_write_map(prios, {"foo": {"bar": {}}}, data)
assert res == {
"writeable": {("foo",), ("foo", "bar"), ("foo", "bar", "baz")},
"non_writeable": set(),
}
def test_non_write_inherited() -> None:
prios = {
"foo": {
"__prio": 50, # <- non writeable: mkForce "foo" = {...}
"bar": {
# Inherits prio from parent <- non writeable
"baz": {"__prio": 1000}, # <- non writeable: mkDefault "foo.bar.baz"
},
},
}
data: dict = {}
res = compute_write_map(prios, {}, data)
assert res == {
"writeable": set(),
"non_writeable": {("foo",), ("foo", "bar", "baz"), ("foo", "bar")},
}
def test_write_list() -> None:
prios = {
"foo": {
"__prio": 100,
},
}
data: dict = {}
default: dict = {
"foo": [
"a",
"b",
], # <- writeable: because lists are merged. Filtering out nix-values comes later
}
res = compute_write_map(prios, default, data)
assert res == {
"writeable": {("foo",)},
"non_writeable": set(),
}
def test_write_because_written() -> None:
# This test is essential to allow removing data that was previously written.
# It may also have the weird effect, that somebody can change data, but others cannot.
# But this might be acceptable, since its rare minority edge-case.
prios = {
"foo": {
"__prio": 100, # <- writeable: "foo"
"bar": {
# Inherits prio from parent <- writeable
"baz": {"__prio": 100}, # <- non writeable usually
"foobar": {"__prio": 100}, # <- non writeable
},
},
}
# Given the following data. {}
# Check that the non-writeable paths are correct.
res = compute_write_map(prios, {"foo": {"bar": {}}}, {})
assert res == {
"writeable": {("foo",), ("foo", "bar")},
"non_writeable": {("foo", "bar", "baz"), ("foo", "bar", "foobar")},
}
data: dict = {
"foo": {
"bar": {
"baz": "foo", # <- written. Since we created the data, we know we can write to it
},
},
}
res = compute_write_map(prios, {}, data)
assert res == {
"writeable": {("foo",), ("foo", "bar"), ("foo", "bar", "baz")},
"non_writeable": {("foo", "bar", "foobar")},
}

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

@@ -14,7 +14,7 @@ from clan_lib.nix_models.clan import (
InventoryInstancesType,
)
from clan_lib.persist.inventory_store import InventoryStore
from clan_lib.persist.util import get_value_by_path, set_value_by_path
from clan_lib.persist.path_utils import get_value_by_path, set_value_by_path
class CategoryInfo(TypedDict):

2
pkgs/clan-cli/clan_lib/tags/list_test.py
View File

@@ -5,7 +5,7 @@ import pytest
from clan_lib.flake import Flake
from clan_lib.nix_models.clan import InventoryMachineTagsType
from clan_lib.persist.inventory_store import InventoryStore
from clan_lib.persist.util import get_value_by_path, set_value_by_path
from clan_lib.persist.path_utils import get_value_by_path, set_value_by_path
from clan_lib.tags.list import list_tags

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

@@ -29,7 +29,7 @@ from clan_lib.nix_models.clan import (
)
from clan_lib.nix_models.clan import InventoryMachineDeploy as MachineDeploy
from clan_lib.persist.inventory_store import InventoryStore
from clan_lib.persist.util import set_value_by_path
from clan_lib.persist.path_utils import set_value_by_path
from clan_lib.services.modules import list_service_modules
from clan_lib.ssh.remote import Remote
from clan_lib.templates.disk import hw_main_disk_options, set_machine_disk_schema