111 lines
3.4 KiB
Python
111 lines
3.4 KiB
Python
"""``ProviderManager.add_provider`` happy path：
|
||
- mock 掉具体 Provider 适配器的 ``create_provider``；
|
||
- 验证内存注册表写入 + postgres_database.add_provider_db.remote 被正确调用。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
import pytest
|
||
|
||
from kilostar.core.global_state_machine.provider_manager import ProviderManager
|
||
from kilostar.core.global_state_machine.model_provider.base_provider import Provider
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_provider_happy_path_writes_register_and_db():
|
||
pm = ProviderManager(postgres=None)
|
||
|
||
fake_provider = Provider(
|
||
provider_title="my-openai",
|
||
provider_url="https://api.openai.com",
|
||
provider_apikey="sk-xxx",
|
||
provider_models=["gpt-4o"],
|
||
provider_type="openai",
|
||
provider_owner="alice",
|
||
)
|
||
|
||
pm.provider_mapper["openai"] = MagicMock()
|
||
pm.provider_mapper["openai"].create_provider = AsyncMock(return_value=fake_provider)
|
||
|
||
postgres = MagicMock()
|
||
postgres.add_provider_db = MagicMock()
|
||
postgres.add_provider_db.remote = AsyncMock(return_value=None)
|
||
|
||
await pm.add_provider(
|
||
provider_type="openai",
|
||
provider_title="my-openai",
|
||
provider_url="https://api.openai.com",
|
||
provider_apikey="sk-xxx",
|
||
provider_owner="alice",
|
||
postgres_database=postgres,
|
||
)
|
||
|
||
assert "my-openai" in pm.provider_register
|
||
assert pm.provider_register["my-openai"] is fake_provider
|
||
postgres.add_provider_db.remote.assert_awaited_once()
|
||
kwargs = postgres.add_provider_db.remote.await_args.kwargs
|
||
assert kwargs["provider_title"] == "my-openai"
|
||
assert kwargs["provider_apikey"] == "sk-xxx"
|
||
assert kwargs["provider_models"] == ["gpt-4o"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_provider_unknown_type_returns_none(caplog):
|
||
pm = ProviderManager(postgres=None)
|
||
postgres = MagicMock()
|
||
postgres.add_provider_db = MagicMock()
|
||
postgres.add_provider_db.remote = AsyncMock()
|
||
|
||
result = await pm.add_provider(
|
||
provider_type="not_supported",
|
||
provider_title="x",
|
||
provider_url="u",
|
||
provider_apikey="k",
|
||
provider_owner="o",
|
||
postgres_database=postgres,
|
||
)
|
||
|
||
assert result is None
|
||
assert "x" not in pm.provider_register
|
||
postgres.add_provider_db.remote.assert_not_awaited()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_provider_network_error_raises_retryable():
|
||
"""网络异常应被包装为 RetryableError 重抛。"""
|
||
import httpx
|
||
from kilostar.utils.error import RetryableError
|
||
|
||
pm = ProviderManager(postgres=None)
|
||
pm.provider_mapper["openai"] = MagicMock()
|
||
pm.provider_mapper["openai"].create_provider = AsyncMock(
|
||
side_effect=httpx.ConnectError("network down")
|
||
)
|
||
|
||
postgres = MagicMock()
|
||
postgres.add_provider_db = MagicMock()
|
||
postgres.add_provider_db.remote = AsyncMock()
|
||
|
||
with pytest.raises(RetryableError):
|
||
await pm.add_provider(
|
||
provider_type="openai",
|
||
provider_title="x",
|
||
provider_url="u",
|
||
provider_apikey="k",
|
||
provider_owner="o",
|
||
postgres_database=postgres,
|
||
)
|
||
|
||
|
||
def test_get_provider_list_returns_internal_dict():
|
||
pm = ProviderManager(postgres=None)
|
||
pm.provider_register["a"] = "fake"
|
||
assert pm.get_provider_list()["a"] == "fake"
|
||
|
||
|
||
def test_get_provider_returns_none_when_missing():
|
||
pm = ProviderManager(postgres=None)
|
||
assert pm.get_provider("nope") is None
|