1040 lines
49 KiB
Python
1040 lines
49 KiB
Python
"""Tests for the recurring house-deduplication merge (#1772).
|
||
|
||
Convention mirrors tests/tasks/test_cadastral_geo_match.py and
|
||
tests/test_house_imv_backfill_scheduler.py: the merge is SQL-heavy and the CI gate has no
|
||
live Postgres, so the bulk of coverage is STATIC analysis of the emitted SQL (text() clauses
|
||
+ inspect.getsource) plus a fake-db behavioural pass driving the dry_run / idempotent / count
|
||
logic. An OPTIONAL real-Postgres behavioural test asserts the actual merge semantics (re-point
|
||
listings, dedup colliding children WITHOUT a UNIQUE violation, delete loser, keeper survives,
|
||
idempotent second run, dry_run no-write) and self-SKIPS when no DB is reachable.
|
||
|
||
The static block is the migration-133 guard: it asserts EVERY FK child of houses(id) is
|
||
handled, and that the UNIQUE-constrained children dedup by the TARGET key before re-pointing.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import inspect
|
||
import os
|
||
import re
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
# settings needs a DSN at import (same dance as the sibling tests); these are static/fake-db.
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||
|
||
from app.services import house_dedup_merge as hdm
|
||
from app.services import scheduler
|
||
|
||
_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
|
||
_MIGRATION_135 = _SQL_DIR / "135_scrape_schedules_seed_house_dedup_merge.sql"
|
||
|
||
_MAPPING_SQL = str(hdm._BUILD_MAPPING_SQL.text)
|
||
_FIAS_MAPPING_SQL = str(hdm._BUILD_MAPPING_SQL_FIAS.text)
|
||
_CARRY_SQL = str(hdm._CARRY_OVER_IDENTITY_SQL.text)
|
||
_STEP_SQL = {label: sql for label, sql in hdm._STEPS}
|
||
_ALL_STEP_SQL = "\n".join(sql for _, sql in hdm._STEPS)
|
||
_SERVICE_SRC = inspect.getsource(hdm)
|
||
|
||
|
||
def _flat(sql: str) -> str:
|
||
return re.sub(r"\s+", " ", sql)
|
||
|
||
|
||
# ── Cluster key + keeper rule ─────────────────────────────────────────────────
|
||
|
||
|
||
def test_cluster_key_is_canonical_address_not_cadastral() -> None:
|
||
"""Spec: cadastral is unpopulated → cluster by CANONICAL address (tradein_canon_addr).
|
||
|
||
The canon collapses spelling/район variants of the same building (ул→улица, strips
|
||
город/район noise, preserves корпус) so e.g. «ул. Вайнера,66» and «улица Вайнера, 66»
|
||
share one cluster_key.
|
||
"""
|
||
flat = _flat(_MAPPING_SQL)
|
||
# Cluster on the CLEAN address: COALESCE(short_address, full_address, address) — `address`
|
||
# may carry район-noise the canon does not strip (e.g. «… · р-н Центр»), while short_address
|
||
# is clean, so preferring the clean field lets Вайнера 66 cluster with its twin.
|
||
assert "'addr:' || tradein_canon_addr(COALESCE(short_address, full_address, address))" in flat
|
||
# The empty-canon guard still keeps blank-canon houses out of any cluster.
|
||
assert (
|
||
"NULLIF( tradein_canon_addr(COALESCE(short_address, full_address, address)), '' )"
|
||
" IS NOT NULL" in flat
|
||
)
|
||
# Digit guard: canon must carry a house number — excludes degenerate canons like
|
||
# «екатеринбург»/«сооружение» (street-less addresses) from ever forming a cluster.
|
||
assert "tradein_canon_addr(COALESCE(short_address, full_address, address)) ~ '[0-9]'" in flat
|
||
# cadastral_number must NOT be part of the cluster key (it is 100% NULL).
|
||
assert "cadastral_number" not in _flat(
|
||
_MAPPING_SQL[: _MAPPING_SQL.index("ranked")]
|
||
) # not in cluster construction
|
||
# Only canons shared by >1 house form a cluster.
|
||
assert "HAVING count(*) > 1" in flat
|
||
|
||
|
||
def test_geo_guard_blocks_cross_town_over_merge() -> None:
|
||
"""Geo guard: a loser is merged only when within 250 m of the keeper.
|
||
|
||
Because tradein_canon_addr strips город/район, two different «Ленина 5» buildings in
|
||
different towns share a canon. The final loser→keeper mapping is filtered on a 250 m
|
||
ST_DistanceSphere predicate (keeper geom exposed per partition via first_value), so
|
||
same-canon-but-far houses (or NULL geom on either side) are NOT merged.
|
||
"""
|
||
flat = _flat(_MAPPING_SQL)
|
||
# keeper geom exposed per partition with the SAME keeper-priority ordering as keeper_id.
|
||
assert "first_value(h.geom) OVER" in flat
|
||
assert "AS keeper_geom" in flat
|
||
assert "h.geom AS loser_geom" in flat
|
||
# the 250 m predicate + the NULL-geom safety guards on the final mapping.
|
||
assert "keeper_geom IS NOT NULL" in flat
|
||
assert "loser_geom IS NOT NULL" in flat
|
||
assert "ST_DistanceSphere(loser_geom, keeper_geom) <= 250" in flat
|
||
|
||
|
||
def test_keeper_rule_priority_order() -> None:
|
||
"""Keeper: geom NOT NULL → most linked listings → most-populated → min(id)."""
|
||
order = _flat(hdm._KEEPER_ORDER)
|
||
assert "(h.geom IS NOT NULL) DESC" in order
|
||
assert "listing_cnt DESC" in order
|
||
assert "h.id ASC" in order
|
||
# geom precedes listing count precedes id (ordering matters).
|
||
assert order.index("geom IS NOT NULL") < order.index("listing_cnt")
|
||
assert order.index("listing_cnt") < order.index("h.id ASC")
|
||
# completeness is counted from the live houses columns.
|
||
for col in ("year_built", "house_type", "total_floors", "developer_name", "rating_score"):
|
||
assert col in hdm._COMPLETENESS_EXPR
|
||
|
||
|
||
def test_mapping_selects_losers_only() -> None:
|
||
"""The mapping is loser_id→keeper_id; the keeper itself is excluded (rn>1, id<>keeper)."""
|
||
flat = _flat(_MAPPING_SQL)
|
||
assert "SELECT id AS loser_id, keeper_id" in flat
|
||
assert "WHERE rn > 1" in flat
|
||
assert "id <> keeper_id" in flat
|
||
|
||
|
||
def test_mapping_temp_table_drops_on_commit() -> None:
|
||
"""TEMP ... ON COMMIT DROP — the mapping never leaks across runs (idempotent setup)."""
|
||
assert "CREATE TEMP TABLE _1772_dup_mapping ON COMMIT DROP" in _flat(_MAPPING_SQL)
|
||
|
||
|
||
# ── EVERY FK child of houses(id) is handled (the migration-133 guard) ──────────
|
||
|
||
# The authoritative FK-child set (grep "REFERENCES houses" data/sql/*.sql). If a new child
|
||
# is ever added this list must grow with it — the test then forces a matching merge step.
|
||
_FK_CHILDREN = {
|
||
"listings": "house_id_fk",
|
||
"house_placement_history": "house_id",
|
||
"house_reviews": "house_id",
|
||
"house_reliability_checks": "house_id",
|
||
"external_valuations": "house_id",
|
||
"house_sources": "house_id",
|
||
"house_address_aliases": "house_id",
|
||
"houses_price_dynamics": "house_id",
|
||
"house_imv_evaluations": "house_id",
|
||
"house_suggestions": "house_id",
|
||
"address_mismatch_audit": "house_id",
|
||
}
|
||
|
||
|
||
def test_every_fk_child_of_houses_is_repointed() -> None:
|
||
"""Each FK child must be re-pointed loser→keeper somewhere in the merge steps."""
|
||
for table in _FK_CHILDREN:
|
||
assert table in _ALL_STEP_SQL, f"FK child {table} is never touched by the merge"
|
||
|
||
|
||
def test_unique_constrained_children_dedup_before_repoint() -> None:
|
||
"""UNIQUE-constrained children must DELETE colliding losers before re-pointing survivors.
|
||
|
||
Missing one of these is exactly what broke migration 133. We assert each has both a
|
||
collision/dedup DELETE step and a re-point UPDATE step, and that the DELETE precedes
|
||
the UPDATE in step order.
|
||
"""
|
||
unique_children = [
|
||
("house_sources", "house_sources(collision-delete)", "house_sources(re-point)"),
|
||
(
|
||
"house_address_aliases",
|
||
"house_address_aliases(collision-delete)",
|
||
"house_address_aliases(re-point)",
|
||
),
|
||
(
|
||
"houses_price_dynamics",
|
||
"houses_price_dynamics(dedup)",
|
||
"houses_price_dynamics(re-point)",
|
||
),
|
||
(
|
||
"house_imv_evaluations",
|
||
"house_imv_evaluations(dedup)",
|
||
"house_imv_evaluations(re-point)",
|
||
),
|
||
("house_suggestions", "house_suggestions(dedup)", "house_suggestions(re-point)"),
|
||
(
|
||
"address_mismatch_audit",
|
||
"address_mismatch_audit(dedup)",
|
||
"address_mismatch_audit(re-point)",
|
||
),
|
||
]
|
||
labels = [label for label, _ in hdm._STEPS]
|
||
for _table, del_label, rep_label in unique_children:
|
||
assert del_label in _STEP_SQL, f"{del_label} missing"
|
||
assert rep_label in _STEP_SQL, f"{rep_label} missing"
|
||
assert "DELETE FROM" in _STEP_SQL[del_label]
|
||
assert "UPDATE" in _STEP_SQL[rep_label]
|
||
assert labels.index(del_label) < labels.index(rep_label)
|
||
|
||
|
||
def test_price_dynamics_dedup_uses_live_6col_key_not_stale_3col() -> None:
|
||
"""Migration 029 replaced the price_dynamics UNIQUE with a 6-col key — dedup on it.
|
||
|
||
Deduping on 108's stale (house_id, month_date, source) would still violate the live
|
||
6-col constraint on re-point. The dedup PARTITION must include all 6 dimension columns.
|
||
"""
|
||
sql = _flat(_STEP_SQL["houses_price_dynamics(dedup)"])
|
||
for col in ("source", "room_count", "prices_type", "period", "month_date"):
|
||
assert col in sql, f"price_dynamics dedup missing dimension column {col}"
|
||
# partition over the TARGET keeper (COALESCE(keeper_id, house_id)) so dup-vs-dup is caught.
|
||
assert "PARTITION BY COALESCE(m.keeper_id, t2.house_id)" in sql
|
||
|
||
|
||
def test_collision_dedup_partitions_on_target_keeper() -> None:
|
||
"""All dedup steps partition on the keeper target so cluster-of-3 dup-vs-dup collisions
|
||
are caught, with the keeper's own row surviving (keeper_id IS NULL DESC)."""
|
||
for label in (
|
||
"houses_price_dynamics(dedup)",
|
||
"house_imv_evaluations(dedup)",
|
||
"house_suggestions(dedup)",
|
||
"address_mismatch_audit(dedup)",
|
||
):
|
||
sql = _flat(_STEP_SQL[label])
|
||
assert "PARTITION BY COALESCE(m.keeper_id, t2.house_id)" in sql
|
||
assert "(m.keeper_id IS NULL) DESC" in sql
|
||
|
||
|
||
def test_losers_deleted_and_keeper_backfilled() -> None:
|
||
"""Loser houses deleted; keeper backfilled with sources + aliases to reduce recurrence."""
|
||
assert "DELETE FROM houses" in _flat(str(hdm._DELETE_LOSERS_SQL.text))
|
||
bf_sources = _flat(str(hdm._BACKFILL_SOURCES_SQL.text))
|
||
assert "INSERT INTO house_sources" in bf_sources
|
||
assert "ON CONFLICT (ext_source, ext_id) DO NOTHING" in bf_sources
|
||
bf_aliases = _flat(str(hdm._BACKFILL_ALIASES_SQL.text))
|
||
assert "INSERT INTO house_address_aliases" in bf_aliases
|
||
assert "ON CONFLICT (normalized_address) DO NOTHING" in bf_aliases
|
||
|
||
|
||
def test_no_psycopg_v3_colon_colon_cast() -> None:
|
||
"""psycopg v3: never :param::type — must use CAST(:param AS type)."""
|
||
assert not re.search(r":\w+::", _MAPPING_SQL)
|
||
assert not re.search(r":\w+::", _ALL_STEP_SQL)
|
||
assert not re.search(r":\w+::", _SERVICE_SRC)
|
||
|
||
|
||
def test_no_fstring_or_format_in_parametrised_sql() -> None:
|
||
"""SQL identifiers are static; no .format()/f-string interpolation of *values* into SQL."""
|
||
# The only f-strings build static column-list fragments (_KEEPER_ORDER / completeness),
|
||
# never user/runtime values — assert no '.format(' value-injection into SQL text.
|
||
assert ".format(" not in _SERVICE_SRC
|
||
|
||
|
||
# ── Fias-key pass + cross-fias guard + identity carry-over (#1772 follow-up) ──
|
||
|
||
|
||
def test_fias_pass_clusters_by_house_fias_id() -> None:
|
||
"""A second cluster key: lower(NULLIF(house_fias_id,'')) — the ФИАС/ГАР building UUID.
|
||
|
||
The fias pass catches duplicates the address canon misses (slash-collapse «Сулимова, 32» vs
|
||
«Сулимова, 3/2»; посёлок truncation). Only non-null fias shared by >1 house_id clusters.
|
||
"""
|
||
flat = _flat(_FIAS_MAPPING_SQL)
|
||
assert "'fias:' || lower(house_fias_id)" in flat
|
||
assert "NULLIF(house_fias_id, '') IS NOT NULL" in flat
|
||
# same collision-safe pipeline (same temp table + >1 grouping), NOT the canon cluster key.
|
||
assert "CREATE TEMP TABLE _1772_dup_mapping" in flat
|
||
assert "HAVING count(*) > 1" in flat
|
||
# the fias pass keys on the fias UUID, never on the 'addr:' canon key.
|
||
assert "'addr:'" not in flat
|
||
|
||
|
||
def test_fias_pass_drops_geo_guard_canon_pass_keeps_it() -> None:
|
||
"""#2187: ФИАС identity strictly outranks geo-proximity, so the fias-pass mapping must NOT
|
||
carry the 250 m distance guard (same-fias rows merge even with NULL geom on a side or >250 m
|
||
apart), while the canon-pass mapping MUST keep it (cross-town over-merge protection)."""
|
||
fias = _flat(_FIAS_MAPPING_SQL)
|
||
canon = _flat(_MAPPING_SQL)
|
||
# canon pass keeps the full geo guard (distance + NULL-geom safety on both sides).
|
||
assert "ST_DistanceSphere(loser_geom, keeper_geom) <= 250" in canon
|
||
assert "keeper_geom IS NOT NULL" in canon
|
||
assert "loser_geom IS NOT NULL" in canon
|
||
# fias pass drops the distance guard AND the NULL-geom exclusions entirely.
|
||
assert "ST_DistanceSphere" not in fias
|
||
assert "loser_geom IS NOT NULL" not in fias
|
||
assert "keeper_geom IS NOT NULL" not in fias
|
||
# the cross-fias anti-over-merge guard is untouched in the canon pass.
|
||
assert "lower(loser_fias) <> lower(keeper_fias)" in canon
|
||
|
||
|
||
def test_mapping_sql_geo_guard_param_toggles_only_distance_filter() -> None:
|
||
"""_mapping_sql(apply_geo_guard=...) toggles ONLY the 250 m distance filter; the cross-fias
|
||
guard is emitted regardless, and the default is True (canon-safe)."""
|
||
with_guard = _flat(hdm._mapping_sql(hdm._CANON_KEY_EXPR, apply_geo_guard=True))
|
||
without_guard = _flat(hdm._mapping_sql(hdm._CANON_KEY_EXPR, apply_geo_guard=False))
|
||
assert "ST_DistanceSphere" in with_guard
|
||
assert "ST_DistanceSphere" not in without_guard
|
||
# default = True (the canon pass must never lose its guard by omission).
|
||
assert "ST_DistanceSphere" in _flat(hdm._mapping_sql(hdm._CANON_KEY_EXPR))
|
||
# cross-fias guard present in BOTH renderings (independent of the geo guard).
|
||
assert "lower(loser_fias) <> lower(keeper_fias)" in with_guard
|
||
assert "lower(loser_fias) <> lower(keeper_fias)" in without_guard
|
||
|
||
|
||
def test_both_passes_share_one_pipeline_no_copy_paste() -> None:
|
||
"""The two mappings differ ONLY in the cluster key + the (canon-only) geo guard — no copy-paste.
|
||
|
||
#2187: the 250 m distance guard is emitted for the canon pass ONLY; the fias pass drops it (a
|
||
shared ГАР UUID IS the building identity and outranks proximity). Everything else — the CTE
|
||
skeleton, keeper rule, per-partition geom/fias exposure and the cross-fias guard — is identical.
|
||
"""
|
||
canon = _flat(_MAPPING_SQL)
|
||
fias = _flat(_FIAS_MAPPING_SQL)
|
||
assert "'addr:'" in canon and "'addr:'" not in fias
|
||
assert "'fias:'" in fias and "'fias:'" not in canon
|
||
# both carry the same downstream skeleton (keeper rule, per-partition geom, cross-fias guard).
|
||
for token in (
|
||
"clusters_with_count",
|
||
"listing_counts",
|
||
"ranked AS",
|
||
"AS keeper_geom",
|
||
"lower(loser_fias) <> lower(keeper_fias)",
|
||
):
|
||
assert token in canon and token in fias
|
||
# the 250 m distance guard is CANON-ONLY (#2187) — fias identity outranks proximity.
|
||
assert "ST_DistanceSphere(loser_geom, keeper_geom) <= 250" in canon
|
||
assert "ST_DistanceSphere" not in fias
|
||
|
||
|
||
def test_cross_fias_guard_blocks_slash_collapse_over_merge() -> None:
|
||
"""Canon pass must NOT merge two rows that BOTH carry a non-null but DIFFERENT house_fias_id.
|
||
|
||
This is the critical anti-over-merge fix for the canon slash-collapse class: «Сулимова, 32»
|
||
and «Сулимова, 3/2» share the canon «сулимова32» but are different buildings — their distinct
|
||
fias UUIDs veto the merge.
|
||
"""
|
||
flat = _flat(_MAPPING_SQL)
|
||
assert "h.house_fias_id AS loser_fias" in flat
|
||
assert "AS keeper_fias" in flat
|
||
assert "first_value(h.house_fias_id) OVER" in flat
|
||
# the guard: NOT (both non-null AND lower-different).
|
||
assert "NULLIF(loser_fias, '') IS NOT NULL" in flat
|
||
assert "NULLIF(keeper_fias, '') IS NOT NULL" in flat
|
||
assert "lower(loser_fias) <> lower(keeper_fias)" in flat
|
||
|
||
|
||
def test_identity_carry_over_fills_all_enrichment_fields() -> None:
|
||
"""Before deleting losers, the keeper's NULL identity / geo-QC fields are filled from losers."""
|
||
flat = _flat(_CARRY_SQL)
|
||
assert "UPDATE houses k" in flat
|
||
for col in (
|
||
"house_fias_id",
|
||
"cadastral_number",
|
||
"gar_house_guid",
|
||
"gar_flat_count",
|
||
"gar_matched_at",
|
||
"gar_match_method",
|
||
"dadata_qc_geo",
|
||
"dadata_qc_house",
|
||
"dadata_enriched_at",
|
||
):
|
||
# COALESCE keeper-first ⇒ a populated keeper field is NEVER overwritten by a loser.
|
||
assert f"COALESCE(k.{col}, d.{col})" in flat, f"identity field {col} not carried over"
|
||
# deterministic donor selection: loser with geom first, then min id.
|
||
assert "ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC" in flat
|
||
assert "FROM _1772_dup_mapping m" in flat
|
||
assert "JOIN houses l ON l.id = m.loser_id" in flat
|
||
|
||
|
||
def test_carry_over_runs_before_delete_losers_in_each_pass() -> None:
|
||
"""Identity carry-over must execute BEFORE the loser rows are deleted, in every pass."""
|
||
src = inspect.getsource(hdm._run_merge_pass)
|
||
assert "_CARRY_OVER_IDENTITY_SQL" in src
|
||
assert src.index("_CARRY_OVER_IDENTITY_SQL") < src.index("_DELETE_LOSERS_SQL")
|
||
|
||
|
||
def test_merge_runs_fias_then_canon_pass_in_one_transaction() -> None:
|
||
"""merge_duplicate_houses runs the fias pass first, then the canon pass, in ONE transaction."""
|
||
src = inspect.getsource(hdm.merge_duplicate_houses)
|
||
assert 'pass_label="fias"' in src and 'pass_label="canon"' in src
|
||
assert src.index('pass_label="fias"') < src.index('pass_label="canon"')
|
||
assert "_BUILD_MAPPING_SQL_FIAS" in src
|
||
# single transaction wrapping BOTH passes: exactly one commit.
|
||
assert src.count("db.commit()") == 1
|
||
|
||
|
||
def test_carry_over_no_psycopg_v3_colon_colon_cast() -> None:
|
||
"""psycopg v3: the carry-over SQL must not use the :param::type bound-param cast form."""
|
||
assert not re.search(r":\w+::", _CARRY_SQL)
|
||
|
||
|
||
# ── Service signature + dry_run / idempotent contract ─────────────────────────
|
||
|
||
|
||
def test_merge_signature_and_dry_run_default() -> None:
|
||
sig = inspect.signature(hdm.merge_duplicate_houses)
|
||
assert "dry_run" in sig.parameters
|
||
assert sig.parameters["dry_run"].default is False
|
||
# dry_run is keyword-only (the `*` in the spec).
|
||
assert sig.parameters["dry_run"].kind is inspect.Parameter.KEYWORD_ONLY
|
||
|
||
|
||
def test_service_logs_every_merge_for_audit() -> None:
|
||
"""Every loser→keeper move is logged with its address (audit trail)."""
|
||
assert "loser_id=%d → keeper_id=%d address=%r" in _SERVICE_SRC
|
||
|
||
|
||
def test_single_transaction_commit_or_rollback() -> None:
|
||
"""One transaction: dry_run → rollback; real → commit; failure → rollback + re-raise."""
|
||
src = inspect.getsource(hdm.merge_duplicate_houses)
|
||
assert "db.rollback()" in src
|
||
assert "db.commit()" in src
|
||
assert "raise" in src # failures re-raise, never swallowed
|
||
|
||
|
||
# ── Fake-db behavioural: dry_run / idempotent / counts (no Postgres) ──────────
|
||
|
||
|
||
class _FakeResult:
|
||
def __init__(self, rowcount: int = 0, rows: list[Any] | None = None, scalar: Any = None):
|
||
self.rowcount = rowcount
|
||
self._rows = rows or []
|
||
self._scalar = scalar
|
||
|
||
def all(self) -> list[Any]:
|
||
return self._rows
|
||
|
||
def scalar(self) -> Any:
|
||
return self._scalar
|
||
|
||
|
||
class _Row:
|
||
def __init__(self, loser_id: int, keeper_id: int, norm_address: str):
|
||
self.loser_id = loser_id
|
||
self.keeper_id = keeper_id
|
||
self.norm_address = norm_address
|
||
|
||
|
||
class _FakeDB:
|
||
"""Session stand-in: build-mapping + a scripted SELECT result, then per-step rowcounts."""
|
||
|
||
def __init__(self, mapping_rows: list[_Row], step_rowcount: int = 1):
|
||
self._mapping_rows = mapping_rows
|
||
self._step_rowcount = step_rowcount
|
||
self._mapping_served = False
|
||
self.commits = 0
|
||
self.rollbacks = 0
|
||
self.executed: list[str] = []
|
||
|
||
def execute(self, clause: Any, params: dict | None = None) -> _FakeResult:
|
||
sql = str(getattr(clause, "text", clause))
|
||
self.executed.append(sql)
|
||
if "CREATE TEMP TABLE" in sql:
|
||
return _FakeResult()
|
||
if "SELECT loser_id, keeper_id, norm_address" in sql:
|
||
# The service now runs TWO passes (fias, then canon). Model «fias pass found the
|
||
# duplicates, canon pass is clean»: serve the scripted mapping once, empty afterwards.
|
||
if self._mapping_served:
|
||
return _FakeResult(rows=[])
|
||
self._mapping_served = True
|
||
return _FakeResult(rows=list(self._mapping_rows))
|
||
# any UPDATE/DELETE/INSERT step (incl. DROP TABLE, carry-over, delete, backfill)
|
||
return _FakeResult(rowcount=self._step_rowcount)
|
||
|
||
def commit(self) -> None:
|
||
self.commits += 1
|
||
|
||
def rollback(self) -> None:
|
||
self.rollbacks += 1
|
||
|
||
|
||
def test_idempotent_no_op_when_no_dups() -> None:
|
||
"""Empty mapping → no commit (only the temp table was opened) → all-zero counters."""
|
||
db = _FakeDB(mapping_rows=[])
|
||
out = hdm.merge_duplicate_houses(db, dry_run=False) # type: ignore[arg-type]
|
||
assert out["clusters_merged"] == 0
|
||
assert out["losers_deleted"] == 0
|
||
assert out["listings_repointed"] == 0
|
||
assert db.commits == 0 # nothing to commit
|
||
assert db.rollbacks == 1 # the no-op rollback
|
||
|
||
|
||
def test_dry_run_computes_counts_but_rolls_back() -> None:
|
||
"""dry_run=True → counts populated, ZERO commits, exactly one rollback (no writes)."""
|
||
rows = [_Row(2, 1, "ул. ленина, 5"), _Row(3, 1, "ул. ленина, 5")]
|
||
db = _FakeDB(mapping_rows=rows, step_rowcount=2)
|
||
out = hdm.merge_duplicate_houses(db, dry_run=True) # type: ignore[arg-type]
|
||
assert out["dry_run"] == 1
|
||
assert out["clusters_merged"] == 1 # both losers → one keeper
|
||
assert out["losers_deleted"] == 2
|
||
assert out["listings_repointed"] == 2 # the 'listings' step rowcount
|
||
assert db.commits == 0 # dry-run NEVER commits
|
||
assert db.rollbacks == 1
|
||
|
||
|
||
def test_real_merge_commits() -> None:
|
||
"""dry_run=False with dups → exactly one commit, no rollback."""
|
||
rows = [_Row(2, 1, "ул. мира, 10")]
|
||
db = _FakeDB(mapping_rows=rows, step_rowcount=1)
|
||
out = hdm.merge_duplicate_houses(db, dry_run=False) # type: ignore[arg-type]
|
||
assert out["dry_run"] == 0
|
||
assert out["losers_deleted"] == 1
|
||
assert db.commits == 1
|
||
assert db.rollbacks == 0
|
||
|
||
|
||
# ── Run-lifecycle wrapper ─────────────────────────────────────────────────────
|
||
|
||
|
||
def test_run_wrapper_marks_done_with_counters(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
marked: dict[str, Any] = {}
|
||
from app.services import scrape_runs as runs_mod
|
||
|
||
monkeypatch.setattr(runs_mod, "update_heartbeat", lambda *a, **k: None)
|
||
monkeypatch.setattr(
|
||
runs_mod,
|
||
"mark_done",
|
||
lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)),
|
||
)
|
||
monkeypatch.setattr(runs_mod, "mark_failed", lambda *a, **k: None)
|
||
monkeypatch.setattr(
|
||
hdm,
|
||
"merge_duplicate_houses",
|
||
lambda _db, dry_run=False: {"clusters_merged": 3, "losers_deleted": 5, "dry_run": 0},
|
||
)
|
||
|
||
out = hdm.run_house_dedup_merge(object(), run_id=42, params={"dry_run": False}) # type: ignore[arg-type]
|
||
assert out["clusters_merged"] == 3
|
||
assert marked["run_id"] == 42
|
||
assert marked["counters"]["losers_deleted"] == 5
|
||
|
||
|
||
def test_run_wrapper_passes_dry_run_param(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
captured: dict[str, Any] = {}
|
||
from app.services import scrape_runs as runs_mod
|
||
|
||
monkeypatch.setattr(runs_mod, "update_heartbeat", lambda *a, **k: None)
|
||
monkeypatch.setattr(runs_mod, "mark_done", lambda *a, **k: None)
|
||
monkeypatch.setattr(runs_mod, "mark_failed", lambda *a, **k: None)
|
||
|
||
def _fake_merge(_db: Any, dry_run: bool = False) -> dict[str, int]:
|
||
captured["dry_run"] = dry_run
|
||
return {"dry_run": int(dry_run)}
|
||
|
||
monkeypatch.setattr(hdm, "merge_duplicate_houses", _fake_merge)
|
||
hdm.run_house_dedup_merge(object(), run_id=1, params={"dry_run": True}) # type: ignore[arg-type]
|
||
assert captured["dry_run"] is True
|
||
|
||
|
||
def test_run_wrapper_marks_failed_on_error(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
failed: dict[str, Any] = {}
|
||
from app.services import scrape_runs as runs_mod
|
||
|
||
monkeypatch.setattr(runs_mod, "update_heartbeat", lambda *a, **k: None)
|
||
monkeypatch.setattr(runs_mod, "mark_done", lambda *a, **k: None)
|
||
monkeypatch.setattr(
|
||
runs_mod,
|
||
"mark_failed",
|
||
lambda _db, run_id, err, counters: failed.update(run_id=run_id, err=err),
|
||
)
|
||
|
||
def _boom(_db: Any, dry_run: bool = False) -> dict[str, int]:
|
||
raise RuntimeError("merge exploded")
|
||
|
||
monkeypatch.setattr(hdm, "merge_duplicate_houses", _boom)
|
||
|
||
class _DB:
|
||
def rollback(self) -> None:
|
||
pass
|
||
|
||
with pytest.raises(RuntimeError):
|
||
hdm.run_house_dedup_merge(_DB(), run_id=9, params={}) # type: ignore[arg-type]
|
||
assert failed["run_id"] == 9
|
||
assert "merge exploded" in failed["err"]
|
||
|
||
|
||
# ── Scheduler wiring ──────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_scheduler_has_trigger_and_dispatch() -> None:
|
||
assert hasattr(scheduler, "trigger_house_dedup_merge_run")
|
||
loop_src = inspect.getsource(scheduler.scheduler_loop)
|
||
assert 'source == "house_dedup_merge"' in loop_src
|
||
assert "trigger_house_dedup_merge_run(db, sch)" in loop_src
|
||
|
||
|
||
def test_trigger_claims_run_and_runs_in_executor() -> None:
|
||
src = inspect.getsource(scheduler.trigger_house_dedup_merge_run)
|
||
assert "_claim_run(db, schedule_row)" in src
|
||
assert "if run_id is None:" in src
|
||
# sync DB-only task → run_in_executor (mirror cadastral_geo_match, not a bare await).
|
||
assert "run_in_executor" in src
|
||
assert "run_house_dedup_merge" in src
|
||
# fresh session inside the spawned task + #1182 P2 detached spawn via _spawn_tracked
|
||
# (registry strong-ref = RUF006 keep-alive + graceful SIGTERM-drain).
|
||
assert "run_db = SessionLocal()" in src
|
||
assert "_spawn_tracked(_run())" in src
|
||
assert "run_db.close()" in src
|
||
|
||
|
||
# ── Migration content ─────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_migration_135_seeds_schedule_disabled() -> None:
|
||
"""The deploy must be NEUTRAL — schedule seeded enabled=false (DORMANT)."""
|
||
sql = _MIGRATION_135.read_text(encoding="utf-8")
|
||
assert "INSERT INTO scrape_schedules" in sql
|
||
assert "'house_dedup_merge'" in sql
|
||
assert "ON CONFLICT (source) DO NOTHING" in sql
|
||
assert "BEGIN;" in sql and "COMMIT;" in sql
|
||
# enabled=false in the VALUES (not enabled=true). Find the source line's VALUES tuple.
|
||
flat = re.sub(r"\s+", " ", sql)
|
||
m = re.search(r"'house_dedup_merge',\s*(\w+)", flat)
|
||
assert m is not None and m.group(1) == "false", "schedule MUST be seeded enabled=false"
|
||
|
||
|
||
def test_migration_135_weekly_window_and_dry_run_param() -> None:
|
||
sql = _MIGRATION_135.read_text(encoding="utf-8")
|
||
# weekly cadence + dry_run default param present.
|
||
assert '"interval_days": 7' in sql
|
||
assert '"dry_run": false' in sql
|
||
# next_run_at NOT NULL = tomorrow (seed convention; harmless while disabled).
|
||
assert "CURRENT_DATE + INTERVAL '1 day'" in sql
|
||
assert "make_interval(hours => 4)" in sql
|
||
|
||
|
||
def test_migration_135_no_psycopg_v3_colon_colon_cast() -> None:
|
||
sql = _MIGRATION_135.read_text(encoding="utf-8")
|
||
assert not re.search(r":\w+::", sql)
|
||
|
||
|
||
# ── Optional real-Postgres behavioural merge test (self-skips without a DB) ────
|
||
|
||
|
||
def _live_session() -> Any | None:
|
||
"""Return a SQLAlchemy Session if a non-placeholder Postgres is reachable, else None."""
|
||
try:
|
||
from sqlalchemy import create_engine
|
||
from sqlalchemy.orm import sessionmaker
|
||
|
||
dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "")
|
||
if not dsn or "localhost:5432/test" in dsn:
|
||
return None
|
||
engine = create_engine(dsn, future=True)
|
||
conn = engine.connect()
|
||
from sqlalchemy import text as _t
|
||
|
||
conn.execute(_t("SELECT 1"))
|
||
conn.close()
|
||
return sessionmaker(bind=engine, future=True)()
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
|
||
def test_real_merge_repoints_dedups_deletes_and_is_idempotent() -> None:
|
||
"""End-to-end on a real DB: 2 houses sharing an address + colliding children →
|
||
merge re-points listings, dedups the colliding children WITHOUT a UNIQUE violation,
|
||
deletes the loser, keeper survives with merged data; second run is a no-op; dry_run
|
||
makes no writes."""
|
||
from sqlalchemy import text as _t
|
||
|
||
db = _live_session()
|
||
assert db is not None
|
||
try:
|
||
# Two houses at the SAME address. Keeper (geom present) should win.
|
||
db.execute(
|
||
_t(
|
||
"INSERT INTO houses (id, source, ext_house_id, address, lat, lon) VALUES "
|
||
"(900001, 'avito', 'EXT-KEEP', 'тестдом 1772, 1', 56.84, 60.60),"
|
||
"(900002, 'cian', 'EXT-LOSE', 'тестдом 1772, 1', NULL, NULL)"
|
||
)
|
||
)
|
||
# listings pointing at BOTH (the loser's must be re-pointed). source_url, dedup_hash,
|
||
# price_rub are NOT NULL with no default (002_core_tables.sql); dedup_hash is also
|
||
# UNIQUE, so each row needs a distinct value.
|
||
db.execute(
|
||
_t(
|
||
"INSERT INTO listings "
|
||
"(id, source, source_url, source_id, dedup_hash, price_rub, house_id_fk) "
|
||
"VALUES "
|
||
"(910001, 'avito', 'http://t/1772/k', 'L-KEEP', 'dh-1772-keep', 5000000, 900001),"
|
||
"(910002, 'cian', 'http://t/1772/l', 'L-LOSE', 'dh-1772-lose', 6000000, 900002)"
|
||
)
|
||
)
|
||
# COLLIDING children: both houses have a price_dynamics row whose 5 dimension fields
|
||
# (source, room_count, prices_type, period, month_date) are IDENTICAL and only house_id
|
||
# differs (loser vs keeper). After the loser re-points to the keeper, both rows would
|
||
# share the LIVE 6-col UNIQUE key → the dedup step must drop the loser's row first.
|
||
# price_per_sqm is the real NOT NULL value column (024_houses_price_dynamics.sql);
|
||
# there is no avg_price column.
|
||
db.execute(
|
||
_t(
|
||
"INSERT INTO houses_price_dynamics "
|
||
"(house_id, month_date, source, room_count, prices_type, period, price_per_sqm) "
|
||
"VALUES "
|
||
"(900001, DATE '2026-01-01', 'cian', 'all', 'priceSqm', 'allTime', 100000),"
|
||
"(900002, DATE '2026-01-01', 'cian', 'all', 'priceSqm', 'allTime', 999999)"
|
||
)
|
||
)
|
||
# house_sources UNIQUE(ext_source, ext_id) is GLOBAL (not scoped to house_id), so two
|
||
# rows can never share the key at insert time — a true same-key collision is impossible
|
||
# to seed. We give the loser and keeper DISTINCT source rows; the merge must re-point the
|
||
# loser's onto the keeper (the collision-delete guard is a defensive no-op here, and both
|
||
# rows survive on the keeper).
|
||
db.execute(
|
||
_t(
|
||
"INSERT INTO house_sources "
|
||
"(house_id, ext_source, ext_id, confidence, matched_method) VALUES "
|
||
"(900001, 'avito', 'SRC-KEEP', 1.0, 'test'),"
|
||
"(900002, 'cian', 'SRC-LOSE', 1.0, 'test')"
|
||
)
|
||
)
|
||
db.commit()
|
||
|
||
# ── dry_run makes NO writes ──
|
||
before = db.execute(_t("SELECT count(*) FROM houses WHERE id IN (900001,900002)")).scalar()
|
||
dry = hdm.merge_duplicate_houses(db, dry_run=True)
|
||
assert dry["losers_deleted"] >= 1
|
||
after_dry = db.execute(
|
||
_t("SELECT count(*) FROM houses WHERE id IN (900001,900002)")
|
||
).scalar()
|
||
assert after_dry == before == 2, "dry_run must not delete anything"
|
||
|
||
# ── real merge ──
|
||
out = hdm.merge_duplicate_houses(db, dry_run=False)
|
||
assert out["losers_deleted"] >= 1
|
||
# loser gone, keeper survives.
|
||
survivors = db.execute(
|
||
_t("SELECT id FROM houses WHERE id IN (900001,900002) ORDER BY id")
|
||
).all()
|
||
assert [r.id for r in survivors] == [900001]
|
||
# loser's listing re-pointed to keeper.
|
||
repointed = db.execute(_t("SELECT house_id_fk FROM listings WHERE id = 910002")).scalar()
|
||
assert repointed == 900001
|
||
# price_dynamics deduped — exactly one row for the keeper at that 6-col key (no dup),
|
||
# and it is the KEEPER's own row (price_per_sqm=100000), not the loser's (999999):
|
||
# the dedup keeps keeper rows (keeper_id IS NULL DESC) over loser rows.
|
||
pd_rows = db.execute(
|
||
_t(
|
||
"SELECT price_per_sqm FROM houses_price_dynamics "
|
||
"WHERE house_id = 900001 AND month_date = DATE '2026-01-01' "
|
||
"AND source='cian' AND room_count='all' AND prices_type='priceSqm' "
|
||
"AND period='allTime'"
|
||
)
|
||
).all()
|
||
assert len(pd_rows) == 1, "colliding price_dynamics must be deduped, not duplicated"
|
||
assert pd_rows[0].price_per_sqm == 100000, "dedup must keep the keeper's own row"
|
||
# house_sources: the loser's source row re-pointed onto the keeper — both now on 900001
|
||
# (plus the backfill row for the keeper's own avito ext_house_id). The loser's row must
|
||
# be present and re-pointed; no row may dangle on the deleted loser.
|
||
hs_on_keeper = db.execute(
|
||
_t(
|
||
"SELECT count(*) FROM house_sources "
|
||
"WHERE house_id = 900001 AND ext_id IN ('SRC-KEEP','SRC-LOSE')"
|
||
)
|
||
).scalar()
|
||
assert hs_on_keeper == 2, "both source rows must be re-pointed onto the keeper"
|
||
hs_dangling = db.execute(
|
||
_t("SELECT count(*) FROM house_sources WHERE house_id = 900002")
|
||
).scalar()
|
||
assert hs_dangling == 0, "no source row may remain on the deleted loser"
|
||
|
||
# ── idempotent second run = no-op ──
|
||
again = hdm.merge_duplicate_houses(db, dry_run=False)
|
||
assert again["losers_deleted"] == 0
|
||
finally:
|
||
# cleanup: drop listings first (house_id_fk has no CASCADE), then the houses — which
|
||
# CASCADE-deletes house_sources / price_dynamics / aliases / etc. Also sweep any
|
||
# backfill house_sources rows the merge may have inserted for the keeper.
|
||
db.rollback()
|
||
db.execute(_t("DELETE FROM listings WHERE id IN (910001,910002)"))
|
||
db.execute(
|
||
_t("DELETE FROM house_sources WHERE ext_id IN ('SRC-KEEP','SRC-LOSE','EXT-KEEP')")
|
||
)
|
||
db.execute(
|
||
_t("DELETE FROM house_address_aliases WHERE normalized_address = 'тестдом 1772, 1'")
|
||
)
|
||
db.execute(_t("DELETE FROM houses WHERE id IN (900001,900002)"))
|
||
db.commit()
|
||
db.close()
|
||
|
||
|
||
@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
|
||
def test_real_canon_clusterkey_and_geo_guard_merge_semantics() -> None:
|
||
"""End-to-end on a real DB for the canon cluster-key + 250 m geo guard:
|
||
|
||
A. «ул. ТестоваяX, 66» + «улица ТестоваяX, 66», ~10 m apart → MERGED (canon collapses
|
||
ул/улица), both listings re-pointed onto the single survivor.
|
||
B. two identical «улица ТестоваяX, 5», ~5 km apart → NOT merged (geo guard >250 m),
|
||
both house rows survive.
|
||
C. «ТестоваяX, 34к2» + «ТестоваяX, 34к4», ~10 m apart → NOT merged (different
|
||
корпус → different canon — корпус is preserved), both survive.
|
||
|
||
Synthetic street «Тестовая1772» avoids colliding with any real prod address/alias.
|
||
Deltas in latitude at ~56.84°: 0.00009° ≈ 10 m (geo passes), 0.045° ≈ 5 km (geo blocks).
|
||
"""
|
||
from sqlalchemy import text as _t
|
||
|
||
db = _live_session()
|
||
assert db is not None
|
||
try:
|
||
db.execute(
|
||
_t(
|
||
"INSERT INTO houses (id, source, ext_house_id, address, lat, lon) VALUES "
|
||
# A — ул/улица spelling variants of the SAME building, ~10 m apart → MERGE
|
||
"(900010, 'avito', 'EXT-T-VK', 'улица Тестовая1772, 66', 56.84000, 60.60000),"
|
||
"(900011, 'cian', 'EXT-T-VL', 'ул. Тестовая1772, 66', 56.84009, 60.60000),"
|
||
# B — same canon (ленина-like) but ~5 km apart → geo guard BLOCKS the merge
|
||
"(900012, 'avito', 'EXT-T-L1', 'улица Тестовая1772, 5', 56.84000, 60.60000),"
|
||
"(900013, 'cian', 'EXT-T-L2', 'улица Тестовая1772, 5', 56.88500, 60.60000),"
|
||
# C — different корпус → different canon, ~10 m apart → NOT merged
|
||
"(900014, 'avito', 'EXT-T-M2', 'Тестовая1772, 34к2', 56.84000, 60.60000),"
|
||
"(900015, 'cian', 'EXT-T-M4', 'Тестовая1772, 34к4', 56.84009, 60.60000)"
|
||
)
|
||
)
|
||
db.execute(
|
||
_t(
|
||
"INSERT INTO listings "
|
||
"(id, source, source_url, source_id, dedup_hash, price_rub, house_id_fk) VALUES "
|
||
"(910010, 'avito', 'http://t/g/k', 'LG-K', 'dh-g-keep', 5000000, 900010),"
|
||
"(910011, 'cian', 'http://t/g/l', 'LG-L', 'dh-g-lose', 6000000, 900011)"
|
||
)
|
||
)
|
||
db.commit()
|
||
|
||
out = hdm.merge_duplicate_houses(db, dry_run=False)
|
||
assert out["losers_deleted"] >= 1
|
||
|
||
ids = [
|
||
r.id
|
||
for r in db.execute(
|
||
_t("SELECT id FROM houses WHERE id BETWEEN 900010 AND 900015 ORDER BY id")
|
||
).all()
|
||
]
|
||
# A: exactly the min-id keeper survives; the ул/улица spelling variant merged away.
|
||
assert 900010 in ids
|
||
assert 900011 not in ids, "ул/улица spelling variant must be merged onto the keeper"
|
||
# B: BOTH survive — same canon but >250 m apart → geo guard blocks cross-town over-merge.
|
||
assert 900012 in ids and 900013 in ids, "same-canon but >250 m apart must NOT merge"
|
||
# C: BOTH survive — different корпус → different canon (корпус preserved).
|
||
assert 900014 in ids and 900015 in ids, "different корпус must NOT merge"
|
||
|
||
# A: the merged loser's listing re-pointed onto the surviving keeper.
|
||
repointed = db.execute(_t("SELECT house_id_fk FROM listings WHERE id = 910011")).scalar()
|
||
assert repointed == 900010
|
||
|
||
# idempotent second run: our merged loser stays gone, the geo-blocked pairs untouched.
|
||
hdm.merge_duplicate_houses(db, dry_run=False)
|
||
ids2 = [
|
||
r.id
|
||
for r in db.execute(
|
||
_t("SELECT id FROM houses WHERE id BETWEEN 900010 AND 900015 ORDER BY id")
|
||
).all()
|
||
]
|
||
assert 900011 not in ids2
|
||
assert {900010, 900012, 900013, 900014, 900015}.issubset(set(ids2))
|
||
finally:
|
||
db.rollback()
|
||
db.execute(_t("DELETE FROM listings WHERE id IN (910010,910011)"))
|
||
db.execute(
|
||
_t(
|
||
"DELETE FROM house_sources WHERE ext_id IN "
|
||
"('EXT-T-VK','EXT-T-VL','EXT-T-L1','EXT-T-L2','EXT-T-M2','EXT-T-M4')"
|
||
)
|
||
)
|
||
db.execute(
|
||
_t(
|
||
"DELETE FROM house_address_aliases WHERE normalized_address IN "
|
||
"('улица тестовая1772, 66','ул. тестовая1772, 66','улица тестовая1772, 5',"
|
||
"'тестовая1772, 34к2','тестовая1772, 34к4')"
|
||
)
|
||
)
|
||
db.execute(_t("DELETE FROM houses WHERE id BETWEEN 900010 AND 900015"))
|
||
db.commit()
|
||
db.close()
|
||
|
||
|
||
@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
|
||
def test_real_fias_pass_cross_guard_and_identity_carryover() -> None:
|
||
"""End-to-end on a real DB for the #1772 follow-up (fias pass + cross-fias guard + carry-over):
|
||
|
||
A. same house_fias_id, DIFFERENT canon (different streets) → MERGED by the FIAS pass;
|
||
the loser's listing re-points onto the single survivor.
|
||
B. same canon (slash-collapse «32»/«3/2»), BOTH fias filled and DIFFERENT → NOT merged
|
||
(cross-fias guard vetoes — provably different buildings), both survive.
|
||
C. same canon, fias only on the LOSER → MERGED by the canon pass; the loser's
|
||
house_fias_id + gar_house_guid CARRY OVER onto the fias-less keeper.
|
||
D. a second full run is idempotent (nothing left to merge).
|
||
|
||
Synthetic streets «*1772» avoid colliding with any real prod address/alias.
|
||
"""
|
||
from sqlalchemy import text as _t
|
||
|
||
db = _live_session()
|
||
assert db is not None
|
||
try:
|
||
db.execute(
|
||
_t(
|
||
"INSERT INTO houses "
|
||
"(id, source, ext_house_id, address, lat, lon, house_fias_id, gar_house_guid, "
|
||
" dadata_enriched_at) VALUES "
|
||
# A — same fias, different canon (different streets), ~10 m apart → FIAS-pass merge
|
||
"(900020,'avito','EXT-F-K','ФиасОдин1772, 10', 56.84000,60.60000,"
|
||
" 'F-SAME-1772',NULL,NULL),"
|
||
"(900021,'cian', 'EXT-F-L','СовсемДругая1772, 77',56.84009,60.60000,"
|
||
" 'F-SAME-1772',NULL,NULL),"
|
||
# B — same canon (slash-collapse), DIFFERENT fias → cross-fias guard BLOCKS
|
||
"(900022,'avito','EXT-B-1','Клара1772, 32',56.84000,60.60000,"
|
||
" 'F-B1-1772',NULL,NULL),"
|
||
"(900023,'cian', 'EXT-B-2','Клара1772, 3/2',56.84009,60.60000,"
|
||
" 'F-B2-1772',NULL,NULL),"
|
||
# C — same canon, fias only on the loser → canon-pass merge + carry-over
|
||
"(900024,'avito','EXT-C-K','Донбасс1772, 8',56.84000,60.60000,NULL,NULL,NULL),"
|
||
"(900025,'cian', 'EXT-C-L','Донбасс1772, 8',56.84009,60.60000,"
|
||
" 'F-CARRY-1772','G-CARRY-1772',NOW())"
|
||
)
|
||
)
|
||
# Listings: A keeper + A loser (loser's must re-point); C keeper (gives it the listing-count
|
||
# tie-break so the fias-LESS row wins the keeper slot and receives the carried-over fias).
|
||
db.execute(
|
||
_t(
|
||
"INSERT INTO listings "
|
||
"(id, source, source_url, source_id, dedup_hash, price_rub, house_id_fk) VALUES "
|
||
"(910020,'avito','http://t/f/k','LF-K','dh-f-keep',5000000,900020),"
|
||
"(910021,'cian', 'http://t/f/l','LF-L','dh-f-lose',6000000,900021),"
|
||
"(910024,'avito','http://t/c/k','LC-K','dh-c-keep',5500000,900024)"
|
||
)
|
||
)
|
||
db.commit()
|
||
|
||
out = hdm.merge_duplicate_houses(db, dry_run=False)
|
||
assert out["losers_deleted"] >= 2 # at least A + C merged
|
||
|
||
ids = [
|
||
r.id
|
||
for r in db.execute(
|
||
_t("SELECT id FROM houses WHERE id BETWEEN 900020 AND 900025 ORDER BY id")
|
||
).all()
|
||
]
|
||
# A: fias merge — different canon, same fias → the loser is gone.
|
||
assert 900020 in ids and 900021 not in ids, "same fias must merge across differing canons"
|
||
# B: cross-fias guard — same canon but different fias → BOTH survive.
|
||
assert 900022 in ids and 900023 in ids, "different fias must veto a same-canon merge"
|
||
# C: canon merge — same canon, one fias → the fias-less keeper survives.
|
||
assert 900024 in ids and 900025 not in ids, "same-canon single-fias pair must merge"
|
||
|
||
# A: the merged loser's listing re-pointed onto the surviving keeper.
|
||
repointed = db.execute(_t("SELECT house_id_fk FROM listings WHERE id = 910021")).scalar()
|
||
assert repointed == 900020
|
||
|
||
# C: identity carry-over — the loser's fias + ГАР guid moved onto the fias-less keeper.
|
||
keeper_c = db.execute(
|
||
_t("SELECT house_fias_id, gar_house_guid FROM houses WHERE id = 900024")
|
||
).one()
|
||
assert keeper_c.house_fias_id == "F-CARRY-1772", "loser's house_fias_id must carry over"
|
||
assert keeper_c.gar_house_guid == "G-CARRY-1772", "loser's gar_house_guid must carry over"
|
||
|
||
# D: idempotent second run — nothing left to merge.
|
||
again = hdm.merge_duplicate_houses(db, dry_run=False)
|
||
assert again["losers_deleted"] == 0
|
||
finally:
|
||
db.rollback()
|
||
db.execute(_t("DELETE FROM listings WHERE id IN (910020,910021,910024)"))
|
||
db.execute(
|
||
_t(
|
||
"DELETE FROM house_sources WHERE ext_id IN "
|
||
"('EXT-F-K','EXT-F-L','EXT-B-1','EXT-B-2','EXT-C-K','EXT-C-L')"
|
||
)
|
||
)
|
||
db.execute(
|
||
_t(
|
||
"DELETE FROM house_address_aliases WHERE normalized_address IN "
|
||
"('фиасодин1772, 10','совсемдругая1772, 77','клара1772, 32','клара1772, 3/2',"
|
||
"'донбасс1772, 8')"
|
||
)
|
||
)
|
||
db.execute(_t("DELETE FROM houses WHERE id BETWEEN 900020 AND 900025"))
|
||
db.commit()
|
||
db.close()
|
||
|
||
|
||
@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
|
||
def test_real_fias_pass_ignores_geo_guard() -> None:
|
||
"""End-to-end on a real DB for #2187 — the FIAS pass drops the 250 m geo guard:
|
||
|
||
A. same house_fias_id, loser has NULL geom (no lat/lon) → MERGED (fias identity outranks
|
||
the missing coordinate; the geom-first keeper survives + the loser's listing re-points).
|
||
B. same house_fias_id, ~5 km apart (>250 m) → MERGED (fias identity outranks
|
||
proximity — a scattered/broken coordinate no longer blocks the merge).
|
||
C. same canon, NO fias, ~5 km apart (>250 m) → NOT merged (the canon pass STILL
|
||
enforces the geo guard — the fix is fias-only, canon behaviour is unchanged).
|
||
|
||
A/B use distinct synthetic streets so ONLY the shared fias groups them (the canon never does);
|
||
C shares one address so only the canon pass can group it. Streets «*2187» avoid prod aliases.
|
||
Latitude delta 0.045° at ~56.84° ≈ 5 km (>250 m).
|
||
"""
|
||
from sqlalchemy import text as _t
|
||
|
||
db = _live_session()
|
||
assert db is not None
|
||
try:
|
||
db.execute(
|
||
_t(
|
||
"INSERT INTO houses "
|
||
"(id, source, ext_house_id, address, lat, lon, house_fias_id) VALUES "
|
||
# A — same fias, loser NULL geom → fias pass merges despite the missing coordinate
|
||
"(900030,'avito','EXT-2187-A-K','ФиасГеоA2187, 1', 56.84000,60.60000,'F-A-2187'),"
|
||
"(900031,'cian', 'EXT-2187-A-L','ФиасГеоAL2187, 2',NULL, NULL, 'F-A-2187'),"
|
||
# B — same fias, ~5 km apart (>250 m) → fias pass merges despite the distance
|
||
"(900032,'avito','EXT-2187-B-K','ФиасГеоB2187, 3', 56.84000,60.60000,'F-B-2187'),"
|
||
"(900033,'cian', 'EXT-2187-B-L','ФиасГеоBL2187, 4',56.88500,60.60000,'F-B-2187'),"
|
||
# C — same canon, NO fias, ~5 km apart → canon pass STILL blocks (guard unchanged)
|
||
"(900034,'avito','EXT-2187-C-1','КанонГео2187, 5', 56.84000,60.60000,NULL),"
|
||
"(900035,'cian', 'EXT-2187-C-2','КанонГео2187, 5', 56.88500,60.60000,NULL)"
|
||
)
|
||
)
|
||
# A loser gets a listing so we prove the re-point still fires with a NULL-geom loser.
|
||
db.execute(
|
||
_t(
|
||
"INSERT INTO listings "
|
||
"(id, source, source_url, source_id, dedup_hash, price_rub, house_id_fk) VALUES "
|
||
"(910031,'cian','http://t/2187/al','L-2187-AL','dh-2187-al',6000000,900031)"
|
||
)
|
||
)
|
||
db.commit()
|
||
|
||
out = hdm.merge_duplicate_houses(db, dry_run=False)
|
||
assert out["losers_deleted"] >= 2 # A + B both merged by the fias pass
|
||
|
||
ids = [
|
||
r.id
|
||
for r in db.execute(
|
||
_t("SELECT id FROM houses WHERE id BETWEEN 900030 AND 900035 ORDER BY id")
|
||
).all()
|
||
]
|
||
# A: same-fias, loser NULL geom → MERGED (the geo guard would have blocked on NULL geom).
|
||
assert 900030 in ids and 900031 not in ids, "same-fias NULL-geom loser must merge"
|
||
# B: same-fias, >250 m apart → MERGED (the geo guard would have blocked on distance).
|
||
assert 900032 in ids and 900033 not in ids, "same-fias >250 m apart must merge"
|
||
# C: same-canon, no fias, >250 m apart → NOT merged (canon geo guard unchanged by #2187).
|
||
assert 900034 in ids and 900035 in ids, "canon-only pair >250 m apart must NOT merge"
|
||
|
||
# A: the NULL-geom loser's listing re-pointed onto the surviving geom-first keeper.
|
||
repointed = db.execute(_t("SELECT house_id_fk FROM listings WHERE id = 910031")).scalar()
|
||
assert repointed == 900030
|
||
finally:
|
||
db.rollback()
|
||
db.execute(_t("DELETE FROM listings WHERE id = 910031"))
|
||
db.execute(_t("DELETE FROM house_sources WHERE house_id BETWEEN 900030 AND 900035"))
|
||
db.execute(_t("DELETE FROM house_address_aliases WHERE house_id BETWEEN 900030 AND 900035"))
|
||
db.execute(_t("DELETE FROM houses WHERE id BETWEEN 900030 AND 900035"))
|
||
db.commit()
|
||
db.close()
|