"""Tests for GET /estimate/{id} revival of "dead" (median_price<=0/NULL) rows. Incident 2026-08-10: a customer opened a saved estimate link and saw «НЕДОСТАТОЧНО ДАННЫХ» — the row was created BEFORE the estimator fix (#oblast-E/#oblast-F, PR #2823/#2825) and is permanently stuck at median_price=0, even though the same address/params now compute a real price. app.api.v1.trade_in::_try_revive_dead_estimate recomputes such a row in place (same id/link) via the same estimate_quality() path as POST /estimate. Also covers migration 255 (relaxations/reliability persistence). Реальная БД не нужна: DB + get_role + estimate_quality мокируются, mirroring test_estimate_idor.py's approach (self-contained, no cross-file fixture imports — this repo has no precedent for importing fixtures across tests/test_*.py modules, only from tests/support/). DB mock dispatches by SQL substring (not call-position): get_estimate's existing rehydrate path calls the real (non-stubbed) _resolve_target_house_id helper, which itself fires 1-2 incidental `SELECT id FROM houses` queries whenever the revival attempt does NOT short-circuit with an early return — hand-counting positional side_effect entries around that would be brittle. """ from __future__ import annotations import inspect import os import sys from datetime import UTC, datetime, timedelta from types import SimpleNamespace from unittest.mock import MagicMock # psycopg v3 driver required; stub DATABASE_URL before any app import os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") # WeasyPrint requires GTK — not present in CI/Windows. Stub before any app import. _wp_mock = MagicMock() sys.modules.setdefault("weasyprint", _wp_mock) sys.modules.setdefault("weasyprint.CSS", _wp_mock) sys.modules.setdefault("weasyprint.HTML", _wp_mock) import pytest # noqa: E402 from fastapi import FastAPI # noqa: E402 from fastapi.testclient import TestClient # noqa: E402 _ESTIMATE_ID = "22222222-2222-2222-2222-222222222222" _TEMP_ID = "33333333-3333-3333-3333-333333333333" @pytest.fixture(autouse=True) def _restore_get_role(): """Restore app.core.auth.get_role after each test (mirror test_estimate_idor).""" from app.core import auth as auth_mod original = auth_mod.get_role yield auth_mod.get_role = original @pytest.fixture() def trade_in_app() -> FastAPI: """Minimal FastAPI app mounting only the trade-in router.""" from app.api.v1 import trade_in as trade_in_module application = FastAPI() application.include_router(trade_in_module.router, prefix="/api/v1/trade-in") return application def _make_dead_row( *, median_price: int | None = 0, house_type: str | None = None, repair_state: str | None = None, relaxations: list[str] | None = None, reliability: str = "ok", ) -> SimpleNamespace: """A trade_in_estimates row in the "dead" state (incident 2026-08-10). house_type/repair_state default to None (valid TradeInEstimateInput input — most revival tests need a row that round-trips through pydantic cleanly); a dedicated test passes legacy Russian literal values to exercise the "invalid persisted value" graceful-degrade path. """ return SimpleNamespace( id=_ESTIMATE_ID, median_price=median_price, range_low=0, range_high=0, median_price_per_m2=0, confidence="low", confidence_explanation="Рядом найдено недостаточно объявлений (4 шт.)", n_analogs=0, analogs=[], actual_deals=[], sources_used=[], data_freshness_minutes=None, expires_at=datetime.now(tz=UTC) + timedelta(hours=12), retain_until=None, address="г Екатеринбург, ул Академика Парина, д 46, к 5", lat=56.8519, lon=60.6122, area_m2=23.1, rooms=1, floor=5, total_floors=16, year_built=2018, house_type=house_type, repair_state=repair_state, has_balcony=None, canonical_address=None, house_cadnum=None, house_fias_id=None, dadata_qc_geo=None, dadata_metro=[], expected_sold_price=None, expected_sold_range_low=None, expected_sold_range_high=None, expected_sold_per_m2=None, asking_to_sold_ratio=None, ratio_basis=None, created_by="kopylov", created_at=datetime(2026, 5, 29, tzinfo=UTC), relaxations=relaxations or [], reliability=reliability, ) def _make_live_row( *, relaxations: list[str] | None = None, reliability: str = "ok" ) -> SimpleNamespace: """A "live" row (median_price>0) — revival must never touch it.""" return SimpleNamespace( id=_ESTIMATE_ID, median_price=4_031_157, range_low=3_700_000, range_high=4_300_000, median_price_per_m2=174_000, confidence="medium", confidence_explanation="Найдено 39 аналогов", n_analogs=39, analogs=[], actual_deals=[], sources_used=["avito", "rosreestr"], data_freshness_minutes=15, expires_at=datetime.now(tz=UTC) + timedelta(hours=12), retain_until=None, address="г Екатеринбург, ул Академика Парина, д 46, к 5", lat=56.8519, lon=60.6122, area_m2=23.1, rooms=1, floor=5, total_floors=16, year_built=2018, house_type=None, repair_state=None, has_balcony=None, canonical_address="г Екатеринбург, ул Академика Парина, д 46, к 5", house_cadnum=None, house_fias_id=None, dadata_qc_geo=0, dadata_metro=[], expected_sold_price=None, expected_sold_range_low=None, expected_sold_range_high=None, expected_sold_per_m2=None, asking_to_sold_ratio=None, ratio_basis=None, created_by="kopylov", created_at=datetime.now(tz=UTC), relaxations=relaxations or [], reliability=reliability, ) def _fake_revived_result(**overrides): """A canned AggregatedEstimate mimicking a successful estimate_quality() call.""" from app.schemas.trade_in import AggregatedEstimate defaults = dict( estimate_id=_TEMP_ID, median_price_rub=4_031_157, range_low_rub=3_700_000, range_high_rub=4_300_000, median_price_per_m2=174_000, confidence="medium", confidence_explanation="Найдено 39 аналогов", n_analogs=39, period_months=12, analogs=[], actual_deals=[], expires_at=datetime.now(tz=UTC) + timedelta(hours=24), target_address="г Екатеринбург, ул Академика Парина, д 46, к 5", target_lat=56.8519, target_lon=60.6122, sources_used=["avito", "rosreestr"], data_freshness_minutes=15, canonical_address="г Екатеринбург, ул Академика Парина, д 46, к 5", relaxations=["снят фильтр по году постройки", "учтены студии"], reliability="low", created_at=datetime.now(tz=UTC), ) defaults.update(overrides) return AggregatedEstimate(**defaults) def _dispatch_db(row: object, claim_result: object = None) -> MagicMock: """DB session mock dispatching fetchone() results by SQL substring. - initial GET SELECT ("SELECT id, median_price ...") -> row - revival claim UPDATE ("SET revival_attempted_at") -> claim_result - everything else (houses lookup, persist UPDATE, DELETE, avito_imv UPDATE — none of which .fetchone() in real code except the two above, but MagicMock tolerates the unused call either way) -> None call_args_list still records every call in order regardless of dispatch, so tests can assert on it directly (grep by substring) without needing to hand-count incidental queries fired by _resolve_target_house_id. """ db = MagicMock() def _execute(clause, params=None, *_a, **_k): sql = getattr(clause, "text", str(clause)) result = MagicMock() if "SET revival_attempted_at" in sql: result.fetchone.return_value = claim_result elif "SELECT id, median_price" in sql: result.fetchone.return_value = row else: result.fetchone.return_value = None return result db.execute.side_effect = _execute return db def _calls_containing(db: MagicMock, needle: str) -> list: return [c for c in db.execute.call_args_list if needle in getattr(c.args[0], "text", "")] def _client_with(app: FastAPI, db_mock: MagicMock, role: str = "pilot") -> TestClient: from app.core.db import get_db def _override_db(): yield db_mock app.dependency_overrides[get_db] = _override_db auth_mod = sys.modules["app.core.auth"] auth_mod.get_role = lambda _u: role # type: ignore[assignment] return TestClient(app) @pytest.fixture() def _estimator_stub(): """Replaces app.services.estimator with a SimpleNamespace stub. Mirrors test_estimate_idor.py::_stub_precision_and_pdf, plus an `estimate_quality` async attribute (revival's own lazy import target). Individual tests overwrite `estimate_quality` per-scenario. """ real_estimator = sys.modules.get("app.services.estimator") async def _default_estimate_quality(*_a, **_k): # pragma: no cover — overridden per test raise AssertionError("estimate_quality stub not configured for this test") stub = SimpleNamespace( _qc_geo_to_precision=lambda _qc: None, _fetch_price_trend=lambda *a, **k: None, _fetch_dkp_corridor=lambda *a, **k: None, _fetch_house_imv_anchor=lambda *a, **k: None, _resolve_target_city=lambda *a, **k: None, _cv_from_ppm2=lambda *a, **k: None, _source_counts=lambda *a, **k: {}, _canonical_sources=lambda *a, **k: [], # #2632: GET-rehydrate реконструирует фактический радиус подбора. У строк # этой фикстуры analogs пусты → None и есть настоящее поведение # (см. tests/test_estimator_search_radius_2632.py). rehydrate_search_radius_m=lambda *a, **k: None, estimate_quality=_default_estimate_quality, ) sys.modules["app.services.estimator"] = stub # type: ignore[assignment] yield stub if real_estimator is not None: sys.modules["app.services.estimator"] = real_estimator else: sys.modules.pop("app.services.estimator", None) # ── Revival success ─────────────────────────────────────────────────────── def test_dead_row_revives_and_updates_db( trade_in_app: FastAPI, _estimator_stub: SimpleNamespace ) -> None: """Dead row (median_price=0) → recomputed, written to the SAME id, returned fresh. Success path returns early (before the pre-existing rehydrate block), so the call count is exactly the 5 documented in _try_revive_dead_estimate's docstring: claim, avito_imv relink, persist UPDATE, DELETE temp — plus the initial SELECT. """ async def _fake_estimate_quality(payload, db, **kwargs): assert payload.address.startswith("г Екатеринбург") assert payload.rooms == 1 assert payload.radius_m is None # never persisted — default cascade return _fake_revived_result() _estimator_stub.estimate_quality = _fake_estimate_quality row = _make_dead_row() db = _dispatch_db(row, claim_result=SimpleNamespace(id=_ESTIMATE_ID)) client = _client_with(trade_in_app, db) resp = client.get( f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}", headers={"X-Authenticated-User": "kopylov"}, ) assert resp.status_code == 200 body = resp.json() # id/link contract: response carries the ORIGINAL id, not the temp uuid # estimate_quality() minted internally. assert body["estimate_id"] == _ESTIMATE_ID assert body["median_price_rub"] == 4_031_157 assert body["n_analogs"] == 39 assert body["insufficient_data"] is False assert body["reliability"] == "low" assert "снят фильтр по году постройки" in body["relaxations"] assert len(db.execute.call_args_list) == 5 persist_calls = _calls_containing(db, "UPDATE trade_in_estimates SET") assert len(persist_calls) == 1 persist_params = persist_calls[0].args[1] assert persist_params["id"] == _ESTIMATE_ID assert persist_params["median_price"] == 4_031_157 assert persist_params["reliability"] == "low" delete_calls = _calls_containing(db, "DELETE FROM trade_in_estimates") assert len(delete_calls) == 1 assert delete_calls[0].args[1]["id"] == _TEMP_ID def test_dead_row_revival_preserves_created_at( trade_in_app: FastAPI, _estimator_stub: SimpleNamespace ) -> None: """#incident-2026-08-11: created_at is the client's ORIGINAL request date (printed on /history, see app/schemas/trade_in.py:317-318) — revival must not clobber it with the throwaway temp row's NOW(). Regression for the prod incident (estimate ff421062-...: created_at jumped from the original 2026-08-10 12:54:47 to the revival moment 2026-08-11 04:30:03). Also asserts: (a) the persist UPDATE never sets created_at at all — the fix removes the column from SET, it doesn't just overwrite it with the right value; (b) migration 256's revival_completed_at IS stamped, as the separate "when did revival last succeed" audit trail; (c) the JSON response mirrors the original created_at, not the temp result's. """ _original_created_at = datetime(2026, 5, 29, tzinfo=UTC) async def _fake_estimate_quality(payload, db, **kwargs): # The temp row estimate_quality() mints internally always carries # NOW() as its created_at — deliberately far from the original, so a # regression (copying result.created_at through) is unmissable. return _fake_revived_result(created_at=datetime(2026, 8, 11, 4, 30, 3, tzinfo=UTC)) _estimator_stub.estimate_quality = _fake_estimate_quality row = _make_dead_row() row.created_at = _original_created_at db = _dispatch_db(row, claim_result=SimpleNamespace(id=_ESTIMATE_ID)) client = _client_with(trade_in_app, db) resp = client.get( f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}", headers={"X-Authenticated-User": "kopylov"}, ) assert resp.status_code == 200 body = resp.json() assert body["created_at"] == "2026-05-29T00:00:00Z" persist_calls = _calls_containing(db, "UPDATE trade_in_estimates SET") assert len(persist_calls) == 1 persist_sql = persist_calls[0].args[0].text persist_params = persist_calls[0].args[1] assert "created_at" not in persist_sql assert "created_at" not in persist_params assert "revival_completed_at = NOW()" in persist_sql # Input-snapshot / TTL columns (what the client originally asked for and # for how long the row is retained) are likewise not recompute outputs — # untouched by the revival persist UPDATE. # NB: "address" is checked via persist_params only (not persist_sql) — # canonical_address (a legitimate recompute output) ends in "address =", # which would false-positive a substring check against the raw SQL text. for protected in ("expires_at", "retain_until", "created_by"): assert f"{protected} =" not in persist_sql assert protected not in persist_params assert "address" not in persist_params # ── Live row is never touched ──────────────────────────────────────────── def test_live_row_never_triggers_revival( trade_in_app: FastAPI, _estimator_stub: SimpleNamespace ) -> None: """median_price>0 → revival branch skipped entirely: estimate_quality is never called and no claim UPDATE fires — the saved price is untouched.""" async def _must_not_be_called(*_a, **_k): raise AssertionError("estimate_quality must not be called for a live row") _estimator_stub.estimate_quality = _must_not_be_called row = _make_live_row() db = _dispatch_db(row) client = _client_with(trade_in_app, db) resp = client.get( f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}", headers={"X-Authenticated-User": "kopylov"}, ) assert resp.status_code == 200 body = resp.json() assert body["median_price_rub"] == 4_031_157 assert _calls_containing(db, "SET revival_attempted_at") == [] # ── Throttle / anti-storm ──────────────────────────────────────────────── def test_dead_row_throttled_does_not_recompute( trade_in_app: FastAPI, _estimator_stub: SimpleNamespace ) -> None: """Claim UPDATE returns no row (recent attempt / lost race) → no recompute, honest still-dead response, no 500, exactly one claim attempt (no retry loop within the same request).""" async def _must_not_be_called(*_a, **_k): raise AssertionError("estimate_quality must not be called when throttled") _estimator_stub.estimate_quality = _must_not_be_called row = _make_dead_row() db = _dispatch_db(row, claim_result=None) # throttled: WHERE matched nothing client = _client_with(trade_in_app, db) resp = client.get( f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}", headers={"X-Authenticated-User": "kopylov"}, ) assert resp.status_code == 200 body = resp.json() assert body["insufficient_data"] is True assert body["median_price_rub"] == 0 assert len(_calls_containing(db, "SET revival_attempted_at")) == 1 # ── Recompute error degrades gracefully (no 500) ───────────────────────── def test_revival_recompute_exception_falls_back_without_500( trade_in_app: FastAPI, _estimator_stub: SimpleNamespace ) -> None: async def _raises(*_a, **_k): raise RuntimeError("geocode timeout") _estimator_stub.estimate_quality = _raises row = _make_dead_row() db = _dispatch_db(row, claim_result=SimpleNamespace(id=_ESTIMATE_ID)) client = _client_with(trade_in_app, db) resp = client.get( f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}", headers={"X-Authenticated-User": "kopylov"}, ) assert resp.status_code == 200 assert resp.json()["insufficient_data"] is True # exception happened AFTER the claim — throttle was still recorded once. assert len(_calls_containing(db, "SET revival_attempted_at")) == 1 # ...but nothing was written back to the row (no persist UPDATE fired). assert _calls_containing(db, "UPDATE trade_in_estimates SET") == [] def test_revival_still_zero_falls_back_without_500( trade_in_app: FastAPI, _estimator_stub: SimpleNamespace ) -> None: """Recompute runs but still finds nothing (median_price_rub=0) — honest insufficient_data, temp throwaway row cleaned up, no crash.""" async def _still_empty(*_a, **_k): return _fake_revived_result( median_price_rub=0, range_low_rub=0, range_high_rub=0, median_price_per_m2=0, n_analogs=0, confidence="low", ) _estimator_stub.estimate_quality = _still_empty row = _make_dead_row() db = _dispatch_db(row, claim_result=SimpleNamespace(id=_ESTIMATE_ID)) client = _client_with(trade_in_app, db) resp = client.get( f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}", headers={"X-Authenticated-User": "kopylov"}, ) assert resp.status_code == 200 assert resp.json()["insufficient_data"] is True delete_calls = _calls_containing(db, "DELETE FROM trade_in_estimates") assert len(delete_calls) == 1 assert delete_calls[0].args[1]["id"] == _TEMP_ID # the still-dead ORIGINAL row was never overwritten with (fresh) zeros. assert _calls_containing(db, "UPDATE trade_in_estimates SET") == [] def test_revival_invalid_persisted_house_type_falls_back_gracefully( trade_in_app: FastAPI, _estimator_stub: SimpleNamespace ) -> None: """Legacy row with a house_type outside the current Literal set — payload reconstruction itself raises (pydantic ValidationError), caught the same way as any other recompute failure. No 500.""" async def _must_not_be_called(*_a, **_k): raise AssertionError("estimate_quality must not be reached — payload build fails first") _estimator_stub.estimate_quality = _must_not_be_called row = _make_dead_row(house_type="монолит", repair_state="хороший") db = _dispatch_db(row, claim_result=SimpleNamespace(id=_ESTIMATE_ID)) client = _client_with(trade_in_app, db) resp = client.get( f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}", headers={"X-Authenticated-User": "kopylov"}, ) assert resp.status_code == 200 assert resp.json()["insufficient_data"] is True # ── relaxations/reliability round-trip (migration 255) ─────────────────── def test_relaxations_reliability_roundtrip_on_get( trade_in_app: FastAPI, _estimator_stub: SimpleNamespace ) -> None: """A live row with persisted relaxations/reliability surfaces them byte-for-byte on GET — the red "точность снижена" banner survives reopening a saved link (previously always reset to ok/[]).""" row = _make_live_row( relaxations=["радиус расширен до 2000 м", "площадь ±25%"], reliability="low" ) db = _dispatch_db(row) client = _client_with(trade_in_app, db) resp = client.get( f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}", headers={"X-Authenticated-User": "kopylov"}, ) assert resp.status_code == 200 body = resp.json() assert body["reliability"] == "low" assert body["relaxations"] == ["радиус расширен до 2000 м", "площадь ±25%"] def test_get_estimate_defaults_relaxations_reliability_when_row_lacks_columns( trade_in_app: FastAPI, _estimator_stub: SimpleNamespace ) -> None: """Defensive getattr fallback: a row/mock without relaxations/reliability attrs (e.g. a stale test double) degrades to schema defaults, not a crash.""" row = _make_live_row() del row.relaxations del row.reliability db = _dispatch_db(row) client = _client_with(trade_in_app, db) resp = client.get( f"/api/v1/trade-in/estimate/{_ESTIMATE_ID}", headers={"X-Authenticated-User": "kopylov"}, ) assert resp.status_code == 200 body = resp.json() assert body["reliability"] == "ok" assert body["relaxations"] == [] # ── estimator.py persists relaxations/reliability on (re)compute ───────── def test_estimate_quality_insert_persists_relaxations_reliability() -> None: """Source guard: the main POST-path INSERT must write relaxations/ reliability, not just return them in the response — regression guard against the exact gap this migration closes (PR #2823 open follow-up).""" from app.services import estimator src = inspect.getsource(estimator.estimate_quality) assert "relaxations_json" in src assert '"reliability": reliability' in src def test_empty_estimate_persists_relaxations_reliability() -> None: """_empty_estimate's INSERT must also set reliability='very_low' (mirrors the Python object it returns) rather than silently defaulting to 'ok'.""" from app.services import estimator src = inspect.getsource(estimator._empty_estimate) assert "relaxations, reliability" in src assert "'very_low'" in src