72 lines
2.0 KiB
Python
72 lines
2.0 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from types import SimpleNamespace
|
||
|
|
from typing import Any, cast
|
||
|
|
from uuid import uuid4
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from v1.auth.schemas import UserByIdResponse
|
||
|
|
from v1.users.contact_resolver import resolve_contacts_by_user_ids
|
||
|
|
|
||
|
|
|
||
|
|
class _AuthGatewayStub:
|
||
|
|
def __init__(self, responses: dict[str, UserByIdResponse]) -> None:
|
||
|
|
self._responses = responses
|
||
|
|
|
||
|
|
async def get_user_by_id(self, user_id: str) -> UserByIdResponse:
|
||
|
|
response = self._responses.get(user_id)
|
||
|
|
if response is None:
|
||
|
|
raise RuntimeError("missing")
|
||
|
|
return response
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_resolve_contacts_by_user_ids_builds_contact_map() -> None:
|
||
|
|
user_id = uuid4()
|
||
|
|
profile = SimpleNamespace(
|
||
|
|
id=user_id,
|
||
|
|
username="alice",
|
||
|
|
avatar_url="https://img.example/a.png",
|
||
|
|
)
|
||
|
|
gateway = _AuthGatewayStub(
|
||
|
|
{
|
||
|
|
str(user_id): UserByIdResponse(
|
||
|
|
id=str(user_id),
|
||
|
|
phone="+8613900001001",
|
||
|
|
created_at="2026-01-01T00:00:00Z",
|
||
|
|
phone_confirmed_at=None,
|
||
|
|
)
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
contacts = await resolve_contacts_by_user_ids(
|
||
|
|
user_ids=[user_id],
|
||
|
|
profiles_by_id=cast(dict[Any, Any], {user_id: profile}),
|
||
|
|
auth_gateway=gateway,
|
||
|
|
)
|
||
|
|
|
||
|
|
assert str(contacts[user_id].user_id) == str(user_id)
|
||
|
|
assert contacts[user_id].username == "alice"
|
||
|
|
assert contacts[user_id].avatar_url == "https://img.example/a.png"
|
||
|
|
assert contacts[user_id].phone == "+8613900001001"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_resolve_contacts_by_user_ids_keeps_profile_on_auth_failure() -> None:
|
||
|
|
user_id = uuid4()
|
||
|
|
profile = SimpleNamespace(
|
||
|
|
id=user_id,
|
||
|
|
username="bob",
|
||
|
|
avatar_url=None,
|
||
|
|
)
|
||
|
|
gateway = _AuthGatewayStub({})
|
||
|
|
|
||
|
|
contacts = await resolve_contacts_by_user_ids(
|
||
|
|
user_ids=[user_id],
|
||
|
|
profiles_by_id=cast(dict[Any, Any], {user_id: profile}),
|
||
|
|
auth_gateway=gateway,
|
||
|
|
)
|
||
|
|
|
||
|
|
assert contacts[user_id].username == "bob"
|
||
|
|
assert contacts[user_id].phone is None
|