222 lines
8.1 KiB
Python
222 lines
8.1 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
import tomllib
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.core.institutional import EvidenceReference
|
|
from govoplan_core.core.payments import (
|
|
ManualPaymentReconciliationCommand,
|
|
PaymentRequestCommand,
|
|
)
|
|
from govoplan_payments.backend.db.models import (
|
|
PaymentEvent,
|
|
PaymentObligation,
|
|
PaymentReconciliation,
|
|
)
|
|
from govoplan_payments.backend.service import (
|
|
PaymentConflict,
|
|
PaymentError,
|
|
SqlPaymentRequestProvider,
|
|
)
|
|
from govoplan_payments.backend.manifest import manifest
|
|
|
|
|
|
NOW = datetime(2026, 8, 19, 10, 0, tzinfo=UTC)
|
|
|
|
|
|
def request_command(**changes) -> PaymentRequestCommand:
|
|
values = {
|
|
"tenant_id": "tenant-1",
|
|
"source_module": "cases",
|
|
"source_resource_type": "case",
|
|
"source_resource_id": "case-rpp-1",
|
|
"amount_minor": 3000,
|
|
"currency": "EUR",
|
|
"subject": "Resident parking permit fee",
|
|
"idempotency_key": "case-rpp-1-payment",
|
|
"requested_at": NOW,
|
|
"requested_by_ref": "account:officer-1",
|
|
"due_at": NOW + timedelta(days=14),
|
|
"context_refs": {
|
|
"case": "case-rpp-1",
|
|
"workflow": "workflow-rpp-1",
|
|
},
|
|
}
|
|
values.update(changes)
|
|
return PaymentRequestCommand(**values)
|
|
|
|
|
|
def reconciliation_command(payment_id: str, **changes) -> ManualPaymentReconciliationCommand:
|
|
values = {
|
|
"tenant_id": "tenant-1",
|
|
"payment_id": payment_id,
|
|
"amount_minor": 3000,
|
|
"currency": "EUR",
|
|
"transaction_reference": "BANK-2026-0001",
|
|
"evidence_ref": EvidenceReference(
|
|
kind="document",
|
|
owner_module="files",
|
|
evidence_id="file-payment-1",
|
|
tenant_id="tenant-1",
|
|
version="1",
|
|
checksum="a" * 64,
|
|
),
|
|
"idempotency_key": "bank-2026-0001",
|
|
"received_at": NOW + timedelta(days=2),
|
|
"recorded_at": NOW + timedelta(days=2, minutes=5),
|
|
"recorded_by_ref": "account:cashier-1",
|
|
}
|
|
values.update(changes)
|
|
return ManualPaymentReconciliationCommand(**values)
|
|
|
|
|
|
class PaymentTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
for table in (
|
|
PaymentObligation.__table__,
|
|
PaymentReconciliation.__table__,
|
|
PaymentEvent.__table__,
|
|
):
|
|
table.create(self.engine)
|
|
self.session = Session(self.engine)
|
|
self.provider = SqlPaymentRequestProvider()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_request_is_source_bound_and_replay_safe(self) -> None:
|
|
first = self.provider.request_payment(self.session, request_command())
|
|
replay = self.provider.request_payment(self.session, request_command())
|
|
|
|
self.assertEqual(first["payment_id"], replay["payment_id"])
|
|
self.assertTrue(replay["replayed"])
|
|
self.assertEqual("requested", first["status"])
|
|
self.assertEqual("case-rpp-1", first["source"]["resource_id"])
|
|
self.assertEqual(1, len(first["events"]))
|
|
with self.assertRaisesRegex(PaymentConflict, "idempotency conflict"):
|
|
self.provider.request_payment(
|
|
self.session,
|
|
request_command(amount_minor=3500),
|
|
)
|
|
self.assertIsNone(
|
|
self.provider.get_payment(
|
|
self.session,
|
|
tenant_id="tenant-2",
|
|
payment_id=str(first["payment_id"]),
|
|
)
|
|
)
|
|
|
|
def test_manual_reconciliation_requires_exact_immutable_evidence(self) -> None:
|
|
requested = self.provider.request_payment(self.session, request_command())
|
|
payment_id = str(requested["payment_id"])
|
|
|
|
with self.assertRaisesRegex(PaymentConflict, "amount and currency"):
|
|
self.provider.reconcile_manual_payment(
|
|
self.session,
|
|
reconciliation_command(payment_id, amount_minor=2999),
|
|
)
|
|
with self.assertRaisesRegex(PaymentError, "another tenant"):
|
|
self.provider.reconcile_manual_payment(
|
|
self.session,
|
|
reconciliation_command(
|
|
payment_id,
|
|
evidence_ref=EvidenceReference(
|
|
kind="document",
|
|
owner_module="files",
|
|
evidence_id="file-payment-1",
|
|
tenant_id="tenant-2",
|
|
version="1",
|
|
),
|
|
),
|
|
)
|
|
with self.assertRaisesRegex(PaymentError, "versioned or checksum"):
|
|
self.provider.reconcile_manual_payment(
|
|
self.session,
|
|
reconciliation_command(
|
|
payment_id,
|
|
evidence_ref=EvidenceReference(
|
|
kind="document",
|
|
owner_module="files",
|
|
evidence_id="file-payment-1",
|
|
tenant_id="tenant-1",
|
|
),
|
|
),
|
|
)
|
|
|
|
paid = self.provider.reconcile_manual_payment(
|
|
self.session,
|
|
reconciliation_command(payment_id),
|
|
)
|
|
replay = self.provider.reconcile_manual_payment(
|
|
self.session,
|
|
reconciliation_command(payment_id),
|
|
)
|
|
self.assertEqual("paid", paid["status"])
|
|
self.assertEqual("BANK-2026-0001", paid["reconciliation"]["transaction_reference"])
|
|
self.assertEqual("file-payment-1", paid["reconciliation"]["evidence_ref"]["evidence_id"])
|
|
self.assertEqual(2, len(paid["events"]))
|
|
self.assertTrue(replay["replayed"])
|
|
with self.assertRaisesRegex(PaymentConflict, "already reconciled"):
|
|
self.provider.reconcile_manual_payment(
|
|
self.session,
|
|
reconciliation_command(
|
|
payment_id,
|
|
idempotency_key="bank-2026-another",
|
|
transaction_reference="BANK-2026-ANOTHER",
|
|
),
|
|
)
|
|
|
|
def test_list_is_tenant_and_state_bounded(self) -> None:
|
|
first = self.provider.request_payment(self.session, request_command())
|
|
self.provider.reconcile_manual_payment(
|
|
self.session,
|
|
reconciliation_command(str(first["payment_id"])),
|
|
)
|
|
self.provider.request_payment(
|
|
self.session,
|
|
request_command(
|
|
source_resource_id="case-rpp-2",
|
|
idempotency_key="case-rpp-2-payment",
|
|
requested_at=NOW + timedelta(minutes=1),
|
|
),
|
|
)
|
|
|
|
paid = self.provider.list_payments(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
status="paid",
|
|
)
|
|
requested = self.provider.list_payments(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
status="requested",
|
|
)
|
|
self.assertEqual(1, len(paid))
|
|
self.assertEqual(1, len(requested))
|
|
self.assertEqual((), self.provider.list_payments(self.session, tenant_id="tenant-2"))
|
|
|
|
def test_manifest_exposes_permission_bounded_operator_workspace(self) -> None:
|
|
project = tomllib.loads((Path(__file__).parents[1] / "pyproject.toml").read_text())
|
|
self.assertEqual(project["project"]["version"], manifest.version)
|
|
self.assertIsNotNone(manifest.frontend)
|
|
assert manifest.frontend is not None
|
|
self.assertEqual("@govoplan/payments-webui", manifest.frontend.package_name)
|
|
self.assertEqual("/payments", manifest.frontend.routes[0].path)
|
|
self.assertEqual(("payments:payment:read",), manifest.frontend.routes[0].required_any)
|
|
self.assertEqual("payments.workspace", manifest.frontend.routes[0].surface_id)
|
|
self.assertEqual("payments.navigation", manifest.frontend.nav_items[0].surface_id)
|
|
surface_ids = {surface.id for surface in manifest.frontend.view_surfaces}
|
|
self.assertIn("payments.request.create", surface_ids)
|
|
self.assertIn("payments.reconciliation.manual", surface_ids)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|