diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..a10d0a0 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1,2 @@ +"""Test helpers.""" + diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..192ca18 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,6 @@ +from gapido_auth.tools.generate_proto import main as generate_proto + + +def pytest_configure() -> None: + generate_proto() + diff --git a/tests/fakes.py b/tests/fakes.py new file mode 100644 index 0000000..4e482da --- /dev/null +++ b/tests/fakes.py @@ -0,0 +1,131 @@ +from datetime import datetime +from itertools import count + +from gapido_auth.domain.entities import RefreshSession, Role, SmsJob, User +from gapido_auth.domain.errors import OtpAttemptsExceeded, OtpExpired + + +class FakeUserRepository: + def __init__(self) -> None: + self._users_by_mobile: dict[str, User] = {} + self._ids = count(1) + + async def get_by_id(self, user_id: str) -> User | None: + return next((user for user in self._users_by_mobile.values() if user.id == user_id), None) + + async def get_by_mobile(self, mobile: str) -> User | None: + return self._users_by_mobile.get(mobile) + + async def get_or_create_user(self, mobile: str, role: Role = Role.USER) -> User: + existing = self._users_by_mobile.get(mobile) + if existing: + return existing + now = datetime.now() + user = User( + id=str(next(self._ids)), + mobile=mobile, + role=role, + is_active=True, + created_at=now, + updated_at=now, + ) + self._users_by_mobile[mobile] = user + return user + + async def ensure_admin_user(self, mobile: str) -> User: + now = datetime.now() + user = User( + id=str(next(self._ids)), + mobile=mobile, + role=Role.ADMIN, + is_active=True, + created_at=now, + updated_at=now, + ) + self._users_by_mobile[mobile] = user + return user + + +class FakeRefreshSessionRepository: + def __init__(self) -> None: + self._sessions: dict[str, RefreshSession] = {} + self._ids = count(1) + + async def create(self, user_id: str, token_hash: str, expires_at: datetime) -> RefreshSession: + session = RefreshSession( + id=str(next(self._ids)), + user_id=user_id, + token_hash=token_hash, + expires_at=expires_at, + revoked_at=None, + replaced_by_hash=None, + created_at=datetime.now(), + ) + self._sessions[token_hash] = session + return session + + async def get_active_by_hash(self, token_hash: str, now: datetime) -> RefreshSession | None: + session = self._sessions.get(token_hash) + if session is None or session.revoked_at is not None or session.expires_at <= now: + return None + return session + + async def revoke( + self, token_hash: str, now: datetime, replaced_by_hash: str | None = None + ) -> None: + session = self._sessions.get(token_hash) + if session is None or session.revoked_at is not None: + return + self._sessions[token_hash] = RefreshSession( + id=session.id, + user_id=session.user_id, + token_hash=session.token_hash, + expires_at=session.expires_at, + revoked_at=now, + replaced_by_hash=replaced_by_hash, + created_at=session.created_at, + ) + + +class FakeOtpStore: + def __init__(self) -> None: + self.requests: dict[str, int] = {} + self.otps: dict[tuple[str, str], tuple[str, int, int]] = {} + + async def allow_request(self, key: str, limit: int, window_seconds: int) -> bool: + self.requests[key] = self.requests.get(key, 0) + 1 + return self.requests[key] <= limit + + async def store_otp( + self, + mobile: str, + purpose: str, + otp_hash: str, + ttl_seconds: int, + max_attempts: int, + ) -> None: + self.otps[(mobile, purpose)] = (otp_hash, 0, max_attempts) + + async def verify_otp(self, mobile: str, purpose: str, candidate_hash: str) -> bool: + key = (mobile, purpose) + if key not in self.otps: + raise OtpExpired("missing") + stored_hash, attempts, max_attempts = self.otps[key] + attempts += 1 + if attempts > max_attempts: + del self.otps[key] + raise OtpAttemptsExceeded("too many") + self.otps[key] = (stored_hash, attempts, max_attempts) + if stored_hash != candidate_hash: + return False + del self.otps[key] + return True + + +class FakeSmsPublisher: + def __init__(self) -> None: + self.jobs: list[SmsJob] = [] + + async def publish(self, job: SmsJob) -> None: + self.jobs.append(job) + diff --git a/tests/test_auth_service.py b/tests/test_auth_service.py new file mode 100644 index 0000000..f5a10ee --- /dev/null +++ b/tests/test_auth_service.py @@ -0,0 +1,100 @@ +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) + diff --git a/tests/test_grpc_auth.py b/tests/test_grpc_auth.py new file mode 100644 index 0000000..85adaa6 --- /dev/null +++ b/tests/test_grpc_auth.py @@ -0,0 +1,75 @@ +import grpc +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.generated import auth_pb2, auth_pb2_grpc +from gapido_auth.transport.grpc.auth_servicer import AuthGrpcServicer +from tests.fakes import ( + FakeOtpStore, + FakeRefreshSessionRepository, + FakeSmsPublisher, + FakeUserRepository, +) + + +def build_grpc_service() -> AuthService: + return AuthService( + users=FakeUserRepository(), + refresh_sessions=FakeRefreshSessionRepository(), + otp_store=FakeOtpStore(), + sms_publisher=FakeSmsPublisher(), + token_codec=JwtTokenCodec("unit-test-secret-key", "tests", 60), + 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", + ), + ) + + +@pytest.mark.asyncio +async def test_grpc_public_and_protected_methods(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(auth_module, "generate_otp_code", lambda: "123456") + server = grpc.aio.server() + auth_pb2_grpc.add_AuthServiceServicer_to_server( + AuthGrpcServicer(build_grpc_service()), server + ) + port = server.add_insecure_port("127.0.0.1:0") + await server.start() + + try: + channel = grpc.aio.insecure_channel(f"127.0.0.1:{port}") + stub = auth_pb2_grpc.AuthServiceStub(channel) + + public = await stub.PublicPing(auth_pb2.PingRequest()) + assert public.message == "public ok" + + with pytest.raises(grpc.aio.AioRpcError) as missing_auth: + await stub.UserOnly(auth_pb2.ProtectedRequest()) + assert missing_auth.value.code() == grpc.StatusCode.UNAUTHENTICATED + + await stub.RequestOtp(auth_pb2.RequestOtpRequest(mobile="989120000000", purpose="login")) + tokens = await stub.VerifyOtp( + auth_pb2.VerifyOtpRequest(mobile="989120000000", code="123456", purpose="login") + ) + user_response = await stub.UserOnly( + auth_pb2.ProtectedRequest(), + metadata=(("authorization", f"Bearer {tokens.access_token}"),), + ) + assert user_response.role == "user" + + with pytest.raises(grpc.aio.AioRpcError) as admin_denied: + await stub.AdminOnly( + auth_pb2.ProtectedRequest(), + metadata=(("authorization", f"Bearer {tokens.access_token}"),), + ) + assert admin_denied.value.code() == grpc.StatusCode.PERMISSION_DENIED + finally: + await server.stop(grace=0) + diff --git a/tests/test_kavenegar_client.py b/tests/test_kavenegar_client.py new file mode 100644 index 0000000..191e2a2 --- /dev/null +++ b/tests/test_kavenegar_client.py @@ -0,0 +1,30 @@ +import httpx +import pytest + +from gapido_auth.domain.errors import ExternalServiceError +from gapido_auth.infrastructure.kavenegar_client import KavenegarSmsClient + + +@pytest.mark.asyncio +async def test_kavenegar_success() -> None: + def handler(request: httpx.Request) -> httpx.Response: + assert request.url.path == "/v1/test-key/verify/lookup.json" + form = dict(item.split("=") for item in request.content.decode().split("&")) + assert form["receptor"] == "989120000000" + assert form["template"] == "login-otp" + return httpx.Response(200, json={"return": {"status": 200}}) + + client = KavenegarSmsClient("test-key", transport=httpx.MockTransport(handler)) + await client.send_otp("989120000000", "123456", "login-otp") + + +@pytest.mark.asyncio +async def test_kavenegar_api_error() -> None: + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"return": {"status": 418}}) + + client = KavenegarSmsClient("test-key", transport=httpx.MockTransport(handler)) + + with pytest.raises(ExternalServiceError): + await client.send_otp("989120000000", "123456", "login-otp") +