feat: implement committee decision workspace
This commit is contained in:
@@ -0,0 +1,240 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
from govoplan_core.core.institutional import (
|
||||
CAPABILITY_DECISION_REGISTRY,
|
||||
CAPABILITY_MANDATE_RESOLVER,
|
||||
ActorRepresentationReference,
|
||||
DecisionEffectReference,
|
||||
EvidenceReference,
|
||||
InformationGovernanceReference,
|
||||
InstitutionalContextError,
|
||||
InstitutionalReference,
|
||||
LegalBasisReference,
|
||||
MandateDefinition,
|
||||
MandateResolution,
|
||||
TemporalRevision,
|
||||
)
|
||||
from govoplan_committee.backend.decision_path import (
|
||||
CommitteeDecisionPath,
|
||||
CommitteeDecisionProposal,
|
||||
)
|
||||
|
||||
|
||||
NOW = datetime(2026, 8, 1, 10, 0, tzinfo=UTC)
|
||||
|
||||
|
||||
def reference(kind: str, object_id: str) -> InstitutionalReference:
|
||||
return InstitutionalReference(
|
||||
kind=kind, # type: ignore[arg-type]
|
||||
owner_module=(
|
||||
"committee"
|
||||
if kind in {"decision", "record"}
|
||||
else "organizations"
|
||||
if kind in {"organization_unit", "function", "jurisdiction"}
|
||||
else "approvals"
|
||||
if kind == "approval"
|
||||
else "cases"
|
||||
),
|
||||
object_id=object_id,
|
||||
tenant_id="tenant-1",
|
||||
valid_at=NOW,
|
||||
)
|
||||
|
||||
|
||||
def mandate() -> MandateDefinition:
|
||||
return MandateDefinition(
|
||||
reference=InstitutionalReference(
|
||||
kind="mandate",
|
||||
owner_module="committee",
|
||||
object_id="mandate-1",
|
||||
tenant_id="tenant-1",
|
||||
version="7",
|
||||
valid_at=NOW,
|
||||
),
|
||||
temporal=TemporalRevision(
|
||||
revision="7",
|
||||
valid_from=NOW - timedelta(days=30),
|
||||
valid_to=NOW + timedelta(days=30),
|
||||
recorded_at=NOW - timedelta(days=40),
|
||||
),
|
||||
task_types=("committee.formal_decision",),
|
||||
authority_types=("committee.resolution",),
|
||||
organization_unit_refs=(reference("organization_unit", "board-1"),),
|
||||
function_refs=(reference("function", "chair"),),
|
||||
jurisdiction_refs=(reference("jurisdiction", "city-1"),),
|
||||
legal_bases=(legal_basis(),),
|
||||
evidence=(evidence("mandate-evidence"),),
|
||||
)
|
||||
|
||||
|
||||
def legal_basis() -> LegalBasisReference:
|
||||
return LegalBasisReference(
|
||||
kind="statute",
|
||||
authority="Example council",
|
||||
reference="rules:committee:12",
|
||||
version="2026-01",
|
||||
effective_from=NOW - timedelta(days=100),
|
||||
)
|
||||
|
||||
|
||||
def evidence(evidence_id: str) -> EvidenceReference:
|
||||
return EvidenceReference(
|
||||
kind="record",
|
||||
owner_module="committee",
|
||||
evidence_id=evidence_id,
|
||||
tenant_id="tenant-1",
|
||||
version="1",
|
||||
checksum="sha256:example",
|
||||
captured_at=NOW,
|
||||
)
|
||||
|
||||
|
||||
def proposal() -> CommitteeDecisionProposal:
|
||||
return CommitteeDecisionProposal(
|
||||
tenant_id="tenant-1",
|
||||
decision_id="decision-1",
|
||||
revision="1",
|
||||
effective_at=NOW,
|
||||
meeting_ref="meeting-4",
|
||||
agenda_item_ref="item-7",
|
||||
decision_type="committee.resolution",
|
||||
subject_refs=(reference("case", "case-1"),),
|
||||
organization_unit_ref=reference("organization_unit", "board-1"),
|
||||
function_ref=reference("function", "chair"),
|
||||
actor=ActorRepresentationReference(
|
||||
tenant_id="tenant-1",
|
||||
account_id="account-1",
|
||||
identity_id="identity-1",
|
||||
represented_function_ref=reference("function", "chair"),
|
||||
mandate_ref=mandate().reference,
|
||||
),
|
||||
approval_refs=(reference("approval", "vote-approval-1"),),
|
||||
fact_evidence=(evidence("minutes-4"),),
|
||||
legal_bases=(legal_basis(),),
|
||||
operative_result="The proposal is accepted.",
|
||||
reasoning="The submitted evidence satisfies the applicable rule.",
|
||||
case_ref=reference("case", "case-1"),
|
||||
jurisdiction_refs=(reference("jurisdiction", "city-1"),),
|
||||
record_refs=(reference("record", "minutes-4"),),
|
||||
requested_effects=(
|
||||
DecisionEffectReference(
|
||||
effect_key="postbox.notify_parties",
|
||||
state="requested",
|
||||
resource_refs=("postbox:case-1",),
|
||||
),
|
||||
),
|
||||
review_refs=("review:administrative-court",),
|
||||
information_governance=InformationGovernanceReference(
|
||||
classification="restricted",
|
||||
purposes=("formal_decision",),
|
||||
legal_basis_refs=("rules:committee:12@2026-01",),
|
||||
disclosure_state="partly_disclosable",
|
||||
),
|
||||
assurance_level="human_reviewed_automation",
|
||||
automation_preparation_refs=("dataflow:recommendation-1",),
|
||||
)
|
||||
|
||||
|
||||
class FakeDecisionRegistry:
|
||||
def __init__(self) -> None:
|
||||
self.recorded = None
|
||||
|
||||
def get_decision(self, session, principal, *, reference):
|
||||
return self.recorded
|
||||
|
||||
def record_decision(self, session, principal, *, decision, expected_revision=None):
|
||||
self.recorded = decision
|
||||
return decision
|
||||
|
||||
|
||||
class FakeMandateResolver:
|
||||
def __init__(self, resolution: MandateResolution) -> None:
|
||||
self.resolution = resolution
|
||||
self.request = None
|
||||
|
||||
def resolve_mandate(self, session, principal, *, request):
|
||||
self.request = request
|
||||
return self.resolution
|
||||
|
||||
|
||||
class FakeRegistry:
|
||||
def __init__(self, capabilities: dict[str, object]) -> None:
|
||||
self.capabilities = capabilities
|
||||
|
||||
def has_capability(self, name: str) -> bool:
|
||||
return name in self.capabilities
|
||||
|
||||
def capability(self, name: str) -> object:
|
||||
return self.capabilities[name]
|
||||
|
||||
|
||||
class CommitteeDecisionPathTests(unittest.TestCase):
|
||||
def test_decision_reconstructs_authority_approval_evidence_and_effects(self) -> None:
|
||||
registry = FakeDecisionRegistry()
|
||||
resolution = MandateResolution(
|
||||
competent=True,
|
||||
mandates=(mandate(),),
|
||||
explanation="Chair is competent for this agenda item.",
|
||||
)
|
||||
mandate_resolver = FakeMandateResolver(resolution)
|
||||
path = CommitteeDecisionPath(
|
||||
FakeRegistry(
|
||||
{
|
||||
CAPABILITY_MANDATE_RESOLVER: mandate_resolver,
|
||||
CAPABILITY_DECISION_REGISTRY: registry,
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
result = path.decide(None, None, proposal=proposal())
|
||||
payload = result.reconstruction_payload()
|
||||
|
||||
self.assertTrue(result.persisted_by_decision_registry)
|
||||
self.assertIs(registry.recorded, result.decision)
|
||||
self.assertEqual("mandate-1", payload["mandate_ref"]["object_id"])
|
||||
self.assertEqual(
|
||||
"vote-approval-1",
|
||||
payload["decision"]["authority_context"]["approval_refs"][0]["object_id"],
|
||||
)
|
||||
self.assertEqual(
|
||||
"postbox.notify_parties",
|
||||
payload["decision"]["requested_effects"][0]["effect_key"],
|
||||
)
|
||||
self.assertEqual(
|
||||
"The submitted evidence satisfies the applicable rule.",
|
||||
payload["decision"]["reasoning"],
|
||||
)
|
||||
self.assertEqual(
|
||||
"city-1", mandate_resolver.request.jurisdiction_refs[0].object_id
|
||||
)
|
||||
self.assertEqual(
|
||||
"human_reviewed_automation",
|
||||
payload["decision"]["assurance_level"],
|
||||
)
|
||||
self.assertEqual(
|
||||
["dataflow:recommendation-1"],
|
||||
payload["decision"]["automation_preparation_refs"],
|
||||
)
|
||||
|
||||
def test_decision_rejects_ambiguous_or_ineffective_mandate(self) -> None:
|
||||
accepted = mandate()
|
||||
with self.assertRaisesRegex(
|
||||
InstitutionalContextError,
|
||||
"exactly one effective active mandate",
|
||||
):
|
||||
CommitteeDecisionPath().decide(
|
||||
None,
|
||||
None,
|
||||
proposal=proposal(),
|
||||
mandate_resolution=MandateResolution(
|
||||
competent=True,
|
||||
mandates=(accepted, accepted),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user