mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
Remove EPC and asset_list changes unrelated to SAL handler
This branch's objective is the SAL ingestion handler (applications/SAL/handler.py) and its dependency tree. Drop work that crept in but is unreferenced by it: - EPC feature: domain/epc, infrastructure/epc (gov_uk + historical clients), tests/infrastructure/epc - datatypes/epc edits (instantaneous_wwhrs Optional) reverted to main - asset_list/app.py local data-file/column tweak reverted to main Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jun-te Kim
parent
a747534f37
commit
96aeed4f2e
25 changed files with 21 additions and 625 deletions
|
|
@ -79,17 +79,17 @@ def app():
|
|||
"""
|
||||
|
||||
data_folder = "/workspaces/model/asset_list"
|
||||
data_filename = "asset_list (8).xlsx"
|
||||
sheet_name = "Standardised Asset List"
|
||||
postcode_column = "postcode"
|
||||
address1_column = "domna_address_1"
|
||||
data_filename = "hyde.xlsx"
|
||||
sheet_name = "AddressProfilingResults"
|
||||
postcode_column = "Postcode"
|
||||
address1_column = "Address"
|
||||
address1_method = None
|
||||
fulladdress_column = "domna_address_1"
|
||||
fulladdress_column = "Postcode"
|
||||
address_cols_to_concat = []
|
||||
missing_postcodes_method = None
|
||||
landlord_year_built = None
|
||||
landlord_os_uprn = None
|
||||
landlord_property_type = "landlord_property_id" # Good to include if landlord gave
|
||||
landlord_property_type = "Property Type" # Good to include if landlord gave
|
||||
landlord_built_form = None # Good to include if landlord gave
|
||||
landlord_wall_construction = None
|
||||
landlord_roof_construction = None
|
||||
|
|
@ -468,3 +468,4 @@ def app():
|
|||
asset_list.duplicated_addresses.to_excel(
|
||||
writer, sheet_name="Duplicate Properties", index=False
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -29,9 +29,7 @@ class MainHeatingDetail:
|
|||
boiler_flue_type: Optional[int] = None # TODO: make enum?
|
||||
boiler_ignition_type: Optional[int] = None # TODO: make enum?
|
||||
central_heating_pump_age: Optional[int] = None
|
||||
central_heating_pump_age_str: Optional[str] = (
|
||||
None # str from site notes e.g. "Unknown", "Pre 2013"
|
||||
)
|
||||
central_heating_pump_age_str: Optional[str] = None # str from site notes e.g. "Unknown", "Pre 2013"
|
||||
main_heating_index_number: Optional[int] = None
|
||||
sap_main_heating_code: Optional[int] = None # TODO: make enum?
|
||||
main_heating_number: Optional[int] = None
|
||||
|
|
@ -56,7 +54,7 @@ class ShowerOutlets:
|
|||
|
||||
@dataclass
|
||||
class SapHeating:
|
||||
instantaneous_wwhrs: Optional[InstantaneousWwhrs]
|
||||
instantaneous_wwhrs: InstantaneousWwhrs
|
||||
main_heating_details: List[MainHeatingDetail]
|
||||
has_fixed_air_conditioning: bool
|
||||
cylinder_size: Optional[Union[int, str]] = (
|
||||
|
|
@ -69,9 +67,7 @@ class SapHeating:
|
|||
cylinder_insulation_type: Optional[Union[int, str]] = None
|
||||
cylinder_thermostat: Optional[str] = None
|
||||
secondary_fuel_type: Optional[int] = None
|
||||
secondary_heating_type: Optional[Union[int, str]] = (
|
||||
None # int from API; str from site notes
|
||||
)
|
||||
secondary_heating_type: Optional[Union[int, str]] = None # int from API; str from site notes
|
||||
cylinder_insulation_thickness_mm: Optional[int] = None
|
||||
|
||||
|
||||
|
|
@ -79,9 +75,7 @@ class SapHeating:
|
|||
class SapVentilation:
|
||||
ventilation_type: Optional[str] = None
|
||||
draught_lobby: Optional[bool] = None
|
||||
pressure_test: Optional[str] = (
|
||||
None # str from site notes e.g. "No test"; int in API via mechanical_ventilation
|
||||
)
|
||||
pressure_test: Optional[str] = None # str from site notes e.g. "No test"; int in API via mechanical_ventilation
|
||||
open_flues_count: Optional[int] = None
|
||||
closed_flues_count: Optional[int] = None
|
||||
boiler_flues_count: Optional[int] = None
|
||||
|
|
@ -225,12 +219,8 @@ class SapBuildingPart:
|
|||
None # TODO: make enum/mapping?
|
||||
)
|
||||
floor_type: Optional[str] = None # str from site notes e.g. "Ground Floor"
|
||||
floor_construction_type: Optional[str] = (
|
||||
None # str from site notes; distinct from floor_construction: int in SapFloorDimension
|
||||
)
|
||||
floor_insulation_type_str: Optional[str] = (
|
||||
None # str from site notes e.g. "As Built"
|
||||
)
|
||||
floor_construction_type: Optional[str] = None # str from site notes; distinct from floor_construction: int in SapFloorDimension
|
||||
floor_insulation_type_str: Optional[str] = None # str from site notes e.g. "As Built"
|
||||
floor_u_value_known: Optional[bool] = None
|
||||
|
||||
roof_construction: Optional[int] = None
|
||||
|
|
|
|||
|
|
@ -37,7 +37,7 @@ class SapHeating:
|
|||
cylinder_size: int
|
||||
water_heating_code: int
|
||||
water_heating_fuel: int
|
||||
instantaneous_wwhrs: Optional[InstantaneousWwhrs]
|
||||
instantaneous_wwhrs: InstantaneousWwhrs
|
||||
main_heating_details: List[MainHeatingDetail]
|
||||
immersion_heating_type: Union[int, str]
|
||||
cylinder_insulation_type: int
|
||||
|
|
|
|||
|
|
@ -41,7 +41,7 @@ class SapHeating:
|
|||
cylinder_size: int
|
||||
water_heating_code: int
|
||||
water_heating_fuel: int
|
||||
instantaneous_wwhrs: Optional[InstantaneousWwhrs]
|
||||
instantaneous_wwhrs: InstantaneousWwhrs
|
||||
main_heating_details: List[MainHeatingDetail]
|
||||
immersion_heating_type: Union[int, str]
|
||||
cylinder_insulation_type: int
|
||||
|
|
|
|||
|
|
@ -41,7 +41,7 @@ class SapHeating:
|
|||
cylinder_size: int
|
||||
water_heating_code: int
|
||||
water_heating_fuel: int
|
||||
instantaneous_wwhrs: Optional[InstantaneousWwhrs]
|
||||
instantaneous_wwhrs: InstantaneousWwhrs
|
||||
main_heating_details: List[MainHeatingDetail]
|
||||
immersion_heating_type: Union[int, str]
|
||||
has_fixed_air_conditioning: str
|
||||
|
|
@ -86,7 +86,6 @@ class SapFloorDimension:
|
|||
@dataclass
|
||||
class SapRoomInRoof:
|
||||
"""Room-in-roof details. floor_area is a Measurement object in schema 18.0."""
|
||||
|
||||
floor_area: Measurement
|
||||
insulation: str
|
||||
roof_room_connected: str
|
||||
|
|
|
|||
|
|
@ -41,7 +41,7 @@ class SapHeating:
|
|||
cylinder_size: int
|
||||
water_heating_code: int
|
||||
water_heating_fuel: int
|
||||
instantaneous_wwhrs: Optional[InstantaneousWwhrs]
|
||||
instantaneous_wwhrs: InstantaneousWwhrs
|
||||
main_heating_details: List[MainHeatingDetail]
|
||||
immersion_heating_type: Union[int, str]
|
||||
has_fixed_air_conditioning: str
|
||||
|
|
|
|||
|
|
@ -49,7 +49,7 @@ class SapHeating:
|
|||
cylinder_size: int
|
||||
water_heating_code: int
|
||||
water_heating_fuel: int
|
||||
instantaneous_wwhrs: Optional[InstantaneousWwhrs]
|
||||
instantaneous_wwhrs: InstantaneousWwhrs
|
||||
main_heating_details: List[MainHeatingDetail]
|
||||
immersion_heating_type: Union[int, str]
|
||||
has_fixed_air_conditioning: str
|
||||
|
|
@ -103,7 +103,6 @@ class SapFloorDimension:
|
|||
@dataclass
|
||||
class SapRoomInRoof:
|
||||
"""Room-in-roof details. floor_area is a plain number in schema 20.0.0 (not a Measurement object)."""
|
||||
|
||||
floor_area: Union[int, float]
|
||||
insulation: str
|
||||
roof_room_connected: str
|
||||
|
|
|
|||
|
|
@ -33,7 +33,6 @@ class ShowerOutlets:
|
|||
@dataclass
|
||||
class InstantaneousWwhrs:
|
||||
"""Changed in 21.0.0: references WWHRS product index numbers instead of room counts."""
|
||||
|
||||
wwhrs_index_number1: Optional[int] = None
|
||||
wwhrs_index_number2: Optional[int] = None
|
||||
|
||||
|
|
@ -62,7 +61,7 @@ class SapHeating:
|
|||
cylinder_size: int
|
||||
water_heating_code: int
|
||||
water_heating_fuel: int
|
||||
instantaneous_wwhrs: Optional[InstantaneousWwhrs]
|
||||
instantaneous_wwhrs: InstantaneousWwhrs
|
||||
main_heating_details: List[MainHeatingDetail]
|
||||
immersion_heating_type: Union[int, str]
|
||||
has_fixed_air_conditioning: str
|
||||
|
|
@ -155,7 +154,6 @@ class SapFloorDimension:
|
|||
@dataclass
|
||||
class SapRoomInRoof:
|
||||
"""Room-in-roof details. insulation and roof_room_connected removed in schema 21.0.0."""
|
||||
|
||||
floor_area: Union[int, float]
|
||||
construction_age_band: str
|
||||
|
||||
|
|
|
|||
|
|
@ -50,7 +50,7 @@ class MainHeatingDetail:
|
|||
main_heating_fraction: int
|
||||
main_heating_data_source: int
|
||||
boiler_flue_type: Optional[int] = None
|
||||
fan_flue_present: Optional[str] = None # TODO: make bool
|
||||
fan_flue_present: Optional[str] = None # TODO: make bool
|
||||
boiler_ignition_type: Optional[int] = None
|
||||
central_heating_pump_age: Optional[int] = None
|
||||
main_heating_index_number: Optional[int] = None
|
||||
|
|
@ -62,7 +62,7 @@ class SapHeating:
|
|||
cylinder_size: int
|
||||
water_heating_code: int
|
||||
water_heating_fuel: int
|
||||
instantaneous_wwhrs: Optional[InstantaneousWwhrs]
|
||||
instantaneous_wwhrs: InstantaneousWwhrs
|
||||
main_heating_details: List[MainHeatingDetail]
|
||||
immersion_heating_type: Union[int, str]
|
||||
has_fixed_air_conditioning: str
|
||||
|
|
|
|||
|
|
@ -1,4 +0,0 @@
|
|||
from domain.epc.epc_record import EpcRecord
|
||||
from domain.epc.property_type import PropertyType
|
||||
|
||||
__all__ = ["EpcRecord", "PropertyType"]
|
||||
|
|
@ -1,21 +0,0 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional
|
||||
|
||||
from domain.epc.property_type import PropertyType
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class EpcRecord:
|
||||
"""A streamlined record of EPC property data.
|
||||
|
||||
A focused subset of the full ``EpcPropertyData``: a property's identity
|
||||
plus its typed property type. Grow this with further fields as the
|
||||
domain needs them.
|
||||
"""
|
||||
|
||||
address_line_1: str
|
||||
postcode: str
|
||||
uprn: Optional[int]
|
||||
property_type: PropertyType
|
||||
|
|
@ -1,9 +0,0 @@
|
|||
from enum import Enum
|
||||
|
||||
|
||||
class PropertyType(Enum):
|
||||
HOUSE = "House"
|
||||
BUNGALOW = "Bungalow"
|
||||
FLAT = "Flat"
|
||||
MAISONETTE = "Maisonette"
|
||||
PARK_HOME = "Park home"
|
||||
|
|
@ -1,13 +0,0 @@
|
|||
from infrastructure.epc.epc_client import EpcClient
|
||||
from infrastructure.epc.exceptions import (
|
||||
EpcApiError,
|
||||
EpcNotFoundError,
|
||||
EpcRateLimitError,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"EpcApiError",
|
||||
"EpcClient",
|
||||
"EpcNotFoundError",
|
||||
"EpcRateLimitError",
|
||||
]
|
||||
|
|
@ -1,41 +0,0 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Optional
|
||||
|
||||
from datatypes.epc.domain.epc_property_data import EpcPropertyData
|
||||
from datatypes.epc.search import EpcSearchResult
|
||||
|
||||
|
||||
class EpcClient(ABC):
|
||||
"""Interface for retrieving EPC (Energy Performance Certificate) data.
|
||||
|
||||
Implementations fetch from a data source and return domain objects;
|
||||
callers depend only on this interface, not on a concrete transport.
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def search_by_postcode(self, postcode: str) -> list[EpcSearchResult]:
|
||||
"""Return the EPC certificates registered at ``postcode``.
|
||||
|
||||
Returns an empty list when the postcode has no certificates.
|
||||
"""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def get_by_certificate_number(
|
||||
self, certificate_number: str
|
||||
) -> EpcPropertyData:
|
||||
"""Return the full EPC record for a certificate number.
|
||||
|
||||
Raises EpcNotFoundError when no such certificate exists.
|
||||
"""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def get_by_uprn(self, uprn: int) -> Optional[EpcPropertyData]:
|
||||
"""Return the most recent EPC record for ``uprn``.
|
||||
|
||||
Returns None when the UPRN has no certificates.
|
||||
"""
|
||||
...
|
||||
|
|
@ -1,17 +0,0 @@
|
|||
from typing import Optional
|
||||
|
||||
|
||||
class EpcApiError(Exception):
|
||||
"""Base for all EPC client errors."""
|
||||
|
||||
|
||||
class EpcNotFoundError(EpcApiError):
|
||||
"""Raised when the API returns 404 for a resource that must exist."""
|
||||
|
||||
|
||||
class EpcRateLimitError(EpcApiError):
|
||||
"""Raised when the API returns 429 and all retries are exhausted."""
|
||||
|
||||
def __init__(self, message: str, retry_after: Optional[float] = None) -> None:
|
||||
super().__init__(message)
|
||||
self.retry_after = retry_after
|
||||
|
|
@ -1,6 +0,0 @@
|
|||
from infrastructure.epc.gov_uk.gov_uk_epc_client import GovUkEpcClient
|
||||
from infrastructure.epc.gov_uk.gov_uk_property_type import (
|
||||
property_type_from_gov_uk_code,
|
||||
)
|
||||
|
||||
__all__ = ["GovUkEpcClient", "property_type_from_gov_uk_code"]
|
||||
|
|
@ -1,34 +0,0 @@
|
|||
import time
|
||||
from typing import Callable, Optional, TypeVar
|
||||
|
||||
from infrastructure.epc.exceptions import EpcRateLimitError
|
||||
|
||||
T = TypeVar("T")
|
||||
|
||||
|
||||
def call_with_retry(
|
||||
fn: Callable[[], T],
|
||||
max_retries: int = 5,
|
||||
backoff_base: float = 1.0,
|
||||
backoff_multiplier: float = 2.0,
|
||||
max_backoff: float = 60.0,
|
||||
) -> T:
|
||||
"""Call ``fn``, retrying on EpcRateLimitError with exponential backoff.
|
||||
|
||||
Honours the API's ``Retry-After`` header when present, otherwise backs off
|
||||
``backoff_base * backoff_multiplier ** attempt`` (capped at ``max_backoff``).
|
||||
"""
|
||||
last_exc: Optional[EpcRateLimitError] = None
|
||||
for attempt in range(max_retries + 1):
|
||||
try:
|
||||
return fn()
|
||||
except EpcRateLimitError as exc:
|
||||
last_exc = exc
|
||||
if attempt < max_retries:
|
||||
if exc.retry_after is not None:
|
||||
delay = exc.retry_after
|
||||
else:
|
||||
delay = backoff_base * (backoff_multiplier**attempt)
|
||||
time.sleep(min(delay, max_backoff))
|
||||
assert last_exc is not None
|
||||
raise last_exc
|
||||
|
|
@ -1,132 +0,0 @@
|
|||
# Spec: https://raw.githubusercontent.com/communitiesuk/epb-data-warehouse/main/api/api.yml
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Optional
|
||||
|
||||
import httpx
|
||||
|
||||
from datatypes.epc.domain.epc_property_data import EpcPropertyData
|
||||
from datatypes.epc.domain.mapper import EpcPropertyDataMapper
|
||||
from datatypes.epc.search import EpcSearchResult
|
||||
from infrastructure.epc.epc_client import EpcClient
|
||||
from infrastructure.epc.exceptions import (
|
||||
EpcApiError,
|
||||
EpcNotFoundError,
|
||||
EpcRateLimitError,
|
||||
)
|
||||
from infrastructure.epc.gov_uk._retry import call_with_retry
|
||||
|
||||
|
||||
class GovUkEpcClient(EpcClient):
|
||||
"""EpcClient backed by the live gov.uk EPC API.
|
||||
|
||||
Endpoint: https://api.get-energy-performance-data.communities.gov.uk
|
||||
"""
|
||||
|
||||
BASE_URL = "https://api.get-energy-performance-data.communities.gov.uk"
|
||||
REQUEST_TIMEOUT = 10.0
|
||||
|
||||
def __init__(self, auth_token: str) -> None:
|
||||
self._headers = {
|
||||
"Authorization": f"Bearer {auth_token}",
|
||||
"Accept": "application/json",
|
||||
}
|
||||
|
||||
def search_by_postcode(self, postcode: str) -> list[EpcSearchResult]:
|
||||
normalised = self._normalise_postcode(postcode)
|
||||
return call_with_retry(lambda: self._search(postcode=normalised))
|
||||
|
||||
def get_by_certificate_number(
|
||||
self, certificate_number: str
|
||||
) -> EpcPropertyData:
|
||||
raw = call_with_retry(lambda: self._fetch_certificate(certificate_number))
|
||||
return EpcPropertyDataMapper.from_api_response(raw)
|
||||
|
||||
def get_by_uprn(self, uprn: int) -> Optional[EpcPropertyData]:
|
||||
results = call_with_retry(lambda: self._search(uprn=uprn))
|
||||
if not results:
|
||||
return None
|
||||
latest = max(results, key=lambda r: r.registration_date)
|
||||
return self.get_by_certificate_number(latest.certificate_number)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Private helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@staticmethod
|
||||
def _normalise_postcode(postcode: str) -> str:
|
||||
"""Return the postcode with all spaces removed and uppercased."""
|
||||
return postcode.replace(" ", "").upper()
|
||||
|
||||
@staticmethod
|
||||
def _parse_retry_after(resp: httpx.Response) -> Optional[float]:
|
||||
header = resp.headers.get("Retry-After")
|
||||
if header is None:
|
||||
return None
|
||||
try:
|
||||
return float(header)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
def _fetch_certificate(self, certificate_number: str) -> dict[str, Any]:
|
||||
resp = httpx.get(
|
||||
f"{self.BASE_URL}/api/certificate",
|
||||
params={"certificate_number": certificate_number},
|
||||
headers=self._headers,
|
||||
timeout=self.REQUEST_TIMEOUT,
|
||||
)
|
||||
if resp.status_code == 404:
|
||||
raise EpcNotFoundError(certificate_number)
|
||||
if resp.status_code == 429:
|
||||
raise EpcRateLimitError(
|
||||
"Rate limited by EPC API",
|
||||
retry_after=self._parse_retry_after(resp),
|
||||
)
|
||||
if not resp.is_success:
|
||||
raise EpcApiError(f"EPC API error {resp.status_code}: {resp.text}")
|
||||
return resp.json()["data"]
|
||||
|
||||
def _search(
|
||||
self,
|
||||
postcode: Optional[str] = None,
|
||||
uprn: Optional[int] = None,
|
||||
) -> list[EpcSearchResult]:
|
||||
params: dict[str, str | int] = {}
|
||||
if postcode:
|
||||
params["postcode"] = postcode
|
||||
if uprn is not None:
|
||||
params["uprn"] = uprn
|
||||
|
||||
resp = httpx.get(
|
||||
f"{self.BASE_URL}/api/domestic/search",
|
||||
params=params,
|
||||
headers=self._headers,
|
||||
timeout=self.REQUEST_TIMEOUT,
|
||||
)
|
||||
if resp.status_code == 404:
|
||||
return []
|
||||
if resp.status_code == 429:
|
||||
raise EpcRateLimitError(
|
||||
"Rate limited by EPC API",
|
||||
retry_after=self._parse_retry_after(resp),
|
||||
)
|
||||
if not resp.is_success:
|
||||
raise EpcApiError(f"EPC API error {resp.status_code}: {resp.text}")
|
||||
|
||||
rows = resp.json().get("data", [])
|
||||
return [self._parse_search_result(row) for row in rows]
|
||||
|
||||
@staticmethod
|
||||
def _parse_search_result(row: dict[str, Any]) -> EpcSearchResult:
|
||||
return EpcSearchResult(
|
||||
certificate_number=row["certificateNumber"],
|
||||
address_line_1=row["addressLine1"],
|
||||
address_line_2=row.get("addressLine2"),
|
||||
address_line_3=row.get("addressLine3"),
|
||||
address_line_4=row.get("addressLine4"),
|
||||
postcode=row["postcode"],
|
||||
post_town=row["postTown"],
|
||||
uprn=row.get("uprn"),
|
||||
current_energy_efficiency_band=row["currentEnergyEfficiencyBand"],
|
||||
registration_date=row["registrationDate"],
|
||||
)
|
||||
|
|
@ -1,25 +0,0 @@
|
|||
from domain.epc.property_type import PropertyType
|
||||
|
||||
# GOV.UK EPC API ``property_type`` integer codes mapped to the domain type.
|
||||
# This translation is GOV.UK-specific and lives in the infrastructure layer so
|
||||
# the domain ``PropertyType`` stays free of any source encoding.
|
||||
_PROPERTY_TYPE_BY_GOV_UK_CODE: dict[int, PropertyType] = {
|
||||
0: PropertyType.HOUSE,
|
||||
1: PropertyType.BUNGALOW,
|
||||
2: PropertyType.FLAT,
|
||||
3: PropertyType.MAISONETTE,
|
||||
4: PropertyType.PARK_HOME,
|
||||
}
|
||||
|
||||
|
||||
def property_type_from_gov_uk_code(code: int) -> PropertyType:
|
||||
"""Translate a GOV.UK EPC ``property_type`` code to the domain PropertyType.
|
||||
|
||||
Raises ValueError for a code GOV.UK has not been mapped here yet.
|
||||
"""
|
||||
try:
|
||||
return _PROPERTY_TYPE_BY_GOV_UK_CODE[code]
|
||||
except KeyError:
|
||||
raise ValueError(
|
||||
f"Unknown GOV.UK EPC property type code: {code}"
|
||||
) from None
|
||||
|
|
@ -1,5 +0,0 @@
|
|||
from infrastructure.epc.historical_open_data_communities.historical_open_data_communities_epc_client import (
|
||||
HistoricalOpenDataCommunitiesEpcClient,
|
||||
)
|
||||
|
||||
__all__ = ["HistoricalOpenDataCommunitiesEpcClient"]
|
||||
|
|
@ -1,24 +0,0 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from typing import Optional
|
||||
|
||||
from domain.epc.epc_record import EpcRecord
|
||||
|
||||
|
||||
class HistoricalOpenDataCommunitiesEpcClient:
|
||||
"""EPC client backed by Open Data Communities' historical EPC data.
|
||||
|
||||
Stub — not yet implemented. Every method raises NotImplementedError for
|
||||
now. Unlike GovUkEpcClient it returns the domain ``EpcRecord`` directly;
|
||||
once the ``EpcClient`` port is migrated to return ``EpcRecord``, this
|
||||
adapter should implement it.
|
||||
"""
|
||||
|
||||
def search_by_postcode(self, postcode: str) -> list[EpcRecord]:
|
||||
raise NotImplementedError
|
||||
|
||||
def get_by_certificate_number(self, certificate_number: str) -> EpcRecord:
|
||||
raise NotImplementedError
|
||||
|
||||
def get_by_uprn(self, uprn: int) -> Optional[EpcRecord]:
|
||||
raise NotImplementedError
|
||||
|
|
@ -1,49 +0,0 @@
|
|||
import json
|
||||
import pathlib
|
||||
|
||||
import pytest
|
||||
|
||||
from infrastructure.epc.gov_uk.gov_uk_epc_client import GovUkEpcClient
|
||||
|
||||
SAMPLES_DIR = pathlib.Path("backend/epc_api/json_samples")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def rdsap_21_0_0_cert():
|
||||
return json.loads((SAMPLES_DIR / "RdSAP-Schema-21.0.0/epc.json").read_text())
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def rdsap_21_0_1_cert():
|
||||
return json.loads((SAMPLES_DIR / "RdSAP-Schema-21.0.1/epc.json").read_text())
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def epc_client():
|
||||
return GovUkEpcClient(auth_token="test-token")
|
||||
|
||||
|
||||
def make_search_row(
|
||||
cert_num="CERT-001",
|
||||
address_line_1="1 Test Street",
|
||||
postcode="SW1A 1AA",
|
||||
post_town="London",
|
||||
uprn=100023336956,
|
||||
band="D",
|
||||
registration_date="2024-01-01",
|
||||
address_line_2=None,
|
||||
address_line_3=None,
|
||||
address_line_4=None,
|
||||
):
|
||||
return {
|
||||
"certificateNumber": cert_num,
|
||||
"addressLine1": address_line_1,
|
||||
"addressLine2": address_line_2,
|
||||
"addressLine3": address_line_3,
|
||||
"addressLine4": address_line_4,
|
||||
"postcode": postcode,
|
||||
"postTown": post_town,
|
||||
"uprn": uprn,
|
||||
"currentEnergyEfficiencyBand": band,
|
||||
"registrationDate": registration_date,
|
||||
}
|
||||
|
|
@ -1,211 +0,0 @@
|
|||
from unittest.mock import MagicMock, call, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from datatypes.epc.domain.epc_property_data import EpcPropertyData
|
||||
from datatypes.epc.search import EpcSearchResult
|
||||
from infrastructure.epc.exceptions import EpcNotFoundError
|
||||
from tests.infrastructure.epc.gov_uk.conftest import make_search_row
|
||||
|
||||
_SLEEP = "infrastructure.epc.gov_uk._retry.time.sleep"
|
||||
|
||||
|
||||
def _mock_response(status_code=200, json_data=None, headers=None):
|
||||
resp = MagicMock()
|
||||
resp.status_code = status_code
|
||||
resp.is_success = 200 <= status_code < 300
|
||||
resp.json.return_value = json_data or {}
|
||||
resp.text = str(json_data)
|
||||
resp.headers = headers or {}
|
||||
return resp
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test 1: get_by_certificate_number happy path
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_get_by_certificate_number_returns_epc_property_data(
|
||||
epc_client, rdsap_21_0_1_cert
|
||||
):
|
||||
cert_response = {"data": rdsap_21_0_1_cert}
|
||||
with patch("httpx.get", return_value=_mock_response(200, cert_response)):
|
||||
result = epc_client.get_by_certificate_number("CERT-001")
|
||||
|
||||
assert isinstance(result, EpcPropertyData)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test 2: get_by_certificate_number 404 -> EpcNotFoundError
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_get_by_certificate_number_404_raises_not_found(epc_client):
|
||||
with patch("httpx.get", return_value=_mock_response(404)):
|
||||
with pytest.raises(EpcNotFoundError):
|
||||
epc_client.get_by_certificate_number("BAD-CERT")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test 3: 429 retried, succeeds on 3rd attempt
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_get_by_certificate_number_retries_on_429_and_succeeds(
|
||||
epc_client, rdsap_21_0_1_cert
|
||||
):
|
||||
cert_response = {"data": rdsap_21_0_1_cert}
|
||||
responses = [
|
||||
_mock_response(429),
|
||||
_mock_response(429),
|
||||
_mock_response(200, cert_response),
|
||||
]
|
||||
with patch("httpx.get", side_effect=responses), patch(_SLEEP):
|
||||
result = epc_client.get_by_certificate_number("CERT-001")
|
||||
|
||||
assert isinstance(result, EpcPropertyData)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test 3b: 429 with Retry-After header -> sleeps for that value
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_429_retry_after_header_drives_sleep_duration(
|
||||
epc_client, rdsap_21_0_1_cert
|
||||
):
|
||||
cert_response = {"data": rdsap_21_0_1_cert}
|
||||
responses = [
|
||||
_mock_response(429, headers={"Retry-After": "7"}),
|
||||
_mock_response(200, cert_response),
|
||||
]
|
||||
with patch("httpx.get", side_effect=responses), patch(_SLEEP) as mock_sleep:
|
||||
epc_client.get_by_certificate_number("CERT-001")
|
||||
|
||||
mock_sleep.assert_called_once_with(7.0)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test 3c: 429 without Retry-After -> falls back to exponential backoff
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_429_without_retry_after_uses_exponential_backoff(
|
||||
epc_client, rdsap_21_0_1_cert
|
||||
):
|
||||
cert_response = {"data": rdsap_21_0_1_cert}
|
||||
responses = [
|
||||
_mock_response(429),
|
||||
_mock_response(429),
|
||||
_mock_response(200, cert_response),
|
||||
]
|
||||
with patch("httpx.get", side_effect=responses), patch(_SLEEP) as mock_sleep:
|
||||
epc_client.get_by_certificate_number("CERT-001")
|
||||
|
||||
assert mock_sleep.call_args_list == [call(1.0), call(2.0)]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test 3d: malformed Retry-After header -> falls back to exponential backoff
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_429_malformed_retry_after_falls_back_to_backoff(
|
||||
epc_client, rdsap_21_0_1_cert
|
||||
):
|
||||
cert_response = {"data": rdsap_21_0_1_cert}
|
||||
responses = [
|
||||
_mock_response(429, headers={"Retry-After": "Wed, 21 Oct 2026 07:28:00 GMT"}),
|
||||
_mock_response(200, cert_response),
|
||||
]
|
||||
with patch("httpx.get", side_effect=responses), patch(_SLEEP) as mock_sleep:
|
||||
epc_client.get_by_certificate_number("CERT-001")
|
||||
|
||||
mock_sleep.assert_called_once_with(1.0)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test 3e: Retry-After capped by max_backoff to avoid hostile/buggy values
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_429_retry_after_capped_by_max_backoff(epc_client, rdsap_21_0_1_cert):
|
||||
cert_response = {"data": rdsap_21_0_1_cert}
|
||||
responses = [
|
||||
_mock_response(429, headers={"Retry-After": "9999"}),
|
||||
_mock_response(200, cert_response),
|
||||
]
|
||||
with patch("httpx.get", side_effect=responses), patch(_SLEEP) as mock_sleep:
|
||||
epc_client.get_by_certificate_number("CERT-001")
|
||||
|
||||
mock_sleep.assert_called_once_with(60.0)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test 4: get_by_uprn empty search -> None
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_get_by_uprn_returns_none_when_no_results(epc_client):
|
||||
with patch("httpx.get", return_value=_mock_response(200, {"data": []})):
|
||||
result = epc_client.get_by_uprn(100023336956)
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test 5: get_by_uprn multiple results -> fetches latest by registration_date
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_get_by_uprn_picks_most_recent_certificate(epc_client, rdsap_21_0_1_cert):
|
||||
search_rows = [
|
||||
make_search_row(cert_num="CERT-OLD", registration_date="2022-01-01"),
|
||||
make_search_row(cert_num="CERT-NEW", registration_date="2024-06-01"),
|
||||
make_search_row(cert_num="CERT-MID", registration_date="2023-03-15"),
|
||||
]
|
||||
cert_response = {"data": rdsap_21_0_1_cert}
|
||||
|
||||
def fake_get(url, params=None, **kwargs):
|
||||
if "search" in url:
|
||||
return _mock_response(200, {"data": search_rows})
|
||||
return _mock_response(200, cert_response)
|
||||
|
||||
with patch("httpx.get", side_effect=fake_get) as mock_get:
|
||||
result = epc_client.get_by_uprn(100023336956)
|
||||
|
||||
assert isinstance(result, EpcPropertyData)
|
||||
# Second call must be for the most recent cert
|
||||
cert_call = mock_get.call_args_list[1]
|
||||
assert cert_call.kwargs["params"]["certificate_number"] == "CERT-NEW"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test 6: search_by_postcode returns list[EpcSearchResult]
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_search_by_postcode_returns_results(epc_client):
|
||||
rows = [
|
||||
make_search_row(cert_num="CERT-A", address_line_1="1 High Street"),
|
||||
make_search_row(cert_num="CERT-B", address_line_1="2 High Street"),
|
||||
]
|
||||
with patch("httpx.get", return_value=_mock_response(200, {"data": rows})):
|
||||
results = epc_client.search_by_postcode("SW1A 1AA")
|
||||
|
||||
assert len(results) == 2
|
||||
assert all(isinstance(r, EpcSearchResult) for r in results)
|
||||
assert results[0].certificate_number == "CERT-A"
|
||||
assert results[1].address_line_1 == "2 High Street"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test 7: search_by_postcode 404 -> empty list
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_search_by_postcode_404_returns_empty_list(epc_client):
|
||||
with patch("httpx.get", return_value=_mock_response(404)):
|
||||
results = epc_client.search_by_postcode("ZZ9 9ZZ")
|
||||
|
||||
assert results == []
|
||||
Loading…
Add table
Reference in a new issue