refactor(persist/util): improve calc_patches

pkgs/clan-cli/clan_lib/persist/util.py

@@ -155,7 +155,7 @@ def calc_patches(
 
     Given its current state and the update to apply.
 
-    Filters out nix-values so it doesnt matter if the anyone sends them.
+    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.
@@ -165,9 +165,9 @@ def calc_patches(
 
     Returns a tuple with the SET and DELETE patches.
     """
-    persisted_flat = flatten_data(persisted)
+    data_all = flatten_data(all_values)
-    update_flat = flatten_data(update)
+    data_all_updated = flatten_data(update)
-    all_values_flat = flatten_data(all_values)
+    data_dyn = flatten_data(persisted)
 
     def is_writeable_key(key: str) -> bool:
         """
@@ -187,49 +187,66 @@ def calc_patches(
         msg = f"Cannot determine writeability for key '{key}'"
         raise ClanError(msg, description="F001")
 
+    all_keys = set(data_all) | set(data_all_updated)
     patchset = {}
-    for update_key, update_data in update_flat.items():
+
-        if not is_writeable_key(update_key):
+    delete_set = find_deleted_paths(all_values, update, parent_key="")
-            if update_data != all_values_flat.get(update_key):
+
-                msg = f"Key '{update_key}' is not writeable."
+    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):
+                msg = f"Key '{key}' is not writeable."
                 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 is_writeable_key(update_key):
+            if old is not None and type(old) is not type(new):
-            prev_value = all_values_flat.get(update_key)
+                if new is None:
-            if prev_value and type(update_data) is not type(prev_value):
+                    # If this is a deleted key, they are handled by 'find_deleted_paths'
-                msg = f"Type mismatch for key '{update_key}'. Cannot update {type(all_values_flat.get(update_key))} with {type(update_data)}"
+                    continue
-                raise ClanError(msg)
 
-            # Handle list separation
+                msg = f"Type mismatch for key '{key}'. Cannot update {type(old)} with {type(new)}"
-            if isinstance(update_data, list):
+                description = f"""
-                duplicates = find_duplicates(update_data)
+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 '{update_key}' contains duplicates: {duplicates}. This not supported yet."
+                    msg = f"Key '{key}' contains duplicates: {duplicates}. This not supported yet."
                     raise ClanError(msg)
                 # List of current values
-                persisted_data = persisted_flat.get(update_key, [])
+                persisted_data = data_dyn.get(key, [])
                 # List including nix values
-                all_list = all_values_flat.get(update_key, [])
+                all_list = data_all.get(key, [])
                 nix_list = unmerge_lists(all_list, persisted_data)
-                if update_data != all_list:
+                if new != all_list:
-                    patchset[update_key] = unmerge_lists(update_data, nix_list)
+                    patchset[key] = unmerge_lists(new, nix_list)
+            else:
+                patchset[key] = new
 
-            elif update_data != persisted_flat.get(update_key, None):
+    # Ensure not inadvertently patching something already marked for deletion
-                patchset[update_key] = update_data
+    conflicts = {key for d in delete_set for key in patchset if key.startswith(d)}
-
+    if conflicts:
-            continue
+        conflict_list = ", ".join(sorted(conflicts))
-
+        msg = (
-        msg = f"Cannot determine writeability for key '{update_key}'"
+            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)
-
-    delete_set = find_deleted_paths(persisted, update)
-
-    for delete_key in delete_set:
-        if not is_writeable_key(delete_key):
-            msg = f"Cannot delete: Key '{delete_key}' is not writeable."
-            raise ClanError(msg)
-
     return patchset, delete_set