101 lines
3.7 KiB
Python
101 lines
3.7 KiB
Python
import pytest
|
|
|
|
from gapido_auth.application import auth_service as auth_module
|
|
from gapido_auth.application.auth_service import AuthConfig, AuthService
|
|
from gapido_auth.application.security import JwtTokenCodec
|
|
from gapido_auth.domain.entities import Role
|
|
from gapido_auth.domain.errors import AuthenticationError, InvalidOtp, PermissionDenied
|
|
from tests.fakes import (
|
|
FakeOtpStore,
|
|
FakeRefreshSessionRepository,
|
|
FakeSmsPublisher,
|
|
FakeUserRepository,
|
|
)
|
|
|
|
|
|
def build_service() -> tuple[
|
|
AuthService, FakeUserRepository, FakeRefreshSessionRepository, FakeOtpStore, FakeSmsPublisher
|
|
]:
|
|
users = FakeUserRepository()
|
|
refresh_sessions = FakeRefreshSessionRepository()
|
|
otp_store = FakeOtpStore()
|
|
sms = FakeSmsPublisher()
|
|
token_codec = JwtTokenCodec("unit-test-secret-key", "tests", 60)
|
|
service = AuthService(
|
|
users=users,
|
|
refresh_sessions=refresh_sessions,
|
|
otp_store=otp_store,
|
|
sms_publisher=sms,
|
|
token_codec=token_codec,
|
|
config=AuthConfig(
|
|
otp_secret="unit-test-secret-key",
|
|
otp_ttl_seconds=120,
|
|
otp_max_attempts=5,
|
|
otp_request_limit=3,
|
|
otp_request_window_seconds=300,
|
|
refresh_token_ttl_seconds=3600,
|
|
sms_template="login-otp",
|
|
),
|
|
)
|
|
return service, users, refresh_sessions, otp_store, sms
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_request_and_verify_otp_issues_tokens(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(auth_module, "generate_otp_code", lambda: "123456")
|
|
service, _, _, _, sms = build_service()
|
|
|
|
await service.request_otp("989120000000", "login", "peer")
|
|
assert sms.jobs[0].mobile == "989120000000"
|
|
assert sms.jobs[0].code == "123456"
|
|
|
|
token_pair = await service.verify_otp("989120000000", "123456", "login")
|
|
assert token_pair.access_token
|
|
assert token_pair.refresh_token
|
|
assert token_pair.role == Role.USER
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_otp_is_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(auth_module, "generate_otp_code", lambda: "123456")
|
|
service, _, _, _, _ = build_service()
|
|
|
|
await service.request_otp("989120000000", "login", "peer")
|
|
|
|
with pytest.raises(InvalidOtp):
|
|
await service.verify_otp("989120000000", "000000", "login")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_token_rotates_and_old_token_stops_working(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(auth_module, "generate_otp_code", lambda: "123456")
|
|
service, _, _, _, _ = build_service()
|
|
await service.request_otp("989120000000", "login", "peer")
|
|
first = await service.verify_otp("989120000000", "123456", "login")
|
|
|
|
second = await service.refresh_token(first.refresh_token)
|
|
|
|
assert second.refresh_token != first.refresh_token
|
|
with pytest.raises(AuthenticationError):
|
|
await service.refresh_token(first.refresh_token)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_role_required(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(auth_module, "generate_otp_code", lambda: "123456")
|
|
service, users, _, _, _ = build_service()
|
|
await users.ensure_admin_user("989120000001")
|
|
await service.request_otp("989120000001", "login", "peer")
|
|
admin_tokens = await service.verify_otp("989120000001", "123456", "login")
|
|
|
|
admin = await service.require_role(admin_tokens.access_token, Role.ADMIN)
|
|
assert admin.role == Role.ADMIN
|
|
|
|
await service.request_otp("989120000002", "login", "peer-2")
|
|
user_tokens = await service.verify_otp("989120000002", "123456", "login")
|
|
with pytest.raises(PermissionDenied):
|
|
await service.require_role(user_tokens.access_token, Role.ADMIN)
|
|
|