feat: add status and payment capability contracts

This commit is contained in:
2026-08-19 12:33:34 +02:00
parent 8a925782ab
commit 5a2e99f496
5 changed files with 332 additions and 0 deletions
+17
View File
@@ -213,8 +213,23 @@ Other stable runtime capabilities currently include:
`calendar.externalProfiles` `calendar.externalProfiles`
- `poll.scheduling` - `poll.scheduling`
- `notifications.dispatch` - `notifications.dispatch`
- `application_status.projection`
- `payments.requests`
- `workflow.definitionContributions` and `workflow.runtimeWorker` - `workflow.definitionContributions` and `workflow.runtimeWorker`
`application_status.projection` lets a presentation module resolve the tenant
and display or request access to an owner-supplied, deliberately bounded
applicant-status view. The provider retains policy, authorization, token, and
record ownership; consumers must not query provider tables or enlarge the
projection.
`payments.requests` carries replay-safe payment obligations and evidence-bound
manual reconciliation across module boundaries. Procedure modules identify the
source Case or Workflow in the command and retain the returned payment ID;
Payments remains authoritative for amount, currency, state, transaction
reference, and reconciliation evidence. Ledger, invoice, and external payment
providers remain separate follow-on contracts.
The provider-neutral `idm.relationships` contract carries tenant-scoped typed The provider-neutral `idm.relationships` contract carries tenant-scoped typed
groups, effective-dated identity relationships, and explicit membership groups, effective-dated identity relationships, and explicit membership
decisions. It deliberately does not expose IDM persistence models or imply an decisions. It deliberately does not expose IDM persistence models or imply an
@@ -305,6 +320,8 @@ contract checks, are:
- `files.access`, `files.campaign_attachments` - `files.access`, `files.campaign_attachments`
- `mail.campaign_delivery` - `mail.campaign_delivery`
- `notifications.dispatch` - `notifications.dispatch`
- `application_status.projection`
- `payments.requests`
- `poll.availability_matrix`, `poll.option_selection`, - `poll.availability_matrix`, `poll.option_selection`,
`poll.response_collection`, `poll.signed_participation`, `poll.response_collection`, `poll.signed_participation`,
`poll.workflow_context` `poll.workflow_context`
@@ -0,0 +1,81 @@
from __future__ import annotations
from collections.abc import Mapping
from datetime import datetime
from typing import Protocol, runtime_checkable
CAPABILITY_APPLICATION_STATUS_PROJECTION = "application_status.projection"
@runtime_checkable
class ApplicationStatusProjectionProvider(Protocol):
"""Bounded applicant-status access without exposing the owning module's data."""
def tenant_id_for_tracking_id(
self,
session: object,
*,
tracking_id: str,
) -> str | None:
...
def public_access_challenge(
self,
session: object,
*,
tracking_id: str,
) -> Mapping[str, object]:
...
def get_authenticated_projection(
self,
session: object,
principal: object,
*,
tracking_id: str,
observed_at: datetime,
) -> Mapping[str, object]:
...
def get_public_projection(
self,
session: object,
*,
tracking_id: str,
token: str | None,
observed_at: datetime,
) -> Mapping[str, object]:
...
def request_email_link(
self,
session: object,
*,
tracking_id: str,
email: str,
requested_at: datetime,
) -> bool:
...
def application_status_projection_provider(
registry: object | None,
) -> ApplicationStatusProjectionProvider | None:
if registry is None or not hasattr(registry, "has_capability"):
return None
if not registry.has_capability(CAPABILITY_APPLICATION_STATUS_PROJECTION):
return None
capability = registry.capability(CAPABILITY_APPLICATION_STATUS_PROJECTION)
return (
capability
if isinstance(capability, ApplicationStatusProjectionProvider)
else None
)
__all__ = [
"ApplicationStatusProjectionProvider",
"CAPABILITY_APPLICATION_STATUS_PROJECTION",
"application_status_projection_provider",
]
+87
View File
@@ -0,0 +1,87 @@
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass, field
from datetime import datetime
from typing import Protocol, runtime_checkable
from govoplan_core.core.institutional import EvidenceReference
CAPABILITY_PAYMENT_REQUESTS = "payments.requests"
@dataclass(frozen=True, slots=True)
class PaymentRequestCommand:
tenant_id: str
source_module: str
source_resource_type: str
source_resource_id: str
amount_minor: int
currency: str
subject: str
idempotency_key: str
requested_at: datetime
requested_by_ref: str
due_at: datetime | None = None
context_refs: Mapping[str, str] = field(default_factory=dict)
metadata: Mapping[str, object] = field(default_factory=dict)
@dataclass(frozen=True, slots=True)
class ManualPaymentReconciliationCommand:
tenant_id: str
payment_id: str
amount_minor: int
currency: str
transaction_reference: str
evidence_ref: EvidenceReference
idempotency_key: str
received_at: datetime
recorded_at: datetime
recorded_by_ref: str
metadata: Mapping[str, object] = field(default_factory=dict)
@runtime_checkable
class PaymentRequestProvider(Protocol):
def request_payment(
self,
session: object,
command: PaymentRequestCommand,
) -> Mapping[str, object]:
...
def get_payment(
self,
session: object,
*,
tenant_id: str,
payment_id: str,
) -> Mapping[str, object] | None:
...
def reconcile_manual_payment(
self,
session: object,
command: ManualPaymentReconciliationCommand,
) -> Mapping[str, object]:
...
def payment_request_provider(registry: object | None) -> PaymentRequestProvider | None:
if registry is None or not hasattr(registry, "has_capability"):
return None
if not registry.has_capability(CAPABILITY_PAYMENT_REQUESTS):
return None
capability = registry.capability(CAPABILITY_PAYMENT_REQUESTS)
return capability if isinstance(capability, PaymentRequestProvider) else None
__all__ = [
"CAPABILITY_PAYMENT_REQUESTS",
"ManualPaymentReconciliationCommand",
"PaymentRequestCommand",
"PaymentRequestProvider",
"payment_request_provider",
]
+64
View File
@@ -0,0 +1,64 @@
from datetime import UTC, datetime
import unittest
from govoplan_core.core.application_status import (
CAPABILITY_APPLICATION_STATUS_PROJECTION,
application_status_projection_provider,
)
class _Provider:
def tenant_id_for_tracking_id(self, session, *, tracking_id):
return "tenant-1"
def public_access_challenge(self, session, *, tracking_id):
return {"tracking_id": tracking_id, "mode": "permanent_link"}
def get_authenticated_projection(
self, session, principal, *, tracking_id, observed_at
):
return {"tracking_id": tracking_id, "status": "submitted"}
def get_public_projection(self, session, *, tracking_id, token, observed_at):
return {"tracking_id": tracking_id, "status": "submitted"}
def request_email_link(self, session, *, tracking_id, email, requested_at):
return True
class _Registry:
def __init__(self, provider):
self.provider = provider
def has_capability(self, name):
return name == CAPABILITY_APPLICATION_STATUS_PROJECTION
def capability(self, name):
return self.provider
class ApplicationStatusContractTests(unittest.TestCase):
def test_resolves_only_structurally_complete_provider(self):
provider = _Provider()
resolved = application_status_projection_provider(_Registry(provider))
self.assertIs(provider, resolved)
self.assertEqual(
"tenant-1",
resolved.tenant_id_for_tracking_id(None, tracking_id="tracking-1"),
)
self.assertTrue(
resolved.request_email_link(
None,
tracking_id="tracking-1",
email="resident@example.test",
requested_at=datetime(2026, 8, 19, tzinfo=UTC),
)
)
def test_rejects_incomplete_provider(self):
self.assertIsNone(application_status_projection_provider(_Registry(object())))
if __name__ == "__main__":
unittest.main()
+83
View File
@@ -0,0 +1,83 @@
from datetime import UTC, datetime
import unittest
from govoplan_core.core.institutional import EvidenceReference
from govoplan_core.core.payments import (
CAPABILITY_PAYMENT_REQUESTS,
ManualPaymentReconciliationCommand,
PaymentRequestCommand,
payment_request_provider,
)
class _Provider:
def request_payment(self, session, command):
return {"payment_id": "payment-1", "status": "requested"}
def get_payment(self, session, *, tenant_id, payment_id):
return {"payment_id": payment_id, "status": "requested"}
def reconcile_manual_payment(self, session, command):
return {"payment_id": command.payment_id, "status": "paid"}
class _Registry:
def __init__(self, provider):
self.provider = provider
def has_capability(self, name):
return name == CAPABILITY_PAYMENT_REQUESTS
def capability(self, name):
return self.provider
class PaymentsContractTests(unittest.TestCase):
def test_provider_preserves_request_and_evidence_contracts(self):
provider = payment_request_provider(_Registry(_Provider()))
now = datetime(2026, 8, 19, tzinfo=UTC)
requested = provider.request_payment(
None,
PaymentRequestCommand(
tenant_id="tenant-1",
source_module="cases",
source_resource_type="case",
source_resource_id="case-1",
amount_minor=3000,
currency="EUR",
subject="Resident parking permit fee",
idempotency_key="case-1-fee",
requested_at=now,
requested_by_ref="account:officer-1",
),
)
paid = provider.reconcile_manual_payment(
None,
ManualPaymentReconciliationCommand(
tenant_id="tenant-1",
payment_id=str(requested["payment_id"]),
amount_minor=3000,
currency="EUR",
transaction_reference="BANK-2026-1",
evidence_ref=EvidenceReference(
kind="document",
owner_module="files",
evidence_id="file-1",
tenant_id="tenant-1",
version="1",
checksum="a" * 64,
),
idempotency_key="bank-2026-1",
received_at=now,
recorded_at=now,
recorded_by_ref="account:officer-1",
),
)
self.assertEqual("paid", paid["status"])
def test_rejects_incomplete_provider(self):
self.assertIsNone(payment_request_provider(_Registry(object())))
if __name__ == "__main__":
unittest.main()