from __future__ import annotations from dataclasses import dataclass, replace from datetime import UTC, datetime, timedelta import unittest from sqlalchemy import create_engine from sqlalchemy.orm import Session from govoplan_core.core.events import EventBus, event_bus_context from govoplan_core.core.institutional import ( ActorRepresentationReference, EvidenceReference, InformationGovernanceReference, InstitutionalReference, LegalBasisReference, MandateDefinition, MandateResolution, TemporalRevision, ) from govoplan_core.core.voting import CAPABILITY_VOTING_BALLOTS, VotingResult from govoplan_committee.backend.db.models import ( CommitteeDecisionProjection, CommitteeWorkspaceEvent, CommitteeWorkspaceRevision, ) from govoplan_committee.backend.ballots import ( BallotFinalizationResult, CommitteeBallotFinalizer, ballot_adapter_capability, ) from govoplan_committee.backend.decision_path import ( CommitteeDecisionPath, CommitteeDecisionProposal, ) from govoplan_committee.backend.workspace import ( CommitteeWorkspaceError, CommitteeWorkspaceRecord, SqlCommitteeWorkspace, get_local_decision, list_workspace_objects, record_workspace_object, workspace_history, ) NOW = datetime(2026, 8, 1, 13, 0, tzinfo=UTC) @dataclass class Principal: tenant_id: str = "tenant-1" account_id: str = "account-1" class BallotAdapter: def finalize_ballot(self, session, principal, *, request): return BallotFinalizationResult( provider_id=request.provider_id, provider_ballot_ref=request.provider_ballot_ref, counts={"yes": 3, "no": 1}, cast_count=4, quorum_met=True, receipt_ref="receipt:secret-ballot-1", result_sha256="a" * 64, evidence=(evidence("secret-ballot-result"),), ) class BallotRegistry: def __init__(self) -> None: self.name = ballot_adapter_capability("secure-vote") self.adapter = BallotAdapter() def has_capability(self, name: str) -> bool: return name == self.name def capability(self, name: str) -> object: if name != self.name: raise KeyError(name) return self.adapter class VotingBallots: def create_ballot(self, *args, **kwargs): raise NotImplementedError def get_ballot(self, session, principal, *, ballot_id): return { "id": ballot_id, "revision": 2, "state": "open", "context": {"module": "committee", "resource_id": "vote-voting"}, } def open_ballot(self, *args, **kwargs): raise NotImplementedError def cast_ballot(self, *args, **kwargs): raise NotImplementedError def close_ballot( self, session, principal, *, ballot_id, expected_revision, idempotency_key ): if expected_revision != 2: raise AssertionError("unexpected Voting revision") return VotingResult( ballot_id=ballot_id, revision=3, counts={"yes": 2, "no": 1}, weighted_counts={"yes": 3, "no": 1}, cast_count=3, cast_weight=4, eligible_count=4, eligible_weight=5, quorum_met=True, threshold_met=True, winning_options=("yes",), result_sha256="c" * 64, ) def certify_ballot(self, *args, **kwargs): raise NotImplementedError class VotingRegistry: def __init__(self) -> None: self.provider = VotingBallots() def has_capability(self, name: str) -> bool: return name == CAPABILITY_VOTING_BALLOTS def capability(self, name: str) -> object: if name != CAPABILITY_VOTING_BALLOTS: raise KeyError(name) return self.provider 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 evidence(evidence_id: str) -> EvidenceReference: return EvidenceReference( kind="record", owner_module="records", evidence_id=evidence_id, tenant_id="tenant-1", version="1", captured_at=NOW, ) def workspace_record( kind: str, object_id: str, *, state: str, attributes: dict[str, object], parent_id: str | None = None, revision: int = 1, evidence_refs: tuple[EvidenceReference, ...] = (), ) -> CommitteeWorkspaceRecord: return CommitteeWorkspaceRecord( tenant_id="tenant-1", object_kind=kind, # type: ignore[arg-type] object_id=object_id, revision=revision, state=state, title=f"{kind.replace('_', ' ').title()} {object_id}", parent_id=parent_id, recorded_at=NOW + timedelta(minutes=revision - 1), change_reason="Governed Committee update.", attributes=attributes, evidence=evidence_refs, ) def mandate() -> MandateDefinition: return MandateDefinition( reference=ref("mandate", "mandate-1", "mandates"), temporal=TemporalRevision( revision="1", valid_from=NOW - timedelta(days=1), recorded_at=NOW - timedelta(days=2), ), task_types=("committee.formal_decision",), authority_types=("committee.resolution",), organization_unit_refs=(ref("organization_unit", "board-1", "organizations"),), function_refs=(ref("function", "chair", "organizations"),), jurisdiction_refs=(ref("jurisdiction", "city-1", "organizations"),), legal_bases=(legal_basis(),), evidence=(evidence("mandate-proof"),), ) def legal_basis() -> LegalBasisReference: return LegalBasisReference( kind="statute", authority="Example council", reference="committee-rules:12", version="2026-01", ) def proposal() -> CommitteeDecisionProposal: function_ref = ref("function", "chair", "organizations") return CommitteeDecisionProposal( tenant_id="tenant-1", decision_id="decision-1", revision="1", effective_at=NOW, meeting_ref="meeting-1", agenda_item_ref="agenda-1", decision_type="committee.resolution", subject_refs=(ref("case", "case-1", "cases"),), organization_unit_ref=ref("organization_unit", "board-1", "organizations"), function_ref=function_ref, actor=ActorRepresentationReference( tenant_id="tenant-1", account_id="account-1", identity_id="identity-1", represented_function_ref=function_ref, mandate_ref=mandate().reference, ), approval_refs=(ref("approval", "vote-1", "approvals"),), fact_evidence=(evidence("vote-result-1"),), legal_bases=(legal_basis(),), operative_result="The application is approved.", reasoning="The evidence and vote meet the applicable rules.", jurisdiction_refs=(ref("jurisdiction", "city-1", "organizations"),), information_governance=InformationGovernanceReference( classification="restricted", purposes=("formal_decision",), ), ) class CommitteeWorkspaceTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite+pysqlite:///:memory:") for table in ( CommitteeWorkspaceRevision.__table__, CommitteeWorkspaceEvent.__table__, CommitteeDecisionProjection.__table__, ): table.create(self.engine) self.session = Session(self.engine) self.principal = Principal() def tearDown(self) -> None: self.session.close() self.engine.dispose() def test_full_workspace_lifecycle_and_local_decision_projection(self) -> None: bus = EventBus() events = [] bus.subscribe("*", events.append) with event_bus_context(bus): body = workspace_record( "body", "body-1", state="active", attributes={ "organization_unit_ref": ref( "organization_unit", "board-1", "organizations", ).to_dict(), "function_refs": [ ref("function", "chair", "organizations").to_dict() ], "quorum": {"minimum_count": 3}, }, ) record_workspace_object( self.session, self.principal, record=body, idempotency_key="body-create", ) meeting = workspace_record( "meeting", "meeting-1", state="scheduled", parent_id="body-1", attributes={ "starts_at": NOW.isoformat(), "ends_at": (NOW + timedelta(hours=2)).isoformat(), "location": "Council chamber", }, ) record_workspace_object( self.session, self.principal, record=meeting, idempotency_key="meeting-create", ) agenda = workspace_record( "agenda_item", "agenda-1", state="scheduled", parent_id="meeting-1", attributes={ "position": 1, "subject_refs": [ref("case", "case-1", "cases").to_dict()], }, ) record_workspace_object( self.session, self.principal, record=agenda, idempotency_key="agenda-create", ) deliberating = replace( agenda, revision=2, state="deliberating", recorded_at=NOW + timedelta(minutes=1), ) record_workspace_object( self.session, self.principal, record=deliberating, expected_revision=1, idempotency_key="agenda-deliberating", ) vote = workspace_record( "vote", "vote-1", state="open", parent_id="agenda-1", attributes={ "method": "recorded", "choices": ["yes", "no", "abstain"], "eligible_count": 5, "cast_count": 0, }, ) record_workspace_object( self.session, self.principal, record=vote, idempotency_key="vote-open", ) closed_vote = replace( vote, revision=2, state="closed", recorded_at=NOW + timedelta(minutes=1), attributes={ **dict(vote.attributes), "cast_count": 5, "counts": {"yes": 4, "no": 1, "abstain": 0}, "quorum_met": True, "approval_ref": ref( "approval", "vote-1", "approvals", ).to_dict(), }, evidence=(evidence("vote-result-1"),), ) record_workspace_object( self.session, self.principal, record=closed_vote, expected_revision=1, idempotency_key="vote-close", ) decision = ( CommitteeDecisionPath() .decide( self.session, self.principal, proposal=proposal(), mandate_resolution=MandateResolution( competent=True, mandates=(mandate(),), ), ) .decision ) workspace = SqlCommitteeWorkspace() projected = workspace.record_local_decision( self.session, self.principal, decision=decision, meeting_id="meeting-1", agenda_item_id="agenda-1", ) decided = replace( deliberating, revision=3, state="decided", recorded_at=NOW + timedelta(minutes=2), attributes={ **dict(deliberating.attributes), "decision_ref": projected.reference.to_dict(), }, ) record_workspace_object( self.session, self.principal, record=decided, expected_revision=2, idempotency_key="agenda-decide", ) open_meeting = replace( meeting, revision=2, state="open", recorded_at=NOW + timedelta(minutes=1), ) record_workspace_object( self.session, self.principal, record=open_meeting, expected_revision=1, idempotency_key="meeting-open", ) closed_meeting = replace( open_meeting, revision=3, state="closed", recorded_at=NOW + timedelta(hours=2), ) record_workspace_object( self.session, self.principal, record=closed_meeting, expected_revision=2, idempotency_key="meeting-close", ) minute = workspace_record( "minute", "minute-1", state="accepted", parent_id="meeting-1", attributes={ "content_ref": ref("record", "minutes-1", "records").to_dict(), "approval_ref": ref( "approval", "minutes-ok", "approvals" ).to_dict(), }, evidence_refs=(evidence("minutes-signature"),), ) record_workspace_object( self.session, self.principal, record=minute, idempotency_key="minute-accept", ) self.assertEqual([], events) self.session.commit() self.assertEqual( "decision-1", get_local_decision( self.session, self.principal, decision_id="decision-1", ).reference.object_id, ) agenda_history = workspace_history( self.session, self.principal, object_kind="agenda_item", object_id="agenda-1", ) self.assertEqual([3, 2, 1], [item.revision for item in agenda_history]) meetings, total = list_workspace_objects( self.session, self.principal, object_kind="meeting", states=("closed",), ) self.assertEqual(1, total) self.assertEqual("meeting-1", meetings[0].object_id) self.assertIn("committee.decision.projected", [item.type for item in events]) def test_parent_state_occ_replay_and_tenant_boundaries_fail_closed(self) -> None: body = workspace_record( "body", "body-1", state="active", attributes={ "organization_unit_ref": ref( "organization_unit", "board-1", "organizations", ).to_dict(), }, ) first = record_workspace_object( self.session, self.principal, record=body, idempotency_key="body-1", ) replay = record_workspace_object( self.session, self.principal, record=body, idempotency_key="body-1", ) self.assertEqual(first, replay) with self.assertRaisesRegex(CommitteeWorkspaceError, "idempotency conflict"): record_workspace_object( self.session, self.principal, record=replace(body, title="Changed"), idempotency_key="body-1", ) with self.assertRaisesRegex(CommitteeWorkspaceError, "existing body"): record_workspace_object( self.session, self.principal, record=workspace_record( "meeting", "meeting-orphan", state="scheduled", parent_id="missing", attributes={ "starts_at": NOW.isoformat(), "ends_at": (NOW + timedelta(hours=1)).isoformat(), }, ), idempotency_key="meeting-orphan", ) with self.assertRaisesRegex(CommitteeWorkspaceError, "stale"): record_workspace_object( self.session, self.principal, record=replace(body, revision=2), expected_revision=99, idempotency_key="body-stale", ) with self.assertRaisesRegex(CommitteeWorkspaceError, "cross tenants"): record_workspace_object( self.session, Principal(tenant_id="tenant-2"), record=body, idempotency_key="cross-tenant", ) def test_provider_ballot_finalization_persists_only_aggregate_evidence( self, ) -> None: records = ( workspace_record( "body", "body-ballot", state="active", attributes={ "organization_unit_ref": ref( "organization_unit", "board-1", "organizations", ).to_dict(), "function_refs": [], }, ), workspace_record( "meeting", "meeting-ballot", state="open", parent_id="body-ballot", attributes={ "starts_at": NOW.isoformat(), "ends_at": (NOW + timedelta(hours=2)).isoformat(), }, ), workspace_record( "agenda_item", "agenda-ballot", state="deliberating", parent_id="meeting-ballot", attributes={ "position": 1, "subject_refs": [ref("case", "case-1", "cases").to_dict()], }, ), workspace_record( "vote", "vote-provider", state="open", parent_id="agenda-ballot", attributes={ "method": "secret", "provider_id": "secure-vote", "choices": ["yes", "no"], "eligible_count": 5, "cast_count": 0, }, ), ) for index, item in enumerate(records): record_workspace_object( self.session, self.principal, record=item, idempotency_key=f"ballot-setup-{index}", ) self.session.commit() with self.assertRaisesRegex(CommitteeWorkspaceError, "ballot adapter"): record_workspace_object( self.session, self.principal, record=replace( records[-1], revision=2, state="closed", recorded_at=NOW + timedelta(minutes=4), evidence=(evidence("manual-result"),), attributes={ **records[-1].attributes, "counts": {"yes": 3, "no": 1}, "cast_count": 4, "quorum_met": True, "provider_receipt_ref": "forged-receipt", "provider_result_sha256": "b" * 64, "approval_ref": ref( "approval", "vote-approval", "approvals" ).to_dict(), }, ), expected_revision=1, idempotency_key="ballot-manual-close", ) closed = CommitteeBallotFinalizer(BallotRegistry()).finalize( self.session, self.principal, vote_id="vote-provider", provider_id="secure-vote", provider_ballot_ref="external-ballot-7", approval_ref=ref("approval", "vote-approval", "approvals"), expected_revision=1, recorded_at=NOW + timedelta(minutes=5), change_reason="Imported verified secret ballot aggregate.", idempotency_key="ballot-finalize-1", ) self.session.commit() self.assertEqual("closed", closed.state) self.assertEqual({"yes": 3, "no": 1}, closed.attributes["counts"]) self.assertEqual("a" * 64, closed.attributes["provider_result_sha256"]) self.assertNotIn("ballots", closed.attributes) self.assertEqual("secret-ballot-result", closed.evidence[0].evidence_id) def test_voting_ballot_finalization_projects_only_aggregate_result(self) -> None: records = ( workspace_record( "body", "body-voting", state="active", attributes={ "organization_unit_ref": ref( "organization_unit", "board-1", "organizations" ).to_dict(), "function_refs": [], }, ), workspace_record( "meeting", "meeting-voting", state="open", parent_id="body-voting", attributes={ "starts_at": NOW.isoformat(), "ends_at": (NOW + timedelta(hours=2)).isoformat(), }, ), workspace_record( "agenda_item", "agenda-voting", state="deliberating", parent_id="meeting-voting", attributes={ "position": 1, "subject_refs": [ref("case", "case-1", "cases").to_dict()], }, ), workspace_record( "vote", "vote-voting", state="open", parent_id="agenda-voting", attributes={ "method": "recorded", "voting_ballot_id": "ballot-1", "choices": ["yes", "no"], "eligible_count": 4, "cast_count": 0, }, ), ) for index, item in enumerate(records): record_workspace_object( self.session, self.principal, record=item, idempotency_key=f"voting-setup-{index}", ) self.session.commit() closed = CommitteeBallotFinalizer(VotingRegistry()).finalize_voting_ballot( self.session, self.principal, vote_id="vote-voting", voting_ballot_id="ballot-1", voting_expected_revision=2, approval_ref=ref("approval", "vote-approval", "approvals"), expected_revision=1, recorded_at=NOW + timedelta(minutes=5), change_reason="Closed the governed Voting ballot.", idempotency_key="voting-finalize-1", ) self.session.commit() self.assertEqual("closed", closed.state) self.assertEqual({"yes": 2, "no": 1}, closed.attributes["counts"]) self.assertEqual("c" * 64, closed.attributes["voting_result_sha256"]) self.assertEqual("voting", closed.evidence[0].owner_module) self.assertNotIn("selections", closed.attributes) if __name__ == "__main__": unittest.main()