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

api: check deletions if possible

Browse Source
This commit is contained in:
Johannes Kirschbauer
2025-10-15 16:15:54 +02:00
parent 24ecdb227e
commit 16d245b179
2 changed files with 99 additions and 260 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
pkgs/clan-cli/clan_lib/persist/patch_engine.py
View File

@@ -6,16 +6,16 @@ from clan_lib.persist.path_utils import (
PathTuple,
flatten_data_structured,
list_difference,
path_to_string,
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 AttributeMap
from clan_lib.persist.write_rules import AttributeMap, PersistenceAttribute
def find_deleted_paths_structured(
@@ -53,37 +53,13 @@ def find_deleted_paths_structured(
# 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
# 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],
@@ -109,11 +85,8 @@ def calc_patches(
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)
persisted_flat = flatten_data_structured(persisted)
update_flat = flatten_data_structured(update)
all_values_flat = flatten_data_structured(all_values)
@@ -123,13 +96,19 @@ def calc_patches(
# 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)
# Validate deletions. Ensure deletions are allowed
# Prior introspection adds PersistenceAttribute.DELETE to paths that can be deleted
for delete_path in delete_paths:
if delete_path not in attribute_props:
msg = f"""Cannot delete path '{path_to_string(delete_path)}' - '{path_to_string(delete_path)}' is not a valid path.
This is an internal error that should not happen.
Please report this issue at https://git.clan.lol/clan/clan-core/issues
"""
raise ClanError(msg)
if PersistenceAttribute.DELETE not in attribute_props.get(delete_path, {}):
msg = f"Cannot delete path '{path_to_string(delete_path)}' - '{path_to_string(delete_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)
@@ -154,7 +133,7 @@ def calc_patches(
continue
# Validate the change is allowed
validate_no_static_deletion(path, new_value, static_data)
# validate_no_static_deletion(path, new_value, static_data)
validate_writeability(path, attribute_props)
validate_type_compatibility(path, old_value, new_value)
validate_list_uniqueness(path, new_value)
@@ -162,7 +141,10 @@ def calc_patches(
patch_value = new_value # Init
if isinstance(new_value, list):
patch_value = list_difference(new_value, static_data.get(path, []))
static_items = list_difference(
old_value or [], persisted_flat.get(path, [])
)
patch_value = list_difference(new_value, static_items)
patches[path] = patch_value

295
pkgs/clan-cli/clan_lib/persist/patch_engine_test.py
View File

@@ -6,7 +6,6 @@ import pytest
from clan_lib.errors import ClanError
from clan_lib.persist.patch_engine import (
calc_patches,
calculate_static_data,
merge_objects,
)
from clan_lib.persist.path_utils import (
@@ -19,195 +18,6 @@ from clan_lib.persist.write_rules import (
compute_attribute_persistence,
)
# --- calculate_static_data ---
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],
},
}
expected_static = {
("settings", "optionB"): False,
("settings", "listSetting"): [1, 4],
("staticOnly",): "staticValue",
}
static_data = calculate_static_data(all_values, persisted)
assert static_data == expected_static
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",
},
},
}
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_dot_in_keys() -> None:
all_values = {
"key.foo": "staticValue",
"key": {
"foo": "anotherStaticValue",
},
}
persisted: dict = {}
expected_static = {
("key.foo",): "staticValue",
("key", "foo"): "anotherStaticValue",
}
static_data = calculate_static_data(all_values, persisted)
assert static_data == expected_static
# --------- calc_patches ---------
def test_update_simple() -> None:
prios = {
@@ -379,34 +189,79 @@ def test_update_parent_non_writeable() -> None:
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": {
# "__this": {"total": True, "prio": 100},
# # We cannot delete children because "foo" is total
# },
# }
def test_calculate_static_data_no_static_sets() -> None:
all_values: dict = {
"instance": {
"hello": {
"static": {},
}
}
}
persisted: dict = {"instance": {"hello": {}}}
attribute_props = {
("instance",): {PersistenceAttribute.WRITE},
("instance", "hello"): {
PersistenceAttribute.WRITE,
PersistenceAttribute.DELETE,
},
("instance", "hello", "static"): {PersistenceAttribute.WRITE},
}
# data_eval: dict = {"foo": {"bar": {}, "baz": {}}}
update: dict = {
"instance": {
# static is a key that is defined in nix, and cannot be deleted
# It needs to be present.
"hello": {} # Error: "static" cannot be deleted
}
}
# data_disk: dict = {}
with pytest.raises(ClanError) as error:
calc_patches(
persisted=persisted,
all_values=all_values,
update=update,
attribute_props=attribute_props,
)
# attribute_props = compute_attribute_persistence(prios, data_eval, data_disk)
assert "Cannot delete path 'instance.hello.static'" in str(error.value)
# 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, attribute_props=attribute_props
# )
# Same as above, but using legacy priorities
# This allows deletion, but is slightly wrong.
# The correct behavior requires nixos >= 25.11
def test_calculate_static_data_no_static_sets_legacy() -> None:
prios = {
"instance": {
"__prio": 100, # <- writeable: "foo"
"hello": {
"__prio": 100,
"static": {"__prio": 100},
},
},
}
# assert "Cannot delete path 'foo.baz'" in str(error.value)
all_values: dict = {
"instance": {
"hello": {
"static": {},
}
}
}
persisted: dict = {"instance": {"hello": {}}}
attribute_props = compute_attribute_persistence(prios, all_values, persisted)
update: dict = {
"instance": {
"hello": {}, # <- user removes "static"
},
}
updates, deletes = calc_patches(
persisted, update, all_values=all_values, attribute_props=attribute_props
)
assert updates == {("instance", "hello"): {}}
assert deletes == {("instance", "hello", "static")}
def test_update_list() -> None:
@@ -526,27 +381,29 @@ def test_dont_persist_defaults() -> None:
def test_set_null() -> None:
data_eval: dict = {
"foo": {},
"bar": {},
"root": {
"foo": {},
"bar": {},
}
}
data_disk = data_eval
# User set Foo to null
# User deleted bar
update = {"foo": None}
update = {"root": {"foo": None}}
patchset, delete_set = calc_patches(
data_disk,
update,
all_values=data_eval,
attribute_props=compute_attribute_persistence(
{"__prio": 100, "foo": {"__prio": 100}, "bar": {"__prio": 100}},
{"root": {"__prio": 100, "foo": {"__prio": 100}, "bar": {"__prio": 100}}},
data_eval,
data_disk,
),
)
assert patchset == {("foo",): None}
assert delete_set == {("bar",)}
assert patchset == {("root", "foo"): None}
assert delete_set == {("root", "bar")}
def test_machine_delete() -> None:
@@ -760,7 +617,7 @@ def test_delete_key_non_writeable() -> None:
data_disk, update, all_values=data_eval, attribute_props=attribute_props
)
assert "Path 'foo' is readonly." in str(error.value)
assert "Cannot delete path 'foo.bar'" in str(error.value)
# --- test merge_objects ---