"""Structural guard: every `EpcPropertyData` field — top-level AND nested — must round-trip. The deep-equality round-trip test (`test_epc_round_trip.py`) only catches a dropped field when a *fixture populates it* — the existing gaps (conservatory, roof windows, solar-HW collector, **PV arrays**, **floor heat-loss flags**) were latent precisely because no fixture exercised them (see `docs/migrations/epc-property-round-trip-fidelity.md`). This guard closes that blind spot structurally. It walks every domain dataclass reachable from `EpcPropertyData` (transitively, through Optionals and lists) and asserts each field is either reconstructed by an `EpcPostgresRepository` `_compose` / `_to_*` mapper or listed — with a reason — in the allow-list below. The earlier version only inspected `EpcPropertyData`'s *top-level* fields, so a dropped field on a *nested* object (`SapEnergySource.photovoltaic_arrays`, `SapFloorDimension.is_above_partially_heated_space`) slipped straight through. Adding a new domain field — at any depth — therefore forces a conscious persist-or-justify decision; persisting a previously-gapped field means deleting its allow-list entry. ADR-0036. """ from __future__ import annotations import ast import dataclasses import inspect import textwrap import typing from datatypes.epc.domain.epc_property_data import EpcPropertyData from repositories.epc.epc_postgres_repository import EpcPostgresRepository # Fields NOT reconstructed by the repo mappers, each with its reason. Keyed by # ``ClassName.field`` (a single field) or ``ClassName`` (the whole dataclass is # an un-persisted, tracked gap awaiting its FE columns / child table). _UNPERSISTED_ALLOWLIST: dict[str, str] = { # Not read by the calculator (dormant); no DB column yet. "EpcPropertyData.air_tightness": "dormant — not read by the calculator; no FE column", "EpcPropertyData.lzc_energy_sources": "dormant — not read by the calculator; no FE column", # Scoring-relevant round-trip gaps awaiting FE columns / child tables # (tracked follow-ups; see docs/migrations/epc-property-round-trip-fidelity.md). "EpcPropertyData.sap_roof_windows": "FE child table pending — tracked round-trip gap", "SapRoofWindow": "FE child table pending — tracked round-trip gap (sap_roof_windows)", "EpcPropertyData.solar_hw_collector_orientation": "FE column pending — tracked round-trip gap", "EpcPropertyData.solar_hw_collector_pitch_deg": "FE column pending — tracked round-trip gap", "EpcPropertyData.solar_hw_overshading": "FE column pending — tracked round-trip gap", "SapAlternativeWall.u_value": "FE column pending — tracked round-trip gap", "SapAlternativeWall.wall_thickness_mm": "FE column pending — tracked round-trip gap", # Room-in-roof detail: only floor_area + age_band reconstruct from the # inlined columns; the geometry / surface detail awaits an FE child table # (tracked round-trip gap, docs/migrations §2). "SapRoomInRoof.common_wall_height_m": "FE child table pending — room-in-roof detail", "SapRoomInRoof.common_wall_length_m": "FE child table pending — room-in-roof detail", "SapRoomInRoof.detailed_surfaces": "FE child table pending — room-in-roof detail", "SapRoomInRoof.gable_1_height_m": "FE child table pending — room-in-roof detail", "SapRoomInRoof.gable_1_length_m": "FE child table pending — room-in-roof detail", "SapRoomInRoof.gable_2_height_m": "FE child table pending — room-in-roof detail", "SapRoomInRoof.gable_2_length_m": "FE child table pending — room-in-roof detail", "SapRoomInRoofSurface": "FE child table pending — room-in-roof surface detail", # --- Nested gaps the calculator does NOT read (dormant); no FE column. "SapAlternativeWall.is_basement": "dormant — not read by the calculator; no FE column", "SapBuildingPart.floor_u_value": "dormant — not read by the calculator (column exists, unread)", "SapBuildingPart.rafter_insulation_thickness": "dormant — not read by the calculator; no FE column", "SapBuildingPart.roof_u_value": "dormant — not read by the calculator; no FE column", "SapBuildingPart.wall_is_basement": "dormant — not read by the calculator; no FE column", "SapBuildingPart.wall_u_value": "dormant — not read by the calculator; no FE column", "SapHeating.cylinder_heat_loss": "dormant — not read by the calculator; no FE column", } def _unwrap(annotation: object) -> list[object]: """The concrete types inside an annotation — peeling `Optional`, `Union` and `list[...]` so a field typed `Optional[list[SapWindow]]` yields `SapWindow`.""" args = typing.get_args(annotation) if not args: return [annotation] origin = typing.get_origin(annotation) out: list[object] = [] if origin is list: out.extend(_unwrap(args[0])) else: # Union / Optional for arg in args: out.extend(_unwrap(arg)) return out def _reachable_dataclasses() -> dict[str, type]: """Every domain dataclass reachable from `EpcPropertyData`, transitively, keyed by class name.""" seen: dict[str, type] = {} stack: list[type] = [EpcPropertyData] while stack: cls = stack.pop() if not dataclasses.is_dataclass(cls) or cls.__name__ in seen: continue seen[cls.__name__] = cls for field in dataclasses.fields(cls): for inner in _unwrap(field.type): if isinstance(inner, type) and dataclasses.is_dataclass(inner): stack.append(inner) return seen def _reconstructed_by_class() -> dict[str, set[str]]: """For each domain-dataclass name, the keyword argument names passed to ANY `ClassName(...)` construction inside `EpcPostgresRepository` — i.e. the fields its `_compose` / `_to_*` mappers reconstruct from the DB rows.""" source = textwrap.dedent(inspect.getsource(EpcPostgresRepository)) tree = ast.parse(source) by_class: dict[str, set[str]] = {} for node in ast.walk(tree): if isinstance(node, ast.Call) and isinstance(node.func, ast.Name): kwargs = {kw.arg for kw in node.keywords if kw.arg is not None} by_class.setdefault(node.func.id, set()).update(kwargs) return by_class def test_every_epc_property_data_field_is_persisted_or_allowlisted() -> None: # Arrange — every field of every reachable domain dataclass, and the fields # the repo mappers reconstruct. reachable = _reachable_dataclasses() reconstructed = _reconstructed_by_class() # Act — fields neither reconstructed nor explicitly excused (per-field or # whole-class allow-list entry). silently_dropped: list[str] = [] for name, cls in reachable.items(): if name in _UNPERSISTED_ALLOWLIST: # whole class excused continue recon = reconstructed.get(name, set()) for field in dataclasses.fields(cls): key = f"{name}.{field.name}" if field.name not in recon and key not in _UNPERSISTED_ALLOWLIST: silently_dropped.append(key) # Assert — nothing falls through the round-trip unnoticed, at any depth. assert silently_dropped == [], ( "EpcPropertyData field(s) are dropped on DB round-trip without a " f"documented reason: {sorted(silently_dropped)}. Either reconstruct them " "in EpcPostgresRepository (_compose / a _to_* mapper), or add them to " "_UNPERSISTED_ALLOWLIST with a justification." ) def test_allowlist_has_no_stale_entries() -> None: # Arrange reachable = _reachable_dataclasses() reconstructed = _reconstructed_by_class() # Act — allow-list entries that are now fully reconstructed, or name a # class/field that no longer exists. redundant: list[str] = [] for entry in _UNPERSISTED_ALLOWLIST: class_name, _, field_name = entry.partition(".") cls = reachable.get(class_name) if cls is None: redundant.append(entry) continue if not field_name: # Whole-class entry: stale once ANY field is reconstructed (the class # is now wired into the round-trip). if reconstructed.get(class_name): redundant.append(entry) continue field_names = {f.name for f in dataclasses.fields(cls)} if field_name not in field_names or field_name in reconstructed.get( class_name, set() ): redundant.append(entry) # Assert — the allow-list stays honest as gaps are closed. assert redundant == [], ( f"stale _UNPERSISTED_ALLOWLIST entries (now persisted or removed): " f"{sorted(redundant)} — delete them." )