367 lines
12 KiB
Python
367 lines
12 KiB
Python
"""Unit tests for cross-source matching service.
|
||
|
||
Covers:
|
||
- normalize_address(): abbreviation expansion, punctuation stripping, edge cases
|
||
- address_fingerprint(): stability on case variation, sensitivity to coords, None handling
|
||
- match_or_create_house() / match_or_create_listing(): mock-DB tier routing
|
||
|
||
Reference: decisions/Cross_Source_Matching_Strategy.md
|
||
"""
|
||
|
||
from unittest.mock import MagicMock
|
||
|
||
from app.services.matching.conflict_resolution import (
|
||
HOUSE_FIELD_PRIORITY,
|
||
LISTING_FIELD_PRIORITY,
|
||
update_canonical_fields,
|
||
)
|
||
from app.services.matching.normalize import address_fingerprint, normalize_address
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# normalize_address
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_normalize_strips_punctuation_and_collapses_whitespace():
|
||
assert normalize_address(" Test, Address! ") == "test address"
|
||
|
||
|
||
def test_normalize_expands_ul_abbreviation():
|
||
result = normalize_address("ул Ленина, 5")
|
||
assert "улица" in result
|
||
assert "5" in result
|
||
|
||
|
||
def test_normalize_expands_pr_abbreviation():
|
||
result = normalize_address("пр Мира 100")
|
||
assert "проспект" in result
|
||
assert "100" in result
|
||
|
||
|
||
def test_normalize_expands_prkT_abbreviation():
|
||
result = normalize_address("пр-кт Ленина 10")
|
||
assert "проспект" in result
|
||
# 'кт' must NOT remain as a stray token after expansion
|
||
assert 'кт' not in result.split()
|
||
|
||
|
||
def test_normalize_prkT_no_leftover_kt():
|
||
"""пр-кт must expand to 'проспект' cleanly — no leftover 'кт' token."""
|
||
result = normalize_address("пр-кт Ленина 10")
|
||
tokens = result.split()
|
||
assert "проспект" in tokens
|
||
assert "пр" not in tokens
|
||
assert "кт" not in tokens
|
||
|
||
|
||
def test_normalize_handles_none():
|
||
assert normalize_address(None) == ""
|
||
|
||
|
||
def test_normalize_handles_empty_string():
|
||
assert normalize_address("") == ""
|
||
|
||
|
||
def test_normalize_lowercases():
|
||
assert normalize_address("ТЕСТ") == "тест"
|
||
|
||
|
||
def test_normalize_unicode_nfc():
|
||
# Composed vs decomposed е (U+0435 vs U+0435)
|
||
s1 = normalize_address("улица")
|
||
s2 = normalize_address("улица")
|
||
assert s1 == s2
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# address_fingerprint
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_fingerprint_stable_on_case_variation():
|
||
f1 = address_fingerprint("Ленина 5", 56.8378, 60.5946)
|
||
f2 = address_fingerprint("ЛЕНИНА 5", 56.8378, 60.5946)
|
||
assert f1 == f2
|
||
|
||
|
||
def test_fingerprint_distinct_on_different_lat():
|
||
f1 = address_fingerprint("Same Address", 56.8, 60.5)
|
||
f2 = address_fingerprint("Same Address", 56.9, 60.5)
|
||
assert f1 != f2
|
||
|
||
|
||
def test_fingerprint_distinct_on_different_lon():
|
||
f1 = address_fingerprint("Same Address", 56.8, 60.5)
|
||
f2 = address_fingerprint("Same Address", 56.8, 60.6)
|
||
assert f1 != f2
|
||
|
||
|
||
def test_fingerprint_stable_within_11m(monkeypatch):
|
||
# 4 decimal places → ~11 m resolution; coords differing in 5th decimal should match
|
||
f1 = address_fingerprint("Addr", 56.83780, 60.59460)
|
||
f2 = address_fingerprint("Addr", 56.83781, 60.59461)
|
||
# 4dp rounding: 56.8378 == 56.8378, 60.5946 == 60.5946 → same fingerprint
|
||
assert f1 == f2
|
||
|
||
|
||
def test_fingerprint_handles_none_coords():
|
||
fp = address_fingerprint("Some Address", None, None)
|
||
assert len(fp) == 32 # truncated sha256 hex
|
||
|
||
|
||
def test_fingerprint_handles_none_address():
|
||
fp = address_fingerprint(None, 56.8, 60.5)
|
||
assert len(fp) == 32
|
||
|
||
|
||
def test_fingerprint_all_none():
|
||
fp = address_fingerprint(None, None, None)
|
||
assert len(fp) == 32
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# match_or_create_house — mock DB tier routing
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _make_db(rows: list[dict | None]) -> MagicMock:
|
||
"""Build a Session mock that returns rows sequentially per execute() call."""
|
||
db = MagicMock()
|
||
side_effects = []
|
||
for row in rows:
|
||
result = MagicMock()
|
||
mappings = MagicMock()
|
||
if row is None:
|
||
mappings.first.return_value = None
|
||
mappings.one.return_value = None
|
||
else:
|
||
mappings.first.return_value = row
|
||
mappings.one.return_value = row
|
||
result.mappings.return_value = mappings
|
||
side_effects.append(result)
|
||
db.execute.side_effect = side_effects
|
||
return db
|
||
|
||
|
||
def test_match_house_tier0_cadastr():
|
||
"""Tier 0: cadastral_number match → returns cadastr_exact without geo query."""
|
||
from app.services.matching.houses import match_or_create_house
|
||
|
||
db = _make_db([
|
||
{'id': 42}, # cadastral match
|
||
None, # _upsert_house_source (INSERT ... ON CONFLICT)
|
||
])
|
||
house_id, conf, method = match_or_create_house(
|
||
db, 'cian', 'ext-001',
|
||
address='ул Ленина 5', lat=56.8, lon=60.5,
|
||
cadastral_number='66:41:0000000:999',
|
||
)
|
||
assert house_id == 42
|
||
assert conf == 1.0
|
||
assert method == 'cadastr_exact'
|
||
|
||
|
||
def test_match_house_tier1_source_exact():
|
||
"""Tier 1: ext_source+ext_id already in house_sources.
|
||
|
||
No cadastral_number supplied → Tier 0 skipped entirely.
|
||
First db.execute call goes to house_sources query.
|
||
source_exact returns immediately without calling _upsert_house_source.
|
||
"""
|
||
from app.services.matching.houses import match_or_create_house
|
||
|
||
db = _make_db([
|
||
{'house_id': 7}, # house_sources hit (first and only execute call)
|
||
])
|
||
house_id, conf, method = match_or_create_house(
|
||
db, 'avito', 'ext-999',
|
||
address='пр Мира 10', lat=56.8, lon=60.5,
|
||
)
|
||
assert house_id == 7
|
||
assert method == 'source_exact'
|
||
|
||
|
||
def test_match_house_tier2_fingerprint():
|
||
"""Tier 2: fingerprint match in house_address_aliases."""
|
||
from app.services.matching.houses import match_or_create_house
|
||
|
||
db = _make_db([
|
||
None, # house_sources miss
|
||
{'house_id': 15}, # fingerprint hit
|
||
None, # _upsert_house_source
|
||
])
|
||
house_id, conf, method = match_or_create_house(
|
||
db, 'yandex', 'ext-123',
|
||
address='Тестовая 1', lat=56.84, lon=60.60,
|
||
)
|
||
assert house_id == 15
|
||
assert conf == 0.9
|
||
assert method == 'fingerprint'
|
||
|
||
|
||
def test_match_house_tier3_geo():
|
||
"""Tier 3: geo-proximity match."""
|
||
from app.services.matching.houses import match_or_create_house
|
||
|
||
db = _make_db([
|
||
None, # house_sources miss
|
||
None, # fingerprint miss
|
||
{'id': 22, 'dist': 15.5}, # geo hit
|
||
None, # _upsert_house_source
|
||
None, # _insert_alias
|
||
])
|
||
house_id, conf, method = match_or_create_house(
|
||
db, 'n1', 'ext-456',
|
||
address='Новая 3', lat=56.83, lon=60.59,
|
||
)
|
||
assert house_id == 22
|
||
assert conf == 0.7
|
||
assert method == 'geo_proximity'
|
||
|
||
|
||
def test_match_house_new():
|
||
"""All tiers miss → new house created."""
|
||
from app.services.matching.houses import match_or_create_house
|
||
|
||
db = _make_db([
|
||
None, # house_sources miss
|
||
None, # fingerprint miss
|
||
None, # geo miss
|
||
{'id': 99}, # INSERT RETURNING id
|
||
None, # _upsert_house_source
|
||
None, # _insert_alias
|
||
])
|
||
house_id, conf, method = match_or_create_house(
|
||
db, 'avito', 'new-ext',
|
||
address='Новостройка 1', lat=56.85, lon=60.61,
|
||
)
|
||
assert house_id == 99
|
||
assert conf == 1.0
|
||
assert method == 'new'
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# match_or_create_listing — mock DB tier routing
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_match_listing_tier0_cadastr():
|
||
from app.services.matching.listings import match_or_create_listing
|
||
|
||
db = _make_db([
|
||
{'id': 200}, # cadastral hit
|
||
None, # _upsert_listing_source
|
||
])
|
||
listing_id, conf, method = match_or_create_listing(
|
||
db, 'cian', 'cian-555', house_id=10,
|
||
floor=5, rooms_count=2, area_m2=48.5, price_rub=5_000_000.0,
|
||
cadastral_number='66:41:0204016:1234',
|
||
)
|
||
assert listing_id == 200
|
||
assert method == 'cadastr_exact'
|
||
|
||
|
||
def test_match_listing_tier1_source_exact():
|
||
"""Tier 1: no cadastral_number → Tier 0 skipped; first execute is listing_sources hit."""
|
||
from app.services.matching.listings import match_or_create_listing
|
||
|
||
db = _make_db([
|
||
{'listing_id': 101}, # listing_sources hit (first and only execute call)
|
||
])
|
||
listing_id, conf, method = match_or_create_listing(
|
||
db, 'avito', 'avito-888', house_id=10,
|
||
)
|
||
assert listing_id == 101
|
||
assert method == 'source_exact'
|
||
|
||
|
||
def test_match_listing_tier2_minhash():
|
||
from app.services.matching.listings import match_or_create_listing
|
||
|
||
db = _make_db([
|
||
None, # listing_sources miss
|
||
{'id': 55}, # minhash hit
|
||
None, # _upsert_listing_source
|
||
])
|
||
listing_id, conf, method = match_or_create_listing(
|
||
db, 'cian', 'cian-999', house_id=10,
|
||
description_minhash='abc123deadbeef',
|
||
)
|
||
assert listing_id == 55
|
||
assert conf == 0.85
|
||
assert method == 'minhash'
|
||
|
||
|
||
def test_match_listing_tier3_composite():
|
||
"""Tier 3: no cadastral + no minhash provided → tiers 0/2 skipped.
|
||
Execute order: listing_sources miss → composite hit → _upsert.
|
||
"""
|
||
from app.services.matching.listings import match_or_create_listing
|
||
|
||
db = _make_db([
|
||
None, # listing_sources miss (Tier 1)
|
||
{'id': 77, 'diff': 0.1}, # composite hit (Tier 3; Tier 2 skipped — no hash)
|
||
None, # _upsert_listing_source
|
||
])
|
||
listing_id, conf, method = match_or_create_listing(
|
||
db, 'cian', 'cian-777', house_id=10,
|
||
floor=3, rooms_count=1, area_m2=36.0,
|
||
)
|
||
assert listing_id == 77
|
||
assert conf == 0.75
|
||
assert method == 'composite'
|
||
|
||
|
||
def test_match_listing_new():
|
||
from app.services.matching.listings import match_or_create_listing
|
||
|
||
db = _make_db([
|
||
None, # listing_sources miss
|
||
None, # composite miss (floor/area/rooms all None → tier skipped)
|
||
])
|
||
listing_id, conf, method = match_or_create_listing(
|
||
db, 'avito', 'avito-new', house_id=10,
|
||
)
|
||
assert listing_id == 0
|
||
assert method == 'new'
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# conflict_resolution
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_field_priority_dicts_not_empty():
|
||
assert HOUSE_FIELD_PRIORITY
|
||
assert LISTING_FIELD_PRIORITY
|
||
|
||
|
||
VALID_STRING_RULES = {"union", "cross_validate", "first_non_null"}
|
||
|
||
|
||
def test_field_priority_sources_are_lists_or_known_string_rules():
|
||
"""Each priority entry is either a source list or a known string rule.
|
||
|
||
PR #471 (main) introduced 'union' / 'cross_validate' / 'first_non_null' string
|
||
rules alongside list entries. Both are valid per LISTING_FIELD_PRIORITY type hint.
|
||
"""
|
||
for col, sources in HOUSE_FIELD_PRIORITY.items():
|
||
assert isinstance(sources, list) or sources in VALID_STRING_RULES, (
|
||
f'HOUSE_FIELD_PRIORITY[{col!r}] must be list or known rule, got {sources!r}'
|
||
)
|
||
for col, sources in LISTING_FIELD_PRIORITY.items():
|
||
assert isinstance(sources, list) or sources in VALID_STRING_RULES, (
|
||
f'LISTING_FIELD_PRIORITY[{col!r}] must be list or known rule, got {sources!r}'
|
||
)
|
||
|
||
|
||
def test_update_canonical_fields_is_callable():
|
||
"""Stage 8 v1 — legacy stub (no-op pass). Full arbitration deferred to Stage 8.x.
|
||
|
||
PR #471 (main) replaced NotImplementedError stub with silent pass to allow
|
||
backward-compat imports. Test updated accordingly.
|
||
"""
|
||
db = MagicMock()
|
||
# Should not raise — no-op stub returns None
|
||
result = update_canonical_fields(db, listing_id=1, ext_source='cian', lot_data=object())
|
||
assert result is None
|