GovOPlaN/govoplan-core
2
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

Add governed form evidence contract

Browse Source
This commit is contained in:
zemion
2026-08-06 12:42:20 +02:00
parent f5949427cc
commit b5f5be15f6
3 changed files with 359 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/RECORDS_FILING_CONTRACT.md
+10
View File
@@ -82,3 +82,13 @@ handling. The simulation is explicitly non-conformant, transfers no custody,
and cannot be used as evidence of an archive handoff. A real provider requires
a selected target/profile, provider-specific recovery declaration, and target
test evidence.
## Form Evidence Boundary
Form attachments use the separate provider-neutral contract in
`govoplan_core.core.form_evidence`. Forms Runtime requests short-lived,
purpose-bound upload grants and re-inspects the exact provider-owned evidence
before final submission. The provider keeps byte storage, quarantine,
classification, and retention ownership; Forms Runtime stores only immutable
evidence references and bounded verification results. This contract is not an
alternative path for Records filing or archive custody.
src/govoplan_core/core/form_evidence.py
+227
View File
@@ -0,0 +1,227 @@
from __future__ import annotations
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from datetime import datetime
from typing import Literal, Protocol, runtime_checkable
from govoplan_core.core.institutional import EvidenceReference, InstitutionalReference
CAPABILITY_FORM_EVIDENCE_PREFIX = "forms_runtime.evidence."
FormEvidenceState = Literal[
"accepted",
"pending",
"rejected",
"expired",
"revoked",
"unavailable",
]
_FORM_EVIDENCE_STATES = {
"accepted",
"pending",
"rejected",
"expired",
"revoked",
"unavailable",
}
class FormEvidenceContractError(ValueError):
"""Stable error for provider-neutral Form evidence operations."""
@dataclass(frozen=True, slots=True)
class FormEvidenceGrantRequest:
"""Request a short-lived, purpose-bound grant from an evidence owner."""
tenant_id: str
instance_id: str
definition_ref: InstitutionalReference
evidence_kind: str
purpose: str
idempotency_key: str
expires_at: datetime
custodian_ref: str | None = None
max_size_bytes: int | None = None
allowed_content_types: tuple[str, ...] = ()
metadata: Mapping[str, object] = field(default_factory=dict)
def __post_init__(self) -> None:
_require_text(self.tenant_id, "Form evidence tenant")
_require_text(self.instance_id, "Form evidence instance")
_require_text(self.evidence_kind, "Form evidence kind")
_require_text(self.purpose, "Form evidence purpose")
_require_text(self.idempotency_key, "Form evidence idempotency key")
if (
self.definition_ref.kind != "form"
or self.definition_ref.tenant_id != self.tenant_id
or not self.definition_ref.version
):
raise FormEvidenceContractError(
"Form evidence grants require an exact same-tenant Form definition."
)
if self.expires_at.tzinfo is None or self.expires_at.utcoffset() is None:
raise FormEvidenceContractError(
"Form evidence grant expiry must include a timezone."
)
if self.max_size_bytes is not None and self.max_size_bytes <= 0:
raise FormEvidenceContractError(
"Form evidence grant size limits must be positive."
)
if any(not item.strip() for item in self.allowed_content_types):
raise FormEvidenceContractError(
"Form evidence content types cannot be empty."
)
@dataclass(frozen=True, slots=True)
class FormEvidenceGrant:
provider_id: str
grant_id: str
upload_token: str | None
upload_url: str
expires_at: datetime
max_size_bytes: int
allowed_content_types: tuple[str, ...] = ()
replayed: bool = False
def __post_init__(self) -> None:
for value, label in (
(self.provider_id, "Form evidence provider"),
(self.grant_id, "Form evidence grant"),
(self.upload_url, "Form evidence upload URL"),
):
_require_text(value, label)
if self.upload_token is not None:
_require_text(self.upload_token, "Form evidence upload token")
if self.max_size_bytes <= 0:
raise FormEvidenceContractError(
"Form evidence grant size limits must be positive."
)
if self.expires_at.tzinfo is None or self.expires_at.utcoffset() is None:
raise FormEvidenceContractError(
"Form evidence grant expiry must include a timezone."
)
@dataclass(frozen=True, slots=True)
class FormEvidenceInspectionRequest:
tenant_id: str
instance_id: str
definition_ref: InstitutionalReference
evidence: EvidenceReference
purpose: str
final: bool
def __post_init__(self) -> None:
_require_text(self.tenant_id, "Form evidence tenant")
_require_text(self.instance_id, "Form evidence instance")
_require_text(self.purpose, "Form evidence purpose")
if self.definition_ref.tenant_id != self.tenant_id:
raise FormEvidenceContractError(
"Form evidence inspection cannot cross tenants."
)
if self.evidence.tenant_id != self.tenant_id:
raise FormEvidenceContractError(
"Form evidence inspection cannot cross tenants."
)
@dataclass(frozen=True, slots=True)
class FormEvidenceInspection:
provider_id: str
reference: EvidenceReference
state: FormEvidenceState
observed_at: datetime
retryable: bool = False
reason: str | None = None
metadata: Mapping[str, object] = field(default_factory=dict)
def __post_init__(self) -> None:
_require_text(self.provider_id, "Form evidence provider")
if self.state not in _FORM_EVIDENCE_STATES:
raise FormEvidenceContractError(
f"Unsupported Form evidence state: {self.state!r}."
)
if self.observed_at.tzinfo is None or self.observed_at.utcoffset() is None:
raise FormEvidenceContractError(
"Form evidence inspection time must include a timezone."
)
if self.state == "accepted" and self.retryable:
raise FormEvidenceContractError(
"Accepted Form evidence cannot require a retry."
)
@property
def accepted(self) -> bool:
return self.state == "accepted"
@runtime_checkable
class FormEvidenceProvider(Protocol):
provider_id: str
def supported_kinds(self) -> Sequence[str]: ...
def create_upload_grant(
self,
session: object,
principal: object,
*,
request: FormEvidenceGrantRequest,
) -> FormEvidenceGrant: ...
def inspect_evidence(
self,
session: object,
principal: object,
*,
request: FormEvidenceInspectionRequest,
) -> FormEvidenceInspection: ...
def form_evidence_capability(provider_id: str) -> str:
normalized = str(provider_id or "").strip().lower().replace("-", "_")
if not normalized or not normalized.replace("_", "").isalnum():
raise FormEvidenceContractError("Invalid Form evidence provider id.")
return f"{CAPABILITY_FORM_EVIDENCE_PREFIX}{normalized}"
def form_evidence_provider(
registry: object | None,
provider_id: str,
) -> FormEvidenceProvider | None:
capability_name = form_evidence_capability(provider_id)
if registry is None or not hasattr(registry, "has_capability"):
return None
if not registry.has_capability(capability_name):
return None
if hasattr(registry, "require_capability"):
provider = registry.require_capability(capability_name)
elif hasattr(registry, "capability"):
provider = registry.capability(capability_name)
else:
return None
return provider if isinstance(provider, FormEvidenceProvider) else None
def _require_text(value: str, label: str) -> None:
if not isinstance(value, str) or not value.strip():
raise FormEvidenceContractError(f"{label} is required.")
__all__ = [
"CAPABILITY_FORM_EVIDENCE_PREFIX",
"FormEvidenceContractError",
"FormEvidenceGrant",
"FormEvidenceGrantRequest",
"FormEvidenceInspection",
"FormEvidenceInspectionRequest",
"FormEvidenceProvider",
"FormEvidenceState",
"form_evidence_capability",
"form_evidence_provider",
]
tests/test_form_evidence_contract.py
+122
View File
@@ -0,0 +1,122 @@
from __future__ import annotations
from datetime import UTC, datetime, timedelta
import unittest
from govoplan_core.core.form_evidence import (
FormEvidenceContractError,
FormEvidenceGrantRequest,
FormEvidenceInspection,
FormEvidenceInspectionRequest,
form_evidence_capability,
form_evidence_provider,
)
from govoplan_core.core.institutional import EvidenceReference, InstitutionalReference
NOW = datetime(2026, 8, 7, 9, 0, tzinfo=UTC)
class Provider:
provider_id = "files"
def supported_kinds(self):
return ("document",)
def create_upload_grant(self, session, principal, *, request):
raise NotImplementedError
def inspect_evidence(self, session, principal, *, request):
return FormEvidenceInspection(
provider_id=self.provider_id,
reference=request.evidence,
state="accepted",
observed_at=NOW,
)
class Registry:
def __init__(self) -> None:
self.provider = Provider()
def has_capability(self, name: str) -> bool:
return name == "forms_runtime.evidence.files"
def require_capability(self, name: str):
return self.provider
class FormEvidenceContractTests(unittest.TestCase):
def setUp(self) -> None:
self.definition = InstitutionalReference(
kind="form",
owner_module="forms",
object_id="permit",
tenant_id="tenant-1",
version="2",
)
self.evidence = EvidenceReference(
kind="document",
owner_module="files",
evidence_id="asset-1",
tenant_id="tenant-1",
version="version-3",
checksum="a" * 64,
)
def test_grants_and_inspections_are_exact_and_tenant_bound(self) -> None:
grant = FormEvidenceGrantRequest(
tenant_id="tenant-1",
instance_id="form-1",
definition_ref=self.definition,
evidence_kind="document",
purpose="attach application evidence",
idempotency_key="grant-1",
expires_at=NOW + timedelta(minutes=10),
max_size_bytes=1024,
)
inspection = FormEvidenceInspectionRequest(
tenant_id="tenant-1",
instance_id="form-1",
definition_ref=self.definition,
evidence=self.evidence,
purpose="submit application",
final=True,
)
self.assertEqual("2", grant.definition_ref.version)
self.assertEqual("version-3", inspection.evidence.version)
def test_cross_tenant_evidence_is_rejected(self) -> None:
with self.assertRaisesRegex(FormEvidenceContractError, "cross tenants"):
FormEvidenceInspectionRequest(
tenant_id="tenant-2",
instance_id="form-1",
definition_ref=self.definition,
evidence=self.evidence,
purpose="submit",
final=True,
)
def test_provider_resolution_is_optional_and_runtime_checked(self) -> None:
registry = Registry()
self.assertEqual(
"forms_runtime.evidence.files",
form_evidence_capability("files"),
)
self.assertIs(registry.provider, form_evidence_provider(registry, "files"))
self.assertIsNone(form_evidence_provider(registry, "missing"))
def test_accepted_evidence_is_not_retryable(self) -> None:
with self.assertRaisesRegex(FormEvidenceContractError, "cannot require"):
FormEvidenceInspection(
provider_id="files",
reference=self.evidence,
state="accepted",
observed_at=NOW,
retryable=True,
)
if __name__ == "__main__":
unittest.main()