mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-30 13:10:47 +00:00
commit
27c9752949
3 changed files with 1 additions and 135 deletions
|
|
@ -41,7 +41,7 @@ class EpcClientService:
|
||||||
|
|
||||||
# ------------------------------------------------------------------
|
# ------------------------------------------------------------------
|
||||||
# Private helperEpcRateLimpolarss
|
# Private helperEpcRateLimpolarss
|
||||||
# ----------------------EpcRateLimpolarsEpcRateLimpolarsEpcRateLimpolarsEpcRateLimpolarsEpcRateLimpolars--------------------------------------------
|
# ------------------------------------------------------------------
|
||||||
|
|
||||||
def _fetch_certificate(self, cert_num: str) -> dict[str, Any]:
|
def _fetch_certificate(self, cert_num: str) -> dict[str, Any]:
|
||||||
resp = httpx.get(
|
resp = httpx.get(
|
||||||
|
|
|
||||||
|
|
@ -2,7 +2,6 @@ from unittest.mock import MagicMock, patch, call
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from backend.epc_client.epc_client_service import EpcClientService
|
from backend.epc_client.epc_client_service import EpcClientService
|
||||||
from backend.utils.epc_address_match import find_best_epc_match
|
|
||||||
from datatypes.epc.search import EpcSearchResult
|
from datatypes.epc.search import EpcSearchResult
|
||||||
from backend.epc_client.exceptions import EpcNotFoundError, EpcRateLimitError
|
from backend.epc_client.exceptions import EpcNotFoundError, EpcRateLimitError
|
||||||
from datatypes.epc.domain.epc_property_data import EpcPropertyData
|
from datatypes.epc.domain.epc_property_data import EpcPropertyData
|
||||||
|
|
@ -132,69 +131,3 @@ def test_search_by_postcode_404_returns_empty_list(epc_service):
|
||||||
results = epc_service.search_by_postcode("ZZ9 9ZZ")
|
results = epc_service.search_by_postcode("ZZ9 9ZZ")
|
||||||
|
|
||||||
assert results == []
|
assert results == []
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
|
||||||
# Tests 8-10: find_best_epc_match — real scoring, only HTTP mocked
|
|
||||||
# ---------------------------------------------------------------------------
|
|
||||||
|
|
||||||
|
|
||||||
def test_find_best_match_clear_winner_on_first_pass(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}
|
|
||||||
|
|
||||||
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):
|
|
||||||
result = find_best_epc_match(epc_service, "SW1A 1AA", "1 High Street")
|
|
||||||
|
|
||||||
assert isinstance(result, EpcPropertyData)
|
|
||||||
|
|
||||||
|
|
||||||
def test_find_best_match_resolves_on_second_pass_using_full_address(
|
|
||||||
epc_service, rdsap_21_0_1_cert
|
|
||||||
):
|
|
||||||
# Both candidates share address_line_1 — round 1 is ambiguous.
|
|
||||||
# Round 2 scores against full_address and picks the correct floor.
|
|
||||||
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}
|
|
||||||
|
|
||||||
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):
|
|
||||||
result = find_best_epc_match(
|
|
||||||
epc_service, "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")]
|
|
||||||
|
|
||||||
with patch("httpx.get", return_value=_mock_response(200, {"data": search_rows})):
|
|
||||||
result = find_best_epc_match(
|
|
||||||
epc_service, "SW1A 1AA", "1 Completely Different Road"
|
|
||||||
)
|
|
||||||
|
|
||||||
assert result is None
|
|
||||||
|
|
|
||||||
|
|
@ -1,67 +0,0 @@
|
||||||
from __future__ import annotations
|
|
||||||
|
|
||||||
from typing import TYPE_CHECKING, Optional
|
|
||||||
|
|
||||||
from backend.utils.addressMatch import AddressMatch
|
|
||||||
from datatypes.epc.domain.epc_property_data import EpcPropertyData
|
|
||||||
from datatypes.epc.search import EpcSearchResult
|
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
|
||||||
from backend.epc_client.epc_client_service import EpcClientService
|
|
||||||
|
|
||||||
_MIN_MATCH_SCORE = 0.6
|
|
||||||
|
|
||||||
|
|
||||||
def find_best_epc_match(
|
|
||||||
service: EpcClientService,
|
|
||||||
postcode: str,
|
|
||||||
address: str,
|
|
||||||
) -> Optional[EpcPropertyData]:
|
|
||||||
candidates = service.search_by_postcode(postcode)
|
|
||||||
if not candidates:
|
|
||||||
return None
|
|
||||||
|
|
||||||
cert_num = _pick_best_cert(candidates, address, use_full_address=False)
|
|
||||||
if cert_num:
|
|
||||||
return _safe_get(service, cert_num)
|
|
||||||
|
|
||||||
cert_num = _pick_best_cert(candidates, address, use_full_address=True)
|
|
||||||
if cert_num:
|
|
||||||
return _safe_get(service, cert_num)
|
|
||||||
|
|
||||||
return None
|
|
||||||
|
|
||||||
|
|
||||||
def _pick_best_cert(
|
|
||||||
candidates: list[EpcSearchResult],
|
|
||||||
user_address: str,
|
|
||||||
use_full_address: bool,
|
|
||||||
) -> Optional[str]:
|
|
||||||
scored: list[tuple[float, str]] = [
|
|
||||||
(
|
|
||||||
AddressMatch.score(
|
|
||||||
user_address,
|
|
||||||
r.full_address if use_full_address else r.address_line_1,
|
|
||||||
),
|
|
||||||
r.certificate_number,
|
|
||||||
)
|
|
||||||
for r in candidates
|
|
||||||
]
|
|
||||||
if not scored:
|
|
||||||
return None
|
|
||||||
best_score = max(s for s, _ in scored)
|
|
||||||
if best_score < _MIN_MATCH_SCORE:
|
|
||||||
return None
|
|
||||||
top = [cert for s, cert in scored if s == best_score]
|
|
||||||
if len(top) != 1:
|
|
||||||
return None
|
|
||||||
return top[0]
|
|
||||||
|
|
||||||
|
|
||||||
def _safe_get(service: EpcClientService, cert_num: str) -> Optional[EpcPropertyData]:
|
|
||||||
from backend.epc_client.exceptions import EpcNotFoundError
|
|
||||||
|
|
||||||
try:
|
|
||||||
return service.get_by_certificate_number(cert_num)
|
|
||||||
except EpcNotFoundError:
|
|
||||||
return None
|
|
||||||
Loading…
Add table
Reference in a new issue