281 lines
8.9 KiB
Python
281 lines
8.9 KiB
Python
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()
|