test: cover auth and SMS flows

2026-07-14 09:33:16 +03:30
parent 0bf7268d9c
commit 2264b7fc95
6 changed files with 344 additions and 0 deletions
--- a/tests/test_grpc_auth.py
+++ 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)
+