from __future__ import annotations from dataclasses import dataclass from datetime import UTC, datetime, timedelta import unittest from sqlalchemy import create_engine from sqlalchemy.orm import Session from govoplan_core.core.institutional import ( GovernedContextEnvelope, InstitutionalReference, TemporalRevision, ) from govoplan_cases.backend.db.models import ( CaseAccessGrant, CaseIdentity, CaseRecordRevision, CaseStatusDefinition, CaseTimelineEntry, CaseTypeDefinition, ) from govoplan_cases.backend.domain import CaseRecord from govoplan_cases.backend.evidence_links import ( link_case_evidence, linked_source, unlink_case_evidence, ) from govoplan_cases.backend.service import ( CaseStoreError, case_history, case_timeline, create_case, update_case, upsert_case_status, upsert_case_type, ) NOW = datetime(2026, 8, 21, 14, 0, tzinfo=UTC) @dataclass class Principal: tenant_id: str = "tenant-1" account_id: str = "account-1" scopes: frozenset[str] = frozenset() def ref(kind: str, object_id: str, owner: str) -> InstitutionalReference: return InstitutionalReference( kind=kind, # type: ignore[arg-type] owner_module=owner, object_id=object_id, tenant_id="tenant-1", version="1", valid_at=NOW, ) def case_record() -> CaseRecord: case_ref = ref("case", "case-1", "cases") return CaseRecord( reference=case_ref, case_number="PERMIT-2026-0001", case_type_key="permit", status_key="open", title="Resident permit", context=GovernedContextEnvelope( tenant_id="tenant-1", temporal=TemporalRevision( revision="1", valid_from=NOW, recorded_at=NOW, change_reason="Case intake.", ), case_ref=case_ref, ), opened_at=NOW, recorded_at=NOW, change_reason="Case created.", ) def file_reference(**changes: str) -> dict[str, str]: result = { "owner_module": "files", "kind": "file-version", "object_id": "version-1", "tenant_id": "tenant-1", "version": "version-1", "path": "/files?fileId=file-1&versionId=version-1", } result.update(changes) return result class CaseEvidenceLinkTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite+pysqlite:///:memory:") for table in ( CaseStatusDefinition.__table__, CaseTypeDefinition.__table__, CaseIdentity.__table__, CaseRecordRevision.__table__, CaseAccessGrant.__table__, CaseTimelineEntry.__table__, ): table.create(self.engine) self.session = Session(self.engine) self.principal = Principal() upsert_case_status( self.session, self.principal, status_key="open", label="Open", ) upsert_case_type( self.session, self.principal, type_key="permit", label="Permit", initial_status_key="open", ) create_case( self.session, self.principal, record=case_record(), idempotency_key="create-case", ) self.session.commit() def tearDown(self) -> None: self.session.close() self.engine.dispose() def link(self, **reference_changes: str) -> CaseRecord: return link_case_evidence( self.session, self.principal, case_id="case-1", expected_revision=1, reference=file_reference(**reference_changes), recorded_at=NOW + timedelta(minutes=1), change_reason="Linked exact file version.", idempotency_key="link-file-1", ) def test_link_appends_minimized_immutable_revision_and_audit_event(self) -> None: linked = self.link() self.session.commit() self.assertEqual(2, linked.revision) self.assertEqual(1, len(linked.evidence_refs)) evidence = linked.evidence_refs[0] self.assertEqual("files", evidence.owner_module) self.assertEqual("version-1", evidence.version) self.assertTrue(evidence.evidence_id.startswith("linked-")) source = linked_source(evidence) self.assertEqual("file-version", source["kind"]) self.assertEqual("/files?fileId=file-1&versionId=version-1", source["path"]) self.assertNotIn("label", evidence.source_ref or "") self.assertNotIn("content", evidence.source_ref or "") history = case_history( self.session, self.principal, case_id="case-1", ) timeline = case_timeline( self.session, self.principal, case_id="case-1", ) self.assertEqual([2, 1], [item.revision for item in history]) self.assertEqual("cases.case.updated", timeline[0]["event_type"]) self.assertNotIn(evidence.evidence_id, str(timeline[0]["payload"])) def test_link_is_replay_safe_and_rejects_duplicate_or_stale_changes(self) -> None: first = self.link() replay = self.link() self.assertEqual(first.reference, replay.reference) self.session.commit() with self.assertRaisesRegex(CaseStoreError, "already linked"): link_case_evidence( self.session, self.principal, case_id="case-1", expected_revision=2, reference=file_reference(), recorded_at=NOW + timedelta(minutes=2), change_reason="Duplicate link.", idempotency_key="duplicate-link", ) updated = update_case( self.session, self.principal, case_id="case-1", expected_revision=2, changes={"title": "Updated permit"}, recorded_at=NOW + timedelta(minutes=2), change_reason="Updated title.", idempotency_key="update-title", ) self.assertEqual(3, updated.revision) with self.assertRaisesRegex(CaseStoreError, "stale"): link_case_evidence( self.session, self.principal, case_id="case-1", expected_revision=2, reference={ "owner_module": "campaigns", "kind": "campaign", "object_id": "campaign-1", "tenant_id": "tenant-1", "version": "campaign-version-1", "path": "/campaigns/campaign-1", }, recorded_at=NOW + timedelta(minutes=3), change_reason="Stale link.", idempotency_key="stale-link", ) def test_rejects_cross_tenant_unsupported_and_malformed_owner_paths(self) -> None: for changes, message in ( ({"tenant_id": "tenant-2"}, "cross Case tenants"), ({"owner_module": "unknown"}, "not supported"), ({"path": "/files?versionId=version-1"}, "exact authorized"), ({"path": "https://example.invalid/file"}, "local platform"), ): with self.subTest(changes=changes): with self.assertRaisesRegex(CaseStoreError, message): self.link(**changes) def test_unlink_only_removes_cases_owned_links_and_is_replay_safe(self) -> None: linked = self.link() self.session.commit() evidence_id = linked.evidence_refs[0].evidence_id unlinked = unlink_case_evidence( self.session, self.principal, case_id="case-1", evidence_id=evidence_id, expected_revision=2, recorded_at=NOW + timedelta(minutes=2), change_reason="Removed obsolete link.", idempotency_key="unlink-file-1", ) replay = unlink_case_evidence( self.session, self.principal, case_id="case-1", evidence_id=evidence_id, expected_revision=2, recorded_at=NOW + timedelta(minutes=2), change_reason="Removed obsolete link.", idempotency_key="unlink-file-1", ) self.assertEqual(3, unlinked.revision) self.assertEqual(unlinked.reference, replay.reference) self.assertEqual((), unlinked.evidence_refs) with self.assertRaisesRegex(CaseStoreError, "not found"): unlink_case_evidence( self.session, self.principal, case_id="case-1", evidence_id="ordinary-evidence", expected_revision=3, recorded_at=NOW + timedelta(minutes=3), change_reason="Invalid unlink.", idempotency_key="invalid-unlink", ) if __name__ == "__main__": unittest.main()