diff --git a/datatypes/epc/domain/historic_epc_matching.py b/datatypes/epc/domain/historic_epc_matching.py index ce6d84f79..d31c918fd 100644 --- a/datatypes/epc/domain/historic_epc_matching.py +++ b/datatypes/epc/domain/historic_epc_matching.py @@ -56,6 +56,19 @@ class HistoricEpcMatches: return next(iter(uprns)) if len(uprns) == 1 else None +def rank_historic_epc( + records: list[HistoricEpc], + user_address: str, + *, + address_column: str = "ADDRESS", + uprn_column: str = "UPRN", +) -> list[ScoredHistoricEpc]: + """Score ``records`` against ``user_address`` (best first), keeping every + record — including hard-zero non-matches. The pure scoring half of the + historic-EPC lookup: no I/O, so it is unit-testable without S3.""" + raise NotImplementedError + + def _sanitise_postcode(postcode: str) -> str: cleaned = (postcode or "").upper().replace(" ", "") if not cleaned: diff --git a/datatypes/epc/domain/tests/test_rank_historic_epc.py b/datatypes/epc/domain/tests/test_rank_historic_epc.py new file mode 100644 index 000000000..8cf102a30 --- /dev/null +++ b/datatypes/epc/domain/tests/test_rank_historic_epc.py @@ -0,0 +1,55 @@ +"""rank_historic_epc scores already-fetched HistoricEpc records by address. + +The pure scoring half of the historic-EPC lookup — it takes records (not a +postcode) so it runs with no S3, and is the piece the HistoricEpcResolver plugs +on top of the repository. +""" + +from __future__ import annotations + +import dataclasses + +import pytest + +from datatypes.epc.domain.historic_epc import HistoricEpc +from datatypes.epc.domain.historic_epc_matching import ( + ScoredHistoricEpc, + rank_historic_epc, +) + + +def _hist(address: str, uprn: str) -> HistoricEpc: + fields = {f.name: "" for f in dataclasses.fields(HistoricEpc)} + fields["address"] = address + fields["uprn"] = uprn + return HistoricEpc(**fields) + + +def test_ranks_records_best_first_keeping_zero_score_rows(): + # Arrange — a near-disjoint non-match (kept) and the exact match (second). + records = [ + _hist("999 SOMEWHERE ELSE", "200"), + _hist("47 GORDON ROAD", "100"), + ] + + # Act + result = rank_historic_epc(records, "47 Gordon Road") + + # Assert + assert all(isinstance(s, ScoredHistoricEpc) for s in result) + assert result[0].record.address == "47 GORDON ROAD" + assert result[0].lexirank == 1 + assert len(result) == 2 # zero-score row is kept, not dropped + scores = [s.lexiscore for s in result] + assert scores == sorted(scores, reverse=True) + + +def test_empty_records_returns_empty_list(): + # Act / Assert + assert rank_historic_epc([], "47 Gordon Road") == [] + + +def test_empty_user_address_raises(): + # Act / Assert + with pytest.raises(ValueError, match="user_address"): + rank_historic_epc([_hist("47 GORDON ROAD", "100")], "")