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, _module_package_catalog_item, ) from govoplan_admin.backend.api.v1.schemas import SystemSettingsItem from govoplan_core.core.modules import ModuleInterfaceProvider, ModuleManifest class CatalogPlanItemTests(unittest.TestCase): def test_system_settings_use_german_reference_default(self) -> None: self.assertEqual("de", SystemSettingsItem().default_locale) 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", "artifact_integrity": { "python": { "url": "https://packages.example.test/calendar.whl", "sha256": "a" * 64, } }, "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.artifact_integrity["python"]["sha256"], "a" * 64) 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") with self.subTest("withdrawn entry"), self.assertRaises(HTTPException) as withdrawn: _catalog_plan_item( "calendar", set(), validation={ "valid": True, "modules": [{ "module_id": "calendar", "action": "install", "availability": "withdrawn", "availability_reason": "Superseded after a security review.", }], }, ) self.assertEqual(withdrawn.exception.status_code, 409) self.assertIn("security review", withdrawn.exception.detail) def test_catalog_item_reports_dependency_interface_and_update_compatibility(self) -> None: raw_item: dict[str, object] = { "module_id": "calendar", "name": "Calendar", "version": "0.2.0", "action": "install", "permissions": [{ "scope": "calendar:event:read", "label": "Read events", "description": "Read calendar events.", "category": "Calendar", "level": "tenant", "resource": "event", "action": "read", "deprecated": False, }], "dependencies": ["access"], "current_version_min": "0.1.5", "current_version_max_exclusive": "0.2.0", "requires_interfaces": [{ "name": "files.storage", "version_min": "2.0.0", "optional": False, }], } available = { "calendar": ModuleManifest(id="calendar", name="Calendar", version="0.1.4"), "files": ModuleManifest( id="files", name="Files", version="0.1.0", provides_interfaces=(ModuleInterfaceProvider(name="files.storage", version="1.0.0"),), ), } with patch( "govoplan_admin.backend.api.v1.routes.module_license_decision", return_value={"allowed": True, "enforced": False, "missing_features": []}, ): item = _module_package_catalog_item(raw_item, available, [raw_item]) self.assertFalse(item.compatible) self.assertFalse(item.plan_allowed) self.assertEqual("blocked", item.catalog_state) self.assertTrue(item.update_available) self.assertEqual("calendar:event:read", item.permissions[0].scope) self.assertEqual(3, len(item.compatibility_reasons)) self.assertTrue(any("update window" in reason for reason in item.compatibility_reasons)) self.assertTrue(any("Required module access" in reason for reason in item.compatibility_reasons)) self.assertTrue(any("Required interface files.storage" in reason for reason in item.compatibility_reasons)) def test_catalog_item_accepts_dependencies_and_interfaces_from_same_catalog(self) -> None: raw_item: dict[str, object] = { "module_id": "calendar", "name": "Calendar", "version": "0.2.0", "action": "install", "dependencies": ["access"], "requires_interfaces": [{ "name": "access.directory", "version_min": "1.0.0", "optional": False, }], } provider: dict[str, object] = { "module_id": "access", "availability": "available", "provides_interfaces": [{"name": "access.directory", "version": "1.1.0"}], } with patch( "govoplan_admin.backend.api.v1.routes.module_license_decision", return_value={"allowed": True, "enforced": False, "missing_features": []}, ): item = _module_package_catalog_item(raw_item, {}, [raw_item, provider]) self.assertTrue(item.compatible) self.assertTrue(item.plan_allowed) self.assertEqual("available", item.catalog_state) self.assertEqual([], item.compatibility_reasons) def test_catalog_item_distinguishes_current_release_and_downgrade(self) -> None: available = { "calendar": ModuleManifest(id="calendar", name="Calendar", version="0.2.0"), } license_decision = {"allowed": True, "enforced": False, "missing_features": []} with patch( "govoplan_admin.backend.api.v1.routes.module_license_decision", return_value=license_decision, ): current = _module_package_catalog_item({ "module_id": "calendar", "name": "Calendar", "version": "0.2.0", "action": "install", }, available) refresh = _module_package_catalog_item({ "module_id": "calendar", "name": "Calendar", "version": "0.2.0", "action": "install", "allow_same_version": True, }, available) downgrade = _module_package_catalog_item({ "module_id": "calendar", "name": "Calendar", "version": "0.1.9", "action": "install", }, available) allowed_downgrade = _module_package_catalog_item({ "module_id": "calendar", "name": "Calendar", "version": "0.1.9", "action": "install", "allow_downgrade": True, }, available) self.assertTrue(current.compatible) self.assertFalse(current.plan_allowed) self.assertFalse(current.update_available) self.assertEqual("installed", current.catalog_state) self.assertTrue(refresh.plan_allowed) self.assertFalse(downgrade.compatible) self.assertFalse(downgrade.plan_allowed) self.assertTrue(any("does not allow a downgrade" in reason for reason in downgrade.compatibility_reasons)) self.assertTrue(allowed_downgrade.compatible) self.assertTrue(allowed_downgrade.plan_allowed) self.assertFalse(allowed_downgrade.update_available) if __name__ == "__main__": unittest.main()