diff --git a/docs/MODULE_ARCHITECTURE.md b/docs/MODULE_ARCHITECTURE.md index b53aa09..519dba2 100644 --- a/docs/MODULE_ARCHITECTURE.md +++ b/docs/MODULE_ARCHITECTURE.md @@ -213,8 +213,23 @@ Other stable runtime capabilities currently include: `calendar.externalProfiles` - `poll.scheduling` - `notifications.dispatch` +- `application_status.projection` +- `payments.requests` - `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 groups, effective-dated identity relationships, and explicit membership decisions. It deliberately does not expose IDM persistence models or imply an @@ -305,6 +320,8 @@ contract checks, are: - `files.access`, `files.campaign_attachments` - `mail.campaign_delivery` - `notifications.dispatch` +- `application_status.projection` +- `payments.requests` - `poll.availability_matrix`, `poll.option_selection`, `poll.response_collection`, `poll.signed_participation`, `poll.workflow_context` diff --git a/src/govoplan_core/core/application_status.py b/src/govoplan_core/core/application_status.py new file mode 100644 index 0000000..f16e519 --- /dev/null +++ b/src/govoplan_core/core/application_status.py @@ -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", +] diff --git a/src/govoplan_core/core/payments.py b/src/govoplan_core/core/payments.py new file mode 100644 index 0000000..0093781 --- /dev/null +++ b/src/govoplan_core/core/payments.py @@ -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", +] diff --git a/tests/test_application_status_contract.py b/tests/test_application_status_contract.py new file mode 100644 index 0000000..1a75876 --- /dev/null +++ b/tests/test_application_status_contract.py @@ -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() diff --git a/tests/test_payments_contract.py b/tests/test_payments_contract.py new file mode 100644 index 0000000..d80fb51 --- /dev/null +++ b/tests/test_payments_contract.py @@ -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()