from __future__ import annotations import os import shutil import subprocess import sys import tempfile import textwrap import unittest from pathlib import Path from types import SimpleNamespace # Keep the default app import side effect from bootstrapping a development DB. _TEST_ROOT = Path(tempfile.mkdtemp(prefix="govoplan-module-tests-")) os.environ.setdefault("APP_ENV", "test") os.environ.setdefault("DATABASE_URL", f"sqlite:///{_TEST_ROOT / 'module-test.db'}") os.environ.setdefault("DEV_BOOTSTRAP_ENABLED", "false") os.environ.setdefault("CELERY_ENABLED", "false") from fastapi.testclient import TestClient from govoplan_core.core.migrations import migration_metadata_plan from govoplan_core.server.app import create_app from govoplan_core.server.config import GovoplanServerConfig from govoplan_core.server.registry import available_module_manifests, build_platform_registry from govoplan_files.backend.storage.backends import LocalFilesystemStorageBackend, StorageBackendError def _settings(root: Path) -> SimpleNamespace: return SimpleNamespace( app_env="test", database_url=f"sqlite:///{root / 'test.db'}", dev_mailbox_api_enabled=False, file_storage_backend="local", file_storage_local_root=str(root / "files"), file_storage_local_fallback_roots="", file_storage_s3_endpoint_url=None, file_storage_s3_region=None, file_storage_s3_bucket=None, file_storage_s3_access_key_id=None, file_storage_s3_secret_access_key=None, s3_endpoint_url=None, s3_region=None, s3_bucket=None, s3_access_key_id=None, s3_secret_access_key=None, file_upload_max_bytes=1024 * 1024, file_upload_zip_max_bytes=10 * 1024 * 1024, celery_enabled=False, redis_url="redis://127.0.0.1:6379/15", mock_mailbox_dir=str(root / "mock-mailbox"), ) class ModuleSystemTests(unittest.TestCase): @classmethod def tearDownClass(cls) -> None: shutil.rmtree(_TEST_ROOT, ignore_errors=True) def _app_for_modules(self, modules: tuple[str, ...]): root = Path(tempfile.mkdtemp(prefix="govoplan-module-app-", dir=_TEST_ROOT)) settings = _settings(root) config = GovoplanServerConfig( title="GovOPlaN Module Test", version="test", settings=settings, enabled_modules=modules, base_routers=(), post_module_routers=(), extra_routers=(), lifespan=None, app_configurators=(), ) return create_app(config), settings def test_discovers_installed_core_and_product_module_manifests(self) -> None: manifests = available_module_manifests() self.assertTrue({"access", "files", "mail", "campaigns"}.issubset(manifests)) self.assertEqual(manifests["campaigns"].dependencies, ("access",)) self.assertEqual(manifests["campaigns"].optional_dependencies, ("files", "mail")) def test_enabled_module_permutations_register_expected_routes(self) -> None: cases = ( ("core_only", (), {"access"}, set()), ("files_only", ("files",), {"access", "files"}, {"/api/v1/files"}), ("mail_only", ("mail",), {"access", "mail"}, {"/api/v1/mail"}), ("campaign_without_files_or_mail", ("campaigns",), {"access", "campaigns"}, {"/api/v1/campaigns"}), ("campaign_without_files", ("campaigns", "mail"), {"access", "campaigns", "mail"}, {"/api/v1/campaigns", "/api/v1/mail"}), ("campaign_with_files_but_no_mail", ("campaigns", "files"), {"access", "campaigns", "files"}, {"/api/v1/campaigns", "/api/v1/files"}), ("full_product", ("campaigns", "files", "mail"), {"access", "campaigns", "files", "mail"}, {"/api/v1/campaigns", "/api/v1/files", "/api/v1/mail"}), ) for name, enabled_modules, expected_modules, expected_prefixes in cases: with self.subTest(name=name): app, _settings_obj = self._app_for_modules(enabled_modules) route_paths = {getattr(route, "path", "") for route in app.routes} self.assertIn("/api/v1/platform/modules", route_paths) for prefix in ("/api/v1/campaigns", "/api/v1/files", "/api/v1/mail"): has_prefix = any(path == prefix or path.startswith(f"{prefix}/") for path in route_paths) self.assertEqual(prefix in expected_prefixes, has_prefix, f"{name}: {prefix}") with TestClient(app) as client: response = client.get("/api/v1/platform/modules") self.assertEqual(response.status_code, 200, response.text) payload_modules = {item["id"] for item in response.json()["modules"]} self.assertEqual(expected_modules, payload_modules) def _run_physical_absence_probe(self, *, enabled_modules: tuple[str, ...], blocked_modules: tuple[str, ...]) -> None: root = Path(tempfile.mkdtemp(prefix="govoplan-absence-probe-", dir=_TEST_ROOT)) enabled = ",".join(("access", *enabled_modules)) if enabled_modules else "access" script = f""" import importlib.abc import json import os import sys from pathlib import Path os.environ["APP_ENV"] = "test" os.environ["DATABASE_URL"] = {f"sqlite:///{root / 'probe.db'}"!r} os.environ["DEV_BOOTSTRAP_ENABLED"] = "false" os.environ["CELERY_ENABLED"] = "false" os.environ["ENABLED_MODULES"] = {enabled!r} BLOCKED = {blocked_modules!r} class Blocker(importlib.abc.MetaPathFinder): def find_spec(self, fullname, path=None, target=None): for prefix in BLOCKED: if fullname == prefix or fullname.startswith(prefix + "."): raise ModuleNotFoundError(f"No module named {{prefix!r}}", name=prefix) return None sys.meta_path.insert(0, Blocker()) from govoplan_core.server.app import create_app from govoplan_core.server.config import GovoplanServerConfig from govoplan_core.server.registry import build_platform_registry registry = build_platform_registry({enabled_modules!r}) config = GovoplanServerConfig( title="absence probe", version="test", enabled_modules={enabled_modules!r}, base_routers=(), post_module_routers=(), extra_routers=(), lifespan=None, app_configurators=(), ) app = create_app(config) route_paths = [getattr(route, "path", "") for route in app.routes] print(json.dumps({{"modules": [item.id for item in registry.manifests()], "routes": route_paths}}, sort_keys=True)) """ result = subprocess.run( [sys.executable, "-c", script], cwd=str(Path(__file__).resolve().parents[1]), env=os.environ.copy(), text=True, capture_output=True, check=False, ) if result.returncode != 0: self.fail(result.stderr or result.stdout) def test_campaign_optional_integrations_follow_enabled_modules(self) -> None: cases = ( (("campaigns",), False, False), (("campaigns", "mail"), False, True), (("campaigns", "files"), True, False), (("campaigns", "files", "mail"), True, True), ) for enabled_modules, files_enabled, mail_enabled in cases: with self.subTest(enabled_modules=enabled_modules): registry = build_platform_registry(enabled_modules) self.assertTrue(registry.has_module("campaigns")) self.assertEqual(files_enabled, registry.integration_enabled("campaigns", "files")) self.assertEqual(mail_enabled, registry.integration_enabled("campaigns", "mail")) def test_module_permutations_start_when_absent_modules_are_physically_unavailable(self) -> None: cases = ( ((), ("govoplan_files", "govoplan_mail", "govoplan_campaign")), (("files",), ("govoplan_mail", "govoplan_campaign")), (("mail",), ("govoplan_files", "govoplan_campaign")), (("campaigns",), ("govoplan_files", "govoplan_mail")), (("campaigns", "files"), ("govoplan_mail",)), (("campaigns", "mail"), ("govoplan_files",)), (("campaigns", "files", "mail"), ()), ) for enabled_modules, blocked_modules in cases: with self.subTest(enabled_modules=enabled_modules, blocked_modules=blocked_modules): self._run_physical_absence_probe(enabled_modules=enabled_modules, blocked_modules=blocked_modules) def test_module_route_factories_receive_runtime_settings(self) -> None: app, settings = self._app_for_modules(("files", "mail")) self.assertIsNotNone(app) from govoplan_files.backend.runtime import get_settings as get_files_settings from govoplan_mail.backend.runtime import get_settings as get_mail_settings self.assertIs(settings, get_files_settings()) self.assertIs(settings, get_mail_settings()) def test_module_migrations_are_registered_only_for_enabled_modules(self) -> None: core_plan = migration_metadata_plan(build_platform_registry(())) self.assertEqual((), core_plan.script_locations) self.assertEqual(1, len(core_plan.metadata)) files_plan = migration_metadata_plan(build_platform_registry(("files",))) self.assertEqual(1, len(files_plan.script_locations)) self.assertTrue(files_plan.script_locations[0].endswith("govoplan_files/backend/migrations/versions")) full_plan = migration_metadata_plan(build_platform_registry(("campaigns", "files", "mail"))) locations = "\n".join(full_plan.script_locations) self.assertIn("govoplan_campaign/backend/migrations/versions", locations) self.assertIn("govoplan_files/backend/migrations/versions", locations) self.assertIn("govoplan_mail/backend/migrations/versions", locations) def test_local_storage_backend_reads_from_fallback_roots_without_copying(self) -> None: with tempfile.TemporaryDirectory(prefix="govoplan-storage-fallback-") as directory: root = Path(directory) / "primary" fallback = Path(directory) / "fallback" key = "tenant-a/files/demo.txt" (fallback / key).parent.mkdir(parents=True) (fallback / key).write_bytes(b"fallback-data") backend = LocalFilesystemStorageBackend(root=root, fallback_roots=(fallback,)) self.assertTrue(backend.exists(key)) self.assertEqual(b"fallback-data", backend.get_bytes(key)) self.assertEqual([b"fallback", b"-data"], list(backend.iter_bytes(key, chunk_size=8))) self.assertFalse((root / key).exists()) backend.delete(key) self.assertTrue((fallback / key).exists()) self.assertEqual(b"fallback-data", backend.get_bytes(key)) with self.assertRaises(StorageBackendError): backend.get_bytes("../escape.txt") if __name__ == "__main__": unittest.main()