Refactor module catalog plan construction

src/govoplan_admin/backend/api/v1/routes.py

@@ -486,44 +486,75 @@ def _catalog_plan_item(
     validation: dict[str, object] | None = None,
 ) -> tuple[ModuleInstallPlanItem, dict[str, object]]:
     result = validation or validate_module_package_catalog()
+    _require_valid_module_catalog(result)
+    raw_item = _catalog_install_or_update_item(result, module_id)
+    action = _catalog_plan_action(raw_item, module_id, available_module_ids)
+    license_decision = module_license_decision(_catalog_license_features(raw_item))
+    _require_catalog_action_license(license_decision, action=action, module_id=module_id)
+    return ModuleInstallPlanItem(
+        module_id=str(raw_item["module_id"]),
+        action=action,
+        source="catalog",
+        catalog=_catalog_plan_metadata(result),
+        python_package=_catalog_string(raw_item, "python_package"),
+        python_ref=_catalog_string(raw_item, "python_ref"),
+        webui_package=_catalog_string(raw_item, "webui_package"),
+        webui_ref=_catalog_string(raw_item, "webui_ref"),
+        notes=_catalog_plan_notes(raw_item, license_decision),
+    ), result
+
+
+def _require_valid_module_catalog(result: dict[str, object]) -> None:
     if not result.get("valid"):
         raise HTTPException(
             status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
             detail=str(result.get("error") or "Module package catalog is invalid."),
         )
+
+
+def _catalog_install_or_update_item(result: dict[str, object], module_id: str) -> dict[str, object]:
     for raw_item in result.get("modules", []):
         if not isinstance(raw_item, dict):
             continue
         if raw_item.get("module_id") != module_id or raw_item.get("action") not in {"install", "update"}:
             continue
-        raw_action = raw_item.get("action")
+        return raw_item
-        action = "update" if raw_action == "update" or module_id in available_module_ids else "install"
-        license_decision = module_license_decision(_catalog_license_features(raw_item))
-        if not license_decision.get("allowed"):
-            missing = license_decision.get("missing_features")
-            missing_text = ", ".join(str(item) for item in missing) if isinstance(missing, list) else "required feature"
-            raise HTTPException(
-                status_code=status.HTTP_403_FORBIDDEN,
-                detail=f"License does not allow {action} for {module_id}: {missing_text}.",
-            )
-        notes = raw_item.get("notes") if isinstance(raw_item.get("notes"), str) else None
-        if license_decision.get("reason") and license_decision.get("missing_features"):
-            prefix = f"{notes}\n" if notes else ""
-            notes = f"{prefix}License warning: {license_decision['reason']}"
-        return ModuleInstallPlanItem(
-            module_id=str(raw_item["module_id"]),
-            action=action,
-            source="catalog",
-            catalog=_catalog_plan_metadata(result),
-            python_package=raw_item.get("python_package") if isinstance(raw_item.get("python_package"), str) else None,
-            python_ref=raw_item.get("python_ref") if isinstance(raw_item.get("python_ref"), str) else None,
-            webui_package=raw_item.get("webui_package") if isinstance(raw_item.get("webui_package"), str) else None,
-            webui_ref=raw_item.get("webui_ref") if isinstance(raw_item.get("webui_ref"), str) else None,
-            notes=notes,
-        ), result
     raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Catalog install/update entry not found: {module_id}")
 
 
+def _catalog_plan_action(raw_item: dict[str, object], module_id: str, available_module_ids: set[str]) -> str:
+    return "update" if raw_item.get("action") == "update" or module_id in available_module_ids else "install"
+
+
+def _require_catalog_action_license(
+    license_decision: dict[str, object],
+    *,
+    action: str,
+    module_id: str,
+) -> None:
+    if license_decision.get("allowed"):
+        return
+    missing = license_decision.get("missing_features")
+    missing_text = ", ".join(str(item) for item in missing) if isinstance(missing, list) else "required feature"
+    raise HTTPException(
+        status_code=status.HTTP_403_FORBIDDEN,
+        detail=f"License does not allow {action} for {module_id}: {missing_text}.",
+    )
+
+
+def _catalog_plan_notes(raw_item: dict[str, object], license_decision: dict[str, object]) -> str | None:
+    notes = _catalog_string(raw_item, "notes")
+    if not (license_decision.get("reason") and license_decision.get("missing_features")):
+        return notes
+    prefix = f"{notes}\n" if notes else ""
+    return f"{prefix}License warning: {license_decision['reason']}"
+
+
+def _catalog_string(raw_item: dict[str, object], field: str) -> str | None:
+    value = raw_item.get(field)
+    return value if isinstance(value, str) else None
+
+
 def _catalog_plan_metadata(validation: dict[str, object]) -> dict[str, object]:
     metadata: dict[str, object] = {
         "source": validation.get("source") or validation.get("path"),

tests/__init__.py

@@ -0,0 +1 @@
+"""GovOPlaN Admin backend tests."""

tests/test_catalog_plan.py

@@ -0,0 +1,108 @@
+from __future__ import annotations
+
+import unittest
+from unittest.mock import patch
+
+from fastapi import HTTPException
+
+from govoplan_admin.backend.api.v1.routes import _catalog_plan_item
+
+
+class CatalogPlanItemTests(unittest.TestCase):
+    def test_builds_catalog_item_and_preserves_catalog_metadata(self) -> None:
+        validation: dict[str, object] = {
+            "valid": True,
+            "source": "https://catalog.example.test/modules.json",
+            "source_type": "remote",
+            "channel": "stable",
+            "signed": True,
+            "trusted": True,
+            "modules": [
+                "ignored",
+                {"module_id": "other", "action": "install"},
+                {
+                    "module_id": "calendar",
+                    "action": "install",
+                    "python_package": "govoplan-calendar",
+                    "python_ref": 42,
+                    "webui_package": "@govoplan/calendar-webui",
+                    "notes": "Catalog note",
+                    "license_features": ["calendar.sync", "", 7],
+                },
+            ],
+        }
+        decision = {
+            "allowed": True,
+            "reason": "Feature expires soon.",
+            "missing_features": ["calendar.future"],
+        }
+
+        with patch(
+            "govoplan_admin.backend.api.v1.routes.module_license_decision",
+            return_value=decision,
+        ) as license_decision:
+            item, returned_validation = _catalog_plan_item(
+                "calendar",
+                {"calendar"},
+                validation=validation,
+            )
+
+        self.assertIs(returned_validation, validation)
+        self.assertEqual(item.module_id, "calendar")
+        self.assertEqual(item.action, "update")
+        self.assertEqual(item.source, "catalog")
+        self.assertEqual(item.python_package, "govoplan-calendar")
+        self.assertIsNone(item.python_ref)
+        self.assertEqual(item.webui_package, "@govoplan/calendar-webui")
+        self.assertIsNone(item.webui_ref)
+        self.assertEqual(item.notes, "Catalog note\nLicense warning: Feature expires soon.")
+        self.assertEqual(item.catalog["source"], "https://catalog.example.test/modules.json")
+        self.assertEqual(item.catalog["channel"], "stable")
+        self.assertTrue(item.catalog["signed"])
+        license_decision.assert_called_once_with(["calendar.sync", "7"])
+
+    def test_rejects_catalog_action_when_license_is_missing(self) -> None:
+        validation: dict[str, object] = {
+            "valid": True,
+            "modules": [{"module_id": "calendar", "action": "install", "license_features": ["calendar.sync"]}],
+        }
+        with patch(
+            "govoplan_admin.backend.api.v1.routes.module_license_decision",
+            return_value={"allowed": False, "missing_features": ["calendar.sync", "calendar.write"]},
+        ), self.assertRaises(HTTPException) as raised:
+            _catalog_plan_item("calendar", set(), validation=validation)
+
+        self.assertEqual(raised.exception.status_code, 403)
+        self.assertEqual(
+            raised.exception.detail,
+            "License does not allow install for calendar: calendar.sync, calendar.write.",
+        )
+
+    def test_rejects_invalid_or_missing_catalog_entries(self) -> None:
+        with self.subTest("invalid catalog"), self.assertRaises(HTTPException) as invalid:
+            _catalog_plan_item(
+                "calendar",
+                set(),
+                validation={"valid": False, "error": "Signature is invalid."},
+            )
+        self.assertEqual(invalid.exception.status_code, 422)
+        self.assertEqual(invalid.exception.detail, "Signature is invalid.")
+
+        with self.subTest("entry not found"), self.assertRaises(HTTPException) as missing:
+            _catalog_plan_item(
+                "calendar",
+                set(),
+                validation={
+                    "valid": True,
+                    "modules": [
+                        {"module_id": "calendar", "action": "remove"},
+                        {"module_id": "other", "action": "install"},
+                    ],
+                },
+            )
+        self.assertEqual(missing.exception.status_code, 404)
+        self.assertEqual(missing.exception.detail, "Catalog install/update entry not found: calendar")
+
+
+if __name__ == "__main__":
+    unittest.main()