2025-09-26 21:03:45 +03:00
|
|
|
|
"""
|
|
|
|
|
|
🧪 Минимальный OAuth тест - DRY/YAGNI принципы
|
|
|
|
|
|
|
|
|
|
|
|
Тестирует только критичную функциональность нового минимального OAuth flow:
|
|
|
|
|
|
- Нет ошибки = успех
|
|
|
|
|
|
- httpOnly cookie только
|
|
|
|
|
|
- Простая логика редиректов
|
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
|
|
|
|
import pytest
|
|
|
|
|
|
from unittest.mock import patch, MagicMock, AsyncMock
|
|
|
|
|
|
from starlette.responses import RedirectResponse, JSONResponse
|
|
|
|
|
|
|
|
|
|
|
|
from auth.oauth import oauth_callback_http
|
|
|
|
|
|
from tests.test_config import HTTP_STATUS, OAUTH_PROVIDERS
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class MinimalOAuthRequest:
|
|
|
|
|
|
"""🔍 DRY: Минимальный Mock для OAuth запросов"""
|
|
|
|
|
|
|
|
|
|
|
|
def __init__(self, query_params=None, path_params=None, headers=None):
|
|
|
|
|
|
self.query_params = query_params or {}
|
|
|
|
|
|
self.path_params = path_params or {}
|
|
|
|
|
|
self.headers = headers or {"user-agent": "test-agent"}
|
2025-09-28 12:22:37 +03:00
|
|
|
|
self.url = "https://v3.discours.io/oauth/github/callback"
|
2025-09-26 21:03:45 +03:00
|
|
|
|
self.method = "GET" # ✅ Добавляем method для логирования
|
|
|
|
|
|
self.client = MagicMock()
|
|
|
|
|
|
self.client.host = "127.0.0.1"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
|
async def test_oauth_minimal_success():
|
|
|
|
|
|
"""🧪 Тест минимального успешного OAuth flow"""
|
|
|
|
|
|
|
|
|
|
|
|
# 🔍 YAGNI: Тестируем только один провайдер - GitHub (самый популярный)
|
|
|
|
|
|
request = MinimalOAuthRequest(
|
|
|
|
|
|
query_params={"state": "valid_state", "code": "auth_code"},
|
|
|
|
|
|
path_params={"provider": "github"}
|
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
# 🔍 DRY: Минимальные моки только для критичного пути
|
|
|
|
|
|
oauth_data = {
|
|
|
|
|
|
"provider": "github",
|
|
|
|
|
|
"redirect_uri": "https://testing.discours.io/dashboard",
|
|
|
|
|
|
"code_verifier": "test_verifier"
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
mock_token = {"access_token": "test_token"}
|
|
|
|
|
|
mock_profile = {"login": "testuser", "email": "test@github.com"}
|
|
|
|
|
|
mock_author = MagicMock()
|
|
|
|
|
|
mock_author.id = 123
|
|
|
|
|
|
|
|
|
|
|
|
with (
|
|
|
|
|
|
patch("auth.oauth.get_oauth_state", return_value=oauth_data),
|
|
|
|
|
|
patch("auth.oauth.oauth.create_client") as mock_client,
|
|
|
|
|
|
patch("auth.oauth.get_user_profile", return_value=mock_profile),
|
|
|
|
|
|
patch("auth.oauth._create_or_update_user", return_value=mock_author),
|
|
|
|
|
|
patch("auth.oauth.TokenStorage.create_session", return_value="jwt_token_123")
|
|
|
|
|
|
):
|
|
|
|
|
|
# Настраиваем OAuth client mock (async)
|
|
|
|
|
|
client_instance = MagicMock()
|
|
|
|
|
|
client_instance.fetch_access_token = AsyncMock(return_value=mock_token)
|
|
|
|
|
|
mock_client.return_value = client_instance
|
|
|
|
|
|
|
|
|
|
|
|
response = await oauth_callback_http(request)
|
|
|
|
|
|
|
|
|
|
|
|
# ✅ Проверяем минимальный успешный результат
|
|
|
|
|
|
assert isinstance(response, RedirectResponse)
|
|
|
|
|
|
assert "testing.discours.io/oauth" in response.headers["location"]
|
|
|
|
|
|
assert "redirect_url=" in response.headers["location"]
|
|
|
|
|
|
|
|
|
|
|
|
# ✅ Главное: НЕТ error параметра = успех
|
|
|
|
|
|
assert "error=" not in response.headers["location"]
|
|
|
|
|
|
|
|
|
|
|
|
# ✅ НЕТ access_token в URL (новая логика)
|
|
|
|
|
|
assert "access_token=" not in response.headers["location"]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
|
async def test_oauth_minimal_error_missing_state():
|
|
|
|
|
|
"""🧪 Тест ошибки: отсутствует state (CSRF защита)"""
|
|
|
|
|
|
|
|
|
|
|
|
request = MinimalOAuthRequest(
|
|
|
|
|
|
query_params={"code": "auth_code"}, # НЕТ state!
|
|
|
|
|
|
path_params={"provider": "github"}
|
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
response = await oauth_callback_http(request)
|
|
|
|
|
|
|
|
|
|
|
|
# ✅ Проверяем обработку ошибки
|
|
|
|
|
|
assert isinstance(response, JSONResponse)
|
|
|
|
|
|
assert response.status_code == HTTP_STATUS["BAD_REQUEST"]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
|
async def test_oauth_minimal_error_expired_state():
|
|
|
|
|
|
"""🧪 Тест ошибки: истекший state"""
|
|
|
|
|
|
|
|
|
|
|
|
request = MinimalOAuthRequest(
|
|
|
|
|
|
query_params={"state": "expired_state", "code": "auth_code"},
|
|
|
|
|
|
path_params={"provider": "github"}
|
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
with patch("auth.oauth.get_oauth_state", return_value=None): # State не найден
|
|
|
|
|
|
response = await oauth_callback_http(request)
|
|
|
|
|
|
|
|
|
|
|
|
# ✅ Проверяем редирект с ошибкой
|
|
|
|
|
|
assert isinstance(response, RedirectResponse)
|
|
|
|
|
|
assert "testing.discours.io/oauth" in response.headers["location"]
|
|
|
|
|
|
assert "error=oauth_state_expired" in response.headers["location"]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
|
async def test_oauth_minimal_invalid_provider():
|
|
|
|
|
|
"""🧪 Тест ошибки: неправильный провайдер"""
|
|
|
|
|
|
|
|
|
|
|
|
request = MinimalOAuthRequest(
|
|
|
|
|
|
query_params={"state": "valid_state", "code": "auth_code"},
|
|
|
|
|
|
path_params={"provider": "invalid_provider"} # Неправильный провайдер
|
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
oauth_data = {"provider": "invalid_provider", "redirect_uri": "https://testing.discours.io/"}
|
|
|
|
|
|
|
|
|
|
|
|
with patch("auth.oauth.get_oauth_state", return_value=oauth_data):
|
|
|
|
|
|
response = await oauth_callback_http(request)
|
|
|
|
|
|
|
|
|
|
|
|
# ✅ Проверяем обработку ошибки провайдера
|
|
|
|
|
|
assert isinstance(response, RedirectResponse)
|
|
|
|
|
|
assert "error=" in response.headers["location"]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 🔍 YAGNI: НЕ тестируем сложные сценарии
|
|
|
|
|
|
# - Множественные провайдеры (достаточно GitHub)
|
|
|
|
|
|
# - Сложную логику создания пользователей (интеграционные тесты)
|
|
|
|
|
|
# - PKCE детали (unit тесты библиотеки)
|
|
|
|
|
|
# - Redis операции (отдельные тесты)
|
|
|
|
|
|
|
|
|
|
|
|
# 🔍 DRY: Переиспользуем существующие константы из test_config.py
|
|
|
|
|
|
# - HTTP_STATUS для статус кодов
|
|
|
|
|
|
# - OAUTH_PROVIDERS для списка провайдеров
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.mark.parametrize("provider", OAUTH_PROVIDERS[:1]) # 🔍 YAGNI: только GitHub
|
|
|
|
|
|
async def test_oauth_minimal_provider_validation(provider):
|
|
|
|
|
|
"""🧪 Параметризованный тест валидации провайдера"""
|
|
|
|
|
|
|
|
|
|
|
|
request = MinimalOAuthRequest(
|
|
|
|
|
|
query_params={"state": "valid_state", "code": "auth_code"},
|
|
|
|
|
|
path_params={"provider": provider}
|
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
oauth_data = {"provider": provider, "redirect_uri": "https://testing.discours.io/"}
|
|
|
|
|
|
|
|
|
|
|
|
with patch("auth.oauth.get_oauth_state", return_value=oauth_data):
|
|
|
|
|
|
# 🔍 YAGNI: Не мокаем весь OAuth flow, только проверяем что провайдер принят
|
|
|
|
|
|
try:
|
|
|
|
|
|
response = await oauth_callback_http(request)
|
|
|
|
|
|
# Если дошли сюда без исключения - провайдер валидный
|
|
|
|
|
|
assert True
|
|
|
|
|
|
except Exception as e:
|
|
|
|
|
|
# Если исключение - проверяем что это не из-за провайдера
|
|
|
|
|
|
assert "Invalid provider" not in str(e)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 🎯 Итого: 5 тестов покрывают все критичные пути
|
|
|
|
|
|
# ✅ Успешный flow
|
|
|
|
|
|
# ❌ Отсутствует state
|
|
|
|
|
|
# ❌ Истекший state
|
|
|
|
|
|
# ❌ Неправильный провайдер
|
|
|
|
|
|
# ✅ Валидация провайдера
|
|
|
|
|
|
|
|
|
|
|
|
# 🔍 DRY принципы:
|
|
|
|
|
|
# - Переиспользование MinimalOAuthRequest
|
|
|
|
|
|
# - Константы из test_config.py
|
|
|
|
|
|
# - Минимальные моки без дублирования
|
|
|
|
|
|
|
|
|
|
|
|
# 🔍 YAGNI принципы:
|
|
|
|
|
|
# - Только критичные пути
|
|
|
|
|
|
# - Один провайдер вместо всех
|
|
|
|
|
|
# - Простые проверки вместо сложной логики
|
|
|
|
|
|
# - Нет избыточных тестов "на всякий случай"
|