test(integration): harden cross-module API coverage

2026-07-20 20:03:27 +02:00
parent 28afc01371
commit 344fc0077f
2 changed files with 147 additions and 3 deletions
--- 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)
--- 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",