mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
224 lines
8.6 KiB
Python
224 lines
8.6 KiB
Python
from unittest.mock import MagicMock, patch, call
|
|
import pytest
|
|
|
|
from backend.epc_client.client import EpcClientService, EpcSearchResult
|
|
from backend.epc_client.exceptions import EpcNotFoundError, EpcRateLimitError
|
|
from datatypes.epc.domain.epc_property_data import EpcPropertyData
|
|
from backend.epc_client.tests.conftest import make_search_row
|
|
|
|
|
|
def _mock_response(status_code=200, json_data=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)
|
|
return resp
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test 1: get_by_certificate_number happy path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_get_by_certificate_number_returns_epc_property_data(epc_service, 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_service.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_service):
|
|
with patch("httpx.get", return_value=_mock_response(404)):
|
|
with pytest.raises(EpcNotFoundError):
|
|
epc_service.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_service, 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("time.sleep"):
|
|
result = epc_service.get_by_certificate_number("CERT-001")
|
|
|
|
assert isinstance(result, EpcPropertyData)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test 4: get_by_uprn empty search → None
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_get_by_uprn_returns_none_when_no_results(epc_service):
|
|
with patch("httpx.get", return_value=_mock_response(200, {"data": []})):
|
|
result = epc_service.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_service, 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_service.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_service):
|
|
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_service.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_service):
|
|
with patch("httpx.get", return_value=_mock_response(404)):
|
|
results = epc_service.search_by_postcode("ZZ9 9ZZ")
|
|
|
|
assert results == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests 8-10: find_best_match
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_scored_df(rows, scores, ranks):
|
|
import pandas as pd
|
|
df = pd.DataFrame(rows)
|
|
df["lexiscore"] = scores
|
|
df["lexirank"] = ranks
|
|
return df.sort_values("lexirank")
|
|
|
|
|
|
def test_find_best_match_round1_clear_winner(epc_service, rdsap_21_0_1_cert):
|
|
search_rows = [
|
|
make_search_row(cert_num="CERT-WIN", address_line_1="1 High Street"),
|
|
make_search_row(cert_num="CERT-LOSE", address_line_1="99 Nowhere Lane"),
|
|
]
|
|
cert_response = {"data": rdsap_21_0_1_cert}
|
|
|
|
df_rows = [
|
|
{"address": "1 High Street", "uprn": "100023336956", "certificate_number": "CERT-WIN"},
|
|
{"address": "99 Nowhere Lane", "uprn": "100023336956", "certificate_number": "CERT-LOSE"},
|
|
]
|
|
scored = _make_scored_df(df_rows, [0.9, 0.1], [1, 2])
|
|
|
|
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), \
|
|
patch("backend.utils.addressMatch.get_uprn_candidates", return_value=scored):
|
|
result = epc_service.find_best_match("SW1A 1AA", "1 High Street")
|
|
|
|
assert isinstance(result, EpcPropertyData)
|
|
|
|
|
|
def test_find_best_match_round1_ambiguous_round2_resolves(epc_service, rdsap_21_0_1_cert):
|
|
search_rows = [
|
|
make_search_row(
|
|
cert_num="CERT-A", address_line_1="1 High Street",
|
|
address_line_2="Ground Floor",
|
|
),
|
|
make_search_row(
|
|
cert_num="CERT-B", address_line_1="1 High Street",
|
|
address_line_2="First Floor",
|
|
),
|
|
]
|
|
cert_response = {"data": rdsap_21_0_1_cert}
|
|
|
|
# Round 1: both score equally — ambiguous (two rank-1s)
|
|
ambiguous = _make_scored_df(
|
|
[
|
|
{"address": "1 High Street", "uprn": "111", "certificate_number": "CERT-A"},
|
|
{"address": "1 High Street", "uprn": "222", "certificate_number": "CERT-B"},
|
|
],
|
|
[0.9, 0.9],
|
|
[1, 1],
|
|
)
|
|
# Round 2: CERT-A wins on full address
|
|
resolved = _make_scored_df(
|
|
[
|
|
{"address": "1 High Street, Ground Floor", "uprn": "111", "certificate_number": "CERT-A"},
|
|
{"address": "1 High Street, First Floor", "uprn": "222", "certificate_number": "CERT-B"},
|
|
],
|
|
[0.85, 0.4],
|
|
[1, 2],
|
|
)
|
|
|
|
call_count = {"n": 0}
|
|
|
|
def fake_candidates(df, user_address, **kwargs):
|
|
call_count["n"] += 1
|
|
return ambiguous if call_count["n"] == 1 else resolved
|
|
|
|
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), \
|
|
patch("backend.utils.addressMatch.get_uprn_candidates", side_effect=fake_candidates):
|
|
result = epc_service.find_best_match("SW1A 1AA", "1 High Street Ground Floor")
|
|
|
|
assert isinstance(result, EpcPropertyData)
|
|
|
|
|
|
def test_find_best_match_returns_none_when_no_good_match(epc_service):
|
|
search_rows = [make_search_row(cert_num="CERT-X", address_line_1="99 Nowhere Lane")]
|
|
|
|
low_score = _make_scored_df(
|
|
[{"address": "99 Nowhere Lane", "uprn": "111", "certificate_number": "CERT-X"}],
|
|
[0.1],
|
|
[1],
|
|
)
|
|
|
|
with patch("httpx.get", return_value=_mock_response(200, {"data": search_rows})), \
|
|
patch("backend.utils.addressMatch.get_uprn_candidates", return_value=low_score):
|
|
result = epc_service.find_best_match("SW1A 1AA", "1 Completely Different Road")
|
|
|
|
assert result is None
|