143 lines
5.0 KiB
Python
143 lines
5.0 KiB
Python
from dataclasses import dataclass
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from gapido_auth.config import Settings
|
|
from gapido_demo.app import app, get_app_settings, get_auth_client, get_debug_redis
|
|
from gapido_demo.grpc_client import DemoTokenResponse
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class FakeAuthClient:
|
|
last_user_token: str | None = None
|
|
last_admin_token: str | None = None
|
|
|
|
async def request_otp(self, mobile: str, purpose: str) -> dict[str, bool]:
|
|
assert mobile == "989120000000"
|
|
assert purpose == "login"
|
|
return {"accepted": True}
|
|
|
|
async def verify_otp(self, mobile: str, code: str, purpose: str) -> DemoTokenResponse:
|
|
assert mobile == "989120000000"
|
|
assert code == "123456"
|
|
assert purpose == "login"
|
|
return DemoTokenResponse(
|
|
"access-token-value-123", "refresh-token-value-123", "Bearer", 900, "user"
|
|
)
|
|
|
|
async def refresh_token(self, refresh_token: str) -> DemoTokenResponse:
|
|
assert refresh_token == "refresh-token-value-123"
|
|
return DemoTokenResponse(
|
|
"new-access-token-value", "new-refresh-token-value", "Bearer", 900, "user"
|
|
)
|
|
|
|
async def revoke_refresh_token(self, access_token: str, refresh_token: str) -> dict[str, bool]:
|
|
assert access_token == "access-token-value-123"
|
|
assert refresh_token == "refresh-token-value-123"
|
|
return {"revoked": True}
|
|
|
|
async def public_ping(self) -> dict[str, str]:
|
|
return {"message": "public ok"}
|
|
|
|
async def user_only(self, access_token: str) -> dict[str, str]:
|
|
self.last_user_token = access_token
|
|
return {"user_id": "1", "role": "user", "message": "authenticated user ok"}
|
|
|
|
async def admin_only(self, access_token: str) -> dict[str, str]:
|
|
self.last_admin_token = access_token
|
|
return {"user_id": "2", "role": "admin", "message": "admin ok"}
|
|
|
|
|
|
class FakeDebugRedis:
|
|
async def get(self, key: str) -> str | None:
|
|
assert key == "debug:sms:last:989120000000"
|
|
return "123456"
|
|
|
|
|
|
def test_demo_auth_flow_forwards_to_grpc_client() -> None:
|
|
fake_client = FakeAuthClient()
|
|
app.dependency_overrides[get_auth_client] = lambda: fake_client
|
|
|
|
client = TestClient(app)
|
|
try:
|
|
request = client.post(
|
|
"/api/auth/request-otp", json={"mobile": "989120000000", "purpose": "login"}
|
|
)
|
|
assert request.status_code == 200
|
|
assert request.json() == {"accepted": True}
|
|
|
|
verify = client.post(
|
|
"/api/auth/verify-otp",
|
|
json={"mobile": "989120000000", "code": "123456", "purpose": "login"},
|
|
)
|
|
assert verify.status_code == 200
|
|
assert verify.json()["access_token"] == "access-token-value-123"
|
|
|
|
public = client.post("/api/demo/public")
|
|
assert public.status_code == 200
|
|
assert public.json() == {"message": "public ok"}
|
|
|
|
user = client.post("/api/demo/user", json={"access_token": "access-token-value-123"})
|
|
assert user.status_code == 200
|
|
assert fake_client.last_user_token == "access-token-value-123"
|
|
|
|
admin = client.post("/api/demo/admin", json={"access_token": "access-token-value-123"})
|
|
assert admin.status_code == 200
|
|
assert fake_client.last_admin_token == "access-token-value-123"
|
|
|
|
refresh = client.post(
|
|
"/api/auth/refresh", json={"refresh_token": "refresh-token-value-123"}
|
|
)
|
|
assert refresh.status_code == 200
|
|
assert refresh.json()["refresh_token"] == "new-refresh-token-value"
|
|
|
|
revoke = client.post(
|
|
"/api/auth/revoke",
|
|
json={
|
|
"access_token": "access-token-value-123",
|
|
"refresh_token": "refresh-token-value-123",
|
|
},
|
|
)
|
|
assert revoke.status_code == 200
|
|
assert revoke.json() == {"revoked": True}
|
|
finally:
|
|
client.close()
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def test_healthz() -> None:
|
|
client = TestClient(app)
|
|
try:
|
|
response = client.get("/healthz")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"status": "ok"}
|
|
finally:
|
|
client.close()
|
|
|
|
|
|
def test_debug_otp_endpoint_reads_redis_when_enabled() -> None:
|
|
app.dependency_overrides[get_app_settings] = lambda: Settings(demo_enable_debug_otp=True)
|
|
app.dependency_overrides[get_debug_redis] = lambda: FakeDebugRedis()
|
|
|
|
client = TestClient(app)
|
|
try:
|
|
response = client.get("/api/debug/otp?mobile=989120000000")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"mobile": "989120000000", "code": "123456"}
|
|
finally:
|
|
client.close()
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def test_debug_otp_endpoint_is_disabled() -> None:
|
|
app.dependency_overrides[get_app_settings] = lambda: Settings(demo_enable_debug_otp=False)
|
|
app.dependency_overrides[get_debug_redis] = lambda: None
|
|
|
|
client = TestClient(app)
|
|
try:
|
|
response = client.get("/api/debug/otp?mobile=989120000000")
|
|
assert response.status_code == 404
|
|
finally:
|
|
client.close()
|
|
app.dependency_overrides.clear()
|