mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-30 13:10:47 +00:00
239 lines
9.4 KiB
Python
239 lines
9.4 KiB
Python
from unittest.mock import patch
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import pytest
|
|
from botocore.exceptions import ClientError
|
|
|
|
from domain.epc import historic_epc_matching as matcher_mod
|
|
from domain.epc.historic_epc_matching import (
|
|
HistoricEpcMatches,
|
|
ScoredHistoricEpc,
|
|
_sanitise_postcode,
|
|
match_addresses_for_postcode,
|
|
)
|
|
|
|
|
|
# Columns required by the HistoricEpc dataclass (lower-cased CSV columns).
|
|
# The matcher only reads ADDRESS + UPRN to score; everything else is filled
|
|
# with "" but must be present for HistoricEpc(**kwargs) to construct.
|
|
_FULL_COLUMN_FIELDS = [
|
|
"LMK_KEY", "ADDRESS1", "ADDRESS2", "ADDRESS3", "POSTCODE",
|
|
"BUILDING_REFERENCE_NUMBER", "CURRENT_ENERGY_RATING", "POTENTIAL_ENERGY_RATING",
|
|
"CURRENT_ENERGY_EFFICIENCY", "POTENTIAL_ENERGY_EFFICIENCY", "PROPERTY_TYPE",
|
|
"BUILT_FORM", "INSPECTION_DATE", "LOCAL_AUTHORITY", "CONSTITUENCY", "COUNTY",
|
|
"LODGEMENT_DATE", "TRANSACTION_TYPE", "ENVIRONMENT_IMPACT_CURRENT",
|
|
"ENVIRONMENT_IMPACT_POTENTIAL", "ENERGY_CONSUMPTION_CURRENT",
|
|
"ENERGY_CONSUMPTION_POTENTIAL", "CO2_EMISSIONS_CURRENT",
|
|
"CO2_EMISS_CURR_PER_FLOOR_AREA", "CO2_EMISSIONS_POTENTIAL",
|
|
"LIGHTING_COST_CURRENT", "LIGHTING_COST_POTENTIAL", "HEATING_COST_CURRENT",
|
|
"HEATING_COST_POTENTIAL", "HOT_WATER_COST_CURRENT", "HOT_WATER_COST_POTENTIAL",
|
|
"TOTAL_FLOOR_AREA", "ENERGY_TARIFF", "MAINS_GAS_FLAG", "FLOOR_LEVEL",
|
|
"FLAT_TOP_STOREY", "FLAT_STOREY_COUNT", "MAIN_HEATING_CONTROLS",
|
|
"MULTI_GLAZE_PROPORTION", "GLAZED_TYPE", "GLAZED_AREA", "EXTENSION_COUNT",
|
|
"NUMBER_HABITABLE_ROOMS", "NUMBER_HEATED_ROOMS", "LOW_ENERGY_LIGHTING",
|
|
"NUMBER_OPEN_FIREPLACES", "HOTWATER_DESCRIPTION", "HOT_WATER_ENERGY_EFF",
|
|
"HOT_WATER_ENV_EFF", "FLOOR_DESCRIPTION", "FLOOR_ENERGY_EFF", "FLOOR_ENV_EFF",
|
|
"WINDOWS_DESCRIPTION", "WINDOWS_ENERGY_EFF", "WINDOWS_ENV_EFF",
|
|
"WALLS_DESCRIPTION", "WALLS_ENERGY_EFF", "WALLS_ENV_EFF",
|
|
"SECONDHEAT_DESCRIPTION", "SHEATING_ENERGY_EFF", "SHEATING_ENV_EFF",
|
|
"ROOF_DESCRIPTION", "ROOF_ENERGY_EFF", "ROOF_ENV_EFF", "MAINHEAT_DESCRIPTION",
|
|
"MAINHEAT_ENERGY_EFF", "MAINHEAT_ENV_EFF", "MAINHEATCONT_DESCRIPTION",
|
|
"MAINHEATC_ENERGY_EFF", "MAINHEATC_ENV_EFF", "LIGHTING_DESCRIPTION",
|
|
"LIGHTING_ENERGY_EFF", "LIGHTING_ENV_EFF", "MAIN_FUEL", "WIND_TURBINE_COUNT",
|
|
"HEAT_LOSS_CORRIDOR", "UNHEATED_CORRIDOR_LENGTH", "FLOOR_HEIGHT",
|
|
"PHOTO_SUPPLY", "SOLAR_WATER_HEATING_FLAG", "MECHANICAL_VENTILATION",
|
|
"ADDRESS", "LOCAL_AUTHORITY_LABEL", "CONSTITUENCY_LABEL", "POSTTOWN",
|
|
"CONSTRUCTION_AGE_BAND", "LODGEMENT_DATETIME", "TENURE",
|
|
"FIXED_LIGHTING_OUTLETS_COUNT", "LOW_ENERGY_FIXED_LIGHT_COUNT", "UPRN",
|
|
"UPRN_SOURCE", "REPORT_TYPE",
|
|
]
|
|
|
|
|
|
def _row(address: str, uprn) -> dict:
|
|
row = {col: "" for col in _FULL_COLUMN_FIELDS}
|
|
row["ADDRESS"] = address
|
|
row["UPRN"] = uprn
|
|
return row
|
|
|
|
|
|
def _build_df(rows: list[dict]) -> pd.DataFrame:
|
|
return pd.DataFrame(rows, columns=_FULL_COLUMN_FIELDS)
|
|
|
|
|
|
@pytest.fixture
|
|
def patch_postcode_valid():
|
|
with patch.object(matcher_mod.AddressMatch, "is_valid_postcode", return_value=True) as m:
|
|
yield m
|
|
|
|
|
|
@pytest.fixture
|
|
def patch_read():
|
|
with patch.object(matcher_mod, "read_csv_gz_from_s3") as m:
|
|
yield m
|
|
|
|
|
|
# ---------- _sanitise_postcode ----------
|
|
|
|
|
|
class TestSanitisePostcode:
|
|
|
|
def test_uppercases_and_strips_spaces(self, patch_postcode_valid):
|
|
assert _sanitise_postcode("ab33 8al") == "AB338AL"
|
|
|
|
def test_empty_raises(self, patch_postcode_valid):
|
|
with pytest.raises(ValueError, match="non-whitespace"):
|
|
_sanitise_postcode("")
|
|
|
|
def test_whitespace_only_raises(self, patch_postcode_valid):
|
|
with pytest.raises(ValueError, match="non-whitespace"):
|
|
_sanitise_postcode(" ")
|
|
|
|
def test_invalid_postcode_raises(self):
|
|
with patch.object(
|
|
matcher_mod.AddressMatch, "is_valid_postcode", return_value=False
|
|
):
|
|
with pytest.raises(ValueError, match="not a valid UK postcode"):
|
|
_sanitise_postcode("NONSENSE")
|
|
|
|
|
|
# ---------- match_addresses_for_postcode ----------
|
|
|
|
|
|
class TestMatchAddressesForPostcode:
|
|
|
|
def test_preserves_row_count_including_zero_score_rows(
|
|
self, patch_read, patch_postcode_valid
|
|
):
|
|
# Disjoint number sets => hard zero. Still kept in matches.
|
|
patch_read.return_value = _build_df([
|
|
_row("47 GORDON ROAD", "100"),
|
|
_row("999 SOMEWHERE ELSE", "200"),
|
|
])
|
|
result = match_addresses_for_postcode("47 Gordon Road", "AB33 8AL")
|
|
assert isinstance(result, HistoricEpcMatches)
|
|
assert len(result.matches) == 2
|
|
|
|
def test_top_has_lexirank_one_and_lexiscore_monotone(
|
|
self, patch_read, patch_postcode_valid
|
|
):
|
|
patch_read.return_value = _build_df([
|
|
_row("48 GORDON ROAD", "200"), # near miss
|
|
_row("47 GORDON ROAD", "100"), # exact (after normalisation)
|
|
])
|
|
result = match_addresses_for_postcode("47 Gordon Road", "AB33 8AL")
|
|
assert result.top().lexirank == 1
|
|
scores = [m.lexiscore for m in result.matches]
|
|
assert scores == sorted(scores, reverse=True)
|
|
|
|
def test_s3_key_built_from_default_root(self, patch_read, patch_postcode_valid):
|
|
patch_read.return_value = _build_df([_row("47 GORDON ROAD", "100")])
|
|
match_addresses_for_postcode("47 Gordon Road", "AB33 8AL")
|
|
patch_read.assert_called_once_with(
|
|
"retrofit-data-dev", "historical_epc/AB338AL/data.csv.gz"
|
|
)
|
|
|
|
def test_s3_key_respects_custom_root_with_trailing_slash(
|
|
self, patch_read, patch_postcode_valid
|
|
):
|
|
patch_read.return_value = _build_df([_row("47 GORDON ROAD", "100")])
|
|
match_addresses_for_postcode(
|
|
"47 Gordon Road",
|
|
"AB33 8AL",
|
|
s3_root="s3://my-bucket/some/prefix/",
|
|
)
|
|
patch_read.assert_called_once_with(
|
|
"my-bucket", "some/prefix/AB338AL/data.csv.gz"
|
|
)
|
|
|
|
def test_no_such_key_translates_to_filenotfound(
|
|
self, patch_read, patch_postcode_valid
|
|
):
|
|
patch_read.side_effect = ClientError(
|
|
{"Error": {"Code": "NoSuchKey", "Message": "missing"}}, "GetObject"
|
|
)
|
|
with pytest.raises(FileNotFoundError):
|
|
match_addresses_for_postcode("47 Gordon Road", "AB33 8AL")
|
|
|
|
def test_other_client_error_propagates(self, patch_read, patch_postcode_valid):
|
|
patch_read.side_effect = ClientError(
|
|
{"Error": {"Code": "AccessDenied", "Message": "nope"}}, "GetObject"
|
|
)
|
|
with pytest.raises(ClientError):
|
|
match_addresses_for_postcode("47 Gordon Road", "AB33 8AL")
|
|
|
|
def test_empty_user_address_raises(self, patch_postcode_valid):
|
|
with pytest.raises(ValueError, match="user_address"):
|
|
match_addresses_for_postcode("", "AB33 8AL")
|
|
|
|
|
|
# ---------- unambiguous_uprn ----------
|
|
|
|
|
|
class TestUnambiguousUprn:
|
|
|
|
def test_exact_match_returns_uprn(self, patch_read, patch_postcode_valid):
|
|
patch_read.return_value = _build_df([
|
|
_row("47 GORDON ROAD", "100"),
|
|
_row("48 GORDON ROAD", "200"),
|
|
])
|
|
result = match_addresses_for_postcode("47 Gordon Road", "AB33 8AL")
|
|
assert result.unambiguous_uprn() == "100"
|
|
|
|
def test_ambiguous_tie_returns_none(self, patch_read, patch_postcode_valid):
|
|
# Two duplicate addresses with different UPRNs share rank-1.
|
|
patch_read.return_value = _build_df([
|
|
_row("47 GORDON ROAD", "100"),
|
|
_row("47 GORDON ROAD", "200"),
|
|
])
|
|
result = match_addresses_for_postcode("47 Gordon Road", "AB33 8AL")
|
|
assert result.unambiguous_uprn() is None
|
|
|
|
def test_all_zero_score_returns_none_even_when_uprn_unique(
|
|
self, patch_read, patch_postcode_valid
|
|
):
|
|
# User address has building number 47; no row has 47 -> all hard-zero.
|
|
patch_read.return_value = _build_df([
|
|
_row("999 ELSEWHERE", "100"),
|
|
_row("888 ELSEWHERE", "200"),
|
|
])
|
|
result = match_addresses_for_postcode("47 Gordon Road", "AB33 8AL")
|
|
assert all(m.lexiscore == 0.0 for m in result.matches)
|
|
assert result.unambiguous_uprn() is None
|
|
|
|
def test_nan_uprn_becomes_empty_string_not_nan(
|
|
self, patch_read, patch_postcode_valid
|
|
):
|
|
# Use a real NaN in the UPRN cell.
|
|
patch_read.return_value = _build_df([
|
|
_row("47 GORDON ROAD", np.nan),
|
|
_row("48 GORDON ROAD", "200"),
|
|
])
|
|
result = match_addresses_for_postcode("47 Gordon Road", "AB33 8AL")
|
|
top = result.top()
|
|
# pandas_cell_to_str must turn NaN/"nan" into "" (not the literal string "nan"),
|
|
# so unambiguous_uprn's truthiness check correctly drops the row.
|
|
assert top.record.uprn == ""
|
|
|
|
|
|
# ---------- top / top_n ----------
|
|
|
|
|
|
class TestTopHelpers:
|
|
|
|
def test_top_n_returns_first_k(self, patch_read, patch_postcode_valid):
|
|
patch_read.return_value = _build_df([
|
|
_row("47 GORDON ROAD", "100"),
|
|
_row("48 GORDON ROAD", "200"),
|
|
_row("49 GORDON ROAD", "300"),
|
|
])
|
|
result = match_addresses_for_postcode("47 Gordon Road", "AB33 8AL")
|
|
top2 = result.top_n(2)
|
|
assert len(top2) == 2
|
|
assert all(isinstance(m, ScoredHistoricEpc) for m in top2)
|
|
|
|
def test_top_on_empty_matches_returns_none(self):
|
|
empty = HistoricEpcMatches(user_address="x", postcode="AB338AL", matches=[])
|
|
assert empty.top() is None
|
|
assert empty.top_n(5) == []
|
|
assert empty.unambiguous_uprn() is None
|