Model/tests/orchestration/test_modelling_solar_threading.py
Khalim Conn-Kowlessar b249f69cb2 feat(modelling): thread SolarPotential into the orchestrator's solar Generator
Slice 8 of the Solar PV Recommendation Generator (ADR-0026). The
ModellingOrchestrator now reads each Property's persisted Google Solar
buildingInsights JSON (uow.solar), projects it once per Property into a typed
SolarPotential via `_solar_potential_for` (None for a missing or error
payload), and threads it into `recommend_solar` alongside planning_restrictions
— mirroring the ASHP wiring. Solar fires only when a feasible potential is
present, so dwellings without fetched solar data are unaffected.

FakeSolarRepo now returns None for an unseeded Property (was raising) and
supports `by_property` seeding, so the orchestrator's new solar read is exercised.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-08 12:22:56 +00:00

115 lines
3.9 KiB
Python

"""Slice 8 — the ModellingOrchestrator threads a Property's Google Solar
potential (SolarRepository → typed SolarPotential) into the solar Recommendation
Generator (ADR-0026), mirroring how planning_restrictions is threaded.
Tests the new branching directly: the projection guard (`_solar_potential_for`)
and the candidate wiring (`_candidate_recommendations` includes a Solar PV
Recommendation only when a feasible potential is present). The end-to-end
run-through-repos path is covered by the DB integration tests; here we keep the
seam fast and isolated.
"""
import json
from pathlib import Path
from typing import Any
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.geospatial.planning_restrictions import PlanningRestrictions
from domain.modelling.product import Product
from domain.modelling.recommendation import Recommendation
from orchestration.modelling_orchestrator import (
_candidate_recommendations, # pyright: ignore[reportPrivateUsage]
_solar_potential_for, # pyright: ignore[reportPrivateUsage]
)
from repositories.product.product_repository import ProductRepository
from tests.domain.modelling._elmhurst_recommendation import (
parse_recommendation_summary,
)
from tests.orchestration.fakes import FakeSolarRepo
_INSIGHTS_FIXTURE: Path = (
Path(__file__).resolve().parents[1]
/ "domain"
/ "modelling"
/ "fixtures"
/ "google_building_insights_001431.json"
)
def _insights() -> dict[str, Any]:
with _INSIGHTS_FIXTURE.open(encoding="utf-8") as handle:
data: dict[str, Any] = json.load(handle)
return data
class _StubProducts(ProductRepository):
def get(self, measure_type: str) -> Product:
return Product(
measure_type=measure_type,
unit_cost_per_m2=0.0,
contingency_rate=0.15,
id=909,
)
def _eligible_house() -> EpcPropertyData:
return parse_recommendation_summary("solar_pv_001431_before.pdf")
def test_solar_potential_for_returns_none_when_no_insights() -> None:
# Arrange — an unseeded Property: Ingestion fetched no solar data.
solar = FakeSolarRepo()
# Act / Assert
assert _solar_potential_for(solar, property_id=42) is None
def test_solar_potential_for_returns_none_for_an_error_payload() -> None:
# Arrange — the Solar API found no building; Ingestion persisted the error
# dict (no `solarPotential` block).
solar = FakeSolarRepo(by_property={7: {"error": "ENTITY_NOT_FOUND"}})
# Act / Assert
assert _solar_potential_for(solar, property_id=7) is None
def test_solar_potential_for_projects_valid_insights() -> None:
# Arrange
solar = FakeSolarRepo(by_property={7: _insights()})
# Act
potential = _solar_potential_for(solar, property_id=7)
# Assert — the real London example projects to the 46-rung ladder.
assert potential is not None
assert abs(potential.panel_capacity_watts - 400.0) <= 1e-4
assert len(potential.configurations) == 46
def test_candidate_recommendations_includes_solar_when_potential_present() -> None:
# Arrange — a solar-eligible house with a feasible potential.
epc = _eligible_house()
potential = _solar_potential_for(
FakeSolarRepo(by_property={1: _insights()}), property_id=1
)
# Act
recommendations: list[Recommendation] = _candidate_recommendations(
epc, _StubProducts(), PlanningRestrictions(), potential
)
# Assert — a "Solar PV" Recommendation is among the candidates.
assert "Solar PV" in {r.surface for r in recommendations}
def test_candidate_recommendations_excludes_solar_without_potential() -> None:
# Arrange — same house, but no solar potential threaded.
epc = _eligible_house()
# Act
recommendations = _candidate_recommendations(
epc, _StubProducts(), PlanningRestrictions(), None
)
# Assert
assert "Solar PV" not in {r.surface for r in recommendations}