zl-q
c9a2c75c35
- JWT 验证器新增 apikey 参数，支持云端 JWKS 认证头 - Auth 网关新增上游超时/错误映射为 503 状态码 - Auth 网关新增重定向 URL 校验，阻断开放重定向风险 - 用户依赖传递 anon_key 给 JWT 验证器 - 新增相关单元测试覆盖 JWKS 头、503 映射、重定向校验 - 新增实现计划文档
124 lines
3.7 KiB
Python
124 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from core.config.settings import SupabaseSettings
|
|
from services.base.supabase import SupabaseService
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initialize_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
service = SupabaseService(
|
|
settings=SupabaseSettings(public_url="https://test.supabase.co")
|
|
)
|
|
anon_client = MagicMock()
|
|
admin_client = MagicMock()
|
|
|
|
create_calls: list[tuple[str, str]] = []
|
|
|
|
def _fake_create_client(url: str, key: str) -> object:
|
|
create_calls.append((url, key))
|
|
return anon_client if len(create_calls) == 1 else admin_client
|
|
|
|
monkeypatch.setattr("services.base.supabase.create_client", _fake_create_client)
|
|
|
|
result = await service.initialize()
|
|
|
|
assert result is True
|
|
assert service.is_initialized is True
|
|
assert service.get_client() is anon_client
|
|
assert service.get_admin_client() is admin_client
|
|
assert len(create_calls) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initialize_failure(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
service = SupabaseService(
|
|
settings=SupabaseSettings(public_url="https://test.supabase.co")
|
|
)
|
|
|
|
def _fake_create_client(_: str, __: str) -> object:
|
|
raise RuntimeError("boom")
|
|
|
|
monkeypatch.setattr("services.base.supabase.create_client", _fake_create_client)
|
|
|
|
result = await service.initialize()
|
|
|
|
assert result is False
|
|
assert service.is_initialized is False
|
|
with pytest.raises(RuntimeError):
|
|
service.get_client()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close_clears_clients(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
service = SupabaseService(
|
|
settings=SupabaseSettings(public_url="https://test.supabase.co")
|
|
)
|
|
|
|
def _fake_create_client(_: str, __: str) -> object:
|
|
return MagicMock()
|
|
|
|
monkeypatch.setattr("services.base.supabase.create_client", _fake_create_client)
|
|
|
|
assert await service.initialize() is True
|
|
assert await service.close() is True
|
|
assert service.is_initialized is False
|
|
with pytest.raises(RuntimeError):
|
|
service.get_client()
|
|
with pytest.raises(RuntimeError):
|
|
service.get_admin_client()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_health_check_uninitialized() -> None:
|
|
service = SupabaseService(
|
|
settings=SupabaseSettings(public_url="https://test.supabase.co")
|
|
)
|
|
|
|
health = await service.health_check()
|
|
|
|
assert health["status"] == "unhealthy"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_health_check_initialized(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
service = SupabaseService(
|
|
settings=SupabaseSettings(public_url="https://test.supabase.co")
|
|
)
|
|
|
|
anon_client = MagicMock()
|
|
anon_client.auth.get_session = MagicMock(return_value=None)
|
|
|
|
admin_list_users = MagicMock(return_value=SimpleNamespace(users=[]))
|
|
admin_client = MagicMock()
|
|
admin_client.auth.admin = SimpleNamespace(list_users=admin_list_users)
|
|
|
|
create_sequence = [anon_client, admin_client]
|
|
|
|
def _fake_create_client(_: str, __: str) -> object:
|
|
return create_sequence.pop(0)
|
|
|
|
monkeypatch.setattr("services.base.supabase.create_client", _fake_create_client)
|
|
|
|
assert await service.initialize() is True
|
|
|
|
health = await service.health_check()
|
|
|
|
assert health["status"] == "healthy"
|
|
admin_list_users.assert_called_once_with(page=1, per_page=1)
|
|
|
|
|
|
def test_get_client_raises_before_init() -> None:
|
|
service = SupabaseService(
|
|
settings=SupabaseSettings(public_url="https://test.supabase.co")
|
|
)
|
|
|
|
with pytest.raises(RuntimeError):
|
|
service.get_client()
|
|
with pytest.raises(RuntimeError):
|
|
service.get_admin_client()
|