qzl
123 lines
3.8 KiB
Python
123 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from v1.auth.schemas import (
|
|
AuthUser,
|
|
OtpSendRequest,
|
|
PhoneSessionCreateRequest,
|
|
SessionRefreshRequest,
|
|
SessionResponse,
|
|
)
|
|
from v1.auth.service import AuthService, AuthServiceGateway
|
|
|
|
|
|
class FakeRegistrationBootstrapper:
|
|
def __init__(self) -> None:
|
|
self.called_user_ids: list[str] = []
|
|
|
|
async def ensure_user_automation_jobs(self, *, user_id: str) -> None:
|
|
self.called_user_ids.append(user_id)
|
|
|
|
|
|
class FakeGateway(AuthServiceGateway):
|
|
def __init__(self, response: SessionResponse) -> None:
|
|
self._response = response
|
|
self.last_send_otp_request: OtpSendRequest | None = None
|
|
self.last_phone_session_request: PhoneSessionCreateRequest | None = None
|
|
|
|
async def send_otp(self, request: OtpSendRequest) -> None:
|
|
self.last_send_otp_request = request
|
|
|
|
async def create_phone_session(
|
|
self, request: PhoneSessionCreateRequest
|
|
) -> SessionResponse:
|
|
self.last_phone_session_request = request
|
|
return self._response
|
|
|
|
async def refresh_session(self, request: SessionRefreshRequest) -> SessionResponse:
|
|
return self._response
|
|
|
|
async def delete_session(self, refresh_token: str | None) -> None:
|
|
return None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_otp_forwards_payload() -> None:
|
|
user = AuthUser(id="user-1", phone="+8613812345678")
|
|
token_response = SessionResponse(
|
|
access_token="access",
|
|
refresh_token="refresh",
|
|
expires_in=3600,
|
|
token_type="bearer",
|
|
user=user,
|
|
)
|
|
gateway = FakeGateway(token_response)
|
|
service = AuthService(gateway=gateway)
|
|
|
|
await service.send_otp(OtpSendRequest(phone="+8613812345678"))
|
|
|
|
assert gateway.last_send_otp_request is not None
|
|
assert gateway.last_send_otp_request.phone == "+8613812345678"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_phone_session_forwards_payload() -> None:
|
|
user = AuthUser(id="user-1", phone="+8613812345678")
|
|
token_response = SessionResponse(
|
|
access_token="access",
|
|
refresh_token="refresh",
|
|
expires_in=3600,
|
|
token_type="bearer",
|
|
user=user,
|
|
)
|
|
gateway = FakeGateway(token_response)
|
|
service = AuthService(gateway=gateway)
|
|
|
|
response = await service.create_phone_session(
|
|
PhoneSessionCreateRequest(phone="+8613812345678", token="123456")
|
|
)
|
|
|
|
assert gateway.last_phone_session_request is not None
|
|
assert gateway.last_phone_session_request.token == "123456"
|
|
assert response.user.phone == "+8613812345678"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_phone_session_bootstraps_automation_job() -> None:
|
|
user = AuthUser(id="b196f8be-c5f4-45d8-8f07-65c0ddf4d3de", phone="+8613812345678")
|
|
token_response = SessionResponse(
|
|
access_token="access",
|
|
refresh_token="refresh",
|
|
expires_in=3600,
|
|
token_type="bearer",
|
|
user=user,
|
|
)
|
|
gateway = FakeGateway(token_response)
|
|
bootstrapper = FakeRegistrationBootstrapper()
|
|
service = AuthService(gateway=gateway, registration_bootstrapper=bootstrapper)
|
|
|
|
await service.create_phone_session(
|
|
PhoneSessionCreateRequest(phone="+8613812345678", token="123456")
|
|
)
|
|
|
|
assert bootstrapper.called_user_ids == ["b196f8be-c5f4-45d8-8f07-65c0ddf4d3de"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_session_forwards_payload() -> None:
|
|
user = AuthUser(id="user-1", phone="+8613812345678")
|
|
token_response = SessionResponse(
|
|
access_token="access",
|
|
refresh_token="refresh",
|
|
expires_in=3600,
|
|
token_type="bearer",
|
|
user=user,
|
|
)
|
|
gateway = FakeGateway(token_response)
|
|
service = AuthService(gateway=gateway)
|
|
|
|
response = await service.refresh_session(SessionRefreshRequest(refresh_token="rt"))
|
|
|
|
assert response.access_token == "access"
|