291 lines
12 KiB
Python
291 lines
12 KiB
Python
"""
|
||
Функциональные тесты OAuth без SessionMiddleware
|
||
"""
|
||
|
||
import pytest
|
||
from unittest.mock import AsyncMock, MagicMock, patch, ANY, call
|
||
from starlette.requests import Request
|
||
from starlette.responses import RedirectResponse, JSONResponse
|
||
|
||
from auth.oauth import oauth_login_http, oauth_callback_http
|
||
|
||
|
||
class TestOAuthFunctional:
|
||
"""Функциональные тесты OAuth системы"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_oauth_provider_not_configured_error(self):
|
||
"""Тест ошибки когда провайдер не настроен"""
|
||
|
||
# Создаем мок запроса
|
||
request = MagicMock(spec=Request)
|
||
request.path_params = {"provider": "google"}
|
||
|
||
# Мокаем что провайдер не найден
|
||
with patch("auth.oauth.oauth.create_client", return_value=None):
|
||
response = await oauth_login_http(request)
|
||
|
||
assert isinstance(response, JSONResponse)
|
||
assert response.status_code == 400
|
||
|
||
# Проверяем содержимое ответа
|
||
body = response.body
|
||
if isinstance(body, memoryview):
|
||
body = bytes(body)
|
||
assert b"Provider not configured" in body
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_oauth_login_success_session_free(self):
|
||
"""Тест успешного OAuth login без использования сессий"""
|
||
|
||
# Создаем мок запроса
|
||
request = MagicMock(spec=Request)
|
||
request.path_params = {"provider": "google"}
|
||
|
||
# Мокаем OAuth клиент
|
||
mock_client = AsyncMock()
|
||
mock_client.create_authorization_url = AsyncMock(return_value={
|
||
"url": "https://accounts.google.com/oauth/authorize?client_id=test&state=abc123"
|
||
})
|
||
|
||
with patch("auth.oauth.oauth.create_client", return_value=mock_client), \
|
||
patch("auth.oauth.store_oauth_state") as mock_store:
|
||
|
||
response = await oauth_login_http(request)
|
||
|
||
# Проверяем что это редирект
|
||
assert isinstance(response, RedirectResponse)
|
||
assert response.status_code == 302
|
||
|
||
# Проверяем что состояние сохранено в Redis
|
||
mock_store.assert_called_once()
|
||
|
||
# Проверяем что URL создан правильно
|
||
mock_client.create_authorization_url.assert_called_once()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_oauth_callback_missing_state(self):
|
||
"""Тест callback без state параметра"""
|
||
|
||
request = MagicMock(spec=Request)
|
||
request.query_params = {} # Нет state
|
||
|
||
response = await oauth_callback_http(request)
|
||
|
||
assert isinstance(response, JSONResponse)
|
||
assert response.status_code == 400
|
||
|
||
body = response.body
|
||
if isinstance(body, memoryview):
|
||
body = bytes(body)
|
||
assert b"Missing OAuth state parameter" in body
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_oauth_callback_invalid_state(self):
|
||
"""Тест callback с неправильным state"""
|
||
|
||
request = MagicMock(spec=Request)
|
||
request.query_params = {"state": "invalid_state"}
|
||
|
||
# Мокаем что состояние не найдено в Redis
|
||
with patch("auth.oauth.get_oauth_state", return_value=None):
|
||
response = await oauth_callback_http(request)
|
||
|
||
assert isinstance(response, JSONResponse)
|
||
assert response.status_code == 400
|
||
|
||
body = response.body
|
||
if isinstance(body, memoryview):
|
||
body = bytes(body)
|
||
assert b"oauth_state_expired" in body # 🔍 Проверяем новый формат ошибки
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_oauth_callback_success_session_free(self):
|
||
"""Тест успешного OAuth callback без сессий"""
|
||
|
||
# Настраиваем мок запроса
|
||
request = MagicMock(spec=Request)
|
||
request.query_params = {
|
||
"state": "valid_state",
|
||
"code": "auth_code_123"
|
||
}
|
||
request.path_params = {"provider": "google"} # Для callback используем path_params
|
||
request.url = "https://localhost:3000/oauth/google/callback?state=valid_state&code=auth_code_123"
|
||
request.headers = {"user-agent": "test-browser"}
|
||
request.client = MagicMock()
|
||
request.client.host = "127.0.0.1"
|
||
|
||
# Мокаем OAuth данные из Redis
|
||
oauth_data = {
|
||
"provider": "google",
|
||
"code_verifier": "test_verifier_123",
|
||
"redirect_uri": "https://localhost:3000"
|
||
}
|
||
|
||
# Мокаем токен от провайдера
|
||
mock_token = {
|
||
"access_token": "access_token_123",
|
||
"userinfo": {
|
||
"sub": "google_user_123",
|
||
"email": "test@gmail.com",
|
||
"name": "Test User",
|
||
"picture": "https://example.com/photo.jpg"
|
||
}
|
||
}
|
||
|
||
# Мокаем профиль пользователя
|
||
user_profile = {
|
||
"id": "google_user_123",
|
||
"email": "test@gmail.com",
|
||
"name": "Test User",
|
||
"picture": "https://example.com/photo.jpg"
|
||
}
|
||
|
||
# Мокаем пользователя
|
||
mock_user = MagicMock()
|
||
mock_user.id = 123
|
||
mock_user.name = "Test User"
|
||
|
||
with patch("auth.oauth.get_oauth_state", return_value=oauth_data), \
|
||
patch("auth.oauth.oauth.create_client") as mock_create_client, \
|
||
patch("auth.oauth.get_user_profile", return_value=user_profile), \
|
||
patch("auth.oauth._create_or_update_user", return_value=mock_user), \
|
||
patch("auth.tokens.storage.TokenStorage.create_session", return_value="session_token_123"):
|
||
|
||
# Настраиваем мок OAuth клиента
|
||
mock_client = AsyncMock()
|
||
mock_client.fetch_access_token = AsyncMock(return_value=mock_token)
|
||
mock_create_client.return_value = mock_client
|
||
|
||
response = await oauth_callback_http(request)
|
||
|
||
# Проверяем что это редирект
|
||
assert isinstance(response, RedirectResponse)
|
||
assert response.status_code == 307
|
||
|
||
# Проверяем URL редиректа (теперь с параметрами)
|
||
redirect_url = response.headers["location"]
|
||
assert redirect_url.startswith("https://localhost:3000")
|
||
assert "access_token=" in redirect_url # 🔍 Проверяем наличие токена
|
||
assert "state=valid_state" in redirect_url # 🔍 Проверяем state
|
||
|
||
# Проверяем что токен получен без использования request.session
|
||
mock_client.fetch_access_token.assert_called_once_with(
|
||
authorization_response=str(request.url),
|
||
code_verifier="test_verifier_123"
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_oauth_redis_state_lifecycle(self):
|
||
"""Тест жизненного цикла OAuth состояния в Redis"""
|
||
from auth.oauth import store_oauth_state, get_oauth_state
|
||
|
||
test_state = "state_lifecycle_test"
|
||
test_data = {
|
||
"provider": "github",
|
||
"code_verifier": "verifier_123",
|
||
"redirect_uri": "https://localhost:3000",
|
||
"created_at": 1234567890
|
||
}
|
||
|
||
# Мокаем Redis операции
|
||
with patch("auth.oauth.redis") as mock_redis:
|
||
# Тест сохранения
|
||
await store_oauth_state(test_state, test_data)
|
||
|
||
# Проверяем что SETEX вызван с правильными параметрами
|
||
mock_redis.execute.assert_called_with(
|
||
"SETEX",
|
||
f"oauth_state:{test_state}",
|
||
600, # TTL 10 минут
|
||
ANY # Сериализованные данные
|
||
)
|
||
|
||
# Тест получения и удаления (one-time use)
|
||
import orjson
|
||
mock_redis.execute.side_effect = [
|
||
orjson.dumps(test_data), # GET возвращает данные
|
||
None # DEL подтверждение
|
||
]
|
||
|
||
result = await get_oauth_state(test_state)
|
||
|
||
# Проверяем что данные корректно получены
|
||
assert result == test_data
|
||
|
||
# Проверяем что GET и DEL были вызваны
|
||
assert mock_redis.execute.call_count == 3 # SETEX + GET + DEL
|
||
|
||
# Проверяем что последние вызовы были GET и DEL
|
||
calls = mock_redis.execute.call_args_list
|
||
assert calls[-2][0][0] == "GET" # Предпоследний вызов - GET
|
||
assert calls[-1][0][0] == "DEL" # Последний вызов - DEL
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_oauth_error_handling_comprehensive(self):
|
||
"""Комплексный тест обработки ошибок OAuth"""
|
||
|
||
# Тест 1: Неправильный провайдер
|
||
request = MagicMock(spec=Request)
|
||
request.path_params = {"provider": "invalid_provider"}
|
||
|
||
response = await oauth_login_http(request)
|
||
assert isinstance(response, JSONResponse)
|
||
assert response.status_code == 400
|
||
|
||
# Тест 2: Провайдер не настроен
|
||
request.path_params = {"provider": "google"}
|
||
with patch("auth.oauth.oauth.create_client", return_value=None):
|
||
response = await oauth_login_http(request)
|
||
assert isinstance(response, JSONResponse)
|
||
assert response.status_code == 400
|
||
|
||
# Тест 3: Callback без code (но с правильным OAuth клиентом)
|
||
request.query_params = {"state": "valid_state"}
|
||
request.path_params = {"provider": "google"} # Для callback используем path_params
|
||
oauth_data = {"provider": "google", "code_verifier": "test"}
|
||
|
||
mock_client = AsyncMock()
|
||
with patch("auth.oauth.get_oauth_state", return_value=oauth_data), \
|
||
patch("auth.oauth.oauth.create_client", return_value=mock_client):
|
||
response = await oauth_callback_http(request)
|
||
assert isinstance(response, JSONResponse)
|
||
assert response.status_code == 400
|
||
|
||
body = response.body
|
||
if isinstance(body, memoryview):
|
||
body = bytes(body)
|
||
assert b"Missing authorization code" in body
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_vk_oauth_without_pkce(self):
|
||
"""Тест VK OAuth без PKCE (VK не поддерживает code_challenge)"""
|
||
|
||
request = MagicMock(spec=Request)
|
||
request.path_params = {"provider": "vk"}
|
||
|
||
mock_client = AsyncMock()
|
||
# VK должен вызываться без code_challenge
|
||
mock_client.create_authorization_url = AsyncMock(return_value={
|
||
"url": "https://oauth.vk.com/authorize?client_id=test&state=abc123"
|
||
})
|
||
|
||
with patch("auth.oauth.oauth.create_client", return_value=mock_client), \
|
||
patch("auth.oauth.store_oauth_state") as mock_store:
|
||
|
||
response = await oauth_login_http(request)
|
||
|
||
assert isinstance(response, RedirectResponse)
|
||
assert response.status_code == 302
|
||
|
||
# Проверяем что create_authorization_url вызван БЕЗ code_challenge для VK
|
||
call_args = mock_client.create_authorization_url.call_args
|
||
assert "code_challenge" not in call_args.kwargs
|
||
assert "code_challenge_method" not in call_args.kwargs
|
||
assert "state" in call_args.kwargs
|
||
|
||
|
||
|
||
if __name__ == "__main__":
|
||
pytest.main([__file__, "-v"])
|