mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
7e06aa63c6
One-time script (dry-run default, --apply in a transaction, idempotent) that NULLs the four lodged_* columns on every predicted-source baseline row — a predicted EPC exists and no lodged one does — leaving Effective, the bill block, and rebaseline_reason intact. Dry-run against the audited DB reports 12,236 rows. Must run after the FE-owned Drizzle ALTER ... DROP NOT NULL lands. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
120 lines
4 KiB
Python
120 lines
4 KiB
Python
"""The one-time backfill nulls a predicted Property's phantom Lodged Performance,
|
|
leaves a lodged Property's intact, and is idempotent (#1361 Class B)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from sqlalchemy import Engine
|
|
from sqlmodel import Session
|
|
|
|
from datatypes.epc.domain.epc import Epc
|
|
from datatypes.epc.domain.epc_property_data import EpcPropertyData
|
|
from datatypes.epc.domain.mapper import EpcPropertyDataMapper
|
|
from domain.property_baseline.performance import Performance
|
|
from domain.property_baseline.property_baseline_performance import (
|
|
PropertyBaselinePerformance,
|
|
)
|
|
from repositories.epc.epc_postgres_repository import EpcPostgresRepository
|
|
from repositories.property_baseline.property_baseline_postgres_repository import (
|
|
PropertyBaselinePostgresRepository,
|
|
)
|
|
from scripts.null_predicted_lodged_performance import backfill
|
|
|
|
_JSON_SAMPLES = Path(__file__).resolve().parents[2] / "backend/epc_api/json_samples"
|
|
|
|
_PHANTOM_LODGED = Performance(
|
|
sap_score=14, epc_band=Epc.G, co2_emissions=5.0, primary_energy_intensity=400
|
|
)
|
|
_EFFECTIVE = Performance(
|
|
sap_score=57, epc_band=Epc.D, co2_emissions=1.5, primary_energy_intensity=160
|
|
)
|
|
|
|
|
|
def _epc() -> EpcPropertyData:
|
|
raw: dict[str, Any] = json.loads(
|
|
(_JSON_SAMPLES / "RdSAP-Schema-21.0.0" / "epc.json").read_text()
|
|
)
|
|
return EpcPropertyDataMapper.from_api_response(raw)
|
|
|
|
|
|
def _baseline(lodged: Performance) -> PropertyBaselinePerformance:
|
|
return PropertyBaselinePerformance(
|
|
lodged=lodged,
|
|
effective=_EFFECTIVE,
|
|
rebaseline_reason="physical_state_changed",
|
|
space_heating_kwh=4200.0,
|
|
water_heating_kwh=1600.0,
|
|
)
|
|
|
|
|
|
def _seed(db_engine: Engine) -> None:
|
|
"""A predicted Property (id 1, predicted EPC only, phantom lodged) and a
|
|
lodged Property (id 2, lodged EPC, real lodged) — both with a baseline row
|
|
carrying a populated lodged half (the pre-fix state)."""
|
|
epc = _epc()
|
|
with Session(db_engine) as session:
|
|
epc_repo = EpcPostgresRepository(session)
|
|
epc_repo.save(epc, property_id=1, source="predicted")
|
|
epc_repo.save(epc, property_id=2, source="lodged")
|
|
baseline_repo = PropertyBaselinePostgresRepository(session)
|
|
baseline_repo.save(_baseline(_PHANTOM_LODGED), property_id=1)
|
|
baseline_repo.save(_baseline(_PHANTOM_LODGED), property_id=2)
|
|
session.commit()
|
|
|
|
|
|
def test_apply_nulls_only_the_predicted_propertys_lodged_half(
|
|
db_engine: Engine,
|
|
) -> None:
|
|
# Arrange
|
|
_seed(db_engine)
|
|
|
|
# Act
|
|
with db_engine.begin() as conn:
|
|
found = backfill(conn, apply=True)
|
|
|
|
# Assert — one phantom found and nulled; the predicted Property loses its
|
|
# lodged half but keeps its Effective half, while the lodged Property is
|
|
# untouched.
|
|
assert found == 1
|
|
with Session(db_engine) as session:
|
|
repo = PropertyBaselinePostgresRepository(session)
|
|
predicted = repo.get_for_property(1)
|
|
lodged = repo.get_for_property(2)
|
|
assert predicted is not None
|
|
assert predicted.lodged is None
|
|
assert predicted.effective == _EFFECTIVE
|
|
assert lodged is not None
|
|
assert lodged.lodged == _PHANTOM_LODGED
|
|
|
|
|
|
def test_dry_run_reports_the_phantom_but_writes_nothing(db_engine: Engine) -> None:
|
|
# Arrange
|
|
_seed(db_engine)
|
|
|
|
# Act
|
|
with db_engine.begin() as conn:
|
|
found = backfill(conn, apply=False)
|
|
|
|
# Assert — the phantom is counted, but the row is left untouched.
|
|
assert found == 1
|
|
with Session(db_engine) as session:
|
|
predicted = PropertyBaselinePostgresRepository(session).get_for_property(1)
|
|
assert predicted is not None
|
|
assert predicted.lodged == _PHANTOM_LODGED
|
|
|
|
|
|
def test_re_running_apply_is_a_no_op(db_engine: Engine) -> None:
|
|
# Arrange — the first apply nulls the phantom.
|
|
_seed(db_engine)
|
|
with db_engine.begin() as conn:
|
|
backfill(conn, apply=True)
|
|
|
|
# Act — a second apply finds nothing left to null.
|
|
with db_engine.begin() as conn:
|
|
found = backfill(conn, apply=True)
|
|
|
|
# Assert
|
|
assert found == 0
|