From 344fc0077f66bfdc656844d37e8a59e15cc12c60 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Mon, 20 Jul 2026 20:03:27 +0200 Subject: [PATCH] test(integration): harden cross-module API coverage --- tests/test_access_contracts.py | 3 +- tests/test_api_smoke.py | 147 ++++++++++++++++++++++++++++++++- 2 files changed, 147 insertions(+), 3 deletions(-) diff --git a/tests/test_access_contracts.py b/tests/test_access_contracts.py index 436c4d8..6a69411 100644 --- a/tests/test_access_contracts.py +++ b/tests/test_access_contracts.py @@ -83,7 +83,6 @@ from govoplan_core.core.organizations import CAPABILITY_ORGANIZATION_DIRECTORY, from govoplan_core.core.registry import PlatformRegistry from govoplan_core.core.runtime import clear_runtime, configure_runtime from govoplan_core.db.base import Base -from govoplan_core.server.app import create_app from govoplan_core.server.config import GovoplanServerConfig from govoplan_core.tenancy.scope import create_scope_tables from govoplan_access.backend.db.models import ( @@ -1382,6 +1381,8 @@ class AccessContractTests(unittest.TestCase): app_configurators=(), ) with temporary_database(f"sqlite:///{root / 'test.db'}") as database: + from govoplan_core.server.app import create_app + app = create_app(config) create_scope_tables(database.engine) diff --git a/tests/test_api_smoke.py b/tests/test_api_smoke.py index f75c130..09bbfa2 100644 --- a/tests/test_api_smoke.py +++ b/tests/test_api_smoke.py @@ -26,10 +26,21 @@ os.environ["FILE_STORAGE_LOCAL_ROOT"] = str(_TEST_ROOT / "files") os.environ["MOCK_MAILBOX_DIR"] = str(_TEST_ROOT / "mock-mailbox") os.environ["DEV_BOOTSTRAP_ENABLED"] = "false" +from govoplan_core.settings import Settings, settings as core_settings + +# unittest discovery imports other contract modules before this smoke module. +# Refresh the shared settings object in place so modules that already imported +# it observe this test's isolated environment before the default app is built. +_isolated_settings = Settings() +for _field_name in Settings.model_fields: + setattr(core_settings, _field_name, getattr(_isolated_settings, _field_name)) + +from alembic import command from fastapi.testclient import TestClient from govoplan_core.db.base import Base from govoplan_core.db.bootstrap import bootstrap_dev_data +from govoplan_core.db.migrations import alembic_config from govoplan_core.db.session import configure_database, set_database from govoplan_core.core.change_sequence import decode_sequence_watermark, prune_sequence_entries from govoplan_core.core.pagination import encode_keyset_cursor, keyset_query_fingerprint @@ -61,6 +72,16 @@ class ApiSmokeTests(unittest.TestCase): Base.metadata.drop_all(bind=engine) create_scope_tables(engine) Base.metadata.create_all(bind=engine) + # This suite builds the current ORM schema directly. Record the matching + # release heads so module-lifecycle tests can safely run Alembic without + # misclassifying the create_all database as a historic dev revision. + if not getattr(type(self), "_migration_heads_stamped", False): + command.stamp( + alembic_config(database_url=os.environ["DATABASE_URL"]), + "heads", + purge=True, + ) + type(self)._migration_heads_stamped = True shutil.rmtree(_TEST_ROOT / "files", ignore_errors=True) shutil.rmtree(_TEST_ROOT / "mock-mailbox", ignore_errors=True) with SessionLocal() as session: @@ -310,6 +331,7 @@ class ApiSmokeTests(unittest.TestCase): roles = self.client.get("/api/v1/admin/system/roles", headers=headers) self.assertEqual(roles.status_code, 200, roles.text) system_admin = next(item for item in roles.json()["roles"] if item["slug"] == "system_admin") + maintenance_operator = next(item for item in roles.json()["roles"] if item["slug"] == "maintenance_operator") created = self.client.post( "/api/v1/admin/system/accounts", headers=headers, @@ -319,7 +341,7 @@ class ApiSmokeTests(unittest.TestCase): "password": "approver-password", "password_reset_required": False, "is_active": True, - "role_ids": [system_admin["id"]], + "role_ids": [system_admin["id"], maintenance_operator["id"]], "memberships": [{ "tenant_id": tenant_id, "is_active": True, @@ -3432,6 +3454,127 @@ class ApiSmokeTests(unittest.TestCase): "202605-010001-90100010-9601741.XLSX", }) + def test_managed_attachment_unlinked_candidates_can_be_linked_on_validation(self) -> None: + headers, login = self._login() + user_id = login["user"]["id"] + campaign_json = { + "version": "1.0", + "campaign": {"id": "managed-unlinked-attachments", "name": "Managed unlinked attachments", "mode": "test"}, + "fields": [{"name": "invoice_number", "type": "string", "required": True}], + "global_values": {}, + "server": { + "smtp": { + "host": "smtp.example.invalid", + "port": 587, + "username": "sender@example.org", + "password": "test-secret", + "security": "starttls", + } + }, + "recipients": { + "from": {"email": "sender@example.org", "name": "Sender", "type": "to"}, + "allow_individual_to": True, + }, + "template": {"subject": "Invoice", "text": "Please see the attached file."}, + "attachments": { + "base_path": "invoices", + "base_paths": [ + { + "id": "personal-invoices", + "name": "My invoices", + "source": f"managed:user:{user_id}", + "path": "invoices", + "allow_individual": True, + } + ], + "global": [], + "allow_individual": True, + }, + "entries": { + "inline": [ + { + "id": "recipient-1", + "to": [{"email": "recipient@example.org", "name": "Recipient", "type": "to"}], + "fields": {"invoice_number": "202605-010002"}, + "attachments": [ + { + "id": "exact-pattern", + "label": "Exact workbook", + "base_path_id": "personal-invoices", + "base_dir": "invoices", + "file_filter": "{{local:invoice_number}}.XLSX", + "required": True, + }, + ], + } + ] + }, + "validation_policy": { + "missing_email": "block", + "template_error": "block", + "missing_required_attachment": "block", + }, + "delivery": {"imap_append_sent": {"enabled": False}}, + } + + created = self.client.post("/api/v1/campaigns", headers=headers, json={"config": campaign_json}) + self.assertEqual(created.status_code, 200, created.text) + campaign_id = created.json()["campaign"]["id"] + version_id = created.json()["version"]["id"] + + uploaded = self.client.post( + "/api/v1/files/upload", + headers=headers, + data={ + "owner_type": "user", + "owner_id": user_id, + "path": "invoices", + }, + files=[("files", ("202605-010002.XLSX", b"unlinked workbook", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"))], + ) + self.assertEqual(uploaded.status_code, 200, uploaded.text) + + preview = self.client.post( + f"/api/v1/campaigns/{campaign_id}/versions/{version_id}/attachments/preview", + headers=headers, + json={"include_unmatched": True, "include_unlinked_candidates": True}, + ) + self.assertEqual(preview.status_code, 200, preview.text) + preview_payload = preview.json() + self.assertEqual(preview_payload["matched_file_count"], 1) + self.assertEqual(preview_payload["unlinked_file_count"], 1) + self.assertEqual(preview_payload["rules"][0]["unlinked_match_count"], 1) + self.assertFalse(preview_payload["rules"][0]["matches"][0]["linked_to_campaign"]) + + dry_run = self.client.post( + f"/api/v1/campaigns/{campaign_id}/versions/{version_id}/attachments/link-matches", + headers=headers, + json={"dry_run": True}, + ) + self.assertEqual(dry_run.status_code, 200, dry_run.text) + self.assertEqual(dry_run.json()["linked_file_count"], 0) + self.assertEqual(len(dry_run.json()["linkable_files"]), 1) + + validated = self.client.post( + f"/api/v1/campaigns/versions/{version_id}/validate", + headers=headers, + json={"check_files": True, "link_unshared_matches": True}, + ) + self.assertEqual(validated.status_code, 200, validated.text) + self.assertTrue(validated.json()["ok"], validated.text) + + linked_preview = self.client.post( + f"/api/v1/campaigns/{campaign_id}/versions/{version_id}/attachments/preview", + headers=headers, + json={"include_unmatched": True, "include_unlinked_candidates": True}, + ) + self.assertEqual(linked_preview.status_code, 200, linked_preview.text) + linked_payload = linked_preview.json() + self.assertEqual(linked_payload["matched_file_count"], 1) + self.assertEqual(linked_payload["linked_file_count"], 1) + self.assertEqual(linked_payload["unlinked_file_count"], 0) + self.assertTrue(linked_payload["rules"][0]["matches"][0]["linked_to_campaign"]) + def test_managed_attachment_send_uses_frozen_build_artifact_after_file_changes(self) -> None: headers, login = self._login() user_id = login["user"]["id"] @@ -5648,7 +5791,7 @@ class ApiSmokeTests(unittest.TestCase): password="reader-password", role_ids=[str(access_role["id"])], ) - other = self._create_user( + self._create_user( owner_headers, email="campaign-other@example.local", password="other-password",