4 Commits
Author SHA1 Message Date
zemion 2340bf53f9 docs(payments): add complete German payment guidance
Module Package Release / publish-packages (push) Successful in 11s
2026-08-23 01:54:19 +02:00
zemion e08bc8b992 feat(payments): add governed DSAR coverage 2026-08-21 04:26:14 +02:00
zemion da0ee0f325 refactor(webui): adopt semantic page actions 2026-08-19 18:47:45 +02:00
zemion 884d068689 Adopt semantic payments action layout 2026-08-19 14:26:26 +02:00
9 changed files with 1222 additions and 10 deletions
+2 -2
View File
@@ -4,12 +4,12 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "govoplan-payments" name = "govoplan-payments"
version = "0.1.20" version = "0.1.21"
description = "Replay-safe payment obligations and reconciliation evidence for GovOPlaN." description = "Replay-safe payment obligations and reconciliation evidence for GovOPlaN."
readme = "README.md" readme = "README.md"
requires-python = ">=3.12" requires-python = ">=3.12"
authors = [{ name = "GovOPlaN" }] authors = [{ name = "GovOPlaN" }]
dependencies = ["govoplan-core>=0.1.18"] dependencies = ["govoplan-core>=0.1.37"]
[tool.setuptools.packages.find] [tool.setuptools.packages.find]
where = ["src"] where = ["src"]
+1 -1
View File
@@ -1,3 +1,3 @@
"""GovOPlaN Payments module.""" """GovOPlaN Payments module."""
__version__ = "0.1.20" __version__ = "0.1.21"
@@ -0,0 +1,550 @@
from __future__ import annotations
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from datetime import datetime, timezone
from sqlalchemy.orm import Session
from govoplan_core.core.dsar import (
DsarErasureActionRef,
DsarExecutionResultRef,
DsarRecordRef,
DsarSubjectRef,
dsar_capability_name,
)
from govoplan_payments.backend.db.models import (
PaymentEvent,
PaymentObligation,
PaymentReconciliation,
)
PAYMENTS_DSAR_CAPABILITY = dsar_capability_name("payments")
_MAX_RECORDS = 5_000
_MAX_CHILD_RECORDS = 1_000
_CONFLICT = object()
_ATTRIBUTION_TYPES = frozenset(
{
"payment_request_attribution",
"payment_reconciliation_attribution",
"payment_event_attribution",
}
)
@dataclass(frozen=True, slots=True)
class _SubjectSelectors:
actor_refs: tuple[str, ...]
payment_row_id: str | None
payment_id: str | None
payment_reference: str | None
@property
def has_payment_selector(self) -> bool:
return bool(self.payment_row_id or self.payment_id or self.payment_reference)
class PaymentsDsarProvider:
provider_id = "payments"
module_id = "payments"
def search_subject(
self,
session: object,
*,
tenant_id: str,
subject: DsarSubjectRef,
) -> Sequence[DsarRecordRef]:
db = _session(session)
selectors = _subject_selectors(subject)
if selectors is None:
return ()
if selectors.has_payment_selector:
query = db.query(PaymentObligation).filter(
PaymentObligation.tenant_id == tenant_id
)
if selectors.payment_row_id:
query = query.filter(PaymentObligation.id == selectors.payment_row_id)
if selectors.payment_id:
query = query.filter(
PaymentObligation.payment_id == selectors.payment_id
)
if selectors.payment_reference:
query = query.filter(
PaymentObligation.payment_reference == selectors.payment_reference
)
rows = (
query.order_by(
PaymentObligation.requested_at,
PaymentObligation.id,
)
.limit(_MAX_RECORDS + 1)
.all()
)
if len(rows) > _MAX_RECORDS:
raise ValueError(
"Payments DSAR result limit exceeded; narrow the identifiers."
)
return tuple(_payment_record(db, row) for row in rows)
if not selectors.actor_refs:
return ()
records: list[DsarRecordRef] = []
obligations = db.query(PaymentObligation).filter(
PaymentObligation.tenant_id == tenant_id,
PaymentObligation.requested_by_ref.in_(selectors.actor_refs),
)
records.extend(
_request_attribution(row)
for row in _limited(
obligations,
PaymentObligation,
"request attribution",
)
)
reconciliations = db.query(PaymentReconciliation).filter(
PaymentReconciliation.tenant_id == tenant_id,
PaymentReconciliation.recorded_by_ref.in_(selectors.actor_refs),
)
records.extend(
_reconciliation_attribution(row)
for row in _limited(
reconciliations,
PaymentReconciliation,
"reconciliation attribution",
)
)
events = db.query(PaymentEvent).filter(
PaymentEvent.tenant_id == tenant_id,
PaymentEvent.actor_ref.in_(selectors.actor_refs),
)
records.extend(
_event_attribution(row)
for row in _limited(events, PaymentEvent, "event attribution")
)
if len(records) > _MAX_RECORDS:
raise ValueError(
"Payments DSAR combined result limit exceeded; narrow the selectors."
)
order = {
"payment_request_attribution": 10,
"payment_reconciliation_attribution": 20,
"payment_event_attribution": 30,
}
return tuple(
sorted(
records,
key=lambda item: (order[item.resource_type], item.resource_id),
)
)
def plan_erasure(
self,
session: object,
*,
tenant_id: str,
subject: DsarSubjectRef,
records: Sequence[DsarRecordRef],
) -> Sequence[DsarErasureActionRef]:
del tenant_id
_session(session)
if _subject_selectors(subject) is None:
raise ValueError("Payments DSAR subject selectors conflict.")
actions: list[DsarErasureActionRef] = []
for record in records:
_validate_record(record)
actions.append(
DsarErasureActionRef(
action_id=(
f"payments:retain:{record.resource_type}:{record.resource_id}"
),
provider_id=self.provider_id,
module_id=self.module_id,
kind="retain",
resource_type=record.resource_type,
resource_id=record.resource_id,
title=f"Retain {record.title}",
rationale=record.retention_reason
or "Financial and reconciliation evidence must be retained.",
executable=False,
)
)
return tuple(actions)
def execute_erasure(
self,
session: object,
*,
tenant_id: str,
subject: DsarSubjectRef,
actions: Sequence[DsarErasureActionRef],
request_id: str,
) -> Sequence[DsarExecutionResultRef]:
del tenant_id
_session(session)
if _subject_selectors(subject) is None:
raise ValueError("Payments DSAR subject selectors conflict.")
results: list[DsarExecutionResultRef] = []
for action in actions:
_validate_action(action)
if action.executable or action.kind != "retain":
raise ValueError("Payments DSAR publishes retain-only actions.")
results.append(
DsarExecutionResultRef(
action_id=action.action_id,
status="blocked",
summary=(
"Payment and reconciliation evidence remains under the "
"configured financial, statutory, and legal-hold policy."
),
evidence={"request_id": request_id},
)
)
return tuple(results)
def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None:
references = subject.external_references
values = {
"account_id": _coalesce(
subject.account_id,
references.get("payments.account"),
references.get("access.account"),
),
"membership_id": _coalesce(
subject.membership_id,
references.get("payments.membership"),
references.get("tenancy.membership"),
),
"identity_id": _coalesce(
subject.identity_id,
references.get("payments.identity"),
references.get("identity.id"),
),
"actor_ref": _coalesce(
references.get("payments.actor"),
references.get("payments.operator"),
),
"payment_row_id": _coalesce(
references.get("payments.obligation"),
references.get("payments.row"),
),
"payment_id": _coalesce(
references.get("payments.payment"),
references.get("payments.payment_id"),
),
"payment_reference": _coalesce(
references.get("payments.reference"),
references.get("payments.payment_reference"),
),
}
if any(value is _CONFLICT for value in values.values()):
return None
account_id = _optional_string(values["account_id"])
membership_id = _optional_string(values["membership_id"])
identity_id = _optional_string(values["identity_id"])
direct_actor = _optional_string(values["actor_ref"])
actor_refs = tuple(
dict.fromkeys(
value
for value in (
account_id,
f"account:{account_id}" if account_id else None,
membership_id,
f"membership:{membership_id}" if membership_id else None,
identity_id,
f"identity:{identity_id}" if identity_id else None,
direct_actor,
)
if value
)
)
selectors = _SubjectSelectors(
actor_refs=actor_refs,
payment_row_id=_optional_string(values["payment_row_id"]),
payment_id=_optional_string(values["payment_id"]),
payment_reference=_optional_string(values["payment_reference"]),
)
if not selectors.actor_refs and not selectors.has_payment_selector:
return None
return selectors
def _coalesce(*values: str | None) -> str | None | object:
normalized = {str(value).strip() for value in values if str(value or "").strip()}
if len(normalized) > 1:
return _CONFLICT
return next(iter(normalized), None)
def _optional_string(value: object) -> str | None:
return value if isinstance(value, str) and value else None
def _limited(query, model, label: str):
rows = query.order_by(model.created_at, model.id).limit(_MAX_RECORDS + 1).all()
if len(rows) > _MAX_RECORDS:
raise ValueError(f"Payments DSAR {label} limit exceeded; narrow the selectors.")
return rows
def _payment_record(
session: Session,
obligation: PaymentObligation,
) -> DsarRecordRef:
reconciliations = _children(
session.query(PaymentReconciliation).filter(
PaymentReconciliation.tenant_id == obligation.tenant_id,
PaymentReconciliation.payment_row_id == obligation.id,
),
PaymentReconciliation,
"reconciliation",
)
events = _children(
session.query(PaymentEvent).filter(
PaymentEvent.tenant_id == obligation.tenant_id,
PaymentEvent.payment_row_id == obligation.id,
),
PaymentEvent,
"event",
)
return DsarRecordRef(
provider_id="payments",
module_id="payments",
resource_type="payment_obligation",
resource_id=obligation.id,
category="financial_obligation_and_evidence",
title=f"Payment obligation {obligation.payment_reference}",
data={
"payment_id": obligation.payment_id,
"payment_reference": obligation.payment_reference,
"source_module": obligation.source_module,
"source_resource_type": obligation.source_resource_type,
"source_resource_id": obligation.source_resource_id,
"amount_minor": obligation.amount_minor,
"currency": obligation.currency,
"subject": obligation.subject[:1_000],
"status": obligation.status,
"requested_at": _iso(obligation.requested_at),
"requested_by_ref": obligation.requested_by_ref,
"due_at": _iso(obligation.due_at),
"settled_at": _iso(obligation.settled_at),
"context_refs": _context_refs(obligation.context_refs),
"reconciliations": [
{
"id": row.id,
"reconciliation_id": row.reconciliation_id,
"mode": row.mode,
"amount_minor": row.amount_minor,
"currency": row.currency,
"transaction_reference": row.transaction_reference,
"evidence_ref": _evidence_reference(row.evidence_ref),
"received_at": _iso(row.received_at),
"recorded_at": _iso(row.recorded_at),
"recorded_by_ref": row.recorded_by_ref,
}
for row in reconciliations
],
"events": [
{
"id": row.id,
"event_id": row.event_id,
"event_type": row.event_type,
"status": row.status,
"occurred_at": _iso(row.occurred_at),
"actor_ref": row.actor_ref,
}
for row in events
],
},
observed_at=_aware(obligation.updated_at),
immutable_evidence=True,
retention_reason=(
"The exact obligation, reconciliation, and lifecycle records are "
"financial evidence. Arbitrary metadata and event payloads are excluded."
),
)
def _children(query, model, label: str):
rows = (
query.order_by(model.created_at, model.id).limit(_MAX_CHILD_RECORDS + 1).all()
)
if len(rows) > _MAX_CHILD_RECORDS:
raise ValueError(
f"Payment {label} history exceeds the DSAR bound; narrow and review the payment."
)
return rows
def _request_attribution(row: PaymentObligation) -> DsarRecordRef:
return _attribution_record(
"payment_request_attribution",
row.id,
"Requested payment obligation",
{
"activity": "requested_payment_obligation",
"payment_id": row.payment_id,
"payment_reference": row.payment_reference,
"source_module": row.source_module,
"source_resource_type": row.source_resource_type,
"source_resource_id": row.source_resource_id,
"amount_minor": row.amount_minor,
"currency": row.currency,
"status": row.status,
"requested_at": _iso(row.requested_at),
},
row.requested_at,
)
def _reconciliation_attribution(row: PaymentReconciliation) -> DsarRecordRef:
return _attribution_record(
"payment_reconciliation_attribution",
row.id,
"Recorded payment reconciliation",
{
"activity": "recorded_payment_reconciliation",
"payment_row_id": row.payment_row_id,
"reconciliation_id": row.reconciliation_id,
"mode": row.mode,
"amount_minor": row.amount_minor,
"currency": row.currency,
"received_at": _iso(row.received_at),
"recorded_at": _iso(row.recorded_at),
},
row.recorded_at,
)
def _event_attribution(row: PaymentEvent) -> DsarRecordRef:
return _attribution_record(
"payment_event_attribution",
row.id,
"Payment lifecycle event attribution",
{
"activity": "recorded_payment_event",
"payment_row_id": row.payment_row_id,
"event_id": row.event_id,
"event_type": row.event_type,
"status": row.status,
"occurred_at": _iso(row.occurred_at),
},
row.occurred_at,
)
def _attribution_record(
resource_type: str,
resource_id: str,
title: str,
data: Mapping[str, object],
observed_at: datetime,
) -> DsarRecordRef:
return DsarRecordRef(
provider_id="payments",
module_id="payments",
resource_type=resource_type,
resource_id=resource_id,
category="operator_accountability_evidence",
title=title,
data=data,
observed_at=_aware(observed_at),
immutable_evidence=True,
retention_reason=(
"Payment operator attribution is financial accountability evidence; "
"arbitrary metadata, hashes, replay keys, and payloads are excluded."
),
)
def _context_refs(value: object) -> dict[str, str]:
if not isinstance(value, Mapping) or len(value) > 100:
raise ValueError("Payment context references exceed the DSAR bound.")
result: dict[str, str] = {}
for raw_key, raw_value in value.items():
key = str(raw_key)
if not key or len(key) > 200:
raise ValueError("Payment context reference key is invalid.")
result[key] = "[redacted]" if _sensitive_key(key) else str(raw_value)[:2_000]
return result
def _evidence_reference(value: object) -> dict[str, object]:
if not isinstance(value, Mapping):
raise ValueError("Payment evidence reference is invalid.")
derived = value.get("derived_from")
if not isinstance(derived, list) or len(derived) > 100:
raise ValueError("Payment evidence derivation exceeds the DSAR bound.")
return {
"kind": _bounded(value.get("kind"), 100),
"owner_module": _bounded(value.get("owner_module"), 100),
"evidence_id": _bounded(value.get("evidence_id"), 255),
"tenant_id": _bounded(value.get("tenant_id"), 36),
"version": _bounded(value.get("version"), 255),
"checksum": _bounded(value.get("checksum"), 255),
"source_ref": _bounded(value.get("source_ref"), 2_000),
"derived_from": [_bounded(item, 2_000) for item in derived],
"responsible_actor_ref": _bounded(
value.get("responsible_actor_ref"),
255,
),
"captured_at": _bounded(value.get("captured_at"), 100),
}
def _bounded(value: object, limit: int) -> str | None:
return str(value)[:limit] if value is not None else None
def _sensitive_key(value: str) -> bool:
normalized = value.strip().casefold().replace("-", "_")
return any(
part in normalized
for part in (
"authorization",
"cookie",
"credential",
"password",
"secret",
"token",
)
)
def _iso(value: datetime | None) -> str | None:
aware = _aware(value)
return aware.isoformat() if aware else None
def _aware(value: datetime | None) -> datetime | None:
if value is None or value.tzinfo is not None:
return value
return value.replace(tzinfo=timezone.utc)
def _session(value: object) -> Session:
if not isinstance(value, Session):
raise TypeError("Payments DSAR requires a SQLAlchemy Session.")
return value
def _validate_record(record: DsarRecordRef) -> None:
if record.provider_id != "payments" or record.module_id != "payments":
raise ValueError("Payments DSAR cannot plan a foreign provider record.")
if record.resource_type not in {"payment_obligation"} | _ATTRIBUTION_TYPES:
raise ValueError("Payments DSAR record type is invalid.")
if not record.resource_id:
raise ValueError("Payments DSAR record identity is incomplete.")
def _validate_action(action: DsarErasureActionRef) -> None:
if action.provider_id != "payments" or action.module_id != "payments":
raise ValueError("Payments DSAR cannot execute a foreign provider action.")
if not action.action_id.startswith("payments:"):
raise ValueError("Payments DSAR action identity is invalid.")
__all__ = ["PAYMENTS_DSAR_CAPABILITY", "PaymentsDsarProvider"]
+214 -2
View File
@@ -8,6 +8,7 @@ from govoplan_core.core.module_guards import (
) )
from govoplan_core.core.modules import ( from govoplan_core.core.modules import (
CapabilityDocumentation, CapabilityDocumentation,
DocumentationCondition,
DocumentationLink, DocumentationLink,
DocumentationTopic, DocumentationTopic,
FrontendModule, FrontendModule,
@@ -26,12 +27,16 @@ from govoplan_core.core.provider_governance import declared_module_architecture
from govoplan_core.core.views import ViewSurface from govoplan_core.core.views import ViewSurface
from govoplan_core.db.base import Base from govoplan_core.db.base import Base
from govoplan_payments.backend.db import models as payment_models from govoplan_payments.backend.db import models as payment_models
from govoplan_payments.backend.dsar_provider import (
PAYMENTS_DSAR_CAPABILITY,
PaymentsDsarProvider,
)
from govoplan_payments.backend.service import SqlPaymentRequestProvider from govoplan_payments.backend.service import SqlPaymentRequestProvider
MODULE_ID = "payments" MODULE_ID = "payments"
MODULE_NAME = "Payments" MODULE_NAME = "Payments"
MODULE_VERSION = "0.1.20" MODULE_VERSION = "0.1.21"
READ_SCOPE = "payments:payment:read" READ_SCOPE = "payments:payment:read"
WRITE_SCOPE = "payments:payment:write" WRITE_SCOPE = "payments:payment:write"
RECONCILE_SCOPE = "payments:payment:reconcile" RECONCILE_SCOPE = "payments:payment:reconcile"
@@ -62,6 +67,10 @@ def _payment_requests(_context: ModuleContext) -> SqlPaymentRequestProvider:
return SqlPaymentRequestProvider() return SqlPaymentRequestProvider()
def _dsar_provider(_context: ModuleContext) -> PaymentsDsarProvider:
return PaymentsDsarProvider()
def _tenant_summary(session: object, tenant_id: str) -> dict[str, int]: def _tenant_summary(session: object, tenant_id: str) -> dict[str, int]:
if not hasattr(session, "query"): if not hasattr(session, "query"):
return {"payment_requests": 0, "paid_payments": 0, "reconciliations": 0} return {"payment_requests": 0, "paid_payments": 0, "reconciliations": 0}
@@ -191,14 +200,26 @@ manifest = ModuleManifest(
), ),
provides_interfaces=( provides_interfaces=(
ModuleInterfaceProvider(name=CAPABILITY_PAYMENT_REQUESTS, version="1.0.0"), ModuleInterfaceProvider(name=CAPABILITY_PAYMENT_REQUESTS, version="1.0.0"),
ModuleInterfaceProvider(name=PAYMENTS_DSAR_CAPABILITY, version="0.1.0"),
), ),
capability_factories={CAPABILITY_PAYMENT_REQUESTS: _payment_requests}, capability_factories={
CAPABILITY_PAYMENT_REQUESTS: _payment_requests,
PAYMENTS_DSAR_CAPABILITY: _dsar_provider,
},
capability_documentation={ capability_documentation={
CAPABILITY_PAYMENT_REQUESTS: CapabilityDocumentation( CAPABILITY_PAYMENT_REQUESTS: CapabilityDocumentation(
label="Payment request and reconciliation", label="Payment request and reconciliation",
summary="Creates replay-safe obligations and records exact, evidence-bound manual settlement.", summary="Creates replay-safe obligations and records exact, evidence-bound manual settlement.",
contract_version="1.0.0", contract_version="1.0.0",
), ),
PAYMENTS_DSAR_CAPABILITY: CapabilityDocumentation(
label="Payments data-subject request provider",
summary=(
"Exports exact verified payment evidence or minimized operator "
"attribution with retain-only erasure outcomes."
),
contract_version="0.1.0",
),
}, },
route_factory=_router, route_factory=_router,
migration_spec=MigrationSpec( migration_spec=MigrationSpec(
@@ -227,6 +248,101 @@ manifest = ModuleManifest(
), ),
tenant_summary_providers=(_tenant_summary,), tenant_summary_providers=(_tenant_summary,),
documentation=( documentation=(
DocumentationTopic(
id="payments.data-subject-requests",
title="Payment data-subject requests",
summary=(
"Export exact payment obligations and financial evidence without "
"using payment descriptions as an identity search surface."
),
body=(
"Payments has no resident or applicant identity column and does not "
"search payment subjects, context JSON, metadata, or source records "
"for a person. Full financial access therefore requires an exact "
"payment row id, payment id, or human payment reference supplied as "
"a verified external subject reference. The resulting package contains "
"the obligation amount, currency, subject, status, dates, source and "
"bounded context references, reconciliation facts and typed evidence "
"references, and lifecycle-event facts. Reconciliation metadata, event "
"payloads, hashes, replay keys, provider data, inspection URLs, and "
"credentials are excluded. A request containing only an account, "
"membership, identity, or exact actor reference receives minimized "
"request, reconciliation, and event attribution for that operator; it "
"does not expose payment subjects. Every result is exact-tenant and "
"bounded. All erasure actions are retain-only and non-executable because "
"obligations, settlements, evidence links, and lifecycle attribution "
"remain governed financial and statutory evidence."
),
layer="configured",
documentation_types=("admin", "user"),
audience=("data_subject", "operator", "auditor", "module_admin"),
related_modules=("core", "cases", "workflow_engine", "ledger"),
metadata={
"kind": "reference",
"help_contexts": [
"payments.workspace",
"payments.state.requested",
"payments.state.paid",
"privacy.data-subject-requests",
],
"consequence_classes": {
"export_exact_payment": (
"Returns the obligation and bounded financial evidence for a "
"verified exact payment identifier."
),
"export_operator_attribution": (
"Returns minimized financial activity, never arbitrary payment "
"content."
),
"retain_payment_evidence": (
"Keeps financial evidence under configured statutory retention "
"and legal hold."
),
},
},
translations={
"de": {
"title": "Datenschutzanfragen zu Zahlungen",
"summary": (
"Exakte Zahlungsverpflichtungen und Finanznachweise exportieren, ohne "
"Zahlungsbeschreibungen als Identitätssuchfläche zu verwenden."
),
"body": (
"Payments besitzt keine Spalte für Einwohner- oder Antragstelleridentitäten und "
"durchsucht weder Zahlungsbetreffe noch Kontext-JSON, Metadaten oder Quelldatensätze "
"nach einer Person. Eine vollständige Finanzauskunft erfordert deshalb eine exakte "
"Zahlungszeilenkennung, Zahlungskennung oder menschenlesbare Zahlungsreferenz, die als "
"verifizierte externe Betroffenenreferenz bereitgestellt wird. Das Auskunftspaket enthält "
"Verpflichtungsbetrag, Währung, Betreff, Status, Zeitpunkte, Quell- und begrenzte "
"Kontextreferenzen, Abstimmungsfakten, typisierte Nachweisreferenzen und Fakten zu "
"Lebenszyklusereignissen. Abstimmungsmetadaten, Ereignisinhalte, Prüfsummen, "
"Wiederholungsschlüssel, Anbieterdaten, Prüf-URLs und Zugangsdaten bleiben ausgeschlossen. "
"Eine Anfrage nur mit Konto-, Mitgliedschafts-, Identitäts- oder exakter Akteursreferenz "
"liefert minimierte Zuschreibungen zu Anforderung, Abstimmung und Ereignissen dieser "
"bearbeitenden Person; Zahlungsbetreffe werden nicht offengelegt. Jedes Ergebnis ist exakt "
"mandantenbegrenzt. Alle Löschaktionen sind reine Aufbewahrungsergebnisse und nicht "
"ausführbar, weil Verpflichtungen, Erfüllungen, Nachweisverknüpfungen und "
"Lebenszykluszuschreibungen gesteuerte finanzielle und gesetzliche Nachweise bleiben."
),
}
},
structured_translation_version="1",
structured_translations={
"de": {
"consequence_classes": {
"export_exact_payment": (
"Gibt Verpflichtung und begrenzte Finanznachweise für eine verifizierte exakte Zahlungskennung zurück."
),
"export_operator_attribution": (
"Gibt minimierte Finanzaktivität, aber niemals beliebige Zahlungsinhalte zurück."
),
"retain_payment_evidence": (
"Bewahrt Finanznachweise gemäß konfigurierter gesetzlicher Aufbewahrung und Sperre auf."
),
}
}
},
),
DocumentationTopic( DocumentationTopic(
id="payments.requests-and-reconciliation", id="payments.requests-and-reconciliation",
title="Payment requests and manual reconciliation", title="Payment requests and manual reconciliation",
@@ -241,6 +357,8 @@ manifest = ModuleManifest(
layer="configured", layer="configured",
documentation_types=("admin", "user"), documentation_types=("admin", "user"),
audience=("operator", "module_admin", "auditor", "product_owner"), audience=("operator", "module_admin", "auditor", "product_owner"),
conditions=(DocumentationCondition(required_scopes=(READ_SCOPE,)),),
related_modules=("cases", "workflow_engine", "audit", "ledger"),
links=( links=(
DocumentationLink( DocumentationLink(
label="Payments boundary and recovery", label="Payments boundary and recovery",
@@ -249,6 +367,7 @@ manifest = ModuleManifest(
), ),
), ),
metadata={ metadata={
"kind": "workflow",
"help_contexts": [ "help_contexts": [
"payments.request", "payments.request",
"payments.workspace", "payments.workspace",
@@ -257,6 +376,26 @@ manifest = ModuleManifest(
"payments.state.requested", "payments.state.requested",
"payments.state.paid", "payments.state.paid",
], ],
"purpose": (
"Create an exact payment obligation and reconcile it as paid only against matching immutable evidence."
),
"prerequisites": [
"The actor can read Payments; creating and reconciling require their dedicated scopes.",
"The source procedure supplies a stable same-tenant reference and a replay-safe request key.",
"Manual reconciliation has a same-tenant versioned or checksum-bound evidence reference.",
],
"steps": [
"Create the amount, currency, source reference, due date, and human payment reference in the guided dialog.",
"Retain the returned payment ID in the calling Case or Workflow instead of writing Payments tables.",
"Reload the obligation before reconciliation when the workspace reports stale data.",
"Provide the external transaction reference, exact settlement time, and immutable evidence reference.",
"Confirm full amount and currency; any mismatch, duplicate, partial amount, or cross-tenant evidence fails closed.",
"Review the appended lifecycle and audit evidence after the obligation becomes paid.",
],
"limitations": [
"Only full manual reconciliation is supported; partial payment, refund, reversal, and correction need future governed flows.",
"Online checkout, callbacks, Ledger posting, XRechnung, and an applicant payment page are not implemented here.",
],
"privacy_notes": [ "privacy_notes": [
"Procedure context uses stable references; applicant names, bank account details, and submitted form values are not required.", "Procedure context uses stable references; applicant names, bank account details, and submitted form values are not required.",
"The immutable evidence remains owned by its provider; Payments stores only the typed EvidenceReference.", "The immutable evidence remains owned by its provider; Payments stores only the typed EvidenceReference.",
@@ -265,6 +404,79 @@ manifest = ModuleManifest(
"request_payment": "Creates a durable amount/currency obligation and a stable applicant payment reference.", "request_payment": "Creates a durable amount/currency obligation and a stable applicant payment reference.",
"reconcile_manual": "Marks the exact obligation paid and appends evidence; a future governed adjustment is required to reverse it.", "reconcile_manual": "Marks the exact obligation paid and appends evidence; a future governed adjustment is required to reverse it.",
}, },
"verification": [
"The paid obligation retains the original amount, currency, payment ID, and human reference unchanged.",
"Reconciliation names the external transaction and immutable evidence reference.",
"Replay and duplicate checks prove that one external settlement did not create conflicting paid states.",
],
},
translations={
"de": {
"title": "Zahlungsanforderungen und manuelle Abstimmung",
"summary": (
"Eine exakte Verpflichtung anlegen und nur mit passendem unveränderlichem "
"Nachweis als bezahlt kennzeichnen."
),
"body": (
"Payments führt die mandantengebundene Zahlungskennung, die menschenlesbare "
"Zahlungsreferenz, angeforderten Betrag und Währung, Lebenszyklusereignisse und "
"Abstimmungsnachweise. Ein Case, Workflow oder anderes Verfahren ruft die Fähigkeit "
"payments.requests mit eigener Quellreferenz und Wiederholungsschlüssel auf und bewahrt "
"die zurückgegebene Zahlungskennung auf, statt Payments-Tabellen zu schreiben. Der "
"Arbeitsbereich zeigt angeforderte und bezahlte Verpflichtungen mit Quelle, Fälligkeit "
"oder Erfüllungszeit und Abstimmungsnachweis. Schreibberechtigte legen eine Anforderung im "
"geführten Dialog an. Abstimmungsberechtigte verwenden den getrennten folgenreichen Dialog, "
"der Betrag und Währung fixiert und eine externe Transaktionsreferenz sowie eine "
"mandantengleiche versionierte oder prüfsummengebundene EvidenceReference verlangt. Neu "
"laden erhält vorhandene Daten und kennzeichnet sie als veraltet, wenn die Aktualisierung "
"scheitert. Fehlende Anlege- oder Abstimmungsberechtigung bleibt mit erforderlicher "
"Berechtigung und zuständiger Administration sichtbar. Der erste unterstützte Zahlungseingang "
"ist die manuelle Abstimmung einer vollständigen Zahlung. Abweichung, doppelte Erfüllung unter "
"anderem Schlüssel, mandantenfremder Nachweis, Teilbetrag oder Zeitstempel ohne Zeitzone "
"scheitert geschlossen. Erfolgreiche Anforderungen und Abstimmungen fügen "
"Zahlungsereignisse an; API-Aktionen erzeugen bei installiertem Audit Nachweise. Es gibt keine "
"stille Korrektur: Storno, Erstattung, Teilzahlung, Online-Checkout, Anbieter-Callbacks, "
"Ledger-Buchung und XRechnung bleiben ausdrückliche zukünftige Abläufe."
),
}
},
structured_translation_version="1",
structured_translations={
"de": {
"purpose": (
"Eine exakte Zahlungsverpflichtung anlegen und nur anhand passender unveränderlicher Nachweise als bezahlt abstimmen."
),
"prerequisites": [
"Die handelnde Person darf Payments lesen; Anlegen und Abstimmen erfordern ihre jeweils eigenen Berechtigungen.",
"Das Quellverfahren liefert eine stabile mandantengleiche Referenz und einen wiederholungssicheren Anforderungsschlüssel.",
"Für die manuelle Abstimmung liegt eine mandantengleiche versionierte oder prüfsummengebundene Nachweisreferenz vor.",
],
"steps": [
"Betrag, Währung, Quellreferenz, Fälligkeit und menschenlesbare Zahlungsreferenz im geführten Dialog anlegen.",
"Die zurückgegebene Zahlungskennung im aufrufenden Case oder Workflow bewahren, statt Payments-Tabellen zu schreiben.",
"Die Verpflichtung vor der Abstimmung neu laden, wenn der Arbeitsbereich veraltete Daten meldet.",
"Externe Transaktionsreferenz, exakte Erfüllungszeit und unveränderliche Nachweisreferenz angeben.",
"Vollständigen Betrag und Währung bestätigen; Abweichung, Duplikat, Teilbetrag oder mandantenfremder Nachweis scheitert geschlossen.",
"Nach dem Wechsel auf bezahlt die angefügten Lebenszyklus- und Auditnachweise prüfen.",
],
"limitations": [
"Nur vollständige manuelle Abstimmung wird unterstützt; Teilzahlung, Erstattung, Storno und Korrektur benötigen zukünftige gesteuerte Abläufe.",
"Online-Checkout, Callbacks, Ledger-Buchung, XRechnung und eine Antragsteller-Zahlungsseite sind hier nicht implementiert.",
],
"privacy_notes": [
"Verfahrenskontext verwendet stabile Referenzen; Namen von Antragstellern, Bankverbindungen und übermittelte Formularwerte sind nicht erforderlich.",
"Der unveränderliche Nachweis bleibt Eigentum seines Anbieters; Payments speichert nur die typisierte EvidenceReference.",
],
"consequence_classes": {
"request_payment": "Erzeugt eine dauerhafte Betrags- und Währungsverpflichtung sowie eine stabile Zahlungsreferenz für Antragsteller.",
"reconcile_manual": "Kennzeichnet die exakte Verpflichtung als bezahlt und fügt Nachweise an; eine zukünftige gesteuerte Anpassung ist zur Umkehr erforderlich.",
},
"verification": [
"Die bezahlte Verpflichtung bewahrt ursprünglichen Betrag, Währung, Zahlungskennung und menschenlesbare Referenz unverändert.",
"Die Abstimmung nennt externe Transaktion und unveränderliche Nachweisreferenz.",
"Wiederholungs- und Duplikatprüfungen belegen, dass eine externe Erfüllung keine widersprüchlichen Bezahltzustände erzeugt hat.",
],
}
}, },
), ),
), ),
+31
View File
@@ -0,0 +1,31 @@
from __future__ import annotations
import unittest
from govoplan_core.core.modules import (
documentation_structured_translation_issues,
user_workflow_scope_condition_issues,
)
from govoplan_payments.backend.manifest import manifest
class PaymentsDocumentationTests(unittest.TestCase):
def test_public_topics_have_complete_german_reference_content(self) -> None:
self.assertEqual(2, len(manifest.documentation))
for topic in manifest.documentation:
translation = topic.translations.get("de", {})
self.assertTrue(
all(translation.get(key) for key in ("title", "summary", "body"))
)
self.assertEqual((), documentation_structured_translation_issues(topic))
def test_documentation_has_scope_conditioned_workflow_and_reference(self) -> None:
kinds = {topic.metadata.get("kind") for topic in manifest.documentation}
self.assertIn("workflow", kinds)
self.assertIn("reference", kinds)
for topic in manifest.documentation:
self.assertEqual((), user_workflow_scope_condition_issues(topic))
if __name__ == "__main__":
unittest.main()
+418
View File
@@ -0,0 +1,418 @@
from __future__ import annotations
import json
import unittest
from datetime import UTC, datetime
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from govoplan_core.core.dsar import (
DsarErasureActionRef,
DsarProvider,
DsarRecordRef,
DsarSubjectRef,
)
from govoplan_core.db.base import Base
from govoplan_core.privacy.dsar_workflow import (
create_data_subject_request,
search_data_subject_request,
)
from govoplan_payments.backend.db.models import (
PaymentEvent,
PaymentObligation,
PaymentReconciliation,
)
from govoplan_payments.backend.dsar_provider import (
PAYMENTS_DSAR_CAPABILITY,
PaymentsDsarProvider,
)
from govoplan_payments.backend.manifest import manifest
NOW = datetime(2026, 8, 21, 10, 0, tzinfo=UTC)
class _Registry:
def __init__(self, provider: PaymentsDsarProvider, *, active: bool = True) -> None:
self.provider = provider
self.active = active
def capability_names(self):
return (PAYMENTS_DSAR_CAPABILITY,)
def capability_owner(self, name):
self._assert_capability(name)
return "payments"
def tenant_entitlement_resolver(self):
active = self.active
class _Resolver:
@staticmethod
def resolve(session, tenant_id):
del session, tenant_id
return type(
"State",
(),
{"effective_modules": ("payments",) if active else ()},
)()
return _Resolver()
def require_tenant_capability(self, name, session, **kwargs):
del session, kwargs
self._assert_capability(name)
return self.provider
def manifests(self):
return (type("Manifest", (), {"id": "payments"})(),)
@staticmethod
def _assert_capability(name: str) -> None:
if name != PAYMENTS_DSAR_CAPABILITY:
raise KeyError(name)
class PaymentsDsarProviderTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite+pysqlite:///:memory:")
Base.metadata.create_all(self.engine)
self.session = Session(self.engine)
self.provider = PaymentsDsarProvider()
self.assertIsInstance(self.provider, DsarProvider)
self._seed()
self.session.commit()
def tearDown(self) -> None:
self.session.close()
self.engine.dispose()
def _obligation(
self,
row_id: str,
*,
tenant_id: str = "tenant-1",
payment_id: str,
payment_reference: str,
requested_by_ref: str,
subject: str,
) -> PaymentObligation:
return PaymentObligation(
id=row_id,
tenant_id=tenant_id,
payment_id=payment_id,
payment_reference=payment_reference,
source_module="cases",
source_resource_type="case",
source_resource_id=f"case-{row_id}",
amount_minor=12_500,
currency="EUR",
subject=subject,
status="paid",
idempotency_key=f"idempotency-{row_id}-do-not-export",
request_sha256="a" * 64,
requested_at=NOW,
requested_by_ref=requested_by_ref,
settled_at=NOW,
context_refs={
"service": "resident-permit",
"access_token": "payment-secret-do-not-export",
},
details={"private": "obligation-metadata-do-not-export"},
)
def _seed(self) -> None:
subject_payment = self._obligation(
"payment-row-1",
payment_id="payment-1",
payment_reference="PAY-0001",
requested_by_ref="account:account-1",
subject="Resident permit fee",
)
other_payment = self._obligation(
"payment-row-other",
payment_id="payment-other",
payment_reference="PAY-OTHER",
requested_by_ref="account:account-other",
subject="Other person's private payment",
)
other_tenant = self._obligation(
"payment-row-other-tenant",
tenant_id="tenant-2",
payment_id="payment-other-tenant",
payment_reference="PAY-TENANT-2",
requested_by_ref="account:account-1",
subject="Other tenant private payment",
)
self.session.add_all((subject_payment, other_payment, other_tenant))
self.session.flush()
self.session.add_all(
(
PaymentReconciliation(
id="reconciliation-1",
tenant_id="tenant-1",
reconciliation_id="reconciliation-command-1",
payment_row_id="payment-row-1",
mode="manual_full",
amount_minor=12_500,
currency="EUR",
transaction_reference="BANK-REFERENCE-1",
evidence_ref={
"kind": "record",
"owner_module": "records",
"evidence_id": "record-1",
"tenant_id": "tenant-1",
"version": "4",
"checksum": "b" * 64,
"source_ref": "records:record-1:v4",
"derived_from": ["bank-statement-1"],
"responsible_actor_ref": "account:account-1",
"captured_at": NOW.isoformat(),
"inspection_url": "/records/record-1",
},
idempotency_key="reconcile-key-do-not-export",
request_sha256="c" * 64,
received_at=NOW,
recorded_at=NOW,
recorded_by_ref="account:account-1",
details={"private": "reconciliation-metadata-do-not-export"},
),
PaymentEvent(
id="event-1",
tenant_id="tenant-1",
event_id="payment-event-1",
payment_row_id="payment-row-1",
event_type="payment.reconciled",
status="paid",
occurred_at=NOW,
actor_ref="account:account-1",
payload={"secret": "event-payload-do-not-export"},
),
PaymentEvent(
id="event-other",
tenant_id="tenant-1",
event_id="payment-event-other",
payment_row_id="payment-row-other",
event_type="payment.requested",
status="requested",
occurred_at=NOW,
actor_ref="account:account-other",
payload={"private": "other event"},
),
)
)
def test_exact_payment_reference_exports_bounded_financial_evidence(self) -> None:
records = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
external_references={"payments.reference": "PAY-0001"}
),
)
self.assertEqual(["payment-row-1"], [record.resource_id for record in records])
exported = json.dumps([record.to_dict() for record in records])
self.assertIn("Resident permit fee", exported)
self.assertIn("BANK-REFERENCE-1", exported)
self.assertIn("records:record-1:v4", exported)
self.assertIn("payment.reconciled", exported)
self.assertIn("[redacted]", exported)
for excluded in (
"payment-secret-do-not-export",
"obligation-metadata-do-not-export",
"reconciliation-metadata-do-not-export",
"event-payload-do-not-export",
"idempotency-payment-row-1-do-not-export",
"reconcile-key-do-not-export",
"inspection_url",
"Other person's private payment",
"Other tenant private payment",
):
self.assertNotIn(excluded, exported)
def test_actor_search_is_minimized_and_does_not_expose_subject(self) -> None:
records = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(account_id="account-1"),
)
self.assertEqual(
{
"payment_request_attribution",
"payment_reconciliation_attribution",
"payment_event_attribution",
},
{record.resource_type for record in records},
)
exported = json.dumps([record.to_dict() for record in records])
self.assertIn("PAY-0001", exported)
self.assertIn("recorded_payment_reconciliation", exported)
self.assertNotIn("Resident permit fee", exported)
self.assertNotIn("BANK-REFERENCE-1", exported)
self.assertNotIn("Other person's private payment", exported)
self.assertNotIn("Other tenant private payment", exported)
def test_identifiers_corroborate_and_alias_conflicts_fail_closed(self) -> None:
corroborated = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
external_references={
"payments.payment": "payment-1",
"payments.reference": "PAY-0001",
}
),
)
mismatched = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
external_references={
"payments.payment": "payment-1",
"payments.reference": "PAY-OTHER",
}
),
)
conflict = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
account_id="account-1",
external_references={"payments.account": "account-other"},
),
)
self.assertEqual(["payment-row-1"], [item.resource_id for item in corroborated])
self.assertEqual((), mismatched)
self.assertEqual((), conflict)
def test_erasure_is_retain_only_and_foreign_inputs_are_rejected(self) -> None:
subject = DsarSubjectRef(external_references={"payments.reference": "PAY-0001"})
records = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=subject,
)
actions = self.provider.plan_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
records=records,
)
self.assertTrue(all(action.kind == "retain" for action in actions))
self.assertTrue(all(not action.executable for action in actions))
results = self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
actions=actions,
request_id="dsar-1",
)
self.assertTrue(all(result.status == "blocked" for result in results))
self.assertIsNotNone(self.session.get(PaymentObligation, "payment-row-1"))
with self.assertRaisesRegex(ValueError, "foreign provider record"):
self.provider.plan_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
records=(
DsarRecordRef(
provider_id="ledger",
module_id="ledger",
resource_type="payment_obligation",
resource_id="payment-row-1",
category="financial",
title="Foreign payment",
),
),
)
with self.assertRaisesRegex(ValueError, "foreign provider action"):
self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
actions=(
DsarErasureActionRef(
action_id="ledger:retain:payment:payment-row-1",
provider_id="ledger",
module_id="ledger",
kind="retain",
resource_type="payment_obligation",
resource_id="payment-row-1",
title="Retain payment",
rationale="Financial evidence",
executable=False,
),
),
request_id="dsar-1",
)
def test_core_workflow_and_manifest_register_provider(self) -> None:
subject = DsarSubjectRef(external_references={"payments.reference": "PAY-0001"})
row = create_data_subject_request(
self.session,
tenant_id="tenant-1",
reference="DSAR-PAYMENTS-1",
request_kind="access",
subject=subject,
purpose="Respond to a verified request.",
legal_basis="Article 15 GDPR",
due_at=None,
requested_by_account_id="privacy-officer",
)
self.session.commit()
search_data_subject_request(
self.session,
registry=_Registry(self.provider),
row=row,
expected_revision=1,
)
self.assertEqual(
[PAYMENTS_DSAR_CAPABILITY], row.coverage["provider_capabilities"]
)
self.assertEqual(1, row.search_result["record_count"])
inactive = create_data_subject_request(
self.session,
tenant_id="tenant-1",
reference="DSAR-PAYMENTS-2",
request_kind="access",
subject=subject,
purpose="Respond to a verified request.",
legal_basis="Article 15 GDPR",
due_at=None,
requested_by_account_id="privacy-officer",
)
self.session.commit()
search_data_subject_request(
self.session,
registry=_Registry(self.provider, active=False),
row=inactive,
expected_revision=1,
)
self.assertEqual(
[PAYMENTS_DSAR_CAPABILITY],
inactive.coverage["inactive_provider_capabilities"],
)
self.assertIn(PAYMENTS_DSAR_CAPABILITY, manifest.capability_factories)
self.assertIn(PAYMENTS_DSAR_CAPABILITY, manifest.capability_documentation)
self.assertIn(
PAYMENTS_DSAR_CAPABILITY,
{item.name for item in manifest.provides_interfaces},
)
self.assertTrue(
any(
topic.id == "payments.data-subject-requests"
and {"admin", "user"}.issubset(topic.documentation_types)
for topic in manifest.documentation
)
)
if __name__ == "__main__":
unittest.main()
+1 -1
View File
@@ -201,7 +201,7 @@ class PaymentTests(unittest.TestCase):
self.assertEqual((), self.provider.list_payments(self.session, tenant_id="tenant-2")) self.assertEqual((), self.provider.list_payments(self.session, tenant_id="tenant-2"))
def test_manifest_exposes_permission_bounded_operator_workspace(self) -> None: def test_manifest_exposes_permission_bounded_operator_workspace(self) -> None:
self.assertEqual("0.1.20", manifest.version) self.assertEqual("0.1.21", manifest.version)
self.assertIsNotNone(manifest.frontend) self.assertIsNotNone(manifest.frontend)
assert manifest.frontend is not None assert manifest.frontend is not None
self.assertEqual("@govoplan/payments-webui", manifest.frontend.package_name) self.assertEqual("@govoplan/payments-webui", manifest.frontend.package_name)
+1 -1
View File
@@ -1,6 +1,6 @@
{ {
"name": "@govoplan/payments-webui", "name": "@govoplan/payments-webui",
"version": "0.1.20", "version": "0.1.21",
"private": true, "private": true,
"type": "module", "type": "module",
"main": "src/index.ts", "main": "src/index.ts",
+4 -3
View File
@@ -1,4 +1,4 @@
import { CheckCircle2, Plus, RefreshCw } from "lucide-react"; import { CheckCircle2, Plus } from "lucide-react";
import { useEffect, useMemo, useRef, useState } from "react"; import { useEffect, useMemo, useRef, useState } from "react";
import { import {
Button, Button,
@@ -7,7 +7,6 @@ import {
DismissibleAlert, DismissibleAlert,
DocumentationHelpLink, DocumentationHelpLink,
FilterBar, FilterBar,
IconButton,
MetricCard, MetricCard,
MetricGrid, MetricGrid,
PageActionBar, PageActionBar,
@@ -263,6 +262,7 @@ export default function PaymentsPage({ settings, auth }: PlatformRouteContext) {
return ( return (
<WorkspaceFrame as="main" height="viewport" surface="plain" label="Payments workspace" interfaceId="payments.workspace" helpContextId="payments.workspace" helpModuleId="payments"> <WorkspaceFrame as="main" height="viewport" surface="plain" label="Payments workspace" interfaceId="payments.workspace" helpContextId="payments.workspace" helpModuleId="payments">
<PageLayout <PageLayout
archetype="collection"
mode="standalone" mode="standalone"
title="Payment requests" title="Payment requests"
description="Track source-bound obligations and record exact manual receipts against immutable evidence." description="Track source-bound obligations and record exact manual receipts against immutable evidence."
@@ -275,11 +275,12 @@ export default function PaymentsPage({ settings, auth }: PlatformRouteContext) {
actions={( actions={(
<PageActionBar <PageActionBar
variant="collection" variant="collection"
refreshable
label="Payment request actions" label="Payment request actions"
interfaceId="payments.workspace.actions" interfaceId="payments.workspace.actions"
helpContextId="payments.workspace" helpContextId="payments.workspace"
helpModuleId="payments" helpModuleId="payments"
reloadAction={<IconButton label="Reload payment requests" icon={<RefreshCw size={17} />} variant="ghost" onClick={() => void reload()} disabled={refreshing} />} reloadAction={{ onReload: () => void reload(), loading: refreshing, label: "Reload payment requests" }}
helpAction={<DocumentationHelpLink reference={{ topicId: "payments.requests-and-reconciliation", documentationType: "user" }} label="Open Payments documentation" />} helpAction={<DocumentationHelpLink reference={{ topicId: "payments.requests-and-reconciliation", documentationType: "user" }} label="Open Payments documentation" />}
createAction={createButton} createAction={createButton}
/> />