govoplan-core/tests/test_records_contract.py

from __future__ import annotations

from datetime import UTC, datetime
import unittest

from govoplan_core.core.records import (
    RecordArchiveProviderState,
    RecordArchiveReceipt,
    RecordArchiveTransferRequest,
    RecordContractError,
    RecordFilingRequest,
    RecordSourceLocator,
    RecordSourceReference,
    RecordTransferPackage,
    record_archive_capabilities,
    record_archive_capability,
    record_source_capabilities,
    record_source_capability,
)


class _Registry:
    def capability_names(self) -> tuple[str, ...]:
        return (
            "mail.delivery",
            "records.archive.simulation",
            "records.source.cases",
            "records.source.files",
        )


class RecordsContractTests(unittest.TestCase):
    def test_exact_source_and_filing_request_are_tenant_bound(self) -> None:
        locator = RecordSourceLocator(
            tenant_id="tenant-1",
            source_module="files",
            resource_type="file_version",
            resource_id="asset-1",
            source_revision="version-7",
        )
        reference = RecordSourceReference(
            locator=locator,
            label="Evidence.pdf",
            content_sha256="a" * 64,
            size_bytes=12,
            recorded_at=datetime(2026, 8, 6, tzinfo=UTC),
        )
        request = RecordFilingRequest(
            tenant_id="tenant-1",
            record_id="record-1",
            source=locator,
            purpose="process application",
            filing_reason="Evidence received with the application.",
            idempotency_key="filing-1",
        )

        self.assertEqual("version-7", reference.locator.source_revision)
        self.assertEqual("record-1", request.record_id)

    def test_cross_tenant_filing_and_inexact_sources_are_rejected(self) -> None:
        with self.assertRaisesRegex(RecordContractError, "source_revision"):
            RecordSourceLocator(
                tenant_id="tenant-1",
                source_module="files",
                resource_type="file_version",
                resource_id="asset-1",
                source_revision="",
            )

        locator = RecordSourceLocator(
            tenant_id="tenant-1",
            source_module="cases",
            resource_type="case_revision",
            resource_id="case-1",
            source_revision="4",
        )
        with self.assertRaisesRegex(RecordContractError, "cross tenants"):
            RecordFilingRequest(
                tenant_id="tenant-2",
                record_id="record-1",
                source=locator,
                purpose="audit",
                filing_reason="Preserve the decision basis.",
                idempotency_key="filing-2",
            )

    def test_source_capabilities_are_discoverable_without_module_imports(self) -> None:
        self.assertEqual("records.source.files", record_source_capability("files"))
        self.assertEqual(
            "records.source.forms_runtime",
            record_source_capability("forms_runtime"),
        )
        self.assertEqual(
            ("records.source.cases", "records.source.files"),
            record_source_capabilities(_Registry()),
        )

    def test_archive_contract_is_digest_bound_and_discoverable(self) -> None:
        package = RecordTransferPackage(
            tenant_id="tenant-1",
            package_id="package-1",
            record_id="record-1",
            record_revision=3,
            profile="govoplan-simulation-v1",
            manifest_sha256="a" * 64,
            manifest={"record_id": "record-1", "items": []},
        )
        request = RecordArchiveTransferRequest(
            package=package,
            purpose="archive appraisal",
            idempotency_key="transfer-1",
        )
        state = RecordArchiveProviderState(
            provider_id="simulation",
            label="Archive simulation",
            profiles=("govoplan-simulation-v1",),
            authority_modes=("linked_reference",),
            healthy=True,
            checked_at=datetime(2026, 8, 6, tzinfo=UTC),
            simulated=True,
        )

        self.assertEqual("package-1", request.package.package_id)
        self.assertTrue(state.simulated)
        self.assertEqual(
            "records.archive.simulation",
            record_archive_capability("simulation"),
        )
        self.assertEqual(
            ("records.archive.simulation",),
            record_archive_capabilities(_Registry()),
        )

    def test_unknown_archive_outcome_cannot_be_retry_safe(self) -> None:
        with self.assertRaisesRegex(RecordContractError, "retry-safe"):
            RecordArchiveReceipt(
                provider_id="archive-1",
                package_id="package-1",
                outcome="outcome_unknown",
                observed_at=datetime(2026, 8, 6, tzinfo=UTC),
                receipt_sha256="b" * 64,
                retry_safe=True,
            )

    def test_archive_provider_runtime_values_are_validated(self) -> None:
        with self.assertRaisesRegex(RecordContractError, "invalid authority mode"):
            RecordArchiveProviderState(
                provider_id="archive",
                label="Archive",
                profiles=("profile",),
                authority_modes=("untrusted",),  # type: ignore[arg-type]
                healthy=True,
                checked_at=datetime.now(UTC),
            )
        with self.assertRaisesRegex(RecordContractError, "outcome is invalid"):
            RecordArchiveReceipt(
                provider_id="archive",
                package_id="package-1",
                outcome="maybe",  # type: ignore[arg-type]
                observed_at=datetime.now(UTC),
                receipt_sha256="a" * 64,
            )


if __name__ == "__main__":
    unittest.main()