108 lines
3.3 KiB
Python
108 lines
3.3 KiB
Python
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from src.core.security import create_refresh_token, verify_token
|
|
from src.services.auth_service import AuthService
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_authenticate_user_returns_token(monkeypatch):
|
|
db = AsyncMock()
|
|
service = AuthService(db)
|
|
service.user_repo = MagicMock()
|
|
service.user_repo.authenticate = AsyncMock(
|
|
return_value=MagicMock(username="admin", email="admin@example.com")
|
|
)
|
|
|
|
token = await service.authenticate_user("admin", "admin")
|
|
|
|
assert token is not None
|
|
assert token.access_token
|
|
assert token.refresh_token
|
|
assert token.token_type == "bearer"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_token_returns_none_for_invalid():
|
|
db = AsyncMock()
|
|
service = AuthService(db)
|
|
|
|
token = await service.refresh_token("bad-token")
|
|
|
|
assert token is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_token_resolves_user_by_email():
|
|
db = AsyncMock()
|
|
service = AuthService(db)
|
|
service.user_repo = MagicMock()
|
|
service.user_repo.get_by_username = AsyncMock(return_value=None)
|
|
user = MagicMock(username="admin", email="admin@example.com", is_active=True)
|
|
service.user_repo.get_by_email = AsyncMock(return_value=user)
|
|
|
|
token = await service.refresh_token(
|
|
create_refresh_token(data={"sub": "admin@example.com"})
|
|
)
|
|
|
|
assert token is not None
|
|
assert token.token_type == "bearer"
|
|
payload = verify_token(token.refresh_token)
|
|
assert payload is not None
|
|
assert payload["sub"] == "admin@example.com"
|
|
service.user_repo.get_by_email.assert_awaited_once_with("admin@example.com")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_token_resolves_user_by_username():
|
|
db = AsyncMock()
|
|
service = AuthService(db)
|
|
service.user_repo = MagicMock()
|
|
user = MagicMock(username="admin", email="admin@example.com", is_active=True)
|
|
service.user_repo.get_by_username = AsyncMock(return_value=user)
|
|
service.user_repo.get_by_email = AsyncMock(return_value=None)
|
|
|
|
token = await service.refresh_token(create_refresh_token(data={"sub": "admin"}))
|
|
|
|
assert token is not None
|
|
payload = verify_token(token.refresh_token)
|
|
assert payload is not None
|
|
assert payload["sub"] == "admin"
|
|
service.user_repo.get_by_email.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_authenticate_user_via_email_uses_username_subject():
|
|
db = AsyncMock()
|
|
service = AuthService(db)
|
|
service.user_repo = MagicMock()
|
|
service.user_repo.authenticate_via_email = AsyncMock(
|
|
return_value=MagicMock(
|
|
username="admin", email="admin@example.com", is_active=True
|
|
)
|
|
)
|
|
|
|
token = await service.authenticate_user_via_email("admin@example.com")
|
|
|
|
assert token is not None
|
|
payload = verify_token(token.access_token)
|
|
assert payload is not None
|
|
assert payload["sub"] == "admin"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_authenticate_user_via_email_rejects_inactive():
|
|
db = AsyncMock()
|
|
service = AuthService(db)
|
|
service.user_repo = MagicMock()
|
|
service.user_repo.authenticate_via_email = AsyncMock(
|
|
return_value=MagicMock(
|
|
username="admin", email="admin@example.com", is_active=False
|
|
)
|
|
)
|
|
|
|
token = await service.authenticate_user_via_email("admin@example.com")
|
|
|
|
assert token is None
|