87 lines
2.8 KiB
Python
87 lines
2.8 KiB
Python
"""Тесты RBAC-зависимостей `api/deps.py`: user / guest / admin.
|
|
|
|
Зависимости вызываются напрямую (минуя HTTP), т.к. отдельного
|
|
admin-only эндпоинта нет; `get_current_user` дополнительно
|
|
покрыт HTTP-тестом в `test_auth.py` (закрытый эндпоинт без JWT → 401).
|
|
"""
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from api.deps import get_current_user, get_current_user_optional, require_admin
|
|
from core.security import create_access_token, hash_password
|
|
from models.user import User
|
|
|
|
|
|
async def _make_user(session: AsyncSession, *, role: str = "user") -> User:
|
|
user = User(
|
|
email=f"{uuid.uuid4()}@example.com",
|
|
name_user="Test User",
|
|
password_hash=hash_password("password123"),
|
|
role=role,
|
|
email_verified=True,
|
|
)
|
|
session.add(user)
|
|
await session.flush()
|
|
return user
|
|
|
|
|
|
async def test_get_current_user_optional_returns_none_without_token(
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
"""Отсутствие токена трактуется как guest (`None`), без ошибки."""
|
|
assert await get_current_user_optional(None, db_session) is None
|
|
|
|
|
|
async def test_get_current_user_raises_401_without_token(db_session: AsyncSession) -> None:
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await get_current_user(None, db_session)
|
|
assert exc_info.value.status_code == 401
|
|
|
|
|
|
async def test_get_current_user_rejects_garbage_token(db_session: AsyncSession) -> None:
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await get_current_user("not-a-jwt", db_session)
|
|
assert exc_info.value.status_code == 401
|
|
|
|
|
|
async def test_get_current_user_resolves_valid_access_token(db_session: AsyncSession) -> None:
|
|
user = await _make_user(db_session)
|
|
token = create_access_token(user.id, user.role)
|
|
|
|
resolved = await get_current_user(token, db_session)
|
|
|
|
assert resolved.id == user.id
|
|
|
|
|
|
async def test_get_current_user_optional_resolves_valid_access_token(
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
user = await _make_user(db_session)
|
|
token = create_access_token(user.id, user.role)
|
|
|
|
resolved = await get_current_user_optional(token, db_session)
|
|
|
|
assert resolved is not None
|
|
assert resolved.id == user.id
|
|
|
|
|
|
async def test_require_admin_passes_for_admin(db_session: AsyncSession) -> None:
|
|
admin = await _make_user(db_session, role="admin")
|
|
|
|
resolved = await require_admin(admin)
|
|
|
|
assert resolved.id == admin.id
|
|
|
|
|
|
async def test_require_admin_forbidden_for_regular_user(db_session: AsyncSession) -> None:
|
|
user = await _make_user(db_session, role="user")
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await require_admin(user)
|
|
|
|
assert exc_info.value.status_code == 403
|