Source code for aeat.domain.modelos._verification_report

"""Structured verification report produced by ``aeat app modelo verify``.

A :class:`VerificationReport` is the decision artifact the verify
command persists for every run. It captures whether the target
calculation :class:`CalculationRevision` meets the
``verificado_completo`` contract,
which blocking findings prevent that transition, which inputs are
missing, which :class:`CasillaId` identifiers are unresolved, which
waivers were accepted, and what the operator should do next.

The report is bucket-scoped and content-addressed by the
verification outcome (parent calculation revision, completeness
status, findings, and actor); the run timestamp is a non-identity
last-seen body field. Failed verification attempts produce a
persisted report (so the audit trail explains why a transition was
refused) without mutating the target revision, and an identical-outcome
retry collapses onto the same report rather than accumulating.
"""

from __future__ import annotations

from collections.abc import Iterator, Mapping
from datetime import datetime
from enum import StrEnum
from typing import Annotated, override

from pydantic import BaseModel, Field, StringConstraints, model_validator

from ...core import STRICT_FROZEN_CONFIG
from ...core.hashing import content_hash_hex
from ..calculations.registry import CasillaId, LegalRefId, SourceRefId, VerificationExpectationId
from ._errors import ModeloValidationError
from ._ids import VerificationReportId

_HEX_64_PATTERN = r"^[0-9a-f]{64}$"

_ReportId = Annotated[
    str,
    StringConstraints(strip_whitespace=True, min_length=64, max_length=64, pattern=_HEX_64_PATTERN),
]
_CalculationRevisionId = _ReportId
ModeloActorLabel = Annotated[
    str,
    StringConstraints(strip_whitespace=True, min_length=1, max_length=64),
]
"""Validated string identifying the operator who triggered a verification run.

Strips surrounding whitespace; must be 1–64 characters after stripping.
Used as ``verified_by`` on :class:`VerificationReport` to record the actor
label fed into the content-addressed id derivation.
"""
_FindingMessage = Annotated[
    str,
    StringConstraints(strip_whitespace=True, min_length=1, max_length=500),
]


[docs] class VerificationCompletenessStatus(StrEnum): """Top-level verdict from one verification run. * ``COMPLETE`` — every required input resolved, zero blocking findings. The revision transitions to ``VERIFICADO_COMPLETO``. * ``INCOMPLETE`` — required inputs missing or unresolved casillas remain. Soft refusal: operator can act on the missing-inputs list. * ``BLOCKED`` — blocking validation findings exist (e.g. reconciliation total mismatch over tolerance, schema violation). Hard refusal: operator must fix the finding before the revision can be verified. """ COMPLETE = "complete" INCOMPLETE = "incomplete" BLOCKED = "blocked"
[docs] class ModeloVerificationFindingKind(StrEnum): """Closed catalogue of verification-finding kinds. Maps to the readiness vocabulary mandated by the verify ADR: bucket / ledger source / profile fact / prior filed revision / live observation / casilla / waiver / blocking finding. """ MISSING_REQUIRED_CASILLA = "missing_required_casilla" RECONCILIATION_MISMATCH = "reconciliation_mismatch" UNRESOLVED_BINDING = "unresolved_binding" CROSS_PERIOD_DEPENDENCY_UNCLEAN = "cross_period_dependency_unclean" INVALID_WAIVER = "invalid_waiver" BLOCKING_RULE = "blocking_rule" ADVISORY = "advisory"
[docs] class ModeloVerificationFindingSeverity(StrEnum): """Severity of one verification finding.""" BLOCKING = "blocking" WARNING = "warning"
[docs] class ModeloVerificationFinding(BaseModel): """One verification finding. Findings of ``BLOCKING`` severity force ``BLOCKED`` completeness status. ``WARNING`` severity findings surface in the report but do not block ``COMPLETE`` status on their own. A finding may point at the affected :class:`CasillaId`, the registry :class:`VerificationExpectationId` that raised it, and the :class:`LegalRefId` / :class:`SourceRefId` provenance that grounds the operator-facing message. """ model_config = STRICT_FROZEN_CONFIG kind: ModeloVerificationFindingKind severity: ModeloVerificationFindingSeverity casilla_id: CasillaId | None = None expectation_id: VerificationExpectationId | None = None message: _FindingMessage next_action: _FindingMessage | None = None legal_refs: tuple[LegalRefId, ...] = Field(min_length=1) source_refs: tuple[SourceRefId, ...] = ()
[docs] def derive_verification_report_id( *, calculation_revision_id: str, completeness_status: VerificationCompletenessStatus, findings: tuple[ModeloVerificationFinding, ...], verified_by: str, ) -> str: """Deterministic 64-char SHA-256 id for a verification report. Content-addressed by the verification *outcome* - the parent :class:`CalculationRevision` id, the ``completeness_status``, the ordered ``findings`` tuple, and the ``verified_by`` actor. ``run_at`` is deliberately excluded from the identity so two retries of an identical-outcome verify collapse to one report on upsert (the id is clock-free); a re-verify whose findings change produces a new distinct report, the audit-meaningful granularity. """ payload = { "calculation_revision_id": calculation_revision_id.strip(), "completeness_status": completeness_status.value, "findings": [finding.model_dump(mode="json") for finding in findings], "verified_by": verified_by.strip(), } return content_hash_hex(payload)
[docs] class VerificationReport(BaseModel): """Decision record of one verification run against a :class:`CalculationRevision`. The id is content-addressed by the verification outcome (``calculation_revision_id``, ``completeness_status``, ``findings``, and ``verified_by``) via :func:`derive_verification_report_id`; ``run_at`` is a non-identity last-seen timestamp. A ``model_validator`` enforces the derivation on construction so no :class:`VerificationReport` can carry an inconsistent id. ``granted_verificado_completo`` is ``True`` if and only if ``completeness_status`` is ``COMPLETE`` and no blocking findings exist. The model validator enforces this invariant bidirectionally. """ model_config = STRICT_FROZEN_CONFIG verification_report_id: VerificationReportId calculation_revision_id: _CalculationRevisionId completeness_status: VerificationCompletenessStatus findings: tuple[ModeloVerificationFinding, ...] = Field(default_factory=tuple) resolved_casilla_ids: tuple[CasillaId, ...] = Field(default_factory=tuple) missing_required_casilla_ids: tuple[CasillaId, ...] = Field(default_factory=tuple) run_at: datetime verified_by: ModeloActorLabel granted_verificado_completo: bool @model_validator(mode="after") def _enforce_invariants(self) -> VerificationReport: derived = derive_verification_report_id( calculation_revision_id=self.calculation_revision_id, completeness_status=self.completeness_status, findings=self.findings, verified_by=self.verified_by, ) if derived != self.verification_report_id: raise ModeloValidationError( f"verification_report_id {self.verification_report_id!r} does not match the derived id {derived!r}", ) # granted_verificado_completo is a True iff completeness_status is COMPLETE # AND no blocking findings exist. has_blocking = any(finding.severity is ModeloVerificationFindingSeverity.BLOCKING for finding in self.findings) if self.granted_verificado_completo: if self.completeness_status is not VerificationCompletenessStatus.COMPLETE: raise ModeloValidationError("granted_verificado_completo=True requires completeness_status=COMPLETE") if has_blocking: raise ModeloValidationError("granted_verificado_completo=True requires no blocking findings") else: if self.completeness_status is VerificationCompletenessStatus.COMPLETE and not has_blocking: raise ModeloValidationError( "completeness_status=COMPLETE with no blocking findings must set granted_verificado_completo=True", ) # Required-casilla sets must be disjoint from resolved. overlap = set(self.resolved_casilla_ids) & set(self.missing_required_casilla_ids) if overlap: raise ModeloValidationError(f"casillas cannot be both resolved and missing: {sorted(overlap)!r}") return self
[docs] class VerificationReportCatalogue(BaseModel): """Immutable catalogue of every verification report in a bucket's storage. Keyed by ``verification_report_id``; the model validator enforces that every key equals the id of the :class:`VerificationReport` it maps to. Iteration yields :class:`VerificationReport` values (not key–value pairs), which diverges from the standard ``Mapping`` contract — the override is annotated with a suppression comment on ``__iter__``. """ model_config = STRICT_FROZEN_CONFIG reports: Mapping[str, VerificationReport] = Field(default_factory=dict) @model_validator(mode="after") def _enforce_keys_match(self) -> VerificationReportCatalogue: for key, report in self.reports.items(): if key != report.verification_report_id: raise ModeloValidationError( f"catalogue key {key!r} does not match verification_report_id {report.verification_report_id!r}", ) return self
[docs] def get(self, verification_report_id: str) -> VerificationReport | None: """Return the :class:`VerificationReport` for ``verification_report_id``, or ``None``.""" return self.reports.get(verification_report_id)
[docs] def for_calculation_revision(self, calculation_revision_id: str) -> tuple[VerificationReport, ...]: """Return every :class:`VerificationReport` against one calculation revision, ordered by run_at.""" matching = tuple(r for r in self.reports.values() if r.calculation_revision_id == calculation_revision_id) return tuple(sorted(matching, key=lambda r: r.run_at))
[docs] def values(self): """Return a view of all :class:`VerificationReport` values in the catalogue.""" return self.reports.values()
@override def __iter__(self) -> Iterator[VerificationReport]: # pyright: ignore[reportIncompatibleMethodOverride] # ty: ignore[invalid-method-override] # pyrefly: ignore[bad-override] # reason: intentional pydantic catalogue iteration adapter — yields domain items not field-value tuples """Iterate over :class:`VerificationReport` values (not ``(key, value)`` pairs).""" return iter(self.reports.values()) def __len__(self) -> int: """Return the number of reports in the catalogue.""" return len(self.reports)
__all__ = [ "ModeloVerificationFinding", "ModeloVerificationFindingKind", "ModeloVerificationFindingSeverity", "VerificationCompletenessStatus", "VerificationReport", "VerificationReportCatalogue", "derive_verification_report_id", ]