208 lines
8.2 KiB
Python
208 lines
8.2 KiB
Python
"""Tests for #6 same-building analog matching via canonical house_id.
|
|
|
|
Two pieces:
|
|
1. matching.houses.match_house_readonly — read-only target→house resolver
|
|
(cadastr → fingerprint → geo, никогда не INSERT'ит).
|
|
2. estimator._fetch_analogs canonical Tier S — when target_house_id задан,
|
|
матчит listings по house_id_fk; иначе fallback на address-prefix Tier S.
|
|
"""
|
|
|
|
import os
|
|
|
|
# Settings requires DATABASE_URL at init time. Set dummy DSN before any app import.
|
|
os.environ.setdefault("DATABASE_URL", "postgresql://test:test@localhost/test_db")
|
|
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
from unittest.mock import MagicMock
|
|
|
|
from app.services.estimator import _fetch_analogs
|
|
from app.services.matching.houses import match_house_readonly
|
|
|
|
|
|
# ── Helpers ───────────────────────────────────────────────────────────────────
|
|
def _listing(**over: Any) -> dict[str, Any]:
|
|
base = {
|
|
"source": "cian",
|
|
"source_url": "https://cian.ru/offer/1",
|
|
"address": "Екатеринбург, ул Малышева, д 125",
|
|
"lat": 56.8408,
|
|
"lon": 60.6540,
|
|
"rooms": 2,
|
|
"area_m2": 50.0,
|
|
"floor": 3,
|
|
"total_floors": 9,
|
|
"price_rub": 6_000_000.0,
|
|
"price_per_m2": 120_000.0,
|
|
"listing_date": datetime(2026, 5, 1),
|
|
"days_on_market": 10,
|
|
"photo_urls": [],
|
|
"scraped_at": datetime(2026, 5, 20, tzinfo=UTC),
|
|
"distance_m": 0.0,
|
|
"relevance_score": 0.0,
|
|
}
|
|
base.update(over)
|
|
return base
|
|
|
|
|
|
def _db_returning(rows: list[dict[str, Any]]) -> MagicMock:
|
|
"""Session mock where every db.execute().mappings().all() returns rows."""
|
|
db = MagicMock()
|
|
db.execute.return_value.mappings.return_value.all.return_value = rows
|
|
return db
|
|
|
|
|
|
def _executed_sqls(db: MagicMock) -> list[str]:
|
|
return [str(call.args[0]) for call in db.execute.call_args_list if call.args]
|
|
|
|
|
|
# ── match_house_readonly ────────────────────────────────────────────────────
|
|
def test_readonly_cadastr_exact_hit():
|
|
db = MagicMock()
|
|
db.execute.return_value.mappings.return_value.first.side_effect = [{"id": 42}]
|
|
result = match_house_readonly(
|
|
db, address="ул Малышева 125", cadastral_number="66:41:0704045:350"
|
|
)
|
|
assert result == (42, 1.0, "cadastr_exact")
|
|
|
|
|
|
def test_readonly_fias_exact_hit():
|
|
"""Tier 0.5: house_fias_id match → fias_exact (conf 0.95), no cad supplied."""
|
|
db = MagicMock()
|
|
db.execute.return_value.mappings.return_value.first.side_effect = [{"id": 55}]
|
|
result = match_house_readonly(
|
|
db,
|
|
address="ул Малышева 125",
|
|
house_fias_id="0a1b2c3d-0000-4000-8000-000000000001",
|
|
)
|
|
assert result == (55, 0.95, "fias_exact")
|
|
|
|
|
|
def test_readonly_fias_skipped_when_absent():
|
|
"""No house_fias_id → fias tier skipped; first .first() is fingerprint tier."""
|
|
db = MagicMock()
|
|
db.execute.return_value.mappings.return_value.first.side_effect = [{"house_id": 7}]
|
|
result = match_house_readonly(db, address="Екатеринбург, ул Малышева, д 125")
|
|
assert result == (7, 0.9, "fingerprint")
|
|
|
|
|
|
def test_readonly_fingerprint_hit():
|
|
db = MagicMock()
|
|
# cadastral_number=None → cadastr tier skipped; first .first() is fingerprint
|
|
db.execute.return_value.mappings.return_value.first.side_effect = [{"house_id": 7}]
|
|
result = match_house_readonly(db, address="Екатеринбург, ул Малышева, д 125")
|
|
assert result == (7, 0.9, "fingerprint")
|
|
|
|
|
|
def test_readonly_geo_proximity_hit():
|
|
db = MagicMock()
|
|
# fingerprint miss → geo hit
|
|
db.execute.return_value.mappings.return_value.first.side_effect = [
|
|
None,
|
|
{"id": 9, "dist": 12.3},
|
|
]
|
|
result = match_house_readonly(db, address="неизвестный адрес", lat=56.84, lon=60.65)
|
|
assert result == (9, 0.7, "geo_proximity")
|
|
|
|
|
|
def test_readonly_geo_rejected_when_house_number_differs():
|
|
"""P2: geo candidate within 50 m but with a DIFFERENT house number is rejected.
|
|
|
|
Target 'Малышева 125' must not resolve to a nearby '... 200' on a coarse geocode.
|
|
fingerprint miss → geo hit (wrong number) → reject → None.
|
|
"""
|
|
db = MagicMock()
|
|
db.execute.return_value.mappings.return_value.first.side_effect = [
|
|
None,
|
|
{"id": 9, "dist": 12.3, "h_addr": "улица малышева 200"},
|
|
]
|
|
result = match_house_readonly(db, address="ул Малышева 125", lat=56.84, lon=60.65)
|
|
assert result is None
|
|
|
|
|
|
def test_readonly_no_match_returns_none():
|
|
db = MagicMock()
|
|
db.execute.return_value.mappings.return_value.first.side_effect = [None, None]
|
|
result = match_house_readonly(db, address="нигде", lat=56.84, lon=60.65)
|
|
assert result is None
|
|
|
|
|
|
def test_readonly_never_inserts_or_writes():
|
|
"""Критично: read-only matcher не должен ничего создавать/писать."""
|
|
db = MagicMock()
|
|
db.execute.return_value.mappings.return_value.first.side_effect = [None, None]
|
|
match_house_readonly(db, address="нигде", lat=56.84, lon=60.65)
|
|
for sql in _executed_sqls(db):
|
|
upper = sql.upper()
|
|
assert "INSERT" not in upper
|
|
assert "UPDATE" not in upper
|
|
assert "ADVISORY" not in upper # no pg_advisory_xact_lock
|
|
|
|
|
|
def test_readonly_no_coords_skips_geo_tier():
|
|
"""Без lat/lon geo-tier не запускается; при fingerprint-miss → None."""
|
|
db = MagicMock()
|
|
db.execute.return_value.mappings.return_value.first.side_effect = [None]
|
|
result = match_house_readonly(db, address="только адрес без координат")
|
|
assert result is None
|
|
|
|
|
|
# ── _fetch_analogs canonical Tier S ──────────────────────────────────────────
|
|
def test_canonical_tier_s_fires_with_house_id():
|
|
rows = [_listing(source=s) for s in ("cian", "avito", "yandex", "n1")]
|
|
db = _db_returning(rows)
|
|
listings, _fallback, tier = _fetch_analogs(
|
|
db,
|
|
lat=56.84,
|
|
lon=60.65,
|
|
rooms=2,
|
|
area=50.0,
|
|
radius_m=1000,
|
|
full_address="Екатеринбург, ул Малышева, д 125",
|
|
target_house_id=555,
|
|
)
|
|
assert tier == "S"
|
|
assert len(listings) >= 3
|
|
# Канонический Tier S — первый запрос должен матчить по house_id_fk
|
|
assert "house_id_fk" in _executed_sqls(db)[0]
|
|
|
|
|
|
def test_falls_back_to_address_tier_s_without_house_id():
|
|
rows = [_listing(source=s) for s in ("cian", "avito", "yandex", "n1")]
|
|
db = _db_returning(rows)
|
|
_listings, _fallback, tier = _fetch_analogs(
|
|
db,
|
|
lat=56.84,
|
|
lon=60.65,
|
|
rooms=2,
|
|
area=50.0,
|
|
radius_m=1000,
|
|
full_address="Екатеринбург, ул Малышева, д 125",
|
|
target_house_id=None,
|
|
)
|
|
assert tier == "S"
|
|
# Без target_house_id канонический tier пропущен — первый запрос по address ILIKE
|
|
first_sql = _executed_sqls(db)[0]
|
|
assert "house_id_fk" not in first_sql
|
|
assert "ILIKE" in first_sql
|
|
|
|
|
|
def test_canonical_tier_s_too_few_falls_through():
|
|
"""<3 в том же доме → не возвращаем Tier S(canonical), идём дальше."""
|
|
rows = [_listing()] # только 1 в доме
|
|
db = _db_returning(rows)
|
|
# full_address без распознаваемого short_addr И year/floors None →
|
|
# address-Tier S и Tier H пропустятся, упадём в Tier W (тоже вернёт rows mock).
|
|
_listings, _fallback, tier = _fetch_analogs(
|
|
db,
|
|
lat=56.84,
|
|
lon=60.65,
|
|
rooms=2,
|
|
area=50.0,
|
|
radius_m=1000,
|
|
full_address=None,
|
|
target_house_id=555,
|
|
)
|
|
# canonical Tier S не сработал (1 < 3); tier — не S(canonical) early-return.
|
|
# Mock возвращает те же rows для Tier W, так что итог не пустой, tier='W'.
|
|
assert tier == "W"
|