63 lines
1.9 KiB
Python
63 lines
1.9 KiB
Python
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from fastapi_oidc_op.models import User
|
|
from fastapi_oidc_op.userid import generate_unique_userid, generate_userid
|
|
|
|
|
|
def test_generate_userid_format() -> None:
|
|
userid = generate_userid()
|
|
# 32-bit proquint format: xxxxx-xxxxx
|
|
parts = userid.split("-")
|
|
assert len(parts) == 2
|
|
for part in parts:
|
|
assert len(part) == 5
|
|
|
|
|
|
def test_generate_userid_uniqueness() -> None:
|
|
ids = {generate_userid() for _ in range(100)}
|
|
assert len(ids) == 100 # All unique
|
|
|
|
|
|
def test_generate_userid_is_lowercase() -> None:
|
|
userid = generate_userid()
|
|
assert userid == userid.lower()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_generate_unique_userid_no_collision() -> None:
|
|
"""Returns a userid when no collision occurs."""
|
|
user_repo = AsyncMock()
|
|
user_repo.get_by_userid.return_value = None
|
|
|
|
result = await generate_unique_userid(user_repo)
|
|
|
|
parts = result.split("-")
|
|
assert len(parts) == 2
|
|
user_repo.get_by_userid.assert_called_once_with(result)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_generate_unique_userid_retries_on_collision() -> None:
|
|
"""Retries when a collision is detected and succeeds on second attempt."""
|
|
existing_user = User(userid="taken-xxxxx", username="existing")
|
|
user_repo = AsyncMock()
|
|
user_repo.get_by_userid.side_effect = [existing_user, None]
|
|
|
|
result = await generate_unique_userid(user_repo)
|
|
|
|
assert user_repo.get_by_userid.call_count == 2
|
|
parts = result.split("-")
|
|
assert len(parts) == 2
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_generate_unique_userid_raises_after_max_retries() -> None:
|
|
"""Raises RuntimeError when max retries are exhausted."""
|
|
existing_user = User(userid="taken-xxxxx", username="existing")
|
|
user_repo = AsyncMock()
|
|
user_repo.get_by_userid.return_value = existing_user
|
|
|
|
with pytest.raises(RuntimeError, match="Failed to generate unique userid"):
|
|
await generate_unique_userid(user_repo)
|