gendesign/tradein-mvp/backend/tests/test_house_dedup_merge.py

"""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)
_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


# ── 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.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:
            return _FakeResult(rows=list(self._mapping_rows))
        # any UPDATE/DELETE/INSERT step
        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()