lekss361
921aba04b8
Some checks failed
Deploy Trade-In / deploy (push) Blocked by required conditions
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 56s
Deploy Trade-In / build-backend (push) Has been cancelled
263 lines
9.1 KiB
Python
263 lines
9.1 KiB
Python
"""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(autouse=True)
|
||
def _restore_get_role():
|
||
"""Restore app.core.auth.get_role after each test.
|
||
|
||
`_client_with` monkeypatches `get_role` via raw attribute assignment; without
|
||
this teardown the patch leaked into the whole-suite run and broke test_rbac.py
|
||
(kopylov resolved as 'admin', unknown users no longer raised KeyError).
|
||
"""
|
||
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_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
|
||
|
||
|
||
def test_history_returns_full_input_params(trade_in_app: FastAPI) -> None:
|
||
"""#2417: /history проецирует floor/total_floors/year_built/house_type/
|
||
|
||
repair_state/has_balcony в дополнение к базовому подмножеству, включая
|
||
NULL-passthrough для строк, где эти поля никогда не заполнялись.
|
||
"""
|
||
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",
|
||
"sources_used": ["avito", "cian"],
|
||
"floor": 3,
|
||
"total_floors": 9,
|
||
"year_built": 2005,
|
||
"house_type": "panel",
|
||
"repair_state": "good",
|
||
"has_balcony": True,
|
||
},
|
||
{
|
||
"id": "2",
|
||
"address": "B",
|
||
"rooms": 1,
|
||
"area_m2": 33,
|
||
"median_price": 3_000_000,
|
||
"confidence": "low",
|
||
"n_analogs": 2,
|
||
"created_at": "2026-05-28T00:00:00",
|
||
"sources_used": [],
|
||
# legacy row / поля никогда не заполнялись пользователем.
|
||
"floor": None,
|
||
"total_floors": None,
|
||
"year_built": None,
|
||
"house_type": None,
|
||
"repair_state": None,
|
||
"has_balcony": None,
|
||
},
|
||
]
|
||
)
|
||
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)
|
||
for col in (
|
||
"floor",
|
||
"total_floors",
|
||
"year_built",
|
||
"house_type",
|
||
"repair_state",
|
||
"has_balcony",
|
||
):
|
||
assert col in sql
|
||
|
||
body = resp.json()
|
||
assert len(body) == 2
|
||
|
||
filled, legacy = body[0], body[1]
|
||
assert filled["floor"] == 3
|
||
assert filled["total_floors"] == 9
|
||
assert filled["year_built"] == 2005
|
||
assert filled["house_type"] == "panel"
|
||
assert filled["repair_state"] == "good"
|
||
assert filled["has_balcony"] is True
|
||
|
||
# NULL passthrough — не должно рейзить/подменяться дефолтами.
|
||
assert legacy["floor"] is None
|
||
assert legacy["total_floors"] is None
|
||
assert legacy["year_built"] is None
|
||
assert legacy["house_type"] is None
|
||
assert legacy["repair_state"] is None
|
||
assert legacy["has_balcony"] is None
|