76 lines
2.7 KiB
Python
76 lines
2.7 KiB
Python
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)
|
|
|