test(tradein): fix real-DB merge test schema (#1772, #1933 review)

The optional real-DB end-to-end test self-skipped without a DB but would
have errored WITH one (the safety fixture never actually ran). Fixed two
schema bugs + one impossible-collision setup, validated against a live
PostGIS 16 container (test now passes end-to-end, not just collected):

- houses_price_dynamics insert used a non-existent `avg_price` column →
  `price_per_sqm` (the real NOT NULL value column, migration 024). The two
  rows share all 5 dim fields with differing house_id → a genuine LIVE
  6-col UNIQUE collision the dedup step must resolve. Also assert the
  surviving row is the keeper's own (price_per_sqm=100000, not 999999).
- listings insert omitted NOT NULL no-default columns → added source_url,
  dedup_hash (also UNIQUE → distinct per row), price_rub (migration 002).
- house_sources UNIQUE(ext_source,ext_id) is GLOBAL, so the original
  same-key-on-both-houses setup violated at INSERT time (before the merge).
  Replaced with distinct loser/keeper source rows + assert the loser's row
  re-points onto the keeper and none dangle on the deleted loser.

Still self-skips in CI (no Postgres) — correct; now valid against a
tunnel/local DB. Full suite 2444 passed / 0 failed; ruff clean.

tradein-mvp/backend/tests/test_house_dedup_merge.py

@@ -475,30 +475,44 @@ def test_real_merge_repoints_dedups_deletes_and_is_idempotent() -> None:
                 "(900002, 'cian',  'EXT-LOSE', 'тестдом 1772, 1', NULL, NULL)"
             )
         )
-        # listings pointing at BOTH (the loser's must be re-pointed).
+        # 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_id, house_id_fk) VALUES "
-                "(910001, 'avito', 'L-KEEP', 900001),"
-                "(910002, 'cian',  'L-LOSE', 900002)"
+                "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 with the SAME 6-col key.
+        # 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, avg_price) VALUES "
+                "(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)"
             )
         )
-        # COLLIDING house_sources: same (ext_source, ext_id) on both would violate on re-point.
+        # 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-DUP', 1.0, 'test'),"
-                "(900002, 'avito', 'SRC-DUP', 1.0, 'test')"
+                "(900001, 'avito', 'SRC-KEEP', 1.0, 'test'),"
+                "(900002, 'cian',  'SRC-LOSE', 1.0, 'test')"
             )
         )
         db.commit()
@@ -525,30 +539,52 @@ def test_real_merge_repoints_dedups_deletes_and_is_idempotent() -> None:
             _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).
-        pd_cnt = db.execute(
+        # 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 count(*) FROM houses_price_dynamics "
+                "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 pd_cnt == 1, "colliding price_dynamics must be deduped, not duplicated"
-        # house_sources collision resolved — one SRC-DUP row on the keeper.
-        hs_cnt = db.execute(
-            _t("SELECT count(*) FROM house_sources WHERE ext_source='avito' AND ext_id='SRC-DUP'")
+        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_cnt == 1
+        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 (children CASCADE on houses delete).
+        # 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.execute(_t("DELETE FROM house_sources WHERE ext_id='SRC-DUP'"))
         db.commit()
         db.close()