129 lines
5.4 KiB
Python
129 lines
5.4 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 (
|
|
EvidenceReference,
|
|
InstitutionalReference,
|
|
PartyRepresentation,
|
|
PartySubjectReference,
|
|
ProcedureParty,
|
|
TemporalRevision,
|
|
)
|
|
from govoplan_core.core.temporal import (
|
|
TemporalDataContext,
|
|
bind_temporal_data_context,
|
|
reset_temporal_data_context,
|
|
)
|
|
from govoplan_parties.backend.db.models import ProcedurePartyRevision
|
|
from govoplan_parties.backend.service import (
|
|
PartyStoreError,
|
|
SqlPartyResolver,
|
|
get_procedure_party,
|
|
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"))
|
|
|
|
def test_read_context_can_reconstruct_recorded_state(self) -> None:
|
|
record_procedure_party(self.session, self.principal, party=party())
|
|
revoked = representation(revision="2", revoked_at=NOW + timedelta(hours=1))
|
|
record_procedure_party(
|
|
self.session,
|
|
self.principal,
|
|
party=party(revision="2", representations=(revoked,)),
|
|
expected_revision="1",
|
|
)
|
|
token = bind_temporal_data_context(
|
|
TemporalDataContext(
|
|
validity_mode="at",
|
|
valid_at=NOW + timedelta(hours=1),
|
|
recorded_at=NOW + timedelta(seconds=30),
|
|
)
|
|
)
|
|
try:
|
|
result = get_procedure_party(
|
|
self.session,
|
|
self.principal,
|
|
party_id="representative",
|
|
)
|
|
self.assertEqual("1", result.temporal.revision if result else None)
|
|
finally:
|
|
reset_temporal_data_context(token)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|