feat: implement procedure party registry

2026-08-01 17:48:37 +02:00
commit 87c705b30e
24 changed files with 1310 additions and 0 deletions
@@ -0,0 +1,92 @@
+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 (
+    EvidenceReference,
+    InstitutionalReference,
+    PartyRepresentation,
+    PartySubjectReference,
+    ProcedureParty,
+    TemporalRevision,
+)
+from govoplan_parties.backend.db.models import ProcedurePartyRevision
+from govoplan_parties.backend.service import PartyStoreError, SqlPartyResolver, record_procedure_party
+
+
+NOW = datetime(2026, 8, 1, 11, 0, tzinfo=UTC)
+
+
+@dataclass
+class Principal:
+    tenant_id: str = "tenant-1"
+    account_id: str = "account-1"
+
+
+def ref(kind: str, object_id: str, owner: str, version: str | None = "1") -> InstitutionalReference:
+    return InstitutionalReference(kind=kind, owner_module=owner, object_id=object_id, tenant_id="tenant-1", version=version, valid_at=NOW)  # type: ignore[arg-type]
+
+
+def representation(*, revision: str = "1", revoked_at: datetime | None = None) -> PartyRepresentation:
+    return PartyRepresentation(
+        representative_party_ref=ref("party", "representative", "parties", "1"),
+        represented_party_ref=ref("party", "applicant", "parties", "1"),
+        power_ref="power-1",
+        permitted_actions=("receive",),
+        temporal=TemporalRevision(revision=revision, valid_from=NOW, recorded_at=NOW + timedelta(minutes=int(revision) - 1), change_reason="Power recorded." if revision == "1" else "Power revoked."),
+        evidence=(EvidenceReference(kind="document", owner_module="files", evidence_id="power.pdf", tenant_id="tenant-1", version=revision),),
+        revoked_at=revoked_at,
+    )
+
+
+def party(*, revision: str = "1", representations: tuple[PartyRepresentation, ...] | None = None) -> ProcedureParty:
+    return ProcedureParty(
+        reference=ref("party", "representative", "parties", revision),
+        procedure_ref=ref("case", "case-1", "cases"),
+        role="representative",
+        subject=PartySubjectReference(kind="identity", provider="identity", subject_id="person-1", tenant_id="tenant-1", version="1"),
+        temporal=TemporalRevision(revision=revision, valid_from=NOW, recorded_at=NOW + timedelta(minutes=int(revision) - 1), change_reason="Party recorded." if revision == "1" else "Representation changed."),
+        permitted_channels=("postbox",),
+        preferred_channels=("postbox",),
+        delivery_recipient=True,
+        representations=representations if representations is not None else (representation(),),
+        contact_snapshot_refs=("addresses:snapshot-1",),
+    )
+
+
+class PartyTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite+pysqlite:///:memory:")
+        ProcedurePartyRevision.__table__.create(self.engine)
+        self.session = Session(self.engine)
+        self.principal = Principal()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def test_resolve_and_explicit_representation_revocation(self) -> None:
+        record_procedure_party(self.session, self.principal, party=party())
+        resolved = SqlPartyResolver().list_procedure_parties(self.session, self.principal, procedure_ref=ref("case", "case-1", "cases"), effective_at=NOW)
+        self.assertEqual(1, len(resolved))
+
+        revoked = representation(revision="2", revoked_at=NOW + timedelta(hours=1))
+        revised = party(revision="2", representations=(revoked,))
+        record_procedure_party(self.session, self.principal, party=revised, expected_revision="1")
+        self.assertEqual("2", SqlPartyResolver().list_procedure_parties(self.session, self.principal, procedure_ref=ref("case", "case-1", "cases"))[0].temporal.revision)
+
+    def test_representation_cannot_disappear_or_cross_tenants(self) -> None:
+        record_procedure_party(self.session, self.principal, party=party())
+        with self.assertRaisesRegex(PartyStoreError, "cannot be removed"):
+            record_procedure_party(self.session, self.principal, party=party(revision="2", representations=()), expected_revision="1")
+        with self.assertRaisesRegex(Exception, "same-tenant"):
+            SqlPartyResolver().list_procedure_parties(self.session, Principal("tenant-2"), procedure_ref=ref("case", "case-1", "cases"))
+
+
+if __name__ == "__main__":
+    unittest.main()