qzl
105cf82d21
- 恢复 CelerySettings 和相关计算属性 - 修复 celery/app.py 调用 configure_celery_app 参数 - 创建 core/initialization/init_data.py stub - 删除不完整的 test_auth_supabase_gateway.py
75 lines
2.2 KiB
Python
75 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from v1.auth.schemas import (
|
|
AuthTokenResponse,
|
|
AuthUser,
|
|
LoginRequest,
|
|
RefreshRequest,
|
|
SignupRequest,
|
|
)
|
|
from v1.auth.service import AuthService, AuthServiceGateway
|
|
|
|
|
|
class FakeGateway(AuthServiceGateway):
|
|
def __init__(self, response: AuthTokenResponse) -> None:
|
|
self._response = response
|
|
|
|
async def signup(self, request: SignupRequest) -> AuthTokenResponse:
|
|
return self._response
|
|
|
|
async def login(self, request: LoginRequest) -> AuthTokenResponse:
|
|
return self._response
|
|
|
|
async def refresh(self, request: RefreshRequest) -> AuthTokenResponse:
|
|
return self._response
|
|
|
|
async def logout(self, refresh_token: str | None) -> None:
|
|
return None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_signup_maps_response() -> None:
|
|
user = AuthUser(id="user-1", email="user@example.com")
|
|
token_response = AuthTokenResponse(
|
|
access_token="access",
|
|
refresh_token="refresh",
|
|
expires_in=3600,
|
|
token_type="bearer",
|
|
user=user,
|
|
)
|
|
service = AuthService(gateway=FakeGateway(token_response))
|
|
|
|
result = await service.signup(
|
|
SignupRequest(email="user@example.com", password="secret123")
|
|
)
|
|
|
|
assert result.access_token == "access"
|
|
assert result.refresh_token == "refresh"
|
|
assert result.user.id == "user-1"
|
|
|
|
|
|
class LogoutAssertingGateway(AuthServiceGateway):
|
|
def __init__(self, expected_refresh_token: str) -> None:
|
|
self._expected_refresh_token = expected_refresh_token
|
|
|
|
async def signup(self, request: SignupRequest) -> AuthTokenResponse:
|
|
raise NotImplementedError
|
|
|
|
async def login(self, request: LoginRequest) -> AuthTokenResponse:
|
|
raise NotImplementedError
|
|
|
|
async def refresh(self, request: RefreshRequest) -> AuthTokenResponse:
|
|
raise NotImplementedError
|
|
|
|
async def logout(self, refresh_token: str | None) -> None:
|
|
assert refresh_token == self._expected_refresh_token
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_logout_forwards_refresh_token() -> None:
|
|
service = AuthService(gateway=LogoutAssertingGateway("refresh-token"))
|
|
|
|
await service.logout("refresh-token")
|