lekss361
76016fd469
All checks were successful
Deploy Trade-In / changes (push) Successful in 11s
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 4m52s
Deploy Trade-In / build-backend (push) Successful in 1m43s
Deploy Trade-In / deploy (push) Successful in 1m8s
287 lines
10 KiB
Python
287 lines
10 KiB
Python
"""Offline-тесты POST /api/v1/trade-in/lead (#2376, sub-issue родителя #1971).
|
||
|
||
Покрытие (db мокается, NO live network/DB):
|
||
- happy path: consent=True + валидный телефон -> 200, INSERT + commit
|
||
- consent=False -> 422 (Literal[True] guard)
|
||
- невалидный формат телефона -> 422
|
||
- телефон без цифр / слишком мало цифр -> 422 (digit-guard, #2376 hardening)
|
||
- source="landing" (мёртвая воронка) -> 422 (литерал убран из схемы)
|
||
- estimate_id, которого нет в trade_in_estimates -> 404
|
||
- IDOR guard (security-audit, зеркалит #690/test_estimate_idor.py): estimate_id
|
||
чужого пользователя -> 404; admin может привязать любой; нет
|
||
X-Authenticated-User -> 401; неизвестная роль -> 403
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||
|
||
from datetime import UTC, datetime
|
||
from types import SimpleNamespace
|
||
from typing import Any
|
||
from unittest.mock import MagicMock
|
||
from uuid import uuid4
|
||
|
||
import pytest
|
||
from fastapi import FastAPI
|
||
from fastapi.testclient import TestClient
|
||
|
||
|
||
@pytest.fixture
|
||
def db() -> MagicMock:
|
||
return MagicMock()
|
||
|
||
|
||
@pytest.fixture
|
||
def client(db: MagicMock) -> TestClient:
|
||
from app.api.v1 import lead as lead_module
|
||
from app.core.db import get_db
|
||
|
||
app = FastAPI()
|
||
app.include_router(lead_module.router, prefix="/api/v1/trade-in")
|
||
|
||
def fake_db() -> Any:
|
||
yield db
|
||
|
||
app.dependency_overrides[get_db] = fake_db
|
||
return TestClient(app)
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _restore_get_role():
|
||
"""Restore app.core.auth.get_role after each test (mirror test_estimate_idor.py)."""
|
||
from app.core import auth as auth_mod
|
||
|
||
original = auth_mod.get_role
|
||
yield
|
||
auth_mod.get_role = original
|
||
|
||
|
||
def _insert_result(lead_id: str) -> MagicMock:
|
||
result = MagicMock()
|
||
result.mappings.return_value.one.return_value = {
|
||
"id": lead_id,
|
||
"created_at": datetime(2026, 7, 4, 12, 0, tzinfo=UTC),
|
||
}
|
||
return result
|
||
|
||
|
||
def test_lead_happy_path(client: TestClient, db: MagicMock) -> None:
|
||
from app.api.v1 import lead as lead_module
|
||
|
||
lead_id = str(uuid4())
|
||
db.execute.return_value = _insert_result(lead_id)
|
||
|
||
r = client.post(
|
||
"/api/v1/trade-in/lead",
|
||
json={"phone": "+7 (912) 345-67-89", "consent": True},
|
||
headers={"x-forwarded-for": "203.0.113.7, 10.0.0.1"},
|
||
)
|
||
assert r.status_code == 200, r.text
|
||
body = r.json()
|
||
assert body["id"] == lead_id
|
||
assert body["status"] == "received"
|
||
assert db.commit.called
|
||
|
||
params = db.execute.call_args.args[1]
|
||
assert params["phone"] == "+7 (912) 345-67-89"
|
||
assert params["consent"] is True
|
||
assert params["source"] == "result"
|
||
assert params["estimate_id"] is None
|
||
# 152-ФЗ proof-of-consent (migration 182): client_ip / policy version / text
|
||
# snapshot must reach the INSERT params, not just the audit log (#2497 TODO).
|
||
assert params["client_ip"] == "203.0.113.7" # first hop of X-Forwarded-For
|
||
assert params["consent_policy_version"] == lead_module._CONSENT_POLICY_VERSION
|
||
assert params["consent_text_snapshot"] == lead_module._CONSENT_TEXT_SNAPSHOT
|
||
|
||
|
||
def test_lead_client_ip_falls_back_to_peer_when_no_xff(client: TestClient, db: MagicMock) -> None:
|
||
# No X-Forwarded-For header -> falls back to request.client.host (TestClient
|
||
# reports "testclient"), still persisted rather than silently None.
|
||
lead_id = str(uuid4())
|
||
db.execute.return_value = _insert_result(lead_id)
|
||
|
||
r = client.post(
|
||
"/api/v1/trade-in/lead",
|
||
json={"phone": "+79123456789", "consent": True},
|
||
)
|
||
assert r.status_code == 200, r.text
|
||
|
||
params = db.execute.call_args.args[1]
|
||
assert params["client_ip"] == "testclient"
|
||
|
||
|
||
def test_lead_consent_false_422(client: TestClient, db: MagicMock) -> None:
|
||
r = client.post(
|
||
"/api/v1/trade-in/lead",
|
||
json={"phone": "+79123456789", "consent": False},
|
||
)
|
||
assert r.status_code == 422
|
||
assert not db.execute.called
|
||
|
||
|
||
def test_lead_consent_missing_422(client: TestClient) -> None:
|
||
r = client.post("/api/v1/trade-in/lead", json={"phone": "+79123456789"})
|
||
assert r.status_code == 422
|
||
|
||
|
||
def test_lead_invalid_phone_422(client: TestClient) -> None:
|
||
r = client.post(
|
||
"/api/v1/trade-in/lead",
|
||
json={"phone": "not-a-phone!!", "consent": True},
|
||
)
|
||
assert r.status_code == 422
|
||
|
||
|
||
def test_lead_with_unknown_estimate_id_404(client: TestClient, db: MagicMock) -> None:
|
||
result = MagicMock()
|
||
result.fetchone.return_value = None
|
||
db.execute.return_value = result
|
||
|
||
r = client.post(
|
||
"/api/v1/trade-in/lead",
|
||
json={"phone": "+79123456789", "consent": True, "estimate_id": str(uuid4())},
|
||
)
|
||
assert r.status_code == 404
|
||
assert not db.commit.called
|
||
|
||
|
||
def test_lead_with_known_estimate_id_200(client: TestClient, db: MagicMock) -> None:
|
||
"""Owner привязывает лид к своей же оценке -> 200."""
|
||
from app.core import auth as auth_mod
|
||
|
||
auth_mod.get_role = lambda _u: "pilot" # type: ignore[assignment]
|
||
|
||
lead_id = str(uuid4())
|
||
exists_result = MagicMock()
|
||
exists_result.fetchone.return_value = SimpleNamespace(created_by="kopylov")
|
||
db.execute.side_effect = [exists_result, _insert_result(lead_id)]
|
||
|
||
estimate_id = str(uuid4())
|
||
r = client.post(
|
||
"/api/v1/trade-in/lead",
|
||
json={
|
||
"phone": "+79123456789",
|
||
"consent": True,
|
||
"estimate_id": estimate_id,
|
||
},
|
||
headers={"X-Authenticated-User": "kopylov"},
|
||
)
|
||
assert r.status_code == 200, r.text
|
||
params = db.execute.call_args.args[1]
|
||
assert params["estimate_id"] == estimate_id
|
||
assert params["source"] == "result"
|
||
|
||
|
||
# ── IDOR guard (security-audit): estimate_id ownership ─────────────────────────
|
||
|
||
|
||
def test_lead_estimate_id_owned_by_other_user_gets_404(client: TestClient, db: MagicMock) -> None:
|
||
"""Чужой estimate_id -> 404 (существование не подтверждаем), лид НЕ создаётся."""
|
||
from app.core import auth as auth_mod
|
||
|
||
auth_mod.get_role = lambda _u: "pilot" # type: ignore[assignment]
|
||
|
||
exists_result = MagicMock()
|
||
exists_result.fetchone.return_value = SimpleNamespace(created_by="victim")
|
||
db.execute.return_value = exists_result
|
||
|
||
r = client.post(
|
||
"/api/v1/trade-in/lead",
|
||
json={"phone": "+79123456789", "consent": True, "estimate_id": str(uuid4())},
|
||
headers={"X-Authenticated-User": "attacker"},
|
||
)
|
||
assert r.status_code == 404, r.text
|
||
assert not db.commit.called
|
||
|
||
|
||
def test_lead_estimate_id_admin_can_attach_any_200(client: TestClient, db: MagicMock) -> None:
|
||
"""Admin может привязать лид к чужой оценке (owner-or-admin, зеркалит #690)."""
|
||
from app.core import auth as auth_mod
|
||
|
||
auth_mod.get_role = lambda _u: "admin" # type: ignore[assignment]
|
||
|
||
lead_id = str(uuid4())
|
||
exists_result = MagicMock()
|
||
exists_result.fetchone.return_value = SimpleNamespace(created_by="someone_else")
|
||
db.execute.side_effect = [exists_result, _insert_result(lead_id)]
|
||
|
||
r = client.post(
|
||
"/api/v1/trade-in/lead",
|
||
json={"phone": "+79123456789", "consent": True, "estimate_id": str(uuid4())},
|
||
headers={"X-Authenticated-User": "admin"},
|
||
)
|
||
assert r.status_code == 200, r.text
|
||
|
||
|
||
def test_lead_estimate_id_requires_authenticated_user_401(
|
||
client: TestClient, db: MagicMock
|
||
) -> None:
|
||
"""estimate_id задан, но нет X-Authenticated-User -> 401 (defense-in-depth: в
|
||
проде rbac_guard уже требует заголовок раньше, см. app/main.py)."""
|
||
exists_result = MagicMock()
|
||
exists_result.fetchone.return_value = SimpleNamespace(created_by="kopylov")
|
||
db.execute.return_value = exists_result
|
||
|
||
r = client.post(
|
||
"/api/v1/trade-in/lead",
|
||
json={"phone": "+79123456789", "consent": True, "estimate_id": str(uuid4())},
|
||
)
|
||
assert r.status_code == 401, r.text
|
||
assert not db.commit.called
|
||
|
||
|
||
def test_lead_estimate_id_unknown_role_403(client: TestClient, db: MagicMock) -> None:
|
||
"""Аутентифицирован через Caddy, но роль отсутствует в roles.yaml -> 403."""
|
||
from app.core import auth as auth_mod
|
||
|
||
def _raise_keyerror(_u: str):
|
||
raise KeyError(_u)
|
||
|
||
auth_mod.get_role = _raise_keyerror # type: ignore[assignment]
|
||
|
||
exists_result = MagicMock()
|
||
exists_result.fetchone.return_value = SimpleNamespace(created_by="kopylov")
|
||
db.execute.return_value = exists_result
|
||
|
||
r = client.post(
|
||
"/api/v1/trade-in/lead",
|
||
json={"phone": "+79123456789", "consent": True, "estimate_id": str(uuid4())},
|
||
headers={"X-Authenticated-User": "ghost"},
|
||
)
|
||
assert r.status_code == 403, r.text
|
||
assert not db.commit.called
|
||
|
||
|
||
def test_lead_digit_free_phone_422(client: TestClient, db: MagicMock) -> None:
|
||
# "(()) -- .." проходит regex-маску (только +/скобки/дефисы/точки/пробелы),
|
||
# но содержит 0 цифр -> должно отклоняться digit-валидатором.
|
||
r = client.post(
|
||
"/api/v1/trade-in/lead",
|
||
json={"phone": "(()) -- ..", "consent": True},
|
||
)
|
||
assert r.status_code == 422, r.text
|
||
assert not db.execute.called
|
||
|
||
|
||
def test_lead_too_few_digits_phone_422(client: TestClient, db: MagicMock) -> None:
|
||
# 6 цифр < минимума 10 -> 422 (маску проходит по длине, digit-guard режет).
|
||
r = client.post(
|
||
"/api/v1/trade-in/lead",
|
||
json={"phone": "123-456", "consent": True},
|
||
)
|
||
assert r.status_code == 422, r.text
|
||
assert not db.execute.called
|
||
|
||
|
||
def test_lead_source_landing_rejected_422(client: TestClient, db: MagicMock) -> None:
|
||
# "landing"-воронка недостижима (rbac_guard закрывает /lead) — литерал убран
|
||
# из схемы, поэтому явный source="landing" теперь 422.
|
||
r = client.post(
|
||
"/api/v1/trade-in/lead",
|
||
json={"phone": "+79123456789", "consent": True, "source": "landing"},
|
||
)
|
||
assert r.status_code == 422, r.text
|
||
assert not db.execute.called
|