"""Cross-reference user-profile schema metadata against modelo registry use.
Validates every :class:`ModeloDefinition` in the registry against the
user-profile schema, checking that each :class:`ModeloRevision` binding
selector, filing schedule predicate, and deadline applicability condition
maps to a declared profile fact path.
"""
from __future__ import annotations
from collections.abc import Iterable, Mapping
from enum import StrEnum
from typing import TYPE_CHECKING, Literal
from pydantic import BaseModel, Field
from ._schema import ProfileSchemaDefinition
if TYPE_CHECKING:
from ..calculations.registry import ModeloDefinition, ModeloRevision
from ...core import STRICT_FROZEN_CONFIG as _STRICT_FROZEN
from ...core import BindingSourceKind
[docs]
class UserProfileRegistryContractSeverity(StrEnum):
"""Severity of a user-profile to modelo-registry contract issue."""
ERROR = "error"
WARNING = "warning"
[docs]
class UserProfileRegistryContractIssue(BaseModel):
"""One unresolved cross-reference between user-profile schema and modelos."""
model_config = _STRICT_FROZEN
severity: UserProfileRegistryContractSeverity
modelo_id: str
revision_id: str
surface: Literal[
"binding",
"filing_schedule",
"deadline_window",
"export_layout",
"cross_reference_applicability",
]
construct_id: str
selector: str
message: str = Field(min_length=1)
[docs]
class UserProfileSelectorIndex(BaseModel):
"""Schema-owned selector namespaces consumed by modelo registry surfaces."""
model_config = _STRICT_FROZEN
profile_selectors: frozenset[str]
schedule_predicates: frozenset[str]
export_headers: frozenset[str]
field_paths: frozenset[str]
[docs]
class UserProfileRegistryContractReport(BaseModel):
"""Typed result for profile schema coverage against modelo registry usage."""
model_config = _STRICT_FROZEN
schema_id: str
schema_version: int
checked_modelos: tuple[str, ...]
issues: tuple[UserProfileRegistryContractIssue, ...] = ()
@property
def errors(self) -> tuple[UserProfileRegistryContractIssue, ...]:
"""Return blocking :class:`UserProfileRegistryContractIssue` failures."""
return tuple(issue for issue in self.issues if issue.severity is UserProfileRegistryContractSeverity.ERROR)
@property
def warnings(self) -> tuple[UserProfileRegistryContractIssue, ...]:
"""Return non-blocking :class:`UserProfileRegistryContractIssue` coverage gaps."""
return tuple(issue for issue in self.issues if issue.severity is UserProfileRegistryContractSeverity.WARNING)
@property
def valid(self) -> bool:
"""Return whether all blocking selector references resolve."""
return not self.errors
[docs]
def build_user_profile_selector_index(schema: ProfileSchemaDefinition) -> UserProfileSelectorIndex:
"""Build the profile selector namespaces declared by the TOML schema.
Returns:
A :class:`UserProfileSelectorIndex` with all declared selector namespaces.
"""
profile_selectors: set[str] = set(schema.field_paths)
schedule_predicates: set[str] = set()
export_headers: set[str] = set()
for section in schema.sections:
for field in section.fields:
profile_selectors.update(field.model_selectors)
schedule_predicates.update(field.schedule_predicates)
export_headers.update(field.export_headers)
return UserProfileSelectorIndex(
profile_selectors=frozenset(profile_selectors),
schedule_predicates=frozenset(schedule_predicates),
export_headers=frozenset(export_headers),
field_paths=frozenset(schema.field_paths),
)
[docs]
def validate_user_profile_registry_contract(
modelos: Iterable[ModeloDefinition],
schema: ProfileSchemaDefinition,
) -> UserProfileRegistryContractReport:
"""Validate all profile-facing modelo registry references against the schema.
Blocking errors are reserved for model calculation and schedule selectors
that cannot be resolved from schema-declared profile facts. Export header
gaps are warnings during the rollout because several committed layouts use
non-profile operational headers that will move behind an export-context
backend rather than a live taxpayer profile fact.
Args:
modelos: Iterable of :class:`ModeloDefinition` instances to validate.
schema: The user-profile schema to validate registry references against.
Returns:
A :class:`UserProfileRegistryContractReport` with all issues found.
"""
index = build_user_profile_selector_index(schema)
checked_modelos: list[str] = []
issues: list[UserProfileRegistryContractIssue] = []
for modelo in modelos:
checked_modelos.append(modelo.id)
for revision in modelo.revisions.values():
issues.extend(_binding_issues(modelo.id, revision, index))
issues.extend(_schedule_issues(modelo.id, revision, index))
issues.extend(_deadline_issues(modelo.id, revision, index))
issues.extend(_cross_reference_applicability_issues(modelo.id, revision, index))
issues.extend(_export_issues(modelo.id, revision, index))
return UserProfileRegistryContractReport(
schema_id=schema.id,
schema_version=schema.version,
checked_modelos=tuple(sorted(checked_modelos)),
issues=tuple(issues),
)
def _binding_issues(
modelo_id: str,
revision: ModeloRevision,
index: UserProfileSelectorIndex,
) -> tuple[UserProfileRegistryContractIssue, ...]:
issues: list[UserProfileRegistryContractIssue] = []
for binding in revision.bindings:
if binding.source != BindingSourceKind.PROFILE:
continue
selectors = tuple(profile_binding_selectors(binding.selector))
if not selectors:
issues.append(
_issue(
severity=UserProfileRegistryContractSeverity.ERROR,
modelo_id=modelo_id,
revision_id=revision.id,
surface="binding",
construct_id=binding.id,
selector="<unresolved>",
message="profile binding does not declare a supported profile selector",
),
)
continue
for selector in selectors:
if selector not in index.profile_selectors:
issues.append(
_issue(
severity=UserProfileRegistryContractSeverity.ERROR,
modelo_id=modelo_id,
revision_id=revision.id,
surface="binding",
construct_id=binding.id,
selector=selector,
message="profile binding selector is not declared by user-profile schema",
),
)
return tuple(issues)
def _schedule_issues(
modelo_id: str,
revision: ModeloRevision,
index: UserProfileSelectorIndex,
) -> tuple[UserProfileRegistryContractIssue, ...]:
issues: list[UserProfileRegistryContractIssue] = []
for schedule in revision.filing_schedules:
for condition in schedule.profile_conditions:
if condition.field not in index.schedule_predicates:
issues.append(
_issue(
severity=UserProfileRegistryContractSeverity.ERROR,
modelo_id=modelo_id,
revision_id=revision.id,
surface="filing_schedule",
construct_id=schedule.id,
selector=condition.field,
message="filing schedule predicate is not declared by user-profile schema",
),
)
return tuple(issues)
def _deadline_issues(
modelo_id: str,
revision: ModeloRevision,
index: UserProfileSelectorIndex,
) -> tuple[UserProfileRegistryContractIssue, ...]:
issues: list[UserProfileRegistryContractIssue] = []
for window in revision.deadline_windows:
for condition in window.applicability_conditions:
if condition.field not in index.schedule_predicates:
issues.append(
_issue(
severity=UserProfileRegistryContractSeverity.ERROR,
modelo_id=modelo_id,
revision_id=revision.id,
surface="deadline_window",
construct_id=window.id,
selector=condition.field,
message="deadline applicability predicate is not declared by user-profile schema",
),
)
return tuple(issues)
def _cross_reference_applicability_issues(
modelo_id: str,
revision: ModeloRevision,
index: UserProfileSelectorIndex,
) -> tuple[UserProfileRegistryContractIssue, ...]:
issues: list[UserProfileRegistryContractIssue] = []
for cross_reference in revision.live_cross_references:
for predicate in cross_reference.applicability_predicates:
if predicate.field not in index.schedule_predicates:
issues.append(
_issue(
severity=UserProfileRegistryContractSeverity.ERROR,
modelo_id=modelo_id,
revision_id=revision.id,
surface="cross_reference_applicability",
construct_id=cross_reference.id,
selector=predicate.field,
message="cross-reference applicability predicate is not declared by user-profile schema",
),
)
return tuple(issues)
def _export_issues(
modelo_id: str,
revision: ModeloRevision,
index: UserProfileSelectorIndex,
) -> tuple[UserProfileRegistryContractIssue, ...]:
# Function-local import keeps ``aeat.domain.user_profile`` free of an
# eager calculation-registry load at package import (the lazy-boundary
# contract); this validator is the only runtime consumer of the enum.
from ..calculations.registry import CasillaFieldKind
issues: list[UserProfileRegistryContractIssue] = []
for layout in revision.export_layouts:
for record in layout.records:
for field in record.fields:
if (
field.kind == CasillaFieldKind.DRAFT
and field.draft_attribute == "profile_tax_id"
and "profile_tax_id" not in index.export_headers
):
issues.append(
_issue(
severity=UserProfileRegistryContractSeverity.ERROR,
modelo_id=modelo_id,
revision_id=revision.id,
surface="export_layout",
construct_id=f"{layout.id}.{record.id}.{field.id}",
selector="profile_tax_id",
message="profile_tax_id draft attribute is not declared by user-profile schema",
),
)
if field.kind == CasillaFieldKind.HEADER and field.header_key not in index.export_headers:
issues.append(
_issue(
severity=UserProfileRegistryContractSeverity.WARNING,
modelo_id=modelo_id,
revision_id=revision.id,
surface="export_layout",
construct_id=f"{layout.id}.{record.id}.{field.id}",
selector=field.header_key or "<missing>",
message="export header is not yet classified by user-profile schema",
),
)
return tuple(issues)
[docs]
def profile_binding_selectors(selector: Mapping[str, object] | BaseModel) -> tuple[str, ...]:
if isinstance(selector, BaseModel):
selector = selector.model_dump(exclude={"source"}, exclude_none=True, exclude_unset=True)
selectors: list[str] = []
profile_key = selector.get("profile_key")
if isinstance(profile_key, str):
selectors.append(profile_key)
profile_keys = selector.get("profile_keys")
if isinstance(profile_keys, tuple):
selectors.extend(item for item in profile_keys if isinstance(item, str))
required_when_profile_key = selector.get("required_when_profile_key")
if isinstance(required_when_profile_key, str):
selectors.append(required_when_profile_key)
profile_model = selector.get("profile_model")
profile_field = selector.get("field")
if isinstance(profile_model, str) and isinstance(profile_field, str):
collection = selector.get("collection")
if isinstance(collection, str):
selectors.append(f"{profile_model}.{collection}.{profile_field}")
else:
selectors.append(f"{profile_model}.{profile_field}")
return tuple(dict.fromkeys(selectors))
def _issue(
*,
severity: UserProfileRegistryContractSeverity,
modelo_id: str,
revision_id: str,
surface: Literal[
"binding",
"filing_schedule",
"deadline_window",
"export_layout",
"cross_reference_applicability",
],
construct_id: str,
selector: str,
message: str,
) -> UserProfileRegistryContractIssue:
return UserProfileRegistryContractIssue(
severity=severity,
modelo_id=modelo_id,
revision_id=revision_id,
surface=surface,
construct_id=construct_id,
selector=selector,
message=message,
)