bot-backend
d22e0e00ae
All checks were successful
Deploy Trade-In / changes (push) Successful in 10s
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / build-browser (push) Has been skipped
Deploy Trade-In / test (push) Successful in 45s
Deploy Trade-In / build-backend (push) Successful in 58s
Deploy Trade-In / deploy (push) Successful in 47s
274 lines
13 KiB
Python
274 lines
13 KiB
Python
"""#1186 — query-side guard: novostroyki строки не попадают в comp-пул вторички.
|
||
|
||
Тесты статические (SQL-string inspection) + поведенческие (_fetch_analogs mock).
|
||
|
||
Каноничный предикат: (listing_segment IS NULL OR listing_segment = 'vtorichka')
|
||
NULL = legacy вторичка до миграции 011 — отбрасывать нельзя.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import inspect
|
||
import os
|
||
import re
|
||
from typing import Any
|
||
from unittest.mock import MagicMock
|
||
|
||
# Settings требует DATABASE_URL при инициализации (fail-fast, C-3).
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
|
||
|
||
from app.services import estimator as est_mod
|
||
from app.tasks import asking_to_sold_ratio as ratio_mod
|
||
|
||
# ── Canonical guard predicate ─────────────────────────────────────────────────
|
||
|
||
_GUARD = "(listing_segment IS NULL OR listing_segment = 'vtorichka')"
|
||
_GUARD_RE = re.compile(
|
||
r"\((?:\w+\.)?listing_segment\s+IS\s+NULL\s+OR\s+(?:\w+\.)?listing_segment\s*=\s*'vtorichka'\)",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
|
||
def _norm(s: str) -> str:
|
||
"""Drop -- comments и склей пробелы для устойчивого поиска."""
|
||
no_comments = re.sub(r"--[^\n]*", "", s)
|
||
return re.sub(r"\s+", " ", no_comments).strip()
|
||
|
||
|
||
# ── estimator.py static checks ────────────────────────────────────────────────
|
||
|
||
|
||
def test_common_where_has_canonical_guard() -> None:
|
||
"""_COMMON_WHERE используется Tier S и Tier H — должен содержать канон-предикат."""
|
||
assert _GUARD_RE.search(
|
||
est_mod._COMMON_WHERE
|
||
), "_COMMON_WHERE lacks novostroyki guard — Tier S/H comp set contaminated"
|
||
|
||
|
||
def test_common_where_no_old_neq_form() -> None:
|
||
"""Старый вариант listing_segment <> 'novostroyki' запрещён — не охватывает иные сегменты."""
|
||
assert "listing_segment <> 'novostroyki'" not in est_mod._COMMON_WHERE
|
||
assert "listing_segment != 'novostroyki'" not in est_mod._COMMON_WHERE
|
||
|
||
|
||
def test_tier_w_inline_has_canonical_guard() -> None:
|
||
"""Tier W использует inline SQL, а не _COMMON_WHERE — нужен собственный guard."""
|
||
# Получаем исходник функции _fetch_analogs.
|
||
src = inspect.getsource(est_mod._fetch_analogs)
|
||
# Tier W — inline блок (не через _COMMON_WHERE).
|
||
assert _GUARD_RE.search(src), "_fetch_analogs Tier W inline SQL lacks novostroyki guard"
|
||
# Старой формы быть не должно.
|
||
assert "listing_segment <> 'novostroyki'" not in src
|
||
assert "listing_segment != 'novostroyki'" not in src
|
||
|
||
|
||
def test_tier_a_anchor_gated_python_side_not_sql_guard() -> None:
|
||
"""#1774: Tier A больше НЕ хардкодит SQL-guard — он делает gated Python-side
|
||
relaxation (впускает novostroyki только если в доме есть вторичка/NULL).
|
||
|
||
Assert: (a) Tier C всё ещё несёт канон-guard в SQL, (b) Tier A делает
|
||
Python-side gating — источник содержит hardened secondary_count/primary_count
|
||
gate (#1774, после code-review).
|
||
"""
|
||
src = inspect.getsource(est_mod._fetch_anchor_comps)
|
||
# (a) Tier C SQL guard сохранён (поиск находит единственное вхождение — Tier C).
|
||
assert _GUARD_RE.search(src), "Tier C anchor SQL lost the canonical novostroyki guard"
|
||
# (b) Tier A теперь gated Python-side (#1774).
|
||
# Hardened gate: secondary_count ≥ primary_count (заменил has_secondary).
|
||
assert "secondary_count" in src, "Tier A lacks secondary_count Python-side gate (#1774)"
|
||
assert "primary_count" in src, "Tier A lacks primary_count comparison in hardened gate (#1774)"
|
||
|
||
|
||
def test_tier_c_anchor_has_canonical_guard_not_segment_param() -> None:
|
||
"""Tier C anchor — канон-предикат, не CAST(:segment AS text) (тот вариант отбрасывал NULL)."""
|
||
src = inspect.getsource(est_mod._fetch_anchor_comps)
|
||
# Canonical guard present (Tier C; Tier A guard снят в #1774 → gated Python-side).
|
||
assert _GUARD_RE.search(src)
|
||
# Старый параметрический вариант убран.
|
||
assert "listing_segment = CAST(:segment AS text)" not in src
|
||
|
||
|
||
# ── asking_to_sold_ratio.py static checks ─────────────────────────────────────
|
||
|
||
_REDERIVE_SQL_TEXT = str(ratio_mod._REDERIVE_SQL.text)
|
||
|
||
|
||
def test_ask_side_cte_has_guard() -> None:
|
||
"""ask_side CTE в _REDERIVE_SQL (per-rooms asking медиана) — guard обязателен."""
|
||
assert _GUARD_RE.search(
|
||
_REDERIVE_SQL_TEXT
|
||
), "ask_side CTE in _REDERIVE_SQL lacks novostroyki guard"
|
||
|
||
|
||
def test_ask_global_cte_has_guard() -> None:
|
||
"""ask_global CTE в _REDERIVE_SQL (global fallback asking медиана) — guard обязателен."""
|
||
# _REDERIVE_SQL содержит два `ask_global`-блока; ищем оба через count.
|
||
matches = len(_GUARD_RE.findall(_REDERIVE_SQL_TEXT))
|
||
assert (
|
||
matches >= 2
|
||
), f"Expected ≥2 guard occurrences in _REDERIVE_SQL (ask_side + ask_global), got {matches}"
|
||
|
||
|
||
# ── Поведенческие тесты: _fetch_analogs (mock DB) ────────────────────────────
|
||
|
||
|
||
def _make_listing(
|
||
*,
|
||
source: str = "avito",
|
||
address: str = "ул. Ленина, 1",
|
||
distance_m: float = 100.0,
|
||
relevance_score: float = 0.1,
|
||
price_rub: float = 5_000_000.0,
|
||
area_m2: float = 40.0,
|
||
rooms: int = 2,
|
||
listing_segment: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Минимальный listing-dict, имитирующий вывод DB mapping."""
|
||
return {
|
||
"source": source,
|
||
"source_url": f"https://{source}.ru/offer/1",
|
||
"address": address,
|
||
"lat": 56.838,
|
||
"lon": 60.595,
|
||
"rooms": rooms,
|
||
"area_m2": area_m2,
|
||
"floor": 3,
|
||
"total_floors": 9,
|
||
"price_rub": price_rub,
|
||
"price_per_m2": price_rub / area_m2,
|
||
"listing_date": None,
|
||
"days_on_market": 15,
|
||
"photo_urls": [],
|
||
"scraped_at": None,
|
||
"distance_m": distance_m,
|
||
"relevance_score": relevance_score,
|
||
"building_cadastral_number": None,
|
||
"listing_segment": listing_segment,
|
||
}
|
||
|
||
|
||
def _make_db_mock(rows: list[dict[str, Any]]) -> MagicMock:
|
||
"""Session mock: db.execute().mappings().all() возвращает rows."""
|
||
db = MagicMock()
|
||
db.execute.return_value.mappings.return_value.all.return_value = rows
|
||
return db
|
||
|
||
|
||
def test_fetch_analogs_sql_guard_present_novostroyki_excluded() -> None:
|
||
"""SQL guard корректно записан: listing_segment='novostroyki' не должен присутствовать
|
||
в WHERE-предикате (guard зарыт в SQL, а не в Python post-filter).
|
||
|
||
Тест проверяет СТАТИКУ SQL внутри _fetch_analogs (Tier W path), а не runtime фильтрацию —
|
||
runtime-фильтрацию сделала бы только интеграционная БД. Убеждаемся, что кто-то в будущем
|
||
не переписал SQL, убрав guard.
|
||
"""
|
||
src = inspect.getsource(est_mod._fetch_analogs)
|
||
# Guard должен присутствовать хотя бы один раз в теле функции.
|
||
assert _GUARD_RE.search(
|
||
src
|
||
), "_fetch_analogs SQL no longer contains novostroyki guard — guard was removed!"
|
||
|
||
|
||
def test_null_segment_listing_not_excluded_by_guard() -> None:
|
||
"""listing_segment=NULL (legacy вторичка до м.011) должны ПОПАСТЬ в comps.
|
||
|
||
Поведенческий тест: mock возвращает строку с segment=None, _fetch_analogs
|
||
должен включить её в результат (_stratify_candidates не отфильтровывает NULL).
|
||
"""
|
||
from app.services.estimator import _fetch_analogs
|
||
|
||
null_seg_listing = _make_listing(listing_segment=None, distance_m=50.0, relevance_score=0.05)
|
||
db = _make_db_mock([null_seg_listing])
|
||
|
||
result, _fallback, _tier = _fetch_analogs(
|
||
db, lat=56.838, lon=60.595, rooms=2, area=40.0, radius_m=1000
|
||
)
|
||
|
||
# NULL-segment listing должен присутствовать в результате (не отброшен Python-стороной).
|
||
assert any(
|
||
r.get("source") == "avito" for r in result
|
||
), "NULL-segment listing was unexpectedly excluded from comp set"
|
||
|
||
|
||
def test_vtorichka_segment_listing_included() -> None:
|
||
"""listing_segment='vtorichka' — явный вторичный сегмент — должен попасть в comps."""
|
||
from app.services.estimator import _fetch_analogs
|
||
|
||
vtor_listing = _make_listing(listing_segment="vtorichka", distance_m=80.0)
|
||
db = _make_db_mock([vtor_listing])
|
||
|
||
result, _, _ = _fetch_analogs(db, lat=56.838, lon=60.595, rooms=2, area=40.0, radius_m=1000)
|
||
assert len(result) >= 1, "vtorichka-segment listing was unexpectedly excluded"
|
||
|
||
|
||
# ── Негативные assert'ы: запрещённые формы guard'а во всём estimator'е ────────
|
||
|
||
|
||
def _estimator_full_src() -> str:
|
||
"""Полный исходник модуля estimator (для статической проверки всех SQL-блоков)."""
|
||
return inspect.getsource(est_mod)
|
||
|
||
|
||
def test_no_bare_vtorichka_eq_without_null_guard() -> None:
|
||
"""Запрет голого `listing_segment = 'vtorichka'` без IS NULL OR части.
|
||
|
||
Такой вариант отбросит NULL (legacy вторичку до миграции 011) — баг охвата.
|
||
"""
|
||
src = _estimator_full_src()
|
||
# Ищем вхождения = 'vtorichka' которым НЕ предшествует IS NULL OR на той же строке.
|
||
bad_matches = [
|
||
line
|
||
for line in src.splitlines()
|
||
if "= 'vtorichka'" in line and "IS NULL OR" not in line and "--" not in line.lstrip()[:3]
|
||
]
|
||
assert not bad_matches, (
|
||
"Found bare `= 'vtorichka'` without IS NULL OR guard in estimator.py:\n"
|
||
+ "\n".join(bad_matches)
|
||
)
|
||
|
||
|
||
def test_no_neq_novostroyki_form_in_estimator() -> None:
|
||
"""Запрет формы `listing_segment <> 'novostroyki'` / `!= 'novostroyki'` в estimator.
|
||
|
||
Эта форма пропускает прочие сегменты (коммерция и т.п.) и не охватывает NULL.
|
||
Канон: IS NULL OR = 'vtorichka'.
|
||
"""
|
||
src = _estimator_full_src()
|
||
assert (
|
||
"<> 'novostroyki'" not in src
|
||
), "estimator.py contains deprecated `<> 'novostroyki'` form — use canonical guard"
|
||
assert (
|
||
"!= 'novostroyki'" not in src
|
||
), "estimator.py contains deprecated `!= 'novostroyki'` form — use canonical guard"
|
||
|
||
|
||
def test_anchor_comps_no_dead_listing_segment_param() -> None:
|
||
"""_fetch_anchor_comps не должна принимать listing_segment как параметр.
|
||
|
||
После #1186 guard хардкожен в SQL — параметр убран, чтобы caller не мог
|
||
случайно передать 'novostroyki' или другой сегмент и обойти guard.
|
||
"""
|
||
import inspect as _inspect
|
||
|
||
sig = _inspect.signature(est_mod._fetch_anchor_comps)
|
||
assert "listing_segment" not in sig.parameters, (
|
||
"_fetch_anchor_comps still has `listing_segment` parameter — "
|
||
"it was removed in #1186 review fixup; guard is hardcoded in SQL"
|
||
)
|
||
|
||
|
||
def test_fetch_anchor_comps_tier_c_canonical_guard_not_parametric() -> None:
|
||
"""Tier C в _fetch_anchor_comps использует канон-guard, а не параметрическое сравнение.
|
||
|
||
Параметрическая форма `listing_segment = CAST(:segment AS text)` исключала NULL
|
||
и давала caller'у возможность выбрать произвольный сегмент (включая novostroyki).
|
||
"""
|
||
src = inspect.getsource(est_mod._fetch_anchor_comps)
|
||
assert "CAST(:segment AS text)" not in src, (
|
||
"_fetch_anchor_comps Tier C still uses parametric segment filter — "
|
||
"must be hardcoded canonical guard"
|
||
)
|
||
# И канон-guard на месте (Tier C-блок).
|
||
assert _GUARD_RE.search(
|
||
src
|
||
), "_fetch_anchor_comps lacks canonical guard after removing parametric form"
|