diff --git a/applications/modelling_e2e/handler.py b/applications/modelling_e2e/handler.py index b580a37d4..51c3e8149 100644 --- a/applications/modelling_e2e/handler.py +++ b/applications/modelling_e2e/handler.py @@ -205,13 +205,21 @@ def _flush_writes(engine: Engine, writes: list[_PropertyWrite]) -> None: uow.epc.save_batch(lodged_requests) if predicted_requests: uow.epc.save_batch(predicted_requests) + # First plan wins: a property with no default Plan yet gets this Plan + # as its default whatever the Scenario's flag says (the legacy + # p.is_new rule — readers select one default Plan per property, so a + # first run under a non-default Scenario must still produce one). + # Also heals properties left default-less by runs before this rule. + pids_with_default: set[int] = uow.plan.property_ids_with_default_plans( + [w.property_id for w in writes] + ) plan_requests = [ PlanSaveRequest( w.plan, property_id=w.property_id, scenario_id=w.scenario_id, portfolio_id=w.portfolio_id, - is_default=w.is_default, + is_default=w.is_default or w.property_id not in pids_with_default, ) for w in writes ] diff --git a/backend/address2UPRN/main.py b/backend/address2UPRN/main.py index 599cbaa6b..a3f0b086a 100644 --- a/backend/address2UPRN/main.py +++ b/backend/address2UPRN/main.py @@ -14,8 +14,13 @@ from utils.s3 import ( ) from datetime import datetime +from datatypes.address_match import UprnMatch from backend.utils.addressMatch import AddressMatch -from backend.address2UPRN.scoring import all_uprns_match, rank_address_similarity +from backend.address2UPRN.scoring import ( + all_uprns_match, + rank_address_similarity, + resolve_group_ambiguity, +) from infrastructure.epc_client.epc_client_service import EpcClientService from repositories.historic_epc.historic_epc_resolver import HistoricEpcResolver from repositories.historic_epc.historic_epc_s3_repository import ( @@ -34,32 +39,39 @@ def get_epc_data_with_postcode(postcode: str) -> pd.DataFrame: service = EpcClientService(auth_token=token) results = service.search_by_postcode(postcode) return pd.DataFrame( - [{"address": r.address_line_1, "uprn": r.uprn} for r in results] + [ + { + "address": r.address_line_1, + "uprn": r.uprn, + "certificate_number": r.certificate_number, + } + for r in results + ] ) def get_uprn_from_historic_epc( user_inputed_address: str, postcode: str, -) -> Optional[tuple[str, str, float]]: +) -> Optional[UprnMatch]: """Resolve a UPRN via historic EPC S3 data. - Returns (uprn, address, lexiscore) when the historic dataset agrees on a - single rank-1 UPRN, None otherwise (no stored data, zero score, or - ambiguous top rank). The score gate is `unambiguous_uprn`'s own (score > 0); - the 0.7 heuristic used for the new-EPC source isn't applied here because - historic addresses use a more verbose format that systematically depresses - lexiscores. + Returns (uprn, address, lexiscore, certificate_number) when the historic + dataset agrees on a single rank-1 UPRN, None otherwise (no stored data, + zero score, or ambiguous top rank). The score gate is `unambiguous_uprn`'s + own (score > 0); the 0.7 heuristic used for the new-EPC source isn't + applied here because historic addresses use a more verbose format that + systematically depresses lexiscores. """ repo = HistoricEpcS3Repository.with_default_s3_client() return HistoricEpcResolver(repo).resolve_uprn(user_inputed_address, postcode) -def get_uprn_with_epc_df( +def get_uprn_from_epc_df( user_inputed_address: str, epc_df: pd.DataFrame, verbose: bool = False, -) -> Optional[str | tuple[str, str, float]]: +) -> Optional[str | UprnMatch]: """ Return uprn (str) using a pre-fetched EPC dataframe. This avoids calling the API multiple times for the same postcode. @@ -88,6 +100,7 @@ def get_uprn_with_epc_df( address = top_rank_df["address"].values[0] score = float(top_rank_df["lexiscore"].values[0]) + certificate_number = top_rank_df["certificate_number"].values[0] logger.info(f"Address found to be: {address}, with lexiscore {score}") # Safe to return the agreed UPRN @@ -98,7 +111,7 @@ def get_uprn_with_epc_df( return None if verbose: - return (found_uprn, address, score) + return UprnMatch(found_uprn, address, score, certificate_number) else: return found_uprn @@ -116,11 +129,11 @@ def get_uprn( back to the historic EPC dataset on S3. For processing multiple addresses in the same postcode, use - get_uprn_with_epc_df instead. + get_uprn_from_epc_df instead. """ df = get_epc_data_with_postcode(postcode=postcode) - result: Optional[tuple[str, str, float]] = get_uprn_with_epc_df( + result: Optional[str | UprnMatch] = get_uprn_from_epc_df( user_inputed_address=user_inputed_address, epc_df=df, verbose=True, @@ -393,6 +406,8 @@ def handler(event, context, local=False): "address2uprn_uprn": "invalid postcode", "address2uprn_address": "invalid postcode", "address2uprn_lexiscore": "invalid postcode", + "address2uprn_certificate_number": "invalid postcode", + "address2uprn_status": "invalid_postcode", } ) continue @@ -409,7 +424,13 @@ def handler(event, context, local=False): ) continue - # Process each address in this postcode with the same EPC data + # Match each address in this postcode against the same EPC data, + # collecting per-row results first. Cross-row ambiguity is resolved + # for the whole group afterwards (ADR-0057) — a UPRN that is the + # best match for two *different* addresses is withheld, so distinct + # addresses are never coerced onto one UPRN (which the identity + # insert would then silently merge). + group_records: list[dict] = [] for row in postcode_rows: try: # Concatenate Address columns directly @@ -428,7 +449,7 @@ def handler(event, context, local=False): continue # Get UPRN using the pre-fetched EPC data with all return options - result: Optional[tuple[str, str, float]] = get_uprn_with_epc_df( + result: Optional[UprnMatch] = get_uprn_from_epc_df( user_inputed_address=address2uprn_user_input, epc_df=epc_df, verbose=True, @@ -451,31 +472,40 @@ def handler(event, context, local=False): f"Historic EPC matched {address2uprn_user_input} in {postcode}" ) + norm_address = AddressMatch.normalise_address( + address2uprn_user_input + ) + # Parse result tuple if successful if result: - uprn, found_address, score = result + uprn, found_address, score, certificate_number = result logger.info( f"Found UPRN for {address2uprn_user_input} in {postcode}: {uprn} (score: {score})" ) - - results_data.append( + group_records.append( { - **row, # Include all original data - "address2uprn_uprn": uprn, - "address2uprn_address": found_address, - "address2uprn_lexiscore": score, + "row": row, + "uprn": uprn, + "address": found_address, + "lexiscore": score, + "certificate_number": certificate_number, + "norm_address": norm_address, + "error": None, } ) else: logger.warning( f"No UPRN found for {address2uprn_user_input} in {postcode}" ) - results_data.append( + group_records.append( { - **row, # Include all original data - "address2uprn_uprn": None, - "address2uprn_address": None, - "address2uprn_lexiscore": None, + "row": row, + "uprn": None, + "address": None, + "lexiscore": None, + "certificate_number": None, + "norm_address": norm_address, + "error": None, } ) @@ -483,18 +513,43 @@ def handler(event, context, local=False): logger.error( f"Error processing address {row.get('address2uprn_user_input', 'unknown')}: {e}" ) - # Still add the row with error markers - results_data.append( + # Still record the row with error markers + group_records.append( { - **row, - "address2uprn_uprn": None, - "address2uprn_address": None, - "address2uprn_lexiscore": None, + "row": row, + "uprn": None, + "address": None, + "lexiscore": None, + "certificate_number": None, + "norm_address": "", "error": str(e), } ) continue + # Resolve cross-row ambiguity for the whole postcode group, then + # emit. A withheld (ambiguous) UPRN drops to None but keeps its + # lexiscore so the confirmation page can triage it (ADR-0057). + decisions = resolve_group_ambiguity( + [(rec["uprn"], rec["norm_address"]) for rec in group_records] + ) + for rec, (final_uprn, status) in zip(group_records, decisions): + emitted = { + **rec["row"], # Include all original data + "address2uprn_uprn": final_uprn, + "address2uprn_address": rec["address"] if final_uprn else None, + "address2uprn_lexiscore": rec["lexiscore"], + "address2uprn_certificate_number": ( + rec["certificate_number"] if final_uprn else None + ), + "address2uprn_status": ( + "error" if rec["error"] is not None else status + ), + } + if rec["error"] is not None: + emitted["error"] = rec["error"] + results_data.append(emitted) + # Create results DataFrame result_df = pd.DataFrame(results_data) diff --git a/backend/address2UPRN/scoring.py b/backend/address2UPRN/scoring.py index dcb86d498..d158a9680 100644 --- a/backend/address2UPRN/scoring.py +++ b/backend/address2UPRN/scoring.py @@ -1,8 +1,45 @@ +from collections import defaultdict +from typing import NamedTuple, Optional + import pandas as pd from backend.utils.addressMatch import AddressMatch +class GroupDecision(NamedTuple): + """One row's outcome after cross-row ambiguity resolution (ADR-0057).""" + + uprn: Optional[str] + status: str # "matched" | "ambiguous_duplicate" | "unmatched" + + +def resolve_group_ambiguity( + matches: list[tuple[Optional[str], str]], +) -> list[GroupDecision]: + """Resolve cross-row UPRN ambiguity within one postcode group (ADR-0057). + + ``matches`` is ``(uprn, normalised_address)`` per row. A UPRN that is the + best match for two rows with *different* normalised addresses is withheld + on both (a coarse EPC record absorbing several real addresses, e.g. flats in + a block); a UPRN shared only by identical addresses is a genuine re-listing + and kept. Returns a ``GroupDecision`` per row, in input order. + """ + distinct_addresses: dict[str, set[str]] = defaultdict(set) + for uprn, norm_address in matches: + if uprn: + distinct_addresses[uprn].add(norm_address) + + resolved: list[GroupDecision] = [] + for uprn, _norm_address in matches: + if not uprn: + resolved.append(GroupDecision(None, "unmatched")) + elif len(distinct_addresses[uprn]) > 1: + resolved.append(GroupDecision(None, "ambiguous_duplicate")) + else: + resolved.append(GroupDecision(uprn, "matched")) + return resolved + + def all_uprns_match( df: pd.DataFrame, target_uprn: str, diff --git a/datatypes/address_match.py b/datatypes/address_match.py new file mode 100644 index 000000000..74ed9ce6e --- /dev/null +++ b/datatypes/address_match.py @@ -0,0 +1,13 @@ +from __future__ import annotations + +from typing import NamedTuple, Optional + + +class UprnMatch(NamedTuple): + """A confident address→UPRN match from either EPC source (new API or + historic). Tuple-compatible, so existing unpacking keeps working.""" + + uprn: str + address: str + lexiscore: float + certificate_number: Optional[str] diff --git a/docs/adr/0057-uprn-confirmation-precedes-finalise.md b/docs/adr/0057-uprn-confirmation-precedes-finalise.md new file mode 100644 index 000000000..81eb1185e --- /dev/null +++ b/docs/adr/0057-uprn-confirmation-precedes-finalise.md @@ -0,0 +1,80 @@ +# UPRN confirmation precedes finalise; the bulk-upload review is one two-tab page and the portfolio "unmatched" tab is retired + +## Status + +proposed + +## Context + +The bulk-upload pipeline is a chain of SQS-driven Lambda stages: +`address2uprn → landlord-overrides → combiner → (review) → finaliser`. +`address2uprn` matches each input row independently against per-postcode EPC +candidates, keeping the best candidate above a `lexiscore >= 0.7` floor. There +is **no cross-row uniqueness guard**: nothing stops one UPRN being the best +match for two *different* addresses. The textbook case is a coarse EPC record +`42 Moreton Road` (no flat token) that `Flat 1, 42 Moreton Road` and +`Flat 2, 42 Moreton Road` both clear — the building-number guard passes and the +flat guard never fires, so both distinct flats collapse onto one UPRN. + +Downstream this is corrosive, not just noisy: + +- The `property` identity insert is `on_conflict_do_nothing` on the partial + unique index `(portfolio_id, uprn) WHERE uprn IS NOT NULL`. Two distinct + addresses carrying the same UPRN therefore **merge into a single property** — + one address silently loses its identity. +- `_build_overrides` (the finaliser) keys `property_overrides` on + `(property_id, override_component, building_part)`. Both merged rows emit the + same tuple, so the single `INSERT ... ON CONFLICT DO UPDATE` raises + `CardinalityViolation` ("cannot affect row a second time"). + +The classifier and finaliser were built assuming a UPRN is always intact and +unique; it is not. Separately, a *post-onboarding* portfolio-level "Unmatched +properties" tab let users fix no-UPRN properties after the fact — too late (the +identity merge has already happened) and in a different place from the rest of +review. + +## Decision + +1. **`address2uprn` emits a UPRN only when confident *and* unambiguous.** Within + a postcode group, a UPRN that is the best match for two or more rows whose + *normalised* input addresses differ is **withheld** (dropped to null) on + every such row; a UPRN shared only by rows with the *same* normalised address + is a genuine re-listing and is kept. `address2uprn` emits an + `address2uprn_status` column (`matched | ambiguous_duplicate | unmatched | + invalid_postcode | error`); withheld rows keep their `lexiscore` for triage. + A `lexiscore` review band `[0.7, X)` that also withholds shaky matches is a + follow-on once `X` is calibrated (see Consequences). +2. **One pre-finalise review page, two tabs**, styled like the retired unmatched + tab: + - **Addresses** — the flagged rows (`unmatched` / `ambiguous_duplicate` / + low-score); the user resolves each via OS Places (reusing the existing + `MatchAddress` / `searchAddresses` / `useAssignUprn` flow + `postcode_search` + cache), producing a valid residential UPRN, or explicitly marks *no-UPRN*. + - **Classification** — the existing description→value verify gate, semantics + unchanged (every `Unknown` must be mapped; the finaliser still fails loud on + an unresolved value). + Finalise is enabled only when **both** tabs are clear. +3. The finaliser runs only after confirmation, so every row carries a confirmed + UPRN **or** an explicit no-UPRN (a first-class terminal state: `property` + identity with `uprn NULL`, no `property_overrides`, building-passport shows + the "not matched yet" card). +4. **Retire the portfolio-level "Unmatched properties" tab** — properties now + arrive pre-matched, so post-onboarding UPRN cleanup is unnecessary. +5. `PropertyOverridePostgresRepository.upsert_all` keeps a **defensive + last-write-wins dedup** as a backstop for the single-statement invariant, but + it is no longer the primary fix. + +## Consequences + +- Distinct addresses can no longer silently merge into one property; the + `CardinalityViolation` can no longer arise from real data. +- Onboarding reaching "complete" now implies confirmed identities, so portfolio + pages need no unmatched tab. +- The ambiguous/low-score subset adds bounded user work at review time; confident + rows pass untouched. +- The `lexiscore` review-band threshold `X` needs calibrating against a real + portfolio's score distribution before it is switched on; until then only the + ambiguous-duplicate rule (which needs no threshold) is active. +- Migration is safe stage-by-stage: withheld UPRNs are caught by the *existing* + unmatched tab until the two-tab page ships, and the tab is removed only after + it does — so an ambiguous property always has somewhere to be resolved. diff --git a/repositories/historic_epc/historic_epc_resolver.py b/repositories/historic_epc/historic_epc_resolver.py index 21a7457be..eb7e37af3 100644 --- a/repositories/historic_epc/historic_epc_resolver.py +++ b/repositories/historic_epc/historic_epc_resolver.py @@ -2,6 +2,7 @@ from __future__ import annotations from typing import Optional +from datatypes.address_match import UprnMatch from datatypes.epc.domain.historic_epc import HistoricEpc from datatypes.epc.domain.historic_epc_matching import ( HistoricEpcMatches, @@ -47,11 +48,10 @@ class HistoricEpcResolver: return None return max(certs, key=lambda r: r.lodgement_date) - def resolve_uprn( - self, user_address: str, postcode: str - ) -> Optional[tuple[str, str, float]]: - """``(uprn, matched_address, lexiscore)`` for an unambiguous rank-1 - match, else None (no data / ambiguous tie / zero score).""" + def resolve_uprn(self, user_address: str, postcode: str) -> Optional[UprnMatch]: + """A ``UprnMatch`` for an unambiguous rank-1 match, else None (no data / + ambiguous tie / zero score). ``certificate_number`` is the historic + dataset's ``lmk_key``.""" matches: HistoricEpcMatches = self.match(user_address, postcode) uprn: Optional[str] = matches.unambiguous_uprn() if not uprn or uprn == "nan": @@ -59,4 +59,4 @@ class HistoricEpcResolver: top: Optional[ScoredHistoricEpc] = matches.top() if top is None: return None - return uprn, top.record.address, top.lexiscore + return UprnMatch(uprn, top.record.address, top.lexiscore, top.record.lmk_key) diff --git a/repositories/plan/plan_postgres_repository.py b/repositories/plan/plan_postgres_repository.py index e81e90844..dc3b135a7 100644 --- a/repositories/plan/plan_postgres_repository.py +++ b/repositories/plan/plan_postgres_repository.py @@ -3,7 +3,7 @@ from __future__ import annotations from typing import Any from sqlalchemy import insert as _sa_insert -from sqlmodel import Session, col, update +from sqlmodel import Session, col, select, update from domain.modelling.plan import Plan from infrastructure.postgres.modelling import PlanModel, RecommendationModel @@ -26,8 +26,9 @@ class PlanPostgresRepository(PlanRepository): A re-run INSERTs a fresh Plan rather than deleting the prior one (the cascade delete was slow); when the new Plan is the default it demotes any prior - default Plan for the same (property_id, scenario_id) to ``is_default=False``, - so readers can select the current Plan via ``is_default=True``.""" + default Plan for the same property — across all Scenarios — to + ``is_default=False``, so readers can select the property's one current Plan + via ``is_default=True``.""" def __init__(self, session: Session) -> None: self._session = session @@ -45,6 +46,19 @@ class PlanPostgresRepository(PlanRepository): [PlanSaveRequest(plan, property_id=property_id, scenario_id=scenario_id, portfolio_id=portfolio_id, is_default=is_default)] )[0] + def property_ids_with_default_plans(self, property_ids: list[int]) -> set[int]: + if not property_ids: + return set() + rows = self._session.exec( + select(col(PlanModel.property_id)) + .distinct() + .where( + col(PlanModel.property_id).in_(property_ids), + col(PlanModel.is_default).is_(True), + ) + ).all() + return {int(row) for row in rows} + def save_batch(self, requests: list[PlanSaveRequest]) -> list[int]: """Persist all Plans in three statements regardless of batch size. @@ -56,17 +70,17 @@ class PlanPostgresRepository(PlanRepository): return [] # Demote prior default Plans for every property in the batch that is - # receiving a new default Plan — one UPDATE for the whole batch. + # receiving a new default Plan — one UPDATE for the whole batch, + # across ALL scenarios: readers select a property's default by + # portfolio + is_default alone, so the invariant is one default per + # property. Scenario-scoped demotion left a promoted first Plan + # (non-default scenario) defaulted forever alongside the default + # scenario's later Plan (#1490). default_pids = [r.property_id for r in requests if r.is_default] if default_pids: - # scenario_id is uniform per batch (one scenario per SQS message). - scenario_id = requests[0].scenario_id self._session.exec( # type: ignore[call-overload] update(PlanModel) - .where( - col(PlanModel.property_id).in_(default_pids), - col(PlanModel.scenario_id) == scenario_id, - ) + .where(col(PlanModel.property_id).in_(default_pids)) .values(is_default=False) ) diff --git a/repositories/plan/plan_repository.py b/repositories/plan/plan_repository.py index 6fdc16d63..15cafced1 100644 --- a/repositories/plan/plan_repository.py +++ b/repositories/plan/plan_repository.py @@ -43,7 +43,8 @@ class PlanRepository(ABC): ) -> int: """Persist ``plan`` and return its Plan id. Keeps prior Plans for ``(property_id, scenario_id)`` as history; when ``is_default`` is True, - demotes those prior Plans to ``is_default=False``.""" + demotes the property's prior default Plan — whatever its Scenario — to + ``is_default=False`` (one default Plan per property).""" ... @abstractmethod @@ -54,3 +55,11 @@ class PlanRepository(ABC): UPDATE only when at least one request has ``is_default=True``. Keeps prior Plans as history (ADR-0017).""" ... + + @abstractmethod + def property_ids_with_default_plans(self, property_ids: list[int]) -> set[int]: + """The subset of ``property_ids`` that already have a default Plan. + A property outside this set is receiving its first Plan, which becomes + its default whatever the Scenario says (the legacy ``p.is_new`` rule — + readers need one default Plan per property).""" + ... diff --git a/repositories/property/property_override_postgres_repository.py b/repositories/property/property_override_postgres_repository.py index be69292aa..d44a2ad65 100644 --- a/repositories/property/property_override_postgres_repository.py +++ b/repositories/property/property_override_postgres_repository.py @@ -31,6 +31,17 @@ class PropertyOverridePostgresRepository(PropertyOverrideRepository): return 0 now = datetime.now(timezone.utc) + # Defensive backstop (ADR-0057): a single INSERT ... ON CONFLICT DO + # UPDATE may not touch the same target row twice — two batch rows sharing + # the conflict key (property_id, override_component, building_part) make + # Postgres raise CardinalityViolation ("cannot affect row a second + # time"). ADR-0057 stops distinct addresses being coerced onto one UPRN + # upstream, so this should no longer arise from real data; collapse to + # the last occurrence (last-write-wins, the same outcome DO UPDATE gives + # across separate statements) so the statement can never be invalid. + deduped: dict[tuple[int, str, int], PropertyOverrideInsert] = { + (r.property_id, r.override_component, r.building_part): r for r in rows + } values = [ { "property_id": r.property_id, @@ -42,7 +53,7 @@ class PropertyOverridePostgresRepository(PropertyOverrideRepository): "created_at": now, "updated_at": now, } - for r in rows + for r in deduped.values() ] stmt = pg_insert(self._table).values(values) diff --git a/repositories/property/property_override_repository.py b/repositories/property/property_override_repository.py index 0a1a0a2e8..9a57d3baf 100644 --- a/repositories/property/property_override_repository.py +++ b/repositories/property/property_override_repository.py @@ -34,5 +34,9 @@ class PropertyOverrideRepository(ABC): """Upsert each row on ``(property_id, override_component, building_part)``, refreshing ``override_value`` + ``original_spreadsheet_description`` + ``updated_at`` on conflict (recalculate-on-rerun). Returns the number of - rows affected; an empty list is a no-op returning 0.""" + rows affected; an empty list is a no-op returning 0. + + Rows sharing a conflict key within one call are collapsed to the last + occurrence (last-write-wins) before writing, so a single batch never + upserts the same target row twice (ADR-0057 backstop).""" ... diff --git a/scripts/resolve_uprns_for_finaliser.py b/scripts/resolve_uprns_for_finaliser.py index c01a55ed4..b9a44a2ce 100644 --- a/scripts/resolve_uprns_for_finaliser.py +++ b/scripts/resolve_uprns_for_finaliser.py @@ -47,7 +47,7 @@ sys.path.insert(0, str(_REPO_ROOT)) # worktree root first — avoid the import from backend.address2UPRN.main import ( # noqa: E402 get_epc_data_with_postcode, get_uprn_from_historic_epc, - get_uprn_with_epc_df, + get_uprn_from_epc_df, ) from backend.ordnanceSurvey.helpers import ( # noqa: E402 lookup_os_places, @@ -108,7 +108,7 @@ def resolve_epc( epc_df = get_epc_data_with_postcode(postcode=postcode_clean) epc_cache[postcode_clean] = epc_df - result = get_uprn_with_epc_df( + result = get_uprn_from_epc_df( user_inputed_address=address, epc_df=epc_df, verbose=True ) if isinstance(result, tuple): diff --git a/tests/applications/modelling_e2e/test_handler.py b/tests/applications/modelling_e2e/test_handler.py index f2e33187b..32319bcd5 100644 --- a/tests/applications/modelling_e2e/test_handler.py +++ b/tests/applications/modelling_e2e/test_handler.py @@ -511,6 +511,70 @@ def test_attach_mode_partial_failure_persists_successes_then_records_failure() - ] +def test_first_plan_for_a_property_becomes_its_default() -> None: + """A property's first Plan is its default whatever the Scenario says (the + legacy p.is_new rule — readers select one default Plan per property); a + property that already has a default Plan keeps the Scenario's flag.""" + # Arrange — a non-default scenario; 111 has no default plan yet, 222 does + pid_first, pid_has_default = 111, 222 + mock_engine = _engine_mock( + [pid_first, pid_has_default], [1001, 1002], [POSTCODE, POSTCODE] + ) + scenario = MagicMock() + scenario.is_default = False + + with ExitStack() as stack: + stack.enter_context(patch("applications.modelling_e2e.handler.os.environ", _ENV)) + stack.enter_context( + patch("applications.modelling_e2e.handler._get_engine", return_value=mock_engine) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcClientService") + ).return_value.get_by_uprn.return_value = MagicMock() + stack.enter_context(patch("applications.modelling_e2e.handler.GeospatialS3Repository")) + stack.enter_context(patch("applications.modelling_e2e.handler.GoogleSolarApiClient")) + stack.enter_context( + patch("applications.modelling_e2e.handler._spatial_for", return_value=None) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler._solar_insights_for", return_value=None) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.overlays_from", return_value=[]) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader") + ).return_value.overrides_for_many.return_value = {} + stack.enter_context( + patch("applications.modelling_e2e.handler.ScenarioPostgresRepository") + ).return_value.get_many.return_value = [scenario] + stack.enter_context(patch("applications.modelling_e2e.handler.catalogue_snapshot_with_off_catalogue_overrides")) + stack.enter_context(patch("applications.modelling_e2e.handler.Session")) + stack.enter_context( + patch("applications.modelling_e2e.handler.run_modelling", return_value=_plan_mock()) + ) + MockUoW = stack.enter_context(patch("applications.modelling_e2e.handler.PostgresUnitOfWork")) + mock_uow = MagicMock() + mock_uow.plan.property_ids_with_default_plans.return_value = {pid_has_default} + MockUoW.return_value.__enter__.return_value = mock_uow + MockUoW.return_value.__exit__.return_value = False + + # Act + from applications.modelling_e2e.handler import handler + handler.__wrapped__( # type: ignore[attr-defined] + {"property_ids": [pid_first, pid_has_default], + "portfolio_id": PORTFOLIO_ID, "scenario_id": SCENARIO_ID, + "refetch_solar": False, "dry_run": False}, + None, _mock_orchestrator(), uuid4(), + ) + + # Assert — the first-timer's plan is promoted; the other keeps the flag + plan_requests = mock_uow.plan.save_batch.call_args.args[0] + by_pid = {r.property_id: r for r in plan_requests} + assert by_pid[pid_first].is_default is True + assert by_pid[pid_has_default].is_default is False + + # --------------------------------------------------------------------------- # Lodged EPC path # --------------------------------------------------------------------------- diff --git a/tests/backend/address2UPRN/__init__.py b/tests/backend/address2UPRN/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/backend/address2UPRN/test_resolve_group_ambiguity.py b/tests/backend/address2UPRN/test_resolve_group_ambiguity.py new file mode 100644 index 000000000..1279cd1d6 --- /dev/null +++ b/tests/backend/address2UPRN/test_resolve_group_ambiguity.py @@ -0,0 +1,70 @@ +"""Unit tests for ``resolve_group_ambiguity`` (ADR-0057). + +Each address is matched independently, so one UPRN can be the best match for +two *different* addresses (a coarse EPC record absorbing several real +addresses). Withholding that UPRN is a pure function of a postcode group's +``(uprn, normalised_address)`` pairs — no S3, EPC API, or scoring needed — so +it is unit-tested directly here. +""" + +from __future__ import annotations + +from backend.address2UPRN.scoring import resolve_group_ambiguity + + +def test_distinct_addresses_sharing_one_uprn_are_withheld() -> None: + # Flat 1 and Flat 2 both matched the same coarse EPC UPRN — distinct + # addresses, so both are withheld rather than silently merged. + result = resolve_group_ambiguity( + [ + ("100", "flat 1 42 moreton road"), + ("100", "flat 2 42 moreton road"), + ] + ) + assert result == [(None, "ambiguous_duplicate"), (None, "ambiguous_duplicate")] + + +def test_same_address_listed_twice_keeps_the_uprn() -> None: + # A genuine re-listing of one property (identical normalised address) is a + # real duplicate, not an ambiguous match — the UPRN is kept on both. + result = resolve_group_ambiguity( + [ + ("100", "42 moreton road"), + ("100", "42 moreton road"), + ] + ) + assert result == [("100", "matched"), ("100", "matched")] + + +def test_distinct_uprns_are_each_matched() -> None: + result = resolve_group_ambiguity( + [ + ("100", "42 moreton road"), + ("101", "44 moreton road"), + ] + ) + assert result == [("100", "matched"), ("101", "matched")] + + +def test_none_uprn_is_unmatched() -> None: + result = resolve_group_ambiguity([(None, "99 nowhere lane")]) + assert result == [(None, "unmatched")] + + +def test_order_is_preserved_across_mixed_group() -> None: + # A withheld pair, an unmatched row, and a clean match — all in one group; + # the output aligns positionally with the input. + result = resolve_group_ambiguity( + [ + ("100", "flat 1 42 moreton road"), # ambiguous (with row 3) + (None, "no epc candidate"), # unmatched + ("100", "flat 2 42 moreton road"), # ambiguous (with row 0) + ("200", "sole match road"), # matched + ] + ) + assert result == [ + (None, "ambiguous_duplicate"), + (None, "unmatched"), + (None, "ambiguous_duplicate"), + ("200", "matched"), + ] diff --git a/tests/orchestration/fakes.py b/tests/orchestration/fakes.py index d5fe016a8..4c6d323b9 100644 --- a/tests/orchestration/fakes.py +++ b/tests/orchestration/fakes.py @@ -212,6 +212,7 @@ class FakePlanRepository(PlanRepository): def __init__(self) -> None: self.saved: dict[tuple[int, int], Plan] = {} + self.default_property_ids: set[int] = set() self._next_id = 1 def save( @@ -224,6 +225,8 @@ class FakePlanRepository(PlanRepository): is_default: bool, ) -> int: self.saved[(property_id, scenario_id)] = plan + if is_default: + self.default_property_ids.add(property_id) plan_id = self._next_id self._next_id += 1 return plan_id @@ -240,6 +243,9 @@ class FakePlanRepository(PlanRepository): for r in requests ] + def property_ids_with_default_plans(self, property_ids: list[int]) -> set[int]: + return self.default_property_ids.intersection(property_ids) + class _UnsetProductRepo(ProductRepository): """Default for a `FakeUnitOfWork` built without a catalogue — raises if a diff --git a/tests/repositories/historic_epc/test_historic_epc_resolver.py b/tests/repositories/historic_epc/test_historic_epc_resolver.py index c4b05f8b7..f28787c1b 100644 --- a/tests/repositories/historic_epc/test_historic_epc_resolver.py +++ b/tests/repositories/historic_epc/test_historic_epc_resolver.py @@ -73,10 +73,11 @@ def test_resolve_uprn_returns_unambiguous_match(): # Assert assert result is not None - uprn, address, score = result + uprn, address, score, certificate_number = result assert uprn == "100" assert address == "47 GORDON ROAD" assert score > 0 + assert certificate_number == "" def test_resolve_uprn_is_none_when_postcode_has_no_data(): diff --git a/tests/repositories/plan/test_plan_postgres_repository.py b/tests/repositories/plan/test_plan_postgres_repository.py index 200c38d6b..7198a6926 100644 --- a/tests/repositories/plan/test_plan_postgres_repository.py +++ b/tests/repositories/plan/test_plan_postgres_repository.py @@ -206,3 +206,56 @@ def test_rerun_as_non_default_does_not_demote_the_prior_default( assert len(plan_rows) == 2 assert by_id[first_id].is_default is True + + +def test_reports_which_properties_already_have_a_default_plan( + db_engine: Engine, +) -> None: + # Arrange — property 20 has a default Plan, 21 only a non-default one, + # 22 has no Plans at all + with Session(db_engine) as session: + repo = PlanPostgresRepository(session) + repo.save(_plan(), property_id=20, scenario_id=7, portfolio_id=1, is_default=True) + repo.save(_plan(), property_id=21, scenario_id=7, portfolio_id=1, is_default=False) + session.commit() + + # Act + with Session(db_engine) as session: + have_default = PlanPostgresRepository(session).property_ids_with_default_plans( + [20, 21, 22] + ) + + # Assert — only the property with a default Plan is reported + assert have_default == {20} + + +def test_a_new_default_demotes_the_prior_default_across_scenarios( + db_engine: Engine, +) -> None: + """One default Plan per property: a promoted first Plan (non-default + scenario) must lose the default when the default scenario's Plan lands — + readers filter portfolio + is_default only, never scenario (#1490).""" + # Arrange — the property's first Plan, promoted under non-default scenario 7 + with Session(db_engine) as session: + first_id = PlanPostgresRepository(session).save( + _plan(), property_id=30, scenario_id=7, portfolio_id=1, is_default=True + ) + session.commit() + + # Act — the default scenario (8) models the same property + with Session(db_engine) as session: + second_id = PlanPostgresRepository(session).save( + _plan(), property_id=30, scenario_id=8, portfolio_id=1, is_default=True + ) + session.commit() + + # Assert — exactly one default for the property, the newest plan + with Session(db_engine) as session: + plan_rows = session.exec( + select(PlanModel).where(col(PlanModel.property_id) == 30) + ).all() + by_id = {p.id: p for p in plan_rows} + + assert by_id[first_id].is_default is False # demoted across scenarios + assert by_id[second_id].is_default is True + assert sum(1 for p in plan_rows if p.is_default) == 1 diff --git a/tests/repositories/property/test_property_override_postgres_repository.py b/tests/repositories/property/test_property_override_postgres_repository.py new file mode 100644 index 000000000..635b32ee5 --- /dev/null +++ b/tests/repositories/property/test_property_override_postgres_repository.py @@ -0,0 +1,102 @@ +"""Integration tests for ``PropertyOverridePostgresRepository.upsert_all``. + +The conflict handling lives entirely in SQL (``INSERT ... ON CONFLICT +(property_id, override_component, building_part) DO UPDATE``), so it can only be +verified against a real Postgres -- the ``db_engine`` fixture in +``tests/conftest.py`` spins one up per test. The batch-dedup below is exactly +the case an in-memory fake cannot catch: only a real ``ON CONFLICT`` statement +raises ``CardinalityViolation`` when a batch would touch one target row twice +(the ADR-0057 backstop). +""" + +from __future__ import annotations + +from collections.abc import Iterator + +import pytest +from sqlalchemy import Engine +from sqlmodel import Session, select + +from infrastructure.postgres.property_override_table import PropertyOverrideRow +from repositories.property.property_override_postgres_repository import ( + PropertyOverridePostgresRepository, +) +from repositories.property.property_override_repository import PropertyOverrideInsert + + +@pytest.fixture +def session(db_engine: Engine) -> Iterator[Session]: + with Session(db_engine) as s: + yield s + + +def _insert( + property_id: int, + override_value: str, + description: str, + *, + building_part: int = 0, + component: str = "property_type", +) -> PropertyOverrideInsert: + return PropertyOverrideInsert( + property_id=property_id, + portfolio_id=820, + building_part=building_part, + override_component=component, + override_value=override_value, + original_spreadsheet_description=description, + ) + + +def _all_rows(session: Session) -> list[PropertyOverrideRow]: + return list(session.exec(select(PropertyOverrideRow)).all()) + + +def test_duplicate_conflict_key_in_one_batch_collapses_to_last( + session: Session, +) -> None: + # arrange: two batch rows sharing the conflict key + # (property_id, override_component, building_part) — before the backstop this + # raised CardinalityViolation ("cannot affect row a second time"). + repo = PropertyOverridePostgresRepository(session) + + # act + repo.upsert_all( + [ + _insert(742961, "House", "House: Semi-Detached"), + _insert(742961, "House", "House: Mid-Terrace"), + ] + ) + session.commit() + + # assert: one row, last occurrence wins (matches ON CONFLICT DO UPDATE). + rows = _all_rows(session) + assert len(rows) == 1 + assert rows[0].override_value == "House" + assert rows[0].original_spreadsheet_description == "House: Mid-Terrace" + + +def test_distinct_conflict_keys_are_all_written(session: Session) -> None: + # arrange: dedup must only collapse exact conflict-key duplicates — a + # different property_id, building_part, or component is a distinct row. + repo = PropertyOverridePostgresRepository(session) + + # act + repo.upsert_all( + [ + _insert(742961, "House", "House: Semi-Detached"), + _insert(742962, "House", "House: Mid-Terrace"), # different property + _insert(742961, "House", "ext", building_part=1), # different part + _insert(742961, "SolidBrick", "solid", component="wall_type"), # component + ] + ) + session.commit() + + # assert + assert len(_all_rows(session)) == 4 + + +def test_empty_batch_is_a_noop(session: Session) -> None: + repo = PropertyOverridePostgresRepository(session) + assert repo.upsert_all([]) == 0 + assert _all_rows(session) == []