fix(tradein/tests): тесты авторизации проверяют настоящий guard + реальный рендер PDF #2541
6 changed files with 396 additions and 212 deletions
136
tradein-mvp/backend/app/core/rbac.py
Normal file
136
tradein-mvp/backend/app/core/rbac.py
Normal file
|
|
@ -0,0 +1,136 @@
|
|||
"""RBAC guard middleware — extracted from ``app/main.py``.
|
||||
|
||||
Historically ``rbac_guard`` lived inline in ``app/main.py`` and the test suite
|
||||
(``tests/test_rbac.py``, ``tests/test_internal_auth_secret.py``) kept a
|
||||
hand-maintained *copy* of it, labelled "MIRROR of app.main — keep in sync
|
||||
manually". The copy drifted: it was missing the #2213
|
||||
``X-Internal-Auth-Secret`` defense-in-depth check that the real guard has,
|
||||
so a regression in that check would NOT have failed CI.
|
||||
|
||||
This module holds the real guard with no DB/lifespan/scheduler side effects
|
||||
(only ``app.core.auth`` + ``app.core.config``, both side-effect-free at
|
||||
import time beyond requiring ``DATABASE_URL`` in the environment for
|
||||
``Settings()``). ``app/main.py`` and the test apps both import THIS module,
|
||||
so tests exercise the exact production code path instead of a copy that can
|
||||
silently fall out of sync.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
import secrets
|
||||
from collections.abc import Awaitable, Callable
|
||||
|
||||
from fastapi import Request
|
||||
from fastapi.responses import JSONResponse, Response
|
||||
|
||||
from app.core.auth import get_role, is_path_allowed
|
||||
from app.core.config import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# RBAC: defense-in-depth поверх Caddy basic_auth + X-Authenticated-User
|
||||
# (см. app/core/auth.py + auth/roles.yaml). Правила:
|
||||
# 1) Любой non-public path требует X-Authenticated-User — иначе 401.
|
||||
# 2) Юзер должен быть в roles.yaml — иначе 403 («неизвестный юзер ничего
|
||||
# не видит» — decided 2026-05-25).
|
||||
# 3) /api/v1/admin/* (= внешний /trade-in/api/v1/admin/* после Caddy
|
||||
# `uri strip_prefix /trade-in`) — только role=admin, иначе 403.
|
||||
# Public paths без auth (/health, /docs, /openapi.json) пропускаем —
|
||||
# X-Authenticated-User там не приходит из Caddy.
|
||||
_ADMIN_API_RE = re.compile(r"^/api/v1/admin/")
|
||||
_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"})
|
||||
# #R2-H3: Caddy срезает внешний префикс /trade-in (uri strip_prefix) перед
|
||||
# tradein-backend, а globs в roles.yaml — ВНЕШНИЕ (/trade-in/api/v1/**). Для
|
||||
# scope-проверки восстанавливаем внешний путь.
|
||||
_EXTERNAL_PREFIX = "/trade-in"
|
||||
# Bootstrap-пути, доступные ЛЮБОМУ известному юзеру независимо от роли: /me отдаёт
|
||||
# роль (expired → trial-экран), /brand/* — брендинг login/trial-экрана. Без них
|
||||
# expired (roles.yaml paths:[] deny:/**) не получил бы роль и не увидел trial-экран.
|
||||
_RBAC_BOOTSTRAP_EXEMPT = ("/api/v1/me", "/api/v1/brand")
|
||||
|
||||
|
||||
async def rbac_guard(
|
||||
request: Request,
|
||||
call_next: Callable[[Request], Awaitable[Response]],
|
||||
) -> Response:
|
||||
path = request.url.path
|
||||
if path in _PUBLIC_PATHS:
|
||||
return await call_next(request)
|
||||
|
||||
username = request.headers.get("X-Authenticated-User")
|
||||
if not username:
|
||||
return JSONResponse(
|
||||
status_code=401,
|
||||
content={"detail": "no authenticated user (Caddy basic_auth required)"},
|
||||
)
|
||||
|
||||
# #2213 defense-in-depth: если общий секрет задан — запрос с X-Authenticated-User
|
||||
# ОБЯЗАН нести валидный X-Internal-Auth-Secret (его добавляет Caddy из env).
|
||||
# Иначе это подделка заголовка мимо Caddy (напр. изнутри gendesign_shared) → 401.
|
||||
# Constant-time compare против timing-атак. Пусто = защита не активна (fail-open).
|
||||
secret = settings.tradein_internal_auth_secret
|
||||
if secret:
|
||||
provided = request.headers.get("X-Internal-Auth-Secret", "")
|
||||
if not secrets.compare_digest(provided, secret):
|
||||
logger.warning(
|
||||
"RBAC: X-Authenticated-User=%r без валидного X-Internal-Auth-Secret "
|
||||
"на %s — возможная подделка заголовка мимо Caddy",
|
||||
username,
|
||||
path,
|
||||
)
|
||||
return JSONResponse(
|
||||
status_code=401,
|
||||
content={"detail": "invalid or missing internal auth secret"},
|
||||
)
|
||||
|
||||
try:
|
||||
role = get_role(username)
|
||||
except KeyError:
|
||||
logger.warning("RBAC: unknown user %r tried %s", username, path)
|
||||
return JSONResponse(
|
||||
status_code=403,
|
||||
content={"detail": "user not in roles config"},
|
||||
)
|
||||
|
||||
if _ADMIN_API_RE.match(path) and role != "admin":
|
||||
logger.info("RBAC: blocked %s (role=%s) from %s", username, role, path)
|
||||
return JSONResponse(
|
||||
status_code=403,
|
||||
content={"detail": "admin only"},
|
||||
)
|
||||
|
||||
# #R2-H3: энфорсим roles.yaml scope (paths/deny) для ВСЕХ non-admin путей, а не
|
||||
# только /admin/*. Иначе revoked (role=expired, paths:[] deny:/**) или узко-
|
||||
# скоупленный аккаунт достаёт non-admin API (напр. POST /api/v1/search —
|
||||
# экспорт листингов), который roles.yaml ему запрещает. Bootstrap-пути (/me,
|
||||
# /brand) исключены выше по списку. roles.yaml globs внешние → восстанавливаем
|
||||
# внешний путь (Caddy срезал /trade-in). На сбой парса — fail-open + громкий
|
||||
# лог: не лочим платящего pilot из-за конфиг-бага (admin-гейт выше остаётся).
|
||||
if not path.startswith(_RBAC_BOOTSTRAP_EXEMPT):
|
||||
external_path = _EXTERNAL_PREFIX + path
|
||||
try:
|
||||
allowed = is_path_allowed(role, external_path)
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"RBAC scope-check raised for %s %s (ext=%s) — fail-open",
|
||||
username,
|
||||
path,
|
||||
external_path,
|
||||
)
|
||||
allowed = True
|
||||
if not allowed:
|
||||
logger.info(
|
||||
"RBAC: scope-blocked %s (role=%s) from %s (ext=%s)",
|
||||
username,
|
||||
role,
|
||||
path,
|
||||
external_path,
|
||||
)
|
||||
return JSONResponse(
|
||||
status_code=403,
|
||||
content={"detail": "forbidden for role"},
|
||||
)
|
||||
|
||||
return await call_next(request)
|
||||
|
|
@ -8,15 +8,12 @@ from __future__ import annotations
|
|||
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import secrets
|
||||
from collections.abc import AsyncGenerator, Awaitable, Callable
|
||||
from collections.abc import AsyncGenerator
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
import sentry_sdk
|
||||
from fastapi import FastAPI, Request
|
||||
from fastapi import FastAPI
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.responses import JSONResponse, Response
|
||||
from sentry_sdk.integrations.fastapi import FastApiIntegration
|
||||
from sentry_sdk.integrations.httpx import HttpxIntegration
|
||||
from sentry_sdk.integrations.logging import LoggingIntegration
|
||||
|
|
@ -35,11 +32,11 @@ from app.api.v1 import (
|
|||
support,
|
||||
trade_in,
|
||||
)
|
||||
from app.core.auth import get_role, is_path_allowed
|
||||
from app.core.config import settings
|
||||
from app.core.db import SessionLocal
|
||||
from app.core.fdw import ensure_fdw_user_mapping
|
||||
from app.core.ratelimit import RateLimitMiddleware
|
||||
from app.core.rbac import rbac_guard
|
||||
from app.core.request_audit import RequestAuditMiddleware
|
||||
from app.observability.sentry_scrub import scrub_pii_event
|
||||
|
||||
|
|
@ -138,104 +135,9 @@ app = FastAPI(
|
|||
# не видит» — decided 2026-05-25).
|
||||
# 3) /api/v1/admin/* (= внешний /trade-in/api/v1/admin/* после Caddy
|
||||
# `uri strip_prefix /trade-in`) — только role=admin, иначе 403.
|
||||
# Public paths без auth (/health, /docs, /openapi.json) пропускаем —
|
||||
# X-Authenticated-User там не приходит из Caddy.
|
||||
_ADMIN_API_RE = re.compile(r"^/api/v1/admin/")
|
||||
_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"})
|
||||
# #R2-H3: Caddy срезает внешний префикс /trade-in (uri strip_prefix) перед
|
||||
# tradein-backend, а globs в roles.yaml — ВНЕШНИЕ (/trade-in/api/v1/**). Для
|
||||
# scope-проверки восстанавливаем внешний путь.
|
||||
_EXTERNAL_PREFIX = "/trade-in"
|
||||
# Bootstrap-пути, доступные ЛЮБОМУ известному юзеру независимо от роли: /me отдаёт
|
||||
# роль (expired → trial-экран), /brand/* — брендинг login/trial-экрана. Без них
|
||||
# expired (roles.yaml paths:[] deny:/**) не получил бы роль и не увидел trial-экран.
|
||||
_RBAC_BOOTSTRAP_EXEMPT = ("/api/v1/me", "/api/v1/brand")
|
||||
|
||||
|
||||
@app.middleware("http")
|
||||
async def rbac_guard(
|
||||
request: Request,
|
||||
call_next: Callable[[Request], Awaitable[Response]],
|
||||
) -> Response:
|
||||
path = request.url.path
|
||||
if path in _PUBLIC_PATHS:
|
||||
return await call_next(request)
|
||||
|
||||
username = request.headers.get("X-Authenticated-User")
|
||||
if not username:
|
||||
return JSONResponse(
|
||||
status_code=401,
|
||||
content={"detail": "no authenticated user (Caddy basic_auth required)"},
|
||||
)
|
||||
|
||||
# #2213 defense-in-depth: если общий секрет задан — запрос с X-Authenticated-User
|
||||
# ОБЯЗАН нести валидный X-Internal-Auth-Secret (его добавляет Caddy из env).
|
||||
# Иначе это подделка заголовка мимо Caddy (напр. изнутри gendesign_shared) → 401.
|
||||
# Constant-time compare против timing-атак. Пусто = защита не активна (fail-open).
|
||||
secret = settings.tradein_internal_auth_secret
|
||||
if secret:
|
||||
provided = request.headers.get("X-Internal-Auth-Secret", "")
|
||||
if not secrets.compare_digest(provided, secret):
|
||||
logger.warning(
|
||||
"RBAC: X-Authenticated-User=%r без валидного X-Internal-Auth-Secret "
|
||||
"на %s — возможная подделка заголовка мимо Caddy",
|
||||
username,
|
||||
path,
|
||||
)
|
||||
return JSONResponse(
|
||||
status_code=401,
|
||||
content={"detail": "invalid or missing internal auth secret"},
|
||||
)
|
||||
|
||||
try:
|
||||
role = get_role(username)
|
||||
except KeyError:
|
||||
logger.warning("RBAC: unknown user %r tried %s", username, path)
|
||||
return JSONResponse(
|
||||
status_code=403,
|
||||
content={"detail": "user not in roles config"},
|
||||
)
|
||||
|
||||
if _ADMIN_API_RE.match(path) and role != "admin":
|
||||
logger.info("RBAC: blocked %s (role=%s) from %s", username, role, path)
|
||||
return JSONResponse(
|
||||
status_code=403,
|
||||
content={"detail": "admin only"},
|
||||
)
|
||||
|
||||
# #R2-H3: энфорсим roles.yaml scope (paths/deny) для ВСЕХ non-admin путей, а не
|
||||
# только /admin/*. Иначе revoked (role=expired, paths:[] deny:/**) или узко-
|
||||
# скоупленный аккаунт достаёт non-admin API (напр. POST /api/v1/search —
|
||||
# экспорт листингов), который roles.yaml ему запрещает. Bootstrap-пути (/me,
|
||||
# /brand) исключены выше по списку. roles.yaml globs внешние → восстанавливаем
|
||||
# внешний путь (Caddy срезал /trade-in). На сбой парса — fail-open + громкий
|
||||
# лог: не лочим платящего pilot из-за конфиг-бага (admin-гейт выше остаётся).
|
||||
if not path.startswith(_RBAC_BOOTSTRAP_EXEMPT):
|
||||
external_path = _EXTERNAL_PREFIX + path
|
||||
try:
|
||||
allowed = is_path_allowed(role, external_path)
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"RBAC scope-check raised for %s %s (ext=%s) — fail-open",
|
||||
username,
|
||||
path,
|
||||
external_path,
|
||||
)
|
||||
allowed = True
|
||||
if not allowed:
|
||||
logger.info(
|
||||
"RBAC: scope-blocked %s (role=%s) from %s (ext=%s)",
|
||||
username,
|
||||
role,
|
||||
path,
|
||||
external_path,
|
||||
)
|
||||
return JSONResponse(
|
||||
status_code=403,
|
||||
content={"detail": "forbidden for role"},
|
||||
)
|
||||
|
||||
return await call_next(request)
|
||||
# Guard body живёт в app/core/rbac.py (без DB/lifespan side effects), чтобы
|
||||
# тесты могли импортировать РЕАЛЬНЫЙ guard вместо hand-maintained копии.
|
||||
app.middleware("http")(rbac_guard)
|
||||
|
||||
|
||||
app.add_middleware(
|
||||
|
|
|
|||
18
tradein-mvp/backend/tests/conftest.py
Normal file
18
tradein-mvp/backend/tests/conftest.py
Normal file
|
|
@ -0,0 +1,18 @@
|
|||
"""Repo-wide test config for tradein-mvp/backend.
|
||||
|
||||
Currently only registers custom pytest markers so they don't emit
|
||||
PytestUnknownMarkWarning when used (`--strict-markers` is not enabled in
|
||||
pyproject.toml, so an unregistered marker would only warn, not fail — this
|
||||
just keeps output clean and documents intent in one place).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
|
||||
def pytest_configure(config) -> None:
|
||||
config.addinivalue_line(
|
||||
"markers",
|
||||
"pdf_render: real (non-mocked) WeasyPrint render — needs native "
|
||||
"Pango/cairo/GObject libs, self-skips where unavailable (see "
|
||||
"tests/test_pdf_real_render.py docstring for how to run it for real).",
|
||||
)
|
||||
|
|
@ -2,13 +2,16 @@
|
|||
|
||||
Backend доверяет X-Authenticated-User (его ставит Caddy). На общей docker-сети
|
||||
gendesign_shared любой контейнер мог бы отправить поддельный
|
||||
`X-Authenticated-User: admin` напрямую на tradein-backend:8000. Общий секрет
|
||||
X-Internal-Auth-Secret закрывает дыру: если TRADEIN_INTERNAL_AUTH_SECRET задан,
|
||||
каждый запрос с X-Authenticated-User обязан нести валидный секрет (constant-time),
|
||||
`X-Authenticated-User: admin` напрямую на tradein-backend:8000, минуя Caddy.
|
||||
Общий секрет X-Internal-Auth-Secret закрывает дыру: если TRADEIN_INTERNAL_AUTH_SECRET
|
||||
задан, каждый запрос с X-Authenticated-User обязан нести валидный секрет (constant-time),
|
||||
иначе 401. Пусто = fail-open (backward-compat до провижининга).
|
||||
|
||||
MIRROR of rbac_guard из app/main.py — включая #2213 secret-gate. Держим копию
|
||||
здесь (как test_rbac.py), чтобы не тянуть тяжёлый app.main (lifespan/DB/scheduler).
|
||||
Использует РЕАЛЬНЫЙ rbac_guard (app/core/rbac.py) — тот же, что регистрирует
|
||||
app/main.py в проде. Раньше здесь была hand-maintained копия ("MIRROR of
|
||||
rbac_guard из app/main.py"); соседний test_rbac.py держал СВОЮ отдельную
|
||||
копию, которая успела отстать (потеряла именно этот secret-gate) — регрессия
|
||||
в реальном guard'е могла бы пройти CI незамеченной. См. app/core/rbac.py.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -17,20 +20,13 @@ import os
|
|||
|
||||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||||
|
||||
import re
|
||||
import secrets
|
||||
from collections.abc import Awaitable, Callable
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI, Request
|
||||
from fastapi.responses import JSONResponse, Response
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.core import auth as auth_mod
|
||||
from app.core import config
|
||||
|
||||
_ADMIN_API_RE = re.compile(r"^/api/v1/admin/")
|
||||
_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"})
|
||||
from app.core.rbac import rbac_guard
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
|
|
@ -39,44 +35,9 @@ def _reset_auth_cache() -> None:
|
|||
|
||||
|
||||
def _build_test_app() -> FastAPI:
|
||||
"""Копия rbac_guard из app/main.py (с #2213 secret-gate)."""
|
||||
"""Test app используя РЕАЛЬНЫЙ rbac_guard (с #2213 secret-gate)."""
|
||||
app = FastAPI()
|
||||
|
||||
@app.middleware("http")
|
||||
async def rbac_guard(
|
||||
request: Request,
|
||||
call_next: Callable[[Request], Awaitable[Response]],
|
||||
) -> Response:
|
||||
path = request.url.path
|
||||
if path in _PUBLIC_PATHS:
|
||||
return await call_next(request)
|
||||
|
||||
username = request.headers.get("X-Authenticated-User")
|
||||
if not username:
|
||||
return JSONResponse(
|
||||
status_code=401,
|
||||
content={"detail": "no authenticated user (Caddy basic_auth required)"},
|
||||
)
|
||||
|
||||
secret = config.settings.tradein_internal_auth_secret
|
||||
if secret:
|
||||
provided = request.headers.get("X-Internal-Auth-Secret", "")
|
||||
if not secrets.compare_digest(provided, secret):
|
||||
return JSONResponse(
|
||||
status_code=401,
|
||||
content={"detail": "invalid or missing internal auth secret"},
|
||||
)
|
||||
|
||||
try:
|
||||
role = auth_mod.get_role(username)
|
||||
except KeyError:
|
||||
return JSONResponse(
|
||||
status_code=403,
|
||||
content={"detail": "user not in roles config"},
|
||||
)
|
||||
if _ADMIN_API_RE.match(path) and role != "admin":
|
||||
return JSONResponse(status_code=403, content={"detail": "admin only"})
|
||||
return await call_next(request)
|
||||
app.middleware("http")(rbac_guard)
|
||||
|
||||
@app.get("/api/v1/ping")
|
||||
async def ping() -> dict:
|
||||
|
|
|
|||
209
tradein-mvp/backend/tests/test_pdf_real_render.py
Normal file
209
tradein-mvp/backend/tests/test_pdf_real_render.py
Normal file
|
|
@ -0,0 +1,209 @@
|
|||
"""Real (non-mocked) WeasyPrint render invariants for the Trade-In PDF report.
|
||||
|
||||
tests/test_pdf_security.py exercises only the HTML-builder layer with
|
||||
WeasyPrint stubbed out (``sys.modules['weasyprint'] = MagicMock()``) — by
|
||||
design, so those tests run everywhere without needing WeasyPrint's native
|
||||
GTK/Pango/cairo libs. That design has a blind spot: it can NEVER catch a real
|
||||
WeasyPrint pagination regression — e.g. the bug fixed in commit 42a50cf8
|
||||
("running @page header/footer → ровно 4 страницы (без пустых)"), where the
|
||||
persistent running header/footer produced an extra TRAILING BLANK 5th page
|
||||
instead of the documented "Структура отчёта — 4 страницы" invariant (see
|
||||
app/services/exporters/trade_in_pdf.py module docstring). A mocked
|
||||
WeasyPrint can never lay out/paginate anything, so it structurally cannot see
|
||||
that class of bug — only a real render can.
|
||||
|
||||
Requires WeasyPrint's native dependencies (Pango/cairo/GObject). These are
|
||||
NOT installed in this dev sandbox (Windows, no libgobject-2.0-0) and NOT
|
||||
installed on the bare ``ubuntu-latest`` runner used by
|
||||
.forgejo/workflows/ci-tradein.yml (no apt-get step there — see that file).
|
||||
This whole module self-skips via ``pytest.skip(..., allow_module_level=True)``
|
||||
the moment the real `import weasyprint` fails for ANY reason (missing native
|
||||
lib, etc.), so it is a harmless no-op everywhere it can't actually run.
|
||||
|
||||
How to run it for real:
|
||||
- Inside the prod/dev docker image — the `runner` stage of
|
||||
tradein-mvp/backend/Dockerfile installs libcairo2 + libpango-1.0-0 +
|
||||
libpangoft2-1.0-0 + fonts-dejavu-core, so WeasyPrint's native deps are
|
||||
present there:
|
||||
docker exec tradein-backend python -m pytest -q -m pdf_render \
|
||||
tests/test_pdf_real_render.py
|
||||
- Locally on a Linux box with WeasyPrint's system deps installed (see
|
||||
https://doc.courtbouillon.org/weasyprint/stable/first_steps.html):
|
||||
uv run pytest -q -m pdf_render tests/test_pdf_real_render.py
|
||||
- Marked ``@pytest.mark.pdf_render`` (registered in tests/conftest.py) so it
|
||||
can be explicitly selected/excluded once a CI runner with the native libs
|
||||
exists; today ci-tradein.yml's `ubuntu-latest` runner doesn't have them,
|
||||
so this module simply self-skips there — nothing to deselect.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||||
|
||||
# test_pdf_security.py deliberately does `sys.modules['weasyprint'] = MagicMock()`
|
||||
# for its own unit tests. sys.modules is a process-global cache: if that module
|
||||
# was imported before this one in the same pytest run, `import weasyprint` below
|
||||
# would silently return the Mock instead of the real package. Purge any
|
||||
# existing stub before attempting the real import — independent of collection
|
||||
# order across test modules.
|
||||
for _name in [n for n in sys.modules if n == "weasyprint" or n.startswith("weasyprint.")]:
|
||||
del sys.modules[_name]
|
||||
|
||||
import pytest # noqa: E402
|
||||
|
||||
try:
|
||||
import weasyprint
|
||||
except Exception as exc: # pragma: no cover - env-dependent (native GTK/Pango/cairo libs)
|
||||
pytest.skip(
|
||||
f"WeasyPrint native deps unavailable, skipping real-render tests: {exc}",
|
||||
allow_module_level=True,
|
||||
)
|
||||
|
||||
pytestmark = pytest.mark.pdf_render
|
||||
|
||||
import unittest.mock as _mock # noqa: E402
|
||||
from datetime import UTC, datetime, timedelta # noqa: E402
|
||||
from uuid import uuid4 # noqa: E402
|
||||
|
||||
from app.schemas.trade_in import AggregatedEstimate # noqa: E402
|
||||
from app.services.brand import Brand # noqa: E402
|
||||
from app.services.exporters import trade_in_pdf as mod # noqa: E402
|
||||
|
||||
_GENERIC = Brand(
|
||||
slug="generic",
|
||||
name="Trade-In",
|
||||
logo_url=None,
|
||||
primary_color="#1d4ed8",
|
||||
accent_color="#f59e0b",
|
||||
footer_text=None,
|
||||
pdf_disclaimer=None,
|
||||
)
|
||||
|
||||
_SNAPSHOT = {
|
||||
"address": "Екатеринбург, ул. Ленина, 1",
|
||||
"area_m2": 50.0,
|
||||
"rooms": 2,
|
||||
"floor": 3,
|
||||
"total_floors": 9,
|
||||
"year_built": 2010,
|
||||
"house_type": "panel",
|
||||
"repair_state": "standard",
|
||||
"has_balcony": True,
|
||||
}
|
||||
|
||||
|
||||
def _estimate(**overrides) -> AggregatedEstimate:
|
||||
base = dict(
|
||||
estimate_id=uuid4(),
|
||||
median_price_rub=10_000_000,
|
||||
range_low_rub=9_000_000,
|
||||
range_high_rub=11_000_000,
|
||||
median_price_per_m2=200_000,
|
||||
confidence="high",
|
||||
n_analogs=15,
|
||||
period_months=24,
|
||||
analogs=[],
|
||||
actual_deals=[],
|
||||
expires_at=datetime.now(UTC) + timedelta(days=30),
|
||||
)
|
||||
base.update(overrides)
|
||||
return AggregatedEstimate(**base)
|
||||
|
||||
|
||||
def _zero_estimate(**overrides) -> AggregatedEstimate:
|
||||
"""Оценка с median=0 — insufficient_data=True (одностраничный empty-state)."""
|
||||
base = dict(
|
||||
estimate_id=uuid4(),
|
||||
median_price_rub=0,
|
||||
range_low_rub=0,
|
||||
range_high_rub=0,
|
||||
median_price_per_m2=0,
|
||||
confidence="low",
|
||||
n_analogs=0,
|
||||
period_months=24,
|
||||
analogs=[],
|
||||
actual_deals=[],
|
||||
expires_at=datetime.now(UTC) + timedelta(days=30),
|
||||
)
|
||||
base.update(overrides)
|
||||
return AggregatedEstimate(**base)
|
||||
|
||||
|
||||
def _render_real_document(estimate, snapshot, brand):
|
||||
"""Call the REAL generate_trade_in_pdf (no mocking of WeasyPrint), capturing
|
||||
the actual weasyprint.HTML/CSS instances + html string it constructs via
|
||||
thin capturing subclasses — so we can additionally call .render() on the
|
||||
exact same HTML instance to get a weasyprint.Document (whose .pages is the
|
||||
public, documented page-count API), without duplicating trade_in_pdf.py's
|
||||
own html_str/css_str assembly logic here (that would drift out of sync
|
||||
with the real function, defeating the point of this test).
|
||||
|
||||
Returns (document, pdf_bytes, html_string).
|
||||
"""
|
||||
html_instances: list[weasyprint.HTML] = []
|
||||
css_instances: list[weasyprint.CSS] = []
|
||||
html_strings: list[str] = []
|
||||
|
||||
class _CapturingHTML(weasyprint.HTML):
|
||||
def __init__(self, *a, **kw):
|
||||
html_strings.append(kw.get("string") if "string" in kw else (a[0] if a else None))
|
||||
super().__init__(*a, **kw)
|
||||
html_instances.append(self)
|
||||
|
||||
class _CapturingCSS(weasyprint.CSS):
|
||||
def __init__(self, *a, **kw):
|
||||
super().__init__(*a, **kw)
|
||||
css_instances.append(self)
|
||||
|
||||
with (
|
||||
_mock.patch.object(weasyprint, "HTML", _CapturingHTML),
|
||||
_mock.patch.object(weasyprint, "CSS", _CapturingCSS),
|
||||
):
|
||||
pdf_bytes = mod.generate_trade_in_pdf(estimate, snapshot, brand=brand)
|
||||
|
||||
assert html_instances, "HTML() was never constructed by generate_trade_in_pdf"
|
||||
assert css_instances, "CSS() was never constructed by generate_trade_in_pdf"
|
||||
document = html_instances[-1].render(stylesheets=[css_instances[-1]])
|
||||
return document, pdf_bytes, html_strings[-1]
|
||||
|
||||
|
||||
def test_real_pdf_has_exactly_4_pages_no_trailing_blank() -> None:
|
||||
"""Regression for commit 42a50cf8: the running @page header/footer must
|
||||
NOT produce a 5th trailing blank page. trade_in_pdf.py's own module
|
||||
docstring documents "Структура отчёта — 4 страницы" as the fixed
|
||||
cover/listings/deals/offer composition — that count IS the authoritative
|
||||
"no empty pages" invariant for this report (any blank page shows up as an
|
||||
extra page beyond these 4 named sections)."""
|
||||
est = _estimate(n_analogs=12, sources_used=["avito"])
|
||||
document, pdf_bytes, _html_str = _render_real_document(est, _SNAPSHOT, _GENERIC)
|
||||
assert len(document.pages) == 4, (
|
||||
f"expected exactly 4 pages (cover/listings/deals/offer), got "
|
||||
f"{len(document.pages)} — likely a trailing/leading blank-page regression"
|
||||
)
|
||||
assert pdf_bytes.startswith(b"%PDF-"), "write_pdf must produce a real PDF"
|
||||
|
||||
|
||||
def test_real_pdf_insufficient_data_is_single_page() -> None:
|
||||
"""insufficient_data=True → one-page empty-state, not the 4-page report."""
|
||||
est = _zero_estimate()
|
||||
document, _pdf_bytes, html_str = _render_real_document(est, _SNAPSHOT, _GENERIC)
|
||||
assert len(document.pages) == 1
|
||||
assert "Недостаточно данных" in html_str
|
||||
|
||||
|
||||
def test_real_pdf_key_blocks_present_exactly_once() -> None:
|
||||
"""Key section headings for each of the 4 pages are present in the actual
|
||||
composed HTML that WeasyPrint rendered (integration-level check — the
|
||||
mocked tests in test_pdf_security.py already verify each builder function
|
||||
in isolation; this catches a composition bug where wiring them together
|
||||
in generate_trade_in_pdf silently drops or duplicates a section)."""
|
||||
est = _estimate(n_analogs=12, sources_used=["avito"])
|
||||
document, _pdf_bytes, html_str = _render_real_document(est, _SNAPSHOT, _GENERIC)
|
||||
assert len(document.pages) == 4
|
||||
markers = ["РЫНОК КВАРТИР", "ФОРМИРОВАНИЕ ВЫКУПНОЙ СТОИМОСТИ"]
|
||||
for marker in markers:
|
||||
found = html_str.count(marker)
|
||||
assert found == 1, f"expected exactly one {marker!r} block, found {found}"
|
||||
|
|
@ -1,7 +1,5 @@
|
|||
"""RBAC unit + integration tests for tradein backend.
|
||||
|
||||
MIRROR of backend/tests/test_rbac.py — kept in sync manually.
|
||||
|
||||
Coverage:
|
||||
- get_role / get_user_scope happy + error paths
|
||||
- is_path_allowed glob semantics (admin everywhere, pilot blocked from /admin/**)
|
||||
|
|
@ -14,22 +12,29 @@ Coverage:
|
|||
|
||||
Тесты используют изолированный FastAPI app (см. _build_test_app), чтобы не
|
||||
тянуть тяжёлые модули из app.main (lifespan task + DB session + scheduler).
|
||||
RBAC middleware и /me router импортируются напрямую — это даёт нам ровно
|
||||
тот же поведение что в проде.
|
||||
Guard больше НЕ мокается/копируется вручную — импортируется тот же
|
||||
``app.core.rbac.rbac_guard``, что регистрирует app/main.py в проде. Раньше
|
||||
здесь была hand-maintained "MIRROR of app.main" копия, которая незаметно
|
||||
отстала (не имела #2213 X-Internal-Auth-Secret check) — правки реального
|
||||
guard'а тесты бы не заметили. См. app/core/rbac.py.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from collections.abc import Awaitable, Callable
|
||||
import os
|
||||
|
||||
# Settings (app.core.config, импортируемый через app.core.rbac) требует
|
||||
# DATABASE_URL на конструирование — stub перед любым app-импортом (тот же
|
||||
# паттерн что и в остальных tests/*.py).
|
||||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI, Request
|
||||
from fastapi.responses import JSONResponse, Response
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.api.v1 import me as me_router
|
||||
from app.core import auth as auth_mod
|
||||
from app.core.rbac import rbac_guard
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
|
|
@ -38,60 +43,13 @@ def _reset_auth_cache() -> None:
|
|||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test app — копия rbac_guard из app/main.py.
|
||||
# Test app — использует РЕАЛЬНЫЙ rbac_guard (app/core/rbac.py), а не копию.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
_ADMIN_API_RE = re.compile(r"^/api/v1/admin/")
|
||||
_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"})
|
||||
# MIRROR of app.main (#R2-H3): keep in sync with the real rbac_guard.
|
||||
_EXTERNAL_PREFIX = "/trade-in"
|
||||
_RBAC_BOOTSTRAP_EXEMPT = ("/api/v1/me", "/api/v1/brand")
|
||||
|
||||
|
||||
def _build_test_app() -> FastAPI:
|
||||
app = FastAPI()
|
||||
|
||||
@app.middleware("http")
|
||||
async def rbac_guard(
|
||||
request: Request,
|
||||
call_next: Callable[[Request], Awaitable[Response]],
|
||||
) -> Response:
|
||||
path = request.url.path
|
||||
if path in _PUBLIC_PATHS:
|
||||
return await call_next(request)
|
||||
|
||||
username = request.headers.get("X-Authenticated-User")
|
||||
if not username:
|
||||
return JSONResponse(
|
||||
status_code=401,
|
||||
content={"detail": "no authenticated user (Caddy basic_auth required)"},
|
||||
)
|
||||
try:
|
||||
role = auth_mod.get_role(username)
|
||||
except KeyError:
|
||||
return JSONResponse(
|
||||
status_code=403,
|
||||
content={"detail": "user not in roles config"},
|
||||
)
|
||||
if _ADMIN_API_RE.match(path) and role != "admin":
|
||||
return JSONResponse(
|
||||
status_code=403,
|
||||
content={"detail": "admin only"},
|
||||
)
|
||||
# #R2-H3 mirror: enforce roles.yaml scope on non-bootstrap paths.
|
||||
if not path.startswith(_RBAC_BOOTSTRAP_EXEMPT):
|
||||
external_path = _EXTERNAL_PREFIX + path
|
||||
try:
|
||||
allowed = auth_mod.is_path_allowed(role, external_path)
|
||||
except Exception:
|
||||
allowed = True
|
||||
if not allowed:
|
||||
return JSONResponse(
|
||||
status_code=403,
|
||||
content={"detail": "forbidden for role"},
|
||||
)
|
||||
return await call_next(request)
|
||||
app.middleware("http")(rbac_guard)
|
||||
|
||||
app.include_router(me_router.router, prefix="/api/v1", tags=["me"])
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue