fix the manager

manager-solution/manager/config_model.py

@@ -0,0 +1,189 @@
+#!/usr/bin/env python3
+import json
+from typing import Any, Dict, Iterable, Optional
+
+DEFAULTS: Dict[str, Any] = {
+    "container_prefix": "fis-",
+    "environment": {},
+    "volumes": {},
+    "mem_limit": None,
+    "memswap_limit": None,
+    "privileged": False,
+    "tmpfs": {},
+    "dns": [],
+    "extra_hosts": {},
+}
+
+
+def merge_dict(base: Optional[Dict[str, Any]], override: Optional[Dict[str, Any]]) -> Dict[str, Any]:
+    result = dict(base or {})
+    result.update(override or {})
+    return result
+
+
+def resolve_placeholders(value: Any, variables: Dict[str, Any]) -> Any:
+    if isinstance(value, str):
+        return value.format(**variables)
+    if isinstance(value, dict):
+        return {
+            resolve_placeholders(key, variables): resolve_placeholders(item, variables)
+            for key, item in value.items()
+        }
+    if isinstance(value, list):
+        return [resolve_placeholders(item, variables) for item in value]
+    return value
+
+
+def merge_defaults(config: Dict[str, Any], user: Dict[str, Any]) -> Dict[str, Any]:
+    defaults = merge_dict(DEFAULTS, config.get("defaults"))
+    merged = dict(user)
+
+    merged["id"] = str(user["id"])
+    merged["port"] = int(user["port"])
+    variables = {"id": merged["id"], "port": merged["port"]}
+
+    merged["environment"] = merge_dict(
+        resolve_placeholders(defaults.get("environment", {}), variables),
+        resolve_placeholders(user.get("environment"), variables),
+    )
+    merged["environment"].setdefault("CUSTOM_USER", merged["id"])
+    merged["volumes"] = merge_dict(
+        resolve_placeholders(defaults.get("volumes", {}), variables),
+        resolve_placeholders(user.get("volumes"), variables),
+    )
+    merged["extra_hosts"] = merge_dict(
+        resolve_placeholders(defaults.get("extra_hosts", {}), variables),
+        resolve_placeholders(user.get("extra_hosts"), variables),
+    )
+    merged["tmpfs"] = merge_dict(
+        resolve_placeholders(defaults.get("tmpfs", {}), variables),
+        resolve_placeholders(user.get("tmpfs"), variables),
+    )
+    merged["dns"] = resolve_placeholders(user.get("dns", defaults.get("dns", [])), variables)
+    merged["mem_limit"] = user.get("mem_limit", defaults.get("mem_limit"))
+    merged["memswap_limit"] = user.get("memswap_limit", defaults.get("memswap_limit"))
+    merged["privileged"] = user.get("privileged", defaults.get("privileged", False))
+
+    if not merged.get("container_name"):
+        prefix = defaults.get("container_prefix", "fis-")
+        merged["container_name"] = f"{prefix}{merged['id']}"
+
+    if not merged["volumes"]:
+        merged["volumes"] = {f"v-conf-{merged['id']}": "/config"}
+
+    return merged
+
+
+def validate_users(users: Iterable[Dict[str, Any]]) -> None:
+    seen_ids = set()
+    seen_ports = set()
+    seen_names = set()
+
+    for user in users:
+        uid = user["id"]
+        port = user["port"]
+        name = user["container_name"]
+        mount_targets = list((user.get("volumes") or {}).values())
+
+        if uid in seen_ids:
+            raise ValueError(f"Duplicate user id: {uid}")
+        if port in seen_ports:
+            raise ValueError(f"Duplicate port: {port}")
+        if name in seen_names:
+            raise ValueError(f"Duplicate container_name: {name}")
+        if not user["volumes"]:
+            raise ValueError(f"User {uid} must define at least one volume")
+        if len(mount_targets) != len(set(mount_targets)):
+            raise ValueError(f"User {uid} has duplicate volume mount targets")
+        if "PASSWORD" not in user["environment"] or not user["environment"]["PASSWORD"]:
+            raise ValueError(f"User {uid} must define environment.PASSWORD")
+
+        seen_ids.add(uid)
+        seen_ports.add(port)
+        seen_names.add(name)
+
+
+def normalize_config(raw: Dict[str, Any]) -> Dict[str, Any]:
+    raw_users = raw.get("users", [])
+    if not raw_users:
+        raise ValueError("Configuration must contain at least one user")
+
+    users = [merge_defaults(raw, user) for user in raw_users]
+    validate_users(users)
+
+    return {"defaults": merge_dict(DEFAULTS, raw.get("defaults")), "users": users}
+
+
+def compact_config(raw: Dict[str, Any]) -> Dict[str, Any]:
+    normalized = normalize_config(raw)
+    defaults = merge_dict(DEFAULTS, raw.get("defaults"))
+    compact_users = []
+
+    for user in normalized["users"]:
+        user_id = user["id"]
+        variables = {"id": user_id, "port": user["port"]}
+        resolved_default_env = resolve_placeholders(defaults.get("environment", {}), variables)
+        resolved_default_volumes = resolve_placeholders(defaults.get("volumes", {}), variables)
+        resolved_default_extra_hosts = resolve_placeholders(defaults.get("extra_hosts", {}), variables)
+        resolved_default_tmpfs = resolve_placeholders(defaults.get("tmpfs", {}), variables)
+        resolved_default_dns = resolve_placeholders(defaults.get("dns", []), variables)
+        default_container_name = f"{defaults.get('container_prefix', 'fis-')}{user_id}"
+
+        compact_user: Dict[str, Any] = {
+            "id": user_id,
+            "port": user["port"],
+        }
+
+        if user.get("container_name") and user["container_name"] != default_container_name:
+            compact_user["container_name"] = user["container_name"]
+
+        environment = {
+            key: value
+            for key, value in (user.get("environment") or {}).items()
+            if resolved_default_env.get(key) != value and not (key == "CUSTOM_USER" and value == user_id)
+        }
+        if environment:
+            compact_user["environment"] = environment
+
+        volumes = {
+            key: value
+            for key, value in (user.get("volumes") or {}).items()
+            if resolved_default_volumes.get(key) != value
+        }
+        if volumes:
+            compact_user["volumes"] = volumes
+
+        extra_hosts = {
+            key: value
+            for key, value in (user.get("extra_hosts") or {}).items()
+            if resolved_default_extra_hosts.get(key) != value
+        }
+        if extra_hosts:
+            compact_user["extra_hosts"] = extra_hosts
+
+        tmpfs = {
+            key: value
+            for key, value in (user.get("tmpfs") or {}).items()
+            if resolved_default_tmpfs.get(key) != value
+        }
+        if tmpfs:
+            compact_user["tmpfs"] = tmpfs
+
+        if user.get("dns") != resolved_default_dns and user.get("dns") is not None:
+            compact_user["dns"] = user["dns"]
+        if user.get("mem_limit") != defaults.get("mem_limit"):
+            compact_user["mem_limit"] = user.get("mem_limit")
+        if user.get("memswap_limit") != defaults.get("memswap_limit"):
+            compact_user["memswap_limit"] = user.get("memswap_limit")
+        if user.get("privileged") != defaults.get("privileged", False):
+            compact_user["privileged"] = user.get("privileged", False)
+
+        compact_users.append(compact_user)
+
+    return {"defaults": defaults, "users": compact_users}
+
+
+def load_config(path: str) -> Dict[str, Any]:
+    with open(path, encoding="utf-8") as handle:
+        raw = json.load(handle)
+    return normalize_config(raw)