8472d83ac3
Make the monthly estimate quota limit configurable via Settings.estimate_quota_limit (env ESTIMATE_QUOTA_LIMIT, default 15), replacing the hardcoded 15 in account_quota.py. Change exhausted-quota text (backend 429 detail + frontend banner) to 'За полной версией обращайтесь к Копылову'; banner uses dynamic limit. Closes #658
391 lines
14 KiB
Python
391 lines
14 KiB
Python
"""Tests for app.services.account_quota — monthly estimate quota enforcement.
|
||
|
||
Coverage:
|
||
(a) admin и kopylov unlimited — не блокируются, increment является no-op
|
||
(b) обычный pilot-юзер блокируется на 16-м запросе (429 + нужный detail)
|
||
(c) increment растит used счётчик
|
||
(d) get_status корректен для different сценариев
|
||
(e) отсутствие заголовка X-Authenticated-User = unlimited (fail-open)
|
||
|
||
DB мокируется через MagicMock — реальная БД не требуется.
|
||
"""
|
||
|
||
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
|
||
_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
|
||
|
||
from app.services.account_quota import ( # noqa: E402
|
||
LIMIT_EXHAUSTED_MESSAGE,
|
||
MONTHLY_LIMIT,
|
||
check_and_raise,
|
||
get_status,
|
||
increment,
|
||
is_unlimited,
|
||
)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _db_with_used(used: int) -> MagicMock:
|
||
"""DB session mock whose execute().fetchone() returns row with .used = N."""
|
||
row = MagicMock()
|
||
row.used = used
|
||
execute_result = MagicMock()
|
||
execute_result.fetchone.return_value = row
|
||
db = MagicMock()
|
||
db.execute.return_value = execute_result
|
||
return db
|
||
|
||
|
||
def _db_no_row() -> MagicMock:
|
||
"""DB session mock where no row exists yet (first estimate of the month)."""
|
||
execute_result = MagicMock()
|
||
execute_result.fetchone.return_value = None
|
||
db = MagicMock()
|
||
db.execute.return_value = execute_result
|
||
return db
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# (a) admin and kopylov are unlimited
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_is_unlimited_admin() -> None:
|
||
assert is_unlimited("admin") is True
|
||
|
||
|
||
def test_is_unlimited_kopylov() -> None:
|
||
assert is_unlimited("kopylov") is True
|
||
|
||
|
||
def test_is_unlimited_pilot_user1() -> None:
|
||
assert is_unlimited("user1") is False
|
||
|
||
|
||
def test_is_unlimited_unknown_user() -> None:
|
||
"""Неизвестный пользователь → False (KeyError трактуется как limited)."""
|
||
assert is_unlimited("ghost_unknown_xyz") is False
|
||
|
||
|
||
def test_check_and_raise_admin_not_blocked() -> None:
|
||
"""admin с used=15 не получает 429."""
|
||
db = _db_with_used(MONTHLY_LIMIT)
|
||
# Should not raise
|
||
check_and_raise(db, "admin")
|
||
|
||
|
||
def test_check_and_raise_kopylov_not_blocked() -> None:
|
||
"""kopylov с used=100 не получает 429."""
|
||
db = _db_with_used(100)
|
||
check_and_raise(db, "kopylov")
|
||
|
||
|
||
def test_increment_admin_is_noop() -> None:
|
||
"""increment для admin → никаких db.execute вызовов."""
|
||
db = MagicMock()
|
||
increment(db, "admin")
|
||
db.execute.assert_not_called()
|
||
db.commit.assert_not_called()
|
||
|
||
|
||
def test_increment_kopylov_is_noop() -> None:
|
||
"""increment для kopylov → no-op."""
|
||
db = MagicMock()
|
||
increment(db, "kopylov")
|
||
db.execute.assert_not_called()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# (b) pilot-юзер блокируется на 16-м запросе
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_check_and_raise_pilot_not_blocked_at_14() -> None:
|
||
"""used=14 < 15 → не блокируется."""
|
||
db = _db_with_used(14)
|
||
check_and_raise(db, "user1") # должно пройти без исключения
|
||
|
||
|
||
def test_check_and_raise_pilot_not_blocked_at_15_boundary() -> None:
|
||
"""used=15 == MONTHLY_LIMIT → 429 (15-я — последняя допустимая, 16-я блокируется).
|
||
|
||
Логика: used >= MONTHLY_LIMIT → block. После 15-й успешной оценки
|
||
increment делает used=15, поэтому следующий запрос (16-й) блокируется.
|
||
"""
|
||
db = _db_with_used(MONTHLY_LIMIT)
|
||
from fastapi import HTTPException
|
||
|
||
with pytest.raises(HTTPException) as exc_info:
|
||
check_and_raise(db, "user1")
|
||
assert exc_info.value.status_code == 429
|
||
assert exc_info.value.detail == LIMIT_EXHAUSTED_MESSAGE
|
||
|
||
|
||
def test_check_and_raise_pilot_blocked_exact_detail() -> None:
|
||
"""Проверяем точный текст сообщения 429."""
|
||
db = _db_with_used(MONTHLY_LIMIT)
|
||
from fastapi import HTTPException
|
||
|
||
with pytest.raises(HTTPException) as exc_info:
|
||
check_and_raise(db, "user3")
|
||
assert exc_info.value.detail == (
|
||
f"Лимит из {MONTHLY_LIMIT} оценок в этом месяце исчерпан. "
|
||
"За полной версией обращайтесь к Копылову."
|
||
)
|
||
|
||
|
||
def test_check_and_raise_pilot_blocked_over_limit() -> None:
|
||
"""used=20 тоже блокируется."""
|
||
db = _db_with_used(20)
|
||
from fastapi import HTTPException
|
||
|
||
with pytest.raises(HTTPException) as exc_info:
|
||
check_and_raise(db, "user2")
|
||
assert exc_info.value.status_code == 429
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# (c) increment растит used
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_increment_pilot_calls_upsert() -> None:
|
||
"""increment для pilot → выполняет db.execute (UPSERT) и db.commit."""
|
||
db = MagicMock()
|
||
increment(db, "user1")
|
||
db.execute.assert_called_once()
|
||
db.commit.assert_called_once()
|
||
|
||
# Проверяем что SQL содержит ON CONFLICT ... DO UPDATE
|
||
call_args = db.execute.call_args
|
||
sql_text = str(call_args[0][0]) # first positional arg — text() object
|
||
assert "ON CONFLICT" in sql_text
|
||
assert "used" in sql_text
|
||
|
||
|
||
def test_increment_none_username_is_noop() -> None:
|
||
"""None username → no-op."""
|
||
db = MagicMock()
|
||
increment(db, None)
|
||
db.execute.assert_not_called()
|
||
|
||
|
||
def test_increment_pilot_first_estimate_of_month() -> None:
|
||
"""Первый инкремент (нет строки в БД) — должен всё равно выполнить UPSERT."""
|
||
db = MagicMock()
|
||
increment(db, "user5")
|
||
db.execute.assert_called_once()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# (d) get_status корректен
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_get_status_none_username() -> None:
|
||
"""username None → unlimited, used=0, remaining=15."""
|
||
db = MagicMock()
|
||
status = get_status(db, None)
|
||
assert status["unlimited"] is True
|
||
assert status["used"] == 0
|
||
assert status["remaining"] == MONTHLY_LIMIT
|
||
assert status["limit"] == MONTHLY_LIMIT
|
||
db.execute.assert_not_called()
|
||
|
||
|
||
def test_get_status_admin() -> None:
|
||
"""admin → unlimited, remaining=15 вне зависимости от used."""
|
||
db = _db_with_used(7)
|
||
status = get_status(db, "admin")
|
||
assert status["unlimited"] is True
|
||
assert status["remaining"] == MONTHLY_LIMIT
|
||
assert status["used"] == 7 # фактический used из БД
|
||
|
||
|
||
def test_get_status_pilot_with_used() -> None:
|
||
"""pilot с used=10 → remaining=5."""
|
||
db = _db_with_used(10)
|
||
status = get_status(db, "user2")
|
||
assert status["unlimited"] is False
|
||
assert status["used"] == 10
|
||
assert status["remaining"] == 5
|
||
assert status["limit"] == MONTHLY_LIMIT
|
||
|
||
|
||
def test_get_status_pilot_no_row_yet() -> None:
|
||
"""Новый месяц — строки нет → used=0, remaining=15."""
|
||
db = _db_no_row()
|
||
status = get_status(db, "user3")
|
||
assert status["used"] == 0
|
||
assert status["remaining"] == MONTHLY_LIMIT
|
||
assert status["unlimited"] is False
|
||
|
||
|
||
def test_get_status_pilot_exhausted() -> None:
|
||
"""used=15 → remaining=0."""
|
||
db = _db_with_used(MONTHLY_LIMIT)
|
||
status = get_status(db, "user4")
|
||
assert status["remaining"] == 0
|
||
assert status["unlimited"] is False
|
||
|
||
|
||
def test_get_status_pilot_over_limit_remaining_zero() -> None:
|
||
"""used=20 → remaining=0 (не отрицательное)."""
|
||
db = _db_with_used(20)
|
||
status = get_status(db, "user5")
|
||
assert status["remaining"] == 0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# (e) отсутствие заголовка = unlimited (через /quota endpoint)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_check_and_raise_no_header_is_noop() -> None:
|
||
"""Без заголовка (None) → no-op, не поднимает HTTPException."""
|
||
db = MagicMock()
|
||
check_and_raise(db, None) # не должно поднять исключение
|
||
db.execute.assert_not_called()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Integration: /quota endpoint через TestClient
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture()
|
||
def quota_app() -> FastAPI:
|
||
"""Минимальное FastAPI app с /quota endpoint и мок-БД."""
|
||
from app.api.v1 import trade_in as trade_in_module
|
||
from app.core.db import get_db
|
||
|
||
application = FastAPI()
|
||
application.include_router(trade_in_module.router, prefix="/api/v1/trade-in")
|
||
|
||
def _override_db():
|
||
yield _db_no_row()
|
||
|
||
application.dependency_overrides[get_db] = _override_db
|
||
return application
|
||
|
||
|
||
def test_quota_endpoint_no_header(quota_app: FastAPI) -> None:
|
||
"""GET /quota без заголовка → unlimited=True, remaining=15."""
|
||
client = TestClient(quota_app)
|
||
resp = client.get("/api/v1/trade-in/quota")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["unlimited"] is True
|
||
assert data["remaining"] == MONTHLY_LIMIT
|
||
assert data["used"] == 0
|
||
|
||
|
||
def test_quota_endpoint_with_pilot_user(quota_app: FastAPI) -> None:
|
||
"""GET /quota с user1 (pilot, нет строки в БД) → unlimited=False, used=0."""
|
||
client = TestClient(quota_app)
|
||
resp = client.get(
|
||
"/api/v1/trade-in/quota",
|
||
headers={"X-Authenticated-User": "user1"},
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["unlimited"] is False
|
||
assert data["used"] == 0
|
||
assert data["remaining"] == MONTHLY_LIMIT
|
||
|
||
|
||
def test_quota_endpoint_admin_unlimited(quota_app: FastAPI) -> None:
|
||
"""GET /quota с admin → unlimited=True."""
|
||
client = TestClient(quota_app)
|
||
resp = client.get(
|
||
"/api/v1/trade-in/quota",
|
||
headers={"X-Authenticated-User": "admin"},
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["unlimited"] is True
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Integration: POST /estimate quota enforcement через TestClient
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture()
|
||
def estimate_app_exhausted() -> FastAPI:
|
||
"""FastAPI app где БД возвращает used=15 для pilot → 429 на POST /estimate."""
|
||
from app.api.v1 import trade_in as trade_in_module
|
||
from app.core.db import get_db
|
||
|
||
application = FastAPI()
|
||
application.include_router(trade_in_module.router, prefix="/api/v1/trade-in")
|
||
|
||
def _override_db():
|
||
yield _db_with_used(MONTHLY_LIMIT)
|
||
|
||
application.dependency_overrides[get_db] = _override_db
|
||
return application
|
||
|
||
|
||
def test_estimate_pilot_blocked_429(estimate_app_exhausted: FastAPI) -> None:
|
||
"""POST /estimate с user1 при used=15 → 429 с нужным detail."""
|
||
client = TestClient(estimate_app_exhausted, raise_server_exceptions=False)
|
||
resp = client.post(
|
||
"/api/v1/trade-in/estimate",
|
||
json={
|
||
"address": "г. Екатеринбург, ул. Малышева, 1",
|
||
"area_m2": 50.0,
|
||
"rooms": 2,
|
||
},
|
||
headers={"X-Authenticated-User": "user1"},
|
||
)
|
||
assert resp.status_code == 429
|
||
assert resp.json()["detail"] == LIMIT_EXHAUSTED_MESSAGE
|
||
|
||
|
||
def test_estimate_admin_not_blocked(estimate_app_exhausted: FastAPI) -> None:
|
||
"""POST /estimate с admin при used=15 → не 429 (blocked by quota), но может падать
|
||
с другой ошибкой (нет реального estimator). Ключевое: не 429."""
|
||
client = TestClient(estimate_app_exhausted, raise_server_exceptions=False)
|
||
resp = client.post(
|
||
"/api/v1/trade-in/estimate",
|
||
json={
|
||
"address": "г. Екатеринбург, ул. Малышева, 1",
|
||
"area_m2": 50.0,
|
||
"rooms": 2,
|
||
},
|
||
headers={"X-Authenticated-User": "admin"},
|
||
)
|
||
assert resp.status_code != 429
|
||
|
||
|
||
def test_estimate_no_header_not_blocked(estimate_app_exhausted: FastAPI) -> None:
|
||
"""POST /estimate без заголовка → не 429 (fail-open, dev-режим)."""
|
||
client = TestClient(estimate_app_exhausted, raise_server_exceptions=False)
|
||
resp = client.post(
|
||
"/api/v1/trade-in/estimate",
|
||
json={
|
||
"address": "г. Екатеринбург, ул. Малышева, 1",
|
||
"area_m2": 50.0,
|
||
"rooms": 2,
|
||
},
|
||
)
|
||
assert resp.status_code != 429
|