clan_lib: rename writeability to write_map
This commit is contained in:
@@ -7,7 +7,7 @@ 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.writability import is_writeable_key
|
||||
from clan_lib.persist.write_rules import is_writeable_key
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
@@ -51,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)
|
||||
|
||||
|
||||
@@ -18,7 +18,7 @@ from clan_lib.persist.path_utils import (
|
||||
list_difference,
|
||||
set_value_by_path,
|
||||
)
|
||||
from clan_lib.persist.writability import is_writeable_key
|
||||
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)
|
||||
|
||||
|
||||
@@ -20,7 +20,7 @@ from clan_lib.persist.path_utils import (
|
||||
path_match,
|
||||
set_value_by_path_tuple,
|
||||
)
|
||||
from clan_lib.persist.writability import WriteabilityResult, determine_writeability
|
||||
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: WriteabilityResult
|
||||
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:
|
||||
@@ -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),
|
||||
|
||||
@@ -17,7 +17,7 @@ from clan_lib.persist.validate import (
|
||||
validate_type_compatibility,
|
||||
validate_writeability,
|
||||
)
|
||||
from clan_lib.persist.writability import WriteabilityResult
|
||||
from clan_lib.persist.write_rules import WriteMap
|
||||
|
||||
|
||||
def find_deleted_paths_structured(
|
||||
@@ -90,7 +90,7 @@ def calc_patches(
|
||||
persisted: dict[str, Any],
|
||||
update: dict[str, Any],
|
||||
all_values: dict[str, Any],
|
||||
writeables: WriteabilityResult,
|
||||
writeables: WriteMap,
|
||||
) -> tuple[dict[PathTuple, Any], set[PathTuple]]:
|
||||
"""Calculate the patches to apply to the inventory using structured paths.
|
||||
|
||||
|
||||
@@ -14,7 +14,7 @@ from clan_lib.persist.path_utils import (
|
||||
set_value_by_path,
|
||||
set_value_by_path_tuple,
|
||||
)
|
||||
from clan_lib.persist.writability import determine_writeability
|
||||
from clan_lib.persist.write_rules import compute_write_map
|
||||
|
||||
# --- calculate_static_data ---
|
||||
|
||||
@@ -176,7 +176,7 @@ def test_update_simple() -> 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",), ("foo", "bar")},
|
||||
@@ -212,7 +212,7 @@ 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)
|
||||
|
||||
@@ -253,7 +253,7 @@ 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": {
|
||||
@@ -309,7 +309,7 @@ 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(),
|
||||
@@ -338,7 +338,7 @@ def test_remove_non_writable_attrs() -> None:
|
||||
|
||||
data_disk: dict = {}
|
||||
|
||||
writeables = determine_writeability(prios, data_eval, data_disk)
|
||||
writeables = compute_write_map(prios, data_eval, data_disk)
|
||||
|
||||
update: dict = {
|
||||
"foo": {
|
||||
@@ -367,7 +367,7 @@ 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()}
|
||||
|
||||
@@ -412,7 +412,7 @@ 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()}
|
||||
|
||||
@@ -436,7 +436,7 @@ def test_dont_persist_defaults() -> None:
|
||||
"config": {"foo": "bar"},
|
||||
}
|
||||
data_disk: dict[str, Any] = {}
|
||||
writeables = determine_writeability(prios, data_eval, data_disk)
|
||||
writeables = compute_write_map(prios, data_eval, data_disk)
|
||||
assert writeables == {
|
||||
"writeable": {("config",), ("enabled",)},
|
||||
"non_writeable": set(),
|
||||
@@ -470,7 +470,7 @@ def test_set_null() -> None:
|
||||
data_disk,
|
||||
update,
|
||||
all_values=data_eval,
|
||||
writeables=determine_writeability(
|
||||
writeables=compute_write_map(
|
||||
{"__prio": 100, "foo": {"__prio": 100}},
|
||||
data_eval,
|
||||
data_disk,
|
||||
@@ -493,7 +493,7 @@ def test_machine_delete() -> 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": {("machines",)}, "non_writeable": set()}
|
||||
|
||||
# Delete machine "bar" from the inventory
|
||||
@@ -522,7 +522,7 @@ 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()}
|
||||
|
||||
@@ -549,7 +549,7 @@ 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()}
|
||||
|
||||
@@ -588,7 +588,7 @@ 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()}
|
||||
|
||||
@@ -622,7 +622,7 @@ 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",)}}
|
||||
|
||||
|
||||
@@ -7,7 +7,7 @@ from clan_lib.persist.path_utils import (
|
||||
path_starts_with,
|
||||
path_to_string,
|
||||
)
|
||||
from clan_lib.persist.writability import WriteabilityResult, is_writeable_path
|
||||
from clan_lib.persist.write_rules import WriteMap, 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: WriteabilityResult) -> None:
|
||||
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."
|
||||
|
||||
@@ -6,14 +6,14 @@ from clan_lib.persist.path_utils import PathTuple, path_to_string
|
||||
WRITABLE_PRIORITY_THRESHOLD = 100 # Values below this are not writeable
|
||||
|
||||
|
||||
class WriteabilityResult(TypedDict):
|
||||
class WriteMap(TypedDict):
|
||||
writeable: set[PathTuple]
|
||||
non_writeable: set[PathTuple]
|
||||
|
||||
|
||||
def is_writeable_path(
|
||||
key: PathTuple,
|
||||
writeables: WriteabilityResult,
|
||||
writeables: WriteMap,
|
||||
) -> bool:
|
||||
"""Recursively check if a key is writeable.
|
||||
|
||||
@@ -35,7 +35,7 @@ def is_writeable_path(
|
||||
|
||||
def is_writeable_key(
|
||||
key: str,
|
||||
writeables: WriteabilityResult,
|
||||
writeables: WriteMap,
|
||||
) -> bool:
|
||||
"""Recursively check if a key is writeable.
|
||||
|
||||
@@ -99,14 +99,14 @@ def _determine_writeability_recursive(
|
||||
current_path: PathTuple = (),
|
||||
inherited_priority: int | None = None,
|
||||
parent_non_writeable: bool = False,
|
||||
results: WriteabilityResult | None = None,
|
||||
) -> WriteabilityResult:
|
||||
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 = WriteabilityResult(writeable=set(), non_writeable=set())
|
||||
results = WriteMap(writeable=set(), non_writeable=set())
|
||||
|
||||
for key, value in priorities.items():
|
||||
# Skip metadata keys
|
||||
@@ -168,9 +168,9 @@ def _determine_writeability_recursive(
|
||||
return results
|
||||
|
||||
|
||||
def determine_writeability(
|
||||
def compute_write_map(
|
||||
priorities: dict[str, Any], all_values: dict[str, Any], persisted: dict[str, Any]
|
||||
) -> WriteabilityResult:
|
||||
) -> WriteMap:
|
||||
"""Determine writeability for all paths based on priorities and current data.
|
||||
|
||||
- Priority-based writeability: Values with priority < 100 are not writeable
|
||||
@@ -1,4 +1,4 @@
|
||||
from clan_lib.persist.writability import determine_writeability
|
||||
from clan_lib.persist.write_rules import compute_write_map
|
||||
|
||||
|
||||
def test_write_simple() -> None:
|
||||
@@ -12,7 +12,7 @@ def test_write_simple() -> None:
|
||||
|
||||
default: dict = {"foo": {}}
|
||||
data: dict = {}
|
||||
res = determine_writeability(prios, default, data)
|
||||
res = compute_write_map(prios, default, data)
|
||||
|
||||
assert res == {
|
||||
"writeable": {("foo", "bar"), ("foo",), ("foo.bar",)},
|
||||
@@ -32,7 +32,7 @@ def test_write_inherited() -> None:
|
||||
}
|
||||
|
||||
data: dict = {}
|
||||
res = determine_writeability(prios, {"foo": {"bar": {}}}, data)
|
||||
res = compute_write_map(prios, {"foo": {"bar": {}}}, data)
|
||||
|
||||
assert res == {
|
||||
"writeable": {("foo",), ("foo", "bar"), ("foo", "bar", "baz")},
|
||||
@@ -52,7 +52,7 @@ def test_non_write_inherited() -> None:
|
||||
}
|
||||
|
||||
data: dict = {}
|
||||
res = determine_writeability(prios, {}, data)
|
||||
res = compute_write_map(prios, {}, data)
|
||||
|
||||
assert res == {
|
||||
"writeable": set(),
|
||||
@@ -74,7 +74,7 @@ def test_write_list() -> None:
|
||||
"b",
|
||||
], # <- writeable: because lists are merged. Filtering out nix-values comes later
|
||||
}
|
||||
res = determine_writeability(prios, default, data)
|
||||
res = compute_write_map(prios, default, data)
|
||||
assert res == {
|
||||
"writeable": {("foo",)},
|
||||
"non_writeable": set(),
|
||||
@@ -98,7 +98,7 @@ def test_write_because_written() -> None:
|
||||
|
||||
# Given the following data. {}
|
||||
# Check that the non-writeable paths are correct.
|
||||
res = determine_writeability(prios, {"foo": {"bar": {}}}, {})
|
||||
res = compute_write_map(prios, {"foo": {"bar": {}}}, {})
|
||||
assert res == {
|
||||
"writeable": {("foo",), ("foo", "bar")},
|
||||
"non_writeable": {("foo", "bar", "baz"), ("foo", "bar", "foobar")},
|
||||
@@ -111,7 +111,7 @@ def test_write_because_written() -> None:
|
||||
},
|
||||
},
|
||||
}
|
||||
res = determine_writeability(prios, {}, data)
|
||||
res = compute_write_map(prios, {}, data)
|
||||
assert res == {
|
||||
"writeable": {("foo",), ("foo", "bar"), ("foo", "bar", "baz")},
|
||||
"non_writeable": {("foo", "bar", "foobar")},
|
||||
Reference in New Issue
Block a user