qzl
f0af44d840
- Update agent router/service/repository with new endpoints - Update auth routes with phone-based authentication - Update users service with new phone lookup - Update schedule_items with new schemas - Update message schemas with visibility support - Update settings with new automation scheduler config - Update CLI with new commands - Update tests to match new API contracts
94 lines
2.8 KiB
Python
94 lines
2.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 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_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"
|