gendesign/tradein-mvp/backend/tests/test_estimate_consent_gate.py

"""ЭТАП 4 B2C launch — consent-before-save gate on trade_in_estimates (part A).

Covers:
  - require_consent=True + payload.consent missing/False -> HTTPException(422),
    raised BEFORE geocode() is ever touched and BEFORE any db.execute call
    (gate is literally the first statement in estimate_quality()).
  - require_consent=True + payload.consent=True -> proceeds, and the eventual
    INSERT carries consent=True / client_ip / policy version / text snapshot.
  - require_consent=True + consent=True but geocode fails -> the
    _empty_estimate() fallback INSERT *also* carries the same consent proof
    (both places an address can reach trade_in_estimates are covered).
  - require_consent defaults to False (regression guard): a bare
    estimate_quality(payload, db) call -- exactly the shape used by ~90 other
    estimator tests that don't care about auth/consent at all -- is completely
    unaffected. This is the whole reason the gate keys off an explicit
    keyword-only flag instead of `created_by is None`: the first version of
    this gate DID key off created_by and broke 92 unrelated tests across the
    estimator test suite (every one of them calls estimate_quality(payload, db)
    with created_by defaulting to None, which used to mean nothing).
  - the real (only) production caller, app/api/v1/trade_in.py::estimate(),
    passes require_consent=(x_authenticated_user is None) -- verified via source
    inspection, since exercising it live would require a full FastAPI app
    fixture (rbac_guard is DB/env-heavy and out of scope for this offline test).

Style mirrors tests/test_estimator_client_coords.py + test_estimator_event_loop_2207.py
(offline, db=MagicMock(), downstream helpers patched).
"""

from __future__ import annotations

import contextlib
import inspect
import os
import re
from pathlib import Path
from typing import Any

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

from unittest.mock import AsyncMock, MagicMock, patch

import anyio
import pytest
from fastapi import HTTPException

_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
_MIGRATION_192 = _SQL_DIR / "192_trade_in_estimates_consent_proof.sql"


def _make_payload(**overrides: Any) -> Any:
    from app.schemas.trade_in import TradeInEstimateInput

    base: dict[str, Any] = dict(address="ЕКБ, ул. Тестовая, 1", area_m2=40.0, rooms=1)
    base.update(overrides)
    return TradeInEstimateInput(**base)


def _make_fake_geo() -> Any:
    from app.services.geocoder import GeocodeResult

    return GeocodeResult(
        lat=56.838,
        lon=60.595,
        full_address="Свердловская обл., Екатеринбург, ул. Тестовая, 1",
        provider="nominatim",
    )


def _downstream_patches(geocode_mock: Any) -> tuple[Any, ...]:
    """Offline mocks so estimate_quality runs to completion (mirrors 2207 test's set)."""
    return (
        patch("app.services.estimator.geocode", new=geocode_mock),
        patch("app.services.estimator.dadata_clean_address", new=AsyncMock(return_value=None)),
        patch("app.services.estimator.match_house_readonly", return_value=None),
        patch("app.services.estimator.get_house_metadata", new=AsyncMock(return_value=None)),
        patch("app.services.estimator._fetch_analogs", return_value=([], False, "W")),
        patch("app.services.estimator._fetch_deals", return_value=[]),
        patch("app.services.estimator._get_or_fetch_imv_cached", new=AsyncMock(return_value=None)),
        patch(
            "app.services.estimator._get_or_fetch_yandex_valuation_cached",
            new=AsyncMock(return_value=None),
        ),
        patch(
            "app.services.estimator.estimate_via_cian_valuation",
            new=AsyncMock(return_value=None),
        ),
        patch("app.services.estimator._get_asking_sold_ratio", return_value=(None, None)),
    )


def _find_insert_call(db: MagicMock, table_marker: str) -> dict[str, Any]:
    for call in db.execute.call_args_list:
        stmt = call.args[0]
        sql = str(getattr(stmt, "text", stmt))
        if f"INSERT INTO {table_marker}" in sql:
            return call.args[1]
    raise AssertionError(
        f"no INSERT INTO {table_marker} call captured; calls={db.execute.call_args_list}"
    )


# ── require_consent=True, no consent -> 422 BEFORE any work ───────────────────


def test_require_consent_without_payload_consent_raises_422_before_geocode() -> None:
    from app.services.estimator import estimate_quality

    db = MagicMock()
    payload = _make_payload()  # consent defaults to None

    async def _run() -> None:
        with patch("app.services.estimator.geocode") as geocode_mock:
            with pytest.raises(HTTPException) as exc_info:
                await estimate_quality(payload, db, require_consent=True)
            assert exc_info.value.status_code == 422
            geocode_mock.assert_not_called()

    anyio.run(_run)
    assert not db.execute.called, "gate must precede ANY db write"


def test_require_consent_with_payload_consent_false_raises_422() -> None:
    from app.services.estimator import estimate_quality

    db = MagicMock()
    payload = _make_payload(consent=False)

    async def _run() -> None:
        with pytest.raises(HTTPException) as exc_info:
            await estimate_quality(payload, db, require_consent=True)
        assert exc_info.value.status_code == 422

    anyio.run(_run)
    assert not db.execute.called


# ── require_consent=True, consent given -> proceeds, proof persisted ──────────


def test_require_consent_with_payload_consent_persists_proof() -> None:
    from app.services.estimator import _ESTIMATE_CONSENT_POLICY_VERSION as POLICY_VERSION
    from app.services.estimator import _ESTIMATE_CONSENT_TEXT_SNAPSHOT as TEXT_SNAPSHOT
    from app.services.estimator import estimate_quality

    db = MagicMock()
    payload = _make_payload(consent=True)
    geocode_mock = AsyncMock(return_value=_make_fake_geo())

    async def _run() -> Any:
        with contextlib.ExitStack() as stack:
            for cm in _downstream_patches(geocode_mock):
                stack.enter_context(cm)
            return await estimate_quality(
                payload, db, client_ip="203.0.113.9", require_consent=True
            )

    result = anyio.run(_run)
    assert result.estimate_id is not None

    params = _find_insert_call(db, "trade_in_estimates")
    assert params["consent"] is True
    assert params["client_ip"] == "203.0.113.9"
    assert params["consent_policy_version"] == POLICY_VERSION
    assert params["consent_text_snapshot"] == TEXT_SNAPSHOT
    assert params["created_by"] is None


# ── require_consent=True, consent given, geocode fails -> _empty_estimate too ─


def test_empty_estimate_fallback_persists_proof_when_consent_required() -> None:
    from app.services.estimator import _ESTIMATE_CONSENT_POLICY_VERSION as POLICY_VERSION
    from app.services.estimator import _ESTIMATE_CONSENT_TEXT_SNAPSHOT as TEXT_SNAPSHOT
    from app.services.estimator import estimate_quality

    db = MagicMock()
    payload = _make_payload(consent=True)
    geocode_mock = AsyncMock(return_value=None)  # geocode failure -> _empty_estimate path

    async def _run() -> Any:
        with patch("app.services.estimator.geocode", new=geocode_mock):
            return await estimate_quality(
                payload, db, client_ip="198.51.100.4", require_consent=True
            )

    result = anyio.run(_run)
    assert result.n_analogs == 0

    params = _find_insert_call(db, "trade_in_estimates")
    assert params["consent"] is True
    assert params["client_ip"] == "198.51.100.4"
    assert params["consent_policy_version"] == POLICY_VERSION
    assert params["consent_text_snapshot"] == TEXT_SNAPSHOT


def test_empty_estimate_fallback_gate_still_blocks_without_consent() -> None:
    """Gate precedes _empty_estimate too -- geocode is never even reached."""
    from app.services.estimator import estimate_quality

    db = MagicMock()
    payload = _make_payload()  # no consent

    async def _run() -> None:
        with patch("app.services.estimator.geocode") as geocode_mock:
            with pytest.raises(HTTPException) as exc_info:
                await estimate_quality(payload, db, require_consent=True)
            assert exc_info.value.status_code == 422
            geocode_mock.assert_not_called()

    anyio.run(_run)


# ── require_consent defaults False -> zero blast radius on existing callers ───


def test_require_consent_defaults_false() -> None:
    from app.services.estimator import estimate_quality

    sig = inspect.signature(estimate_quality)
    assert sig.parameters["require_consent"].default is False
    assert sig.parameters["require_consent"].kind == inspect.Parameter.KEYWORD_ONLY


def test_bare_call_without_require_consent_is_unaffected() -> None:
    """The exact call shape used by ~90 other estimator tests
    (estimate_quality(payload, db), no created_by/require_consent at all) --
    must keep working with NO consent field on the payload whatsoever."""
    from app.services.estimator import estimate_quality

    db = MagicMock()
    payload = _make_payload()  # no consent field
    geocode_mock = AsyncMock(return_value=_make_fake_geo())

    async def _run() -> Any:
        with contextlib.ExitStack() as stack:
            for cm in _downstream_patches(geocode_mock):
                stack.enter_context(cm)
            return await estimate_quality(payload, db)

    result = anyio.run(_run)
    assert result.estimate_id is not None

    params = _find_insert_call(db, "trade_in_estimates")
    assert params["consent"] is None
    assert params["client_ip"] is None
    assert params["consent_policy_version"] is None
    assert params["consent_text_snapshot"] is None


# ── B2B pilot (created_by set, require_consent left False) -> unaffected ──────


def test_b2b_pilot_without_consent_field_unaffected() -> None:
    from app.services.estimator import estimate_quality

    db = MagicMock()
    payload = _make_payload()  # no consent field at all -- mirrors real pilot frontend
    geocode_mock = AsyncMock(return_value=_make_fake_geo())

    async def _run() -> Any:
        with contextlib.ExitStack() as stack:
            for cm in _downstream_patches(geocode_mock):
                stack.enter_context(cm)
            return await estimate_quality(payload, db, created_by="kopylov")

    result = anyio.run(_run)
    assert result.estimate_id is not None

    params = _find_insert_call(db, "trade_in_estimates")
    assert params["created_by"] == "kopylov"
    assert params["consent"] is None
    assert params["client_ip"] is None
    assert params["consent_policy_version"] is None
    assert params["consent_text_snapshot"] is None


# ── production wiring: app/api/v1/trade_in.py passes require_consent correctly ─


def test_api_handler_wires_require_consent_from_auth_header() -> None:
    """Source-inspection guard: the ONLY production caller of estimate_quality
    must derive require_consent from the ABSENCE of X-Authenticated-User, not
    hardcode True/False. Cheaper and more robust than spinning up a full app +
    rbac_guard fixture just to exercise one kwarg's wiring."""
    from app.api.v1 import trade_in as trade_in_module

    src = re.sub(r"\s+", " ", inspect.getsource(trade_in_module.estimate))
    assert "require_consent=x_authenticated_user is None" in src


# ── Migration 192 sanity ────────────────────────────────────────────────────────


def test_migration_192_exists() -> None:
    assert _MIGRATION_192.is_file(), f"missing migration: {_MIGRATION_192}"


def test_migration_192_is_transactional() -> None:
    sql = _MIGRATION_192.read_text("utf-8")
    assert "BEGIN;" in sql
    assert "COMMIT;" in sql


def test_migration_192_is_idempotent() -> None:
    sql = _MIGRATION_192.read_text("utf-8")
    assert "ADD COLUMN IF NOT EXISTS consent" in sql
    assert "ADD COLUMN IF NOT EXISTS client_ip" in sql
    assert "CREATE INDEX IF NOT EXISTS" in sql
    assert "pg_constraint" in sql  # DO-block guard, not bare ADD CONSTRAINT


def test_migration_192_no_psycopg_trap() -> None:
    sql = _MIGRATION_192.read_text("utf-8")
    assert not re.search(r":\w+::", sql)


def test_migration_192_check_constraint_allows_null_or_true() -> None:
    sql = _MIGRATION_192.read_text("utf-8")
    assert "consent IS NULL OR consent IS TRUE" in sql