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

Merge pull request 'api: prepare persistence_attributes' (#5453) from persistence_attributes into main

Browse Source 
Reviewed-on: https://git.clan.lol/clan/clan-core/pulls/5453
This commit is contained in:
hsjobeki
2025-10-13 15:23:03 +00:00
parent 254f9b9c5f 40de60946a
commit d9f6b7e3fb
7 changed files with 309 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
lib/introspection/test.nix
View File

@@ -392,21 +392,22 @@ in
);
};
config.foo = {
a.b = {
bar = 1;
# Statically define "foo.a.c"
# This cannot be deleted
a.c = { };
};
imports = [
{
_file = "inventory.json";
config.foo = {
a.c = {
bar = 1;
};
x.y = {
bar = 1;
};
x.z = {
bar = 1;
};
};
}
];
}
];
in
{
inherit evaluated;
@@ -414,74 +415,34 @@ in
expected = {
foo = {
__this = {
files = [ "<unknown-file>" ];
files = [
"inventory.json"
"<unknown-file>"
];
prio = 100;
total = false;
};
a = {
__this = {
files = [ "<unknown-file>" ];
files = [
"inventory.json"
"<unknown-file>"
];
prio = 100;
total = false;
};
b = {
__this = {
files = [ "<unknown-file>" ];
prio = 100;
total = true;
};
bar = {
__this = {
files = [ "<unknown-file>" ];
prio = 100;
total = false;
};
};
};
c = {
__this = {
files = [ "<unknown-file>" ];
files = [
"inventory.json"
"<unknown-file>"
];
prio = 100;
total = true;
};
bar = {
__this = {
files = [ "<unknown-file>" ];
prio = 100;
total = false;
};
};
};
};
x = {
__this = {
files = [ "<unknown-file>" ];
prio = 100;
total = false;
};
y = {
__this = {
files = [ "<unknown-file>" ];
prio = 100;
total = true;
};
bar = {
__this = {
files = [ "<unknown-file>" ];
prio = 100;
total = false;
};
};
};
z = {
__this = {
files = [ "<unknown-file>" ];
prio = 100;
total = true;
};
bar = {
__this = {
files = [ "<unknown-file>" ];
files = [ "inventory.json" ];
prio = 100;
total = false;
};

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

@@ -20,7 +20,7 @@ from clan_lib.persist.path_utils import (
path_match,
set_value_by_path_tuple,
)
from clan_lib.persist.write_rules import WriteMap, compute_write_map
from clan_lib.persist.write_rules import AttributeMap, compute_attribute_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: WriteMap
writeables: AttributeMap
data_eval: "InventorySnapshot"
data_disk: "InventorySnapshot"
@@ -216,7 +216,7 @@ class InventoryStore:
data_eval: InventorySnapshot = self._load_merged_inventory()
data_disk: InventorySnapshot = self._get_persisted()
write_map = compute_write_map(
write_map = compute_attribute_map(
current_priority,
dict(data_eval),
dict(data_disk),

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

@@ -15,7 +15,7 @@ from clan_lib.persist.validate import (
validate_type_compatibility,
validate_writeability,
)
from clan_lib.persist.write_rules import WriteMap
from clan_lib.persist.write_rules import AttributeMap
def find_deleted_paths_structured(
@@ -88,20 +88,19 @@ def calc_patches(
persisted: dict[str, Any],
update: dict[str, Any],
all_values: dict[str, Any],
writeables: WriteMap,
attribute_props: AttributeMap,
) -> 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.
While validating persistence 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'}}
attribute_props: Persistence attribute map, see: 'compute_attribute_map'
Returns:
Tuple of (SET patches dict, DELETE paths set)
@@ -156,7 +155,7 @@ def calc_patches(
# Validate the change is allowed
validate_no_static_deletion(path, new_value, static_data)
validate_writeability(path, writeables)
validate_writeability(path, attribute_props)
validate_type_compatibility(path, old_value, new_value)
validate_list_uniqueness(path, new_value)

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

@@ -14,7 +14,7 @@ from clan_lib.persist.path_utils import (
set_value_by_path,
set_value_by_path_tuple,
)
from clan_lib.persist.write_rules import compute_write_map
from clan_lib.persist.write_rules import PersistenceAttribute, compute_attribute_map
# --- calculate_static_data ---
@@ -219,11 +219,12 @@ def test_update_simple() -> None:
data_disk: dict = {}
writeables = compute_write_map(prios, data_eval, data_disk)
attribute_props = compute_attribute_map(prios, data_eval, data_disk)
assert writeables == {
"writeable": {("foo",), ("foo", "bar")},
"non_writeable": {("foo", "nix")},
assert attribute_props == {
("foo",): {PersistenceAttribute.WRITE},
("foo", "bar"): {PersistenceAttribute.WRITE, PersistenceAttribute.DELETE},
("foo", "nix"): {PersistenceAttribute.READONLY},
}
update = {
"foo": {
@@ -236,7 +237,7 @@ def test_update_simple() -> None:
data_disk,
update,
all_values=data_eval,
writeables=writeables,
attribute_props=attribute_props,
)
assert patchset == {("foo", "bar"): "new value"}
@@ -255,7 +256,7 @@ def test_update_add_empty_dict() -> None:
data_disk: dict = {}
writeables = compute_write_map(prios, data_eval, data_disk)
writeables = compute_attribute_map(prios, data_eval, data_disk)
update = deepcopy(data_eval)
@@ -265,7 +266,7 @@ def test_update_add_empty_dict() -> None:
data_disk,
update,
all_values=data_eval,
writeables=writeables,
attribute_props=writeables,
)
assert patchset == {("foo", "mimi"): {}} # this is what gets persisted
@@ -296,16 +297,19 @@ def test_update_many() -> None:
data_disk = {"foo": {"bar": "baz", "nested": {"x": "x"}}}
writeables = compute_write_map(prios, data_eval, data_disk)
attribute_props = compute_attribute_map(prios, data_eval, data_disk)
assert writeables == {
"writeable": {
("foo",),
("foo", "bar"),
("foo", "nested"),
("foo", "nested", "x"),
assert attribute_props == {
("foo",): {PersistenceAttribute.WRITE},
("foo", "bar"): {PersistenceAttribute.WRITE, PersistenceAttribute.DELETE},
("foo", "nested"): {PersistenceAttribute.WRITE, PersistenceAttribute.DELETE},
("foo", "nested", "x"): {
PersistenceAttribute.WRITE,
PersistenceAttribute.DELETE,
},
"non_writeable": {("foo", "nix"), ("foo", "nested", "y")},
# Readonly
("foo", "nix"): {PersistenceAttribute.READONLY},
("foo", "nested", "y"): {PersistenceAttribute.READONLY},
}
update = {
@@ -322,7 +326,7 @@ def test_update_many() -> None:
data_disk,
update,
all_values=data_eval,
writeables=writeables,
attribute_props=attribute_props,
)
assert patchset == {
@@ -352,11 +356,11 @@ def test_update_parent_non_writeable() -> None:
},
}
writeables = compute_write_map(prios, data_eval, data_disk)
attribute_props = compute_attribute_map(prios, data_eval, data_disk)
assert writeables == {
"writeable": set(),
"non_writeable": {("foo",), ("foo", "bar")},
assert attribute_props == {
("foo",): {PersistenceAttribute.READONLY},
("foo", "bar"): {PersistenceAttribute.READONLY},
}
update = {
@@ -365,7 +369,9 @@ def test_update_parent_non_writeable() -> None:
},
}
with pytest.raises(ClanError) as error:
calc_patches(data_disk, update, all_values=data_eval, writeables=writeables)
calc_patches(
data_disk, update, all_values=data_eval, attribute_props=attribute_props
)
assert "Path 'foo.bar' is readonly." in str(error.value)
@@ -411,9 +417,11 @@ def test_update_list() -> None:
data_disk = {"foo": ["B"]}
writeables = compute_write_map(prios, data_eval, data_disk)
attribute_props = compute_attribute_map(prios, data_eval, data_disk)
assert writeables == {"writeable": {("foo",)}, "non_writeable": set()}
assert attribute_props == {
("foo",): {PersistenceAttribute.WRITE},
}
# Add "C" to the list
update = {"foo": ["A", "B", "C"]} # User wants to add "C"
@@ -422,7 +430,7 @@ def test_update_list() -> None:
data_disk,
update,
all_values=data_eval,
writeables=writeables,
attribute_props=attribute_props,
)
assert patchset == {("foo",): ["B", "C"]}
@@ -436,7 +444,7 @@ def test_update_list() -> None:
data_disk,
update,
all_values=data_eval,
writeables=writeables,
attribute_props=attribute_props,
)
assert patchset == {("foo",): []}
@@ -456,15 +464,19 @@ def test_update_list_duplicates() -> None:
data_disk = {"foo": ["B"]}
writeables = compute_write_map(prios, data_eval, data_disk)
attribute_props = compute_attribute_map(prios, data_eval, data_disk)
assert writeables == {"writeable": {("foo",)}, "non_writeable": set()}
assert attribute_props == {
("foo",): {PersistenceAttribute.WRITE},
}
# Add "A" to the list
update = {"foo": ["A", "B", "A"]} # User wants to add duplicate "A"
with pytest.raises(ClanError) as error:
calc_patches(data_disk, update, all_values=data_eval, writeables=writeables)
calc_patches(
data_disk, update, all_values=data_eval, attribute_props=attribute_props
)
assert "Path 'foo' contains list duplicates: ['A']" in str(error.value)
@@ -480,10 +492,10 @@ def test_dont_persist_defaults() -> None:
"config": {"foo": "bar"},
}
data_disk: dict[str, Any] = {}
writeables = compute_write_map(prios, data_eval, data_disk)
assert writeables == {
"writeable": {("config",), ("enabled",)},
"non_writeable": set(),
attribute_props = compute_attribute_map(prios, data_eval, data_disk)
assert attribute_props == {
("enabled",): {PersistenceAttribute.WRITE},
("config",): {PersistenceAttribute.WRITE},
}
update = deepcopy(data_eval)
@@ -493,7 +505,7 @@ def test_dont_persist_defaults() -> None:
data_disk,
update,
all_values=data_eval,
writeables=writeables,
attribute_props=attribute_props,
)
assert patchset == {("config", "foo"): "foo"}
assert delete_set == set()
@@ -514,7 +526,7 @@ def test_set_null() -> None:
data_disk,
update,
all_values=data_eval,
writeables=compute_write_map(
attribute_props=compute_attribute_map(
{"__prio": 100, "foo": {"__prio": 100}},
data_eval,
data_disk,
@@ -537,8 +549,10 @@ def test_machine_delete() -> None:
}
data_disk = data_eval
writeables = compute_write_map(prios, data_eval, data_disk)
assert writeables == {"writeable": {("machines",)}, "non_writeable": set()}
attribute_props = compute_attribute_map(prios, data_eval, data_disk)
assert attribute_props == {
("machines",): {PersistenceAttribute.WRITE},
}
# Delete machine "bar" from the inventory
update = deepcopy(data_eval)
@@ -548,7 +562,7 @@ def test_machine_delete() -> None:
data_disk,
update,
all_values=data_eval,
writeables=writeables,
attribute_props=attribute_props,
)
assert patchset == {}
@@ -566,15 +580,19 @@ def test_update_mismatching_update_type() -> None:
data_disk: dict = {}
writeables = compute_write_map(prios, data_eval, data_disk)
attribute_props = compute_attribute_map(prios, data_eval, data_disk)
assert writeables == {"writeable": {("foo",)}, "non_writeable": set()}
assert attribute_props == {
("foo",): {PersistenceAttribute.WRITE},
}
# set foo to an int but it is a list
update: dict = {"foo": 1}
with pytest.raises(ClanError) as error:
calc_patches(data_disk, update, all_values=data_eval, writeables=writeables)
calc_patches(
data_disk, update, all_values=data_eval, attribute_props=attribute_props
)
assert (
"Type mismatch for path 'foo'. Cannot update <class 'list'> with <class 'int'>"
@@ -593,9 +611,11 @@ def test_delete_key() -> None:
data_disk = data_eval
writeables = compute_write_map(prios, data_eval, data_disk)
attribute_props = compute_attribute_map(prios, data_eval, data_disk)
assert writeables == {"writeable": {("foo",)}, "non_writeable": set()}
assert attribute_props == {
("foo",): {PersistenceAttribute.WRITE},
}
# remove all keys from foo
update: dict = {"foo": {}}
@@ -604,7 +624,7 @@ def test_delete_key() -> None:
data_disk,
update,
all_values=data_eval,
writeables=writeables,
attribute_props=attribute_props,
)
assert patchset == {("foo",): {}}
@@ -632,17 +652,18 @@ def test_delete_key_intermediate() -> None:
data_disk = data_eval
writeables = compute_write_map(prios, data_eval, data_disk)
assert writeables == {"writeable": {("foo",)}, "non_writeable": set()}
attribute_props = compute_attribute_map(prios, data_eval, data_disk)
assert attribute_props == {
("foo",): {PersistenceAttribute.WRITE},
}
# remove all keys from foo
patchset, delete_set = calc_patches(
data_disk,
update,
all_values=data_eval,
writeables=writeables,
attribute_props=attribute_props,
)
assert patchset == {}
@@ -666,13 +687,17 @@ def test_delete_key_non_writeable() -> None:
data_disk = data_eval
writeables = compute_write_map(prios, data_eval, data_disk)
attribute_props = compute_attribute_map(prios, data_eval, data_disk)
assert writeables == {"writeable": set(), "non_writeable": {("foo",)}}
assert attribute_props == {
("foo",): {PersistenceAttribute.READONLY},
}
# remove all keys from foo
with pytest.raises(ClanError) as error:
calc_patches(data_disk, update, all_values=data_eval, writeables=writeables)
calc_patches(
data_disk, update, all_values=data_eval, attribute_props=attribute_props
)
assert "Path 'foo' is readonly." in str(error.value)

4
pkgs/clan-cli/clan_lib/persist/validate.py
View File

@@ -7,7 +7,7 @@ from clan_lib.persist.path_utils import (
path_starts_with,
path_to_string,
)
from clan_lib.persist.write_rules import WriteMap, is_writeable_path
from clan_lib.persist.write_rules import AttributeMap, is_writeable_path
def validate_no_static_deletion(
@@ -29,7 +29,7 @@ def validate_no_static_deletion(
raise ClanError(msg)
def validate_writeability(path: PathTuple, writeables: WriteMap) -> None:
def validate_writeability(path: PathTuple, writeables: AttributeMap) -> 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."

94
pkgs/clan-cli/clan_lib/persist/write_rules.py
View File

@@ -1,4 +1,5 @@
from typing import Any, TypedDict
from enum import Enum
from typing import Any
from clan_lib.errors import ClanError
from clan_lib.persist.path_utils import PathTuple, path_to_string
@@ -6,14 +7,18 @@ 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]
class PersistenceAttribute(Enum):
WRITE = "write"
READONLY = "readonly"
DELETE = "delete" # can be deleted
type AttributeMap = dict[PathTuple, set[PersistenceAttribute]]
def is_writeable_path(
key: PathTuple,
writeables: WriteMap,
attributes: AttributeMap,
) -> bool:
"""Recursively check if a key is writeable.
@@ -23,10 +28,12 @@ def is_writeable_path(
remaining = key
while remaining:
current_path = remaining
if current_path in writeables["writeable"]:
if current_path in attributes:
if PersistenceAttribute.WRITE in attributes[current_path]:
return True
if current_path in writeables["non_writeable"]:
if PersistenceAttribute.READONLY in attributes[current_path]:
return False
# Check the parent path
remaining = remaining[:-1]
msg = f"Cannot determine writeability for key '{key}'"
@@ -35,7 +42,7 @@ def is_writeable_path(
def is_writeable_key(
key: str,
writeables: WriteMap,
attributes: AttributeMap,
) -> bool:
"""Recursively check if a key is writeable.
@@ -44,7 +51,7 @@ def is_writeable_key(
In case of ambiguity use is_writeable_path with tuple keys.
"""
items = key.split(".")
return is_writeable_path(tuple(items), writeables)
return is_writeable_path(tuple(items), attributes)
def get_priority(value: Any) -> int | None:
@@ -94,21 +101,43 @@ def is_key_writeable(
return is_mergeable_type(value_in_all) or exists_in_persisted
def _determine_writeability_recursive(
def get_inventory_exclusive(value: dict) -> bool | None:
if "__this" not in value:
return None
definition_locations = value.get("__this", {}).get("files")
if not definition_locations:
return None
return (
len(definition_locations) == 1 and definition_locations[0] == "inventory.json"
)
def get_totality(value: dict) -> bool:
if "__this" not in value:
return False
return value.get("__this", {}).get("total", False)
def _determine_props_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:
parent_redonly: bool = False,
results: AttributeMap | None = None,
parent_total: bool = True,
) -> AttributeMap:
"""Recursively determine writeability for all paths in the priority structure.
This is internal recursive function. Use 'determine_writeability' as entry point.
results: AttributeMap that accumulates results, returned at the end.
"""
if results is None:
results = WriteMap(writeable=set(), non_writeable=set())
results = {}
for key, value in priorities.items():
# Skip metadata keys
@@ -117,6 +146,16 @@ def _determine_writeability_recursive(
path = (*current_path, key)
# If the value is defined only in inventory.json, we might be able to delete it.
# If we don't know (None), decide to allow deletion as well. (Backwards compatibility)
# Unless there is a default that applies instead, when removed. Currently we cannot test that.
# So we assume exclusive values can be removed. In reality we might need to check defaults too. (TODO)
# Total parents prevent deletion of immediate children.
is_inventory_exclusive = get_inventory_exclusive(value)
if not parent_total and (
is_inventory_exclusive or is_inventory_exclusive is None
):
results.setdefault(path, set()).add(PersistenceAttribute.DELETE)
# Determine priority for this key
key_priority = get_priority(value)
effective_priority = (
@@ -125,21 +164,23 @@ def _determine_writeability_recursive(
# Check if this should be non-writeable due to inheritance
force_non_writeable = should_inherit_non_writeable(
effective_priority, parent_non_writeable
effective_priority, parent_redonly
)
if force_non_writeable:
results["non_writeable"].add(path)
results.setdefault(path, set()).clear()
results.setdefault(path, set()).add(PersistenceAttribute.READONLY)
# All children are also non-writeable
if isinstance(value, dict):
_determine_writeability_recursive(
_determine_props_recursive(
value,
all_values.get(key, {}),
{}, # Doesn't matter since all children will be non-writeable
path,
effective_priority,
parent_non_writeable=True,
parent_redonly=True,
results=results,
parent_total=value.get("__this", {}).get("total", False),
)
else:
# Determine writeability based on rules
@@ -151,28 +192,31 @@ def _determine_writeability_recursive(
value_in_all = all_values.get(key)
if is_key_writeable(effective_priority, exists_in_persisted, value_in_all):
results["writeable"].add(path)
# TODO: Distinguish between different write types?
results.setdefault(path, set()).add(PersistenceAttribute.WRITE)
else:
results["non_writeable"].add(path)
results.setdefault(path, set()).clear()
results.setdefault(path, set()).add(PersistenceAttribute.READONLY)
# Recurse into children
if isinstance(value, dict):
_determine_writeability_recursive(
_determine_props_recursive(
value,
all_values.get(key, {}),
persisted.get(key, {}),
path,
effective_priority,
parent_non_writeable=False,
parent_redonly=False,
results=results,
parent_total=get_totality(value),
)
return results
def compute_write_map(
def compute_attribute_map(
priorities: dict[str, Any], all_values: dict[str, Any], persisted: dict[str, Any]
) -> WriteMap:
) -> AttributeMap:
"""Determine writeability for all paths based on priorities and current data.
- Priority-based writeability: Values with priority < 100 are not writeable
@@ -188,4 +232,4 @@ def compute_write_map(
Dict with sets of writeable and non-writeable paths using tuple keys
"""
return _determine_writeability_recursive(priorities, all_values, persisted)
return _determine_props_recursive(priorities, all_values, persisted)

155
pkgs/clan-cli/clan_lib/persist/write_rules_test.py
View File

@@ -5,7 +5,7 @@ import pytest
from clan_lib.flake.flake import Flake
from clan_lib.persist.inventory_store import InventoryStore
from clan_lib.persist.write_rules import compute_write_map
from clan_lib.persist.write_rules import PersistenceAttribute, compute_attribute_map
if TYPE_CHECKING:
from clan_lib.nix_models.clan import Clan
@@ -21,15 +21,16 @@ def test_write_integration(clan_flake: Callable[..., Flake]) -> None:
data_eval = cast("dict", inventory_store.read())
prios = flake.select("clanInternals.inventoryClass.introspection")
res = compute_write_map(prios, data_eval, {})
res = compute_attribute_map(prios, data_eval, {})
# We should be able to write to these top-level keys
assert ("machines",) in res["writeable"]
assert ("instances",) in res["writeable"]
assert ("meta",) in res["writeable"]
assert PersistenceAttribute.WRITE in res[("machines",)]
assert PersistenceAttribute.WRITE in res[("instances",)]
assert PersistenceAttribute.WRITE in res[("meta",)]
# Managed by nix
assert ("assertions",) in res["non_writeable"]
# # Managed by nix
assert PersistenceAttribute.WRITE not in res[("assertions",)]
assert PersistenceAttribute.READONLY in res[("assertions",)]
# New style __this.prio
@@ -48,15 +49,18 @@ def test_write_simple() -> None:
default: dict = {"foo": {}}
data: dict = {}
res = compute_write_map(prios, default, data)
res = compute_attribute_map(prios, default, data)
assert res == {
"writeable": {("foo", "bar"), ("foo",), ("foo.bar",)},
"non_writeable": set(),
("foo",): {PersistenceAttribute.WRITE},
# Can be deleted, because it has a parent.
# The parent doesnt set "total", so we assume its not total.
("foo", "bar"): {PersistenceAttribute.WRITE, PersistenceAttribute.DELETE},
("foo.bar",): {PersistenceAttribute.WRITE},
}
# Compatibility test for old __prio style
# ---- Compatibility tests ---- #
def test_write_inherited() -> None:
@@ -71,11 +75,15 @@ def test_write_inherited() -> None:
}
data: dict = {}
res = compute_write_map(prios, {"foo": {"bar": {}}}, data)
res = compute_attribute_map(prios, {"foo": {"bar": {}}}, data)
assert res == {
"writeable": {("foo",), ("foo", "bar"), ("foo", "bar", "baz")},
"non_writeable": set(),
("foo",): {PersistenceAttribute.WRITE},
("foo", "bar"): {PersistenceAttribute.WRITE, PersistenceAttribute.DELETE},
("foo", "bar", "baz"): {
PersistenceAttribute.WRITE,
PersistenceAttribute.DELETE,
},
}
@@ -91,11 +99,12 @@ def test_non_write_inherited() -> None:
}
data: dict = {}
res = compute_write_map(prios, {}, data)
res = compute_attribute_map(prios, {}, data)
assert res == {
"writeable": set(),
"non_writeable": {("foo",), ("foo", "bar", "baz"), ("foo", "bar")},
("foo",): {PersistenceAttribute.READONLY},
("foo", "bar"): {PersistenceAttribute.READONLY},
("foo", "bar", "baz"): {PersistenceAttribute.READONLY},
}
@@ -113,10 +122,10 @@ def test_write_list() -> None:
"b",
], # <- writeable: because lists are merged. Filtering out nix-values comes later
}
res = compute_write_map(prios, default, data)
res = compute_attribute_map(prios, default, data)
assert res == {
"writeable": {("foo",)},
"non_writeable": set(),
("foo",): {PersistenceAttribute.WRITE},
}
@@ -137,12 +146,18 @@ def test_write_because_written() -> None:
# 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")},
}
res = compute_attribute_map(prios, {"foo": {"bar": {}}}, {})
assert res == {
("foo",): {PersistenceAttribute.WRITE},
("foo", "bar"): {PersistenceAttribute.WRITE, PersistenceAttribute.DELETE},
("foo", "bar", "baz"): {
PersistenceAttribute.READONLY,
},
("foo", "bar", "foobar"): {
PersistenceAttribute.READONLY,
},
}
data: dict = {
"foo": {
"bar": {
@@ -150,8 +165,90 @@ def test_write_because_written() -> None:
},
},
}
res = compute_write_map(prios, {}, data)
assert res == {
"writeable": {("foo",), ("foo", "bar"), ("foo", "bar", "baz")},
"non_writeable": {("foo", "bar", "foobar")},
res = compute_attribute_map(prios, {}, data)
assert res[("foo", "bar", "baz")] == {
PersistenceAttribute.WRITE,
PersistenceAttribute.DELETE,
}
### --- NEW API ---
def test_static_object() -> None:
introspection = {
"foo": {
"__this": {
"files": ["inventory.json", "<unknown-file>"],
"prio": 100,
"total": False,
},
"a": {
"__this": {
"files": ["inventory.json", "<unknown-file>"],
"prio": 100,
"total": False,
},
"c": {
"__this": {
"files": ["inventory.json", "<unknown-file>"],
"prio": 100,
"total": False,
},
"bar": {
"__this": {
"files": ["inventory.json"],
"prio": 100,
"total": False,
}
},
},
},
}
}
data_eval: dict = {"foo": {"a": {"c": {"bar": 1}}}}
persisted: dict = {"foo": {"a": {"c": {"bar": 1}}}}
res = compute_attribute_map(introspection, data_eval, persisted)
assert res == {
# We can extend "foo", "foo.a", "foo.a.c"
# That means the user could define "foo.b"
# But they cannot delete "foo.a" or its static subkeys "foo.a.c"
("foo",): {PersistenceAttribute.WRITE},
("foo", "a"): {PersistenceAttribute.WRITE},
("foo", "a", "c"): {PersistenceAttribute.WRITE},
# We can delete "bar"
("foo", "a", "c", "bar"): {
PersistenceAttribute.DELETE,
PersistenceAttribute.WRITE,
},
}
def test_attributes_totality() -> None:
introspection = {
"foo": {
"__this": {
"files": ["inventory.json"],
"prio": 100,
"total": True,
},
"a": { # Cannot delete "a" because parent is total
"__this": {
"files": ["inventory.json", "<unknown-file>"],
"prio": 100,
"total": False,
},
},
}
}
data_eval: dict = {"foo": {"a": {}}}
persisted: dict = {"foo": {"a": {}}}
res = compute_attribute_map(introspection, data_eval, persisted)
assert res == {
("foo",): {PersistenceAttribute.WRITE},
("foo", "a"): {PersistenceAttribute.WRITE},
}