clan_lib/api: init update service instance

pkgs/clan-cli/clan_lib/services/modules.py

@@ -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 set_value_by_path
+from clan_lib.persist.util import get_value_by_path, set_value_by_path
 
 
 class CategoryInfo(TypedDict):
@@ -431,3 +431,64 @@ def list_service_instances(flake: Flake) -> dict[str, InventoryInstanceInfo]:
             roles=instance.get("roles", {}),
         )
     return res
+
+
+@API.register
+def update_service_instance(
+    flake: Flake, instance_ref: str, roles: InventoryInstanceRolesType
+) -> None:
+    """Update the roles of a service instance
+
+    :param instance_ref: The name of the instance to update
+    :param roles: The roles to update
+
+
+    :raises ClanError: If the instance_ref is invalid or missing required fields
+    """
+    inventory_store = InventoryStore(flake)
+    inventory = inventory_store.read()
+
+    instance: InventoryInstance | None = get_value_by_path(
+        inventory, f"instances.{instance_ref}", None
+    )
+
+    if instance is None:
+        msg = f"Instance '{instance_ref}' not found"
+        raise ClanError(msg)
+
+    module_ref = instance.get("module")
+    if module_ref is None:
+        msg = f"Instance '{instance_ref}' seems invalid: Missing module reference"
+        raise ClanError(msg)
+
+    module = resolve_service_module_ref(flake, module_ref)
+    allowed_roles = module.info.roles.keys()
+
+    for role_name in roles:
+        if role_name not in allowed_roles:
+            msg = f"Role '{role_name}' cannot be used in the module"
+            description = f"Allowed roles: {', '.join(allowed_roles)}"
+            raise ClanError(msg, description=description)
+
+    for role_name, role_cfg in roles.items():
+        if forbidden_keys := {"extraModules"} & role_cfg.keys():
+            msg = f"Role '{role_name}' cannot contain {', '.join(f"'{k}'" for k in forbidden_keys)} directly"
+            raise ClanError(msg)
+
+        static = get_value_by_path(
+            instance, f"roles.{role_name}", {}, expected_type=InventoryInstanceRolesType
+        )
+
+        # override settings, machines only if passed
+        merged = {
+            **static,
+            **role_cfg,
+        }
+
+        set_value_by_path(
+            inventory, f"instances.{instance_ref}.roles.{role_name}", merged
+        )
+
+    inventory_store.write(
+        inventory, message=f"Update service instance '{instance_ref}'"
+    )

pkgs/clan-cli/clan_lib/services/modules_test.py

@@ -3,8 +3,14 @@ from typing import TYPE_CHECKING
 
 import pytest
 from clan_cli.tests.fixtures_flakes import nested_dict
+from clan_lib.errors import ClanError
 from clan_lib.flake.flake import Flake
-from clan_lib.services.modules import list_service_instances, list_service_modules
+from clan_lib.nix_models.clan import Inventory
+from clan_lib.services.modules import (
+    list_service_instances,
+    list_service_modules,
+    update_service_instance,
+)
 
 if TYPE_CHECKING:
     from clan_lib.nix_models.clan import Clan
@@ -103,3 +109,115 @@ def test_list_service_modules(
     assert sshd_service
     assert sshd_service.usage_ref == {"name": "sshd", "input": None}
     assert set(sshd_service.instance_refs) == set({})
+
+
+@pytest.mark.with_core
+def test_update_service_instance(
+    clan_flake: Callable[..., Flake],
+) -> None:
+    # Data that can be mutated via API calls
+    mutable_inventory_json: Inventory = {
+        "instances": {
+            "hello-world": {
+                "roles": {
+                    "morning": {
+                        "machines": {
+                            "jon": {
+                                "settings": {  # type: ignore[typeddict-item]
+                                    "greeting": "jon",
+                                },
+                            },
+                            "sara": {
+                                "settings": {  # type: ignore[typeddict-item]
+                                    "greeting": "sara",
+                                },
+                            },
+                        },
+                        "tags": {
+                            "all": {},
+                        },
+                        "settings": {  # type: ignore[typeddict-item]
+                            "greeting": "hello",
+                        },
+                    }
+                }
+            }
+        }
+    }
+    flake = clan_flake({}, mutable_inventory_json=mutable_inventory_json)
+
+    # Ensure preconditions
+    instances = list_service_instances(flake)
+    assert set(instances.keys()) == {"hello-world"}
+
+    # Wrong instance
+    with pytest.raises(ClanError) as excinfo:
+        update_service_instance(
+            flake,
+            "admin",
+            {},
+        )
+    assert "Instance 'admin' not found" in str(excinfo.value)
+
+    # Wrong roles
+    with pytest.raises(ClanError) as excinfo:
+        update_service_instance(
+            flake,
+            "hello-world",
+            {"default": {"machines": {}}},
+        )
+    assert "Role 'default' cannot be used" in str(excinfo.value)
+
+    # Remove 'settings' from jon machine
+    update_service_instance(
+        flake,
+        "hello-world",
+        {
+            "morning": {
+                "machines": {
+                    "jon": {},  # Unset the machine settings
+                    "sara": {
+                        "settings": {  # type: ignore[typeddict-item]
+                            "greeting": "sara",
+                        },
+                    },
+                },
+            }
+        },
+    )
+
+    updated_instances = list_service_instances(flake)
+    updated_machines = (
+        updated_instances["hello-world"].roles.get("morning", {}).get("machines", {})
+    )
+
+    assert updated_machines == {
+        "jon": {"settings": {}},
+        "sara": {"settings": {"greeting": "sara"}},
+    }
+
+    # Remove jon
+    update_service_instance(
+        flake,
+        "hello-world",
+        {
+            "morning": {
+                "machines": {
+                    "sara": {
+                        "settings": {  # type: ignore[typeddict-item]
+                            "greeting": "sara",
+                        },
+                    },
+                },
+            }
+        },
+    )
+
+    updated_instances = list_service_instances(flake)
+    updated_machines = (
+        updated_instances["hello-world"].roles.get("morning", {}).get("machines", {})
+    )
+
+    assert updated_machines == {
+        "sara": {"settings": {"greeting": "sara"}},
+    }