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

"""Tests for GET /api/v1/trade-in/history scoping (#656).

Закрывает cross-pilot data-leak: non-admin видит только свои оценки
(created_by = X-Authenticated-User), admin видит все либо фильтрует по ?account.

Проверяем САМ скоупинг через перехват SQL/params, переданных в db.execute —
реальная БД не нужна (DB + get_role мокируются).
"""

from __future__ import annotations

import os
import sys
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


@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_db_mock(rows: list[dict]) -> MagicMock:
    """DB session mock returning *rows* from .mappings().all()."""
    db = MagicMock()
    mapping_result = MagicMock()
    mapping_result.all.return_value = rows
    execute_result = MagicMock()
    execute_result.mappings.return_value = mapping_result
    db.execute.return_value = execute_result
    return db


def _captured_sql_and_params(db_mock: MagicMock) -> tuple[str, dict]:
    """Extract the SQL text + params dict from the single db.execute call."""
    args, _ = db_mock.execute.call_args
    sql = str(args[0])
    params = args[1]
    return sql, params


def _client_with(app: FastAPI, db_mock: MagicMock, role: str) -> TestClient:
    """Override get_db with *db_mock* and patch get_role to return *role*."""
    from app.api.v1 import trade_in as trade_in_module
    from app.core.db import get_db

    def _override_db():
        yield db_mock

    app.dependency_overrides[get_db] = _override_db
    # estimate_history imports get_role from app.core.auth at call time → patch source.
    trade_in_module_auth = sys.modules["app.core.auth"]
    trade_in_module_auth.get_role = lambda _u: role  # type: ignore[assignment]
    _ = trade_in_module  # keep router import side-effect explicit
    return TestClient(app)


def test_history_requires_authenticated_user(trade_in_app: FastAPI) -> None:
    """No X-Authenticated-User header → 401 (mirror /me)."""
    db_mock = _make_db_mock([])
    from app.core.db import get_db

    def _override_db():
        yield db_mock

    trade_in_app.dependency_overrides[get_db] = _override_db
    client = TestClient(trade_in_app)
    resp = client.get("/api/v1/trade-in/history")
    assert resp.status_code == 401
    # DB must not be touched when unauthenticated.
    db_mock.execute.assert_not_called()


def test_non_admin_sees_only_own_rows(trade_in_app: FastAPI) -> None:
    """Pilot user → WHERE created_by = :owner with owner == header value."""
    db_mock = _make_db_mock([{"id": "1", "address": "A", "rooms": 2,
                             "area_m2": 50, "median_price": 5_000_000,
                             "confidence": "medium", "n_analogs": 7,
                             "created_at": "2026-05-29T00:00:00"}])
    client = _client_with(trade_in_app, db_mock, role="pilot")
    resp = client.get(
        "/api/v1/trade-in/history",
        headers={"X-Authenticated-User": "kopylov"},
    )
    assert resp.status_code == 200
    sql, params = _captured_sql_and_params(db_mock)
    assert "created_by = :owner" in sql
    assert params["owner"] == "kopylov"
    assert "account" not in params


def test_non_admin_account_param_is_ignored(trade_in_app: FastAPI) -> None:
    """Pilot passing ?account=victim must still be scoped to themselves."""
    db_mock = _make_db_mock([])
    client = _client_with(trade_in_app, db_mock, role="pilot")
    resp = client.get(
        "/api/v1/trade-in/history",
        params={"account": "victim"},
        headers={"X-Authenticated-User": "kopylov"},
    )
    assert resp.status_code == 200
    sql, params = _captured_sql_and_params(db_mock)
    assert "created_by = :owner" in sql
    assert params["owner"] == "kopylov"
    # The attacker-supplied account must NOT reach the query.
    assert "account" not in params
    assert "victim" not in params.values()


def test_admin_sees_all_rows(trade_in_app: FastAPI) -> None:
    """Admin without ?account → no WHERE filter (all rows)."""
    db_mock = _make_db_mock([])
    client = _client_with(trade_in_app, db_mock, role="admin")
    resp = client.get(
        "/api/v1/trade-in/history",
        headers={"X-Authenticated-User": "admin"},
    )
    assert resp.status_code == 200
    sql, params = _captured_sql_and_params(db_mock)
    assert "WHERE" not in sql.upper()
    assert "owner" not in params
    assert "account" not in params


def test_admin_filters_by_account(trade_in_app: FastAPI) -> None:
    """Admin with ?account=kopylov → WHERE created_by = :account."""
    db_mock = _make_db_mock([])
    client = _client_with(trade_in_app, db_mock, role="admin")
    resp = client.get(
        "/api/v1/trade-in/history",
        params={"account": "kopylov"},
        headers={"X-Authenticated-User": "admin"},
    )
    assert resp.status_code == 200
    sql, params = _captured_sql_and_params(db_mock)
    assert "created_by = :account" in sql
    assert params["account"] == "kopylov"
    assert "owner" not in params