diff --git a/src/govoplan_admin/backend/api/v1/routes.py b/src/govoplan_admin/backend/api/v1/routes.py index 14f59ca..a7711b1 100644 --- a/src/govoplan_admin/backend/api/v1/routes.py +++ b/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") - 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 + return raw_item 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"), diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..c306c3a --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""GovOPlaN Admin backend tests.""" diff --git a/tests/test_catalog_plan.py b/tests/test_catalog_plan.py new file mode 100644 index 0000000..924d4f6 --- /dev/null +++ b/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()