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()