Merge pull request #2877 from kirs-hi/feat/custom-providers-ui

feat(web): manage multiple custom (OpenAI-compatible) providers in console UI
This commit is contained in:
zhayujie
2026-06-12 11:54:48 +08:00
12 changed files with 1539 additions and 77 deletions

View File

@@ -0,0 +1,240 @@
# encoding:utf-8
"""
Unit tests for multiple custom (OpenAI-compatible) provider support (issue #2838).
Covers models/custom_provider.py:
- Backward compatibility: legacy custom_api_key / custom_api_base fallback
- Multi-provider selection via bot_type "custom:<id>" routing
- parse_custom_bot_type helper
- Robustness against malformed config (missing id, non-dict, non-list)
"""
import sys
import os
import unittest
# Add project root to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
import config as config_module
from config import Config
def set_conf(d):
"""Install a fresh Config as the global config used by conf()."""
config_module.config = Config(d)
class TestParseCustomBotType(unittest.TestCase):
"""parse_custom_bot_type() parsing logic."""
def setUp(self):
from models.custom_provider import parse_custom_bot_type
self.parse = parse_custom_bot_type
def test_legacy_custom(self):
is_custom, pid = self.parse("custom")
self.assertTrue(is_custom)
self.assertEqual(pid, "")
def test_custom_with_id(self):
is_custom, pid = self.parse("custom:3f2a9c1b")
self.assertTrue(is_custom)
self.assertEqual(pid, "3f2a9c1b")
def test_non_custom(self):
is_custom, pid = self.parse("openai")
self.assertFalse(is_custom)
self.assertEqual(pid, "")
def test_empty(self):
is_custom, pid = self.parse("")
self.assertFalse(is_custom)
self.assertEqual(pid, "")
def test_none(self):
is_custom, pid = self.parse(None)
self.assertFalse(is_custom)
self.assertEqual(pid, "")
class TestResolveCustomCredentials(unittest.TestCase):
"""resolve_custom_credentials() resolution order and fallbacks."""
def setUp(self):
from models.custom_provider import resolve_custom_credentials, get_custom_providers
self.resolve = resolve_custom_credentials
self.get_providers = get_custom_providers
# --- Backward compatibility ---
def test_legacy_fallback_when_no_providers(self):
set_conf({
"bot_type": "custom",
"custom_api_key": "legacy-key",
"custom_api_base": "https://legacy.example.com/v1",
})
self.assertEqual(
self.resolve(),
("legacy-key", "https://legacy.example.com/v1", None),
)
def test_empty_config(self):
set_conf({"bot_type": "custom"})
self.assertEqual(self.resolve(), ("", None, None))
# --- Multi-provider selection via bot_type ---
def test_provider_selected_by_id(self):
set_conf({
"bot_type": "custom:abc12345",
"custom_providers": [
{"id": "sf001", "name": "siliconflow", "api_key": "sf-key",
"api_base": "https://api.siliconflow.cn/v1", "model": "deepseek-ai/DeepSeek-V3"},
{"id": "abc12345", "name": "qiniu", "api_key": "qn-key",
"api_base": "https://api.qnaigc.com/v1", "model": "deepseek-v3"},
],
})
self.assertEqual(
self.resolve(),
("qn-key", "https://api.qnaigc.com/v1", "deepseek-v3"),
)
def test_id_not_found_falls_back_to_legacy(self):
set_conf({
"bot_type": "custom:ghost",
"custom_api_key": "legacy-key",
"custom_api_base": "https://legacy.example.com/v1",
"custom_providers": [
{"id": "sf001", "name": "siliconflow", "api_key": "sf-key",
"api_base": "https://api.siliconflow.cn/v1"},
],
})
self.assertEqual(
self.resolve(),
("legacy-key", "https://legacy.example.com/v1", None),
)
def test_provider_without_model_returns_none_model(self):
set_conf({
"bot_type": "custom:local01",
"custom_providers": [
{"id": "local01", "name": "local", "api_key": "", "api_base": "http://localhost:11434/v1"},
],
})
self.assertEqual(
self.resolve(),
("", "http://localhost:11434/v1", None),
)
# --- Robustness against malformed config ---
def test_malformed_entries_filtered_and_fallback(self):
set_conf({
"bot_type": "custom:nope",
"custom_api_key": "legacy-key",
"custom_api_base": "https://legacy.example.com/v1",
"custom_providers": [
{"name": "no-id", "api_key": "no-id-key"}, # invalid: no id
"not-a-dict", # invalid: wrong type
],
})
# All entries invalid -> treated as empty -> legacy fallback
self.assertEqual(
self.resolve(),
("legacy-key", "https://legacy.example.com/v1", None),
)
def test_get_custom_providers_filters_invalid(self):
set_conf({
"bot_type": "custom",
"custom_providers": [
{"id": "ok1", "name": "ok", "api_key": "k", "api_base": "https://x/v1"},
{"name": "no-id", "api_key": "no-id"}, # dropped: no id
123, # dropped
],
})
providers = self.get_providers()
self.assertEqual(len(providers), 1)
self.assertEqual(providers[0]["id"], "ok1")
def test_custom_providers_not_a_list_falls_back(self):
set_conf({
"bot_type": "custom",
"custom_api_key": "legacy-key",
"custom_api_base": "https://legacy.example.com/v1",
"custom_providers": "oops-a-string",
})
self.assertEqual(
self.resolve(),
("legacy-key", "https://legacy.example.com/v1", None),
)
class TestGenerateProviderId(unittest.TestCase):
"""generate_provider_id() produces valid short ids."""
def test_length_and_hex(self):
from models.custom_provider import generate_provider_id
pid = generate_provider_id()
self.assertEqual(len(pid), 8)
# Must be valid hex characters
int(pid, 16)
def test_uniqueness(self):
from models.custom_provider import generate_provider_id
ids = {generate_provider_id() for _ in range(100)}
self.assertEqual(len(ids), 100)
class TestConfigDefaults(unittest.TestCase):
"""The new config fields must exist with safe defaults."""
def test_default_config_has_custom_providers(self):
from config import available_setting
self.assertIn("custom_providers", available_setting)
self.assertEqual(available_setting["custom_providers"], [])
def test_default_config_no_custom_active_provider(self):
"""custom_active_provider was removed — replaced by bot_type routing."""
from config import available_setting
self.assertNotIn("custom_active_provider", available_setting)
class TestDragSensitiveNested(unittest.TestCase):
"""drag_sensitive() must mask api_key in nested structures."""
def test_nested_api_key_masked(self):
from config import drag_sensitive
import json
test_config = {
"open_ai_api_key": "sk-1234567890abcdef",
"custom_providers": [
{"id": "x1", "name": "test", "api_key": "sk-nested-secret-key-long", "api_base": "https://x/v1"}
],
}
result = drag_sensitive(test_config)
# Top-level key should be masked
self.assertNotIn("1234567890abcdef", str(result))
# Nested key should also be masked
self.assertNotIn("nested-secret-key-long", str(result))
# But the id/name/api_base should not be masked
self.assertIn("x1", str(result))
self.assertIn("test", str(result))
self.assertIn("https://x/v1", str(result))
def test_string_config_masked(self):
from config import drag_sensitive
import json
test_str = json.dumps({
"open_ai_api_key": "sk-1234567890abcdef",
"custom_providers": [
{"id": "x1", "api_key": "sk-nested-very-long-secret"}
],
})
result = drag_sensitive(test_str)
self.assertNotIn("1234567890abcdef", result)
self.assertNotIn("nested-very-long-secret", result)
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,289 @@
# encoding:utf-8
"""
Unit tests for the multi custom-provider management API (issue #2838, web UI).
Covers channel/web/web_channel.py::ModelsHandler:
- _custom_provider_cards / _provider_overview expansion
- _handle_set_custom_provider (create / edit / activate)
- _handle_delete_custom_provider
- _handle_set_active_custom_provider
Uses id-based routing (bot_type: "custom:<id>") — no custom_active_provider.
"""
import json
import os
import sys
import types
import unittest
# Add project root to path.
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
# Stub the web.py framework so web_channel imports without the dependency.
if "web" not in sys.modules:
_web_stub = types.ModuleType("web")
_web_stub.header = lambda *a, **k: None
_web_stub.data = lambda: b"{}"
_web_stub.ctx = types.SimpleNamespace()
sys.modules["web"] = _web_stub
import config as config_module
from config import Config
from channel.web.web_channel import ModelsHandler
def set_conf(d):
"""Install a fresh Config as the global config used by conf()."""
config_module.config = Config(d)
class _HandlerHarness:
"""Test double around ModelsHandler that captures persisted config in
memory instead of touching config.json, and no-ops the Bridge reset."""
def __init__(self):
self.handler = ModelsHandler.__new__(ModelsHandler)
self._file_cfg = {}
self.bridge_resets = 0
# Patch the disk + bridge boundary on this instance.
self.handler._read_file_config = lambda: dict(self._file_cfg)
self.handler._write_file_config = self._capture_write
self.handler._reset_bridge = self._capture_reset
def _capture_write(self, data):
self._file_cfg = dict(data)
def _capture_reset(self):
self.bridge_resets += 1
def call(self, **payload):
# Resolve the bound method by action for convenience.
action = payload.get("action")
method = {
"set_custom_provider": self.handler._handle_set_custom_provider,
"delete_custom_provider": self.handler._handle_delete_custom_provider,
"set_active_custom_provider": self.handler._handle_set_active_custom_provider,
}[action]
return json.loads(method(payload))
class TestSetCustomProvider(unittest.TestCase):
def setUp(self):
set_conf({"bot_type": "custom", "custom_providers": []})
self.h = _HandlerHarness()
def test_create_provider_does_not_hijack_bot_type(self):
"""Creating a provider without make_active must not change bot_type."""
res = self.h.call(action="set_custom_provider", name="siliconflow",
api_base="https://api.siliconflow.cn/v1", api_key="sf-key")
self.assertEqual(res["status"], "success")
self.assertTrue(res["created"])
self.assertIn("id", res)
# bot_type must remain unchanged — no auto-activation.
bot_type = config_module.conf().get("bot_type")
self.assertEqual(bot_type, "custom") # unchanged from setUp
providers = config_module.conf().get("custom_providers")
self.assertEqual(len(providers), 1)
self.assertEqual(providers[0]["id"], res["id"])
self.assertEqual(providers[0]["name"], "siliconflow")
self.assertEqual(self.h.bridge_resets, 1)
def test_create_with_make_active_switches_bot_type(self):
"""Creating a provider with make_active=true must switch bot_type."""
res = self.h.call(action="set_custom_provider", name="siliconflow",
api_base="https://api.siliconflow.cn/v1", api_key="sf-key",
make_active=True)
self.assertEqual(res["status"], "success")
bot_type = config_module.conf().get("bot_type")
self.assertEqual(bot_type, f"custom:{res['id']}")
def test_create_requires_api_base(self):
res = self.h.call(action="set_custom_provider", name="x", api_key="k")
self.assertEqual(res["status"], "error")
self.assertIn("api_base", res["message"])
def test_create_requires_name(self):
res = self.h.call(action="set_custom_provider", name="", api_base="https://x/v1")
self.assertEqual(res["status"], "error")
def test_second_provider_does_not_steal_active(self):
# Explicitly activate the first provider.
res1 = self.h.call(action="set_custom_provider", name="a",
api_base="https://a/v1", api_key="ak", make_active=True)
res2 = self.h.call(action="set_custom_provider", name="b",
api_base="https://b/v1", api_key="bk")
self.assertTrue(res2["created"])
# First provider stays active — second creation doesn't steal it.
bot_type = config_module.conf().get("bot_type")
self.assertEqual(bot_type, f"custom:{res1['id']}")
def test_make_active_flag(self):
self.h.call(action="set_custom_provider", name="a",
api_base="https://a/v1", api_key="ak")
res2 = self.h.call(action="set_custom_provider", name="b",
api_base="https://b/v1", api_key="bk", make_active=True)
bot_type = config_module.conf().get("bot_type")
self.assertEqual(bot_type, f"custom:{res2['id']}")
def test_edit_keeps_key_when_omitted(self):
res = self.h.call(action="set_custom_provider", name="a",
api_base="https://a/v1", api_key="secret")
pid = res["id"]
# Edit only the base; omit api_key.
res2 = self.h.call(action="set_custom_provider", name="a",
id=pid, api_base="https://a2/v1")
self.assertEqual(res2["status"], "success")
self.assertFalse(res2["created"])
providers = config_module.conf().get("custom_providers")
self.assertEqual(providers[0]["api_base"], "https://a2/v1")
self.assertEqual(providers[0]["api_key"], "secret") # preserved
def test_edit_can_rename(self):
res = self.h.call(action="set_custom_provider", name="old",
api_base="https://a/v1", api_key="ak")
pid = res["id"]
res2 = self.h.call(action="set_custom_provider", name="new",
id=pid, api_base="https://a/v1")
self.assertEqual(res2["status"], "success")
providers = config_module.conf().get("custom_providers")
self.assertEqual(providers[0]["name"], "new")
# ID stays the same
self.assertEqual(providers[0]["id"], pid)
def test_edit_clears_model_when_empty(self):
res = self.h.call(action="set_custom_provider", name="a",
api_base="https://a/v1", api_key="ak", model="m1")
pid = res["id"]
self.assertEqual(config_module.conf().get("custom_providers")[0]["model"], "m1")
self.h.call(action="set_custom_provider", name="a", id=pid,
api_base="https://a/v1", model="")
self.assertNotIn("model", config_module.conf().get("custom_providers")[0])
class TestDeleteCustomProvider(unittest.TestCase):
def setUp(self):
set_conf({"bot_type": "custom", "custom_providers": []})
self.h = _HandlerHarness()
self.res_a = self.h.call(action="set_custom_provider", name="a",
api_base="https://a/v1", api_key="ak",
make_active=True)
self.res_b = self.h.call(action="set_custom_provider", name="b",
api_base="https://b/v1", api_key="bk")
def test_delete_unknown(self):
res = self.h.call(action="delete_custom_provider", id="ghost")
self.assertEqual(res["status"], "error")
def test_delete_non_active(self):
res = self.h.call(action="delete_custom_provider", id=self.res_b["id"])
self.assertEqual(res["status"], "success")
ids = [p["id"] for p in config_module.conf().get("custom_providers")]
self.assertEqual(ids, [self.res_a["id"]])
# bot_type unchanged (still pointing to a)
self.assertEqual(config_module.conf().get("bot_type"), f"custom:{self.res_a['id']}")
def test_delete_active_falls_back_to_first_remaining(self):
# 'a' is active (created first); deleting it should re-point to 'b'.
self.assertEqual(config_module.conf().get("bot_type"), f"custom:{self.res_a['id']}")
res = self.h.call(action="delete_custom_provider", id=self.res_a["id"])
self.assertEqual(res["status"], "success")
self.assertEqual(config_module.conf().get("bot_type"), f"custom:{self.res_b['id']}")
def test_delete_last_reverts_to_legacy(self):
self.h.call(action="delete_custom_provider", id=self.res_a["id"])
self.h.call(action="delete_custom_provider", id=self.res_b["id"])
self.assertEqual(config_module.conf().get("custom_providers"), [])
# When all providers deleted, reverts to legacy "custom"
self.assertEqual(config_module.conf().get("bot_type"), "custom")
class TestSetActiveCustomProvider(unittest.TestCase):
def setUp(self):
set_conf({"bot_type": "custom", "custom_providers": []})
self.h = _HandlerHarness()
self.res_a = self.h.call(action="set_custom_provider", name="a",
api_base="https://a/v1", api_key="ak",
make_active=True)
self.res_b = self.h.call(action="set_custom_provider", name="b",
api_base="https://b/v1", api_key="bk")
def test_set_active_valid(self):
res = self.h.call(action="set_active_custom_provider", id=self.res_b["id"])
self.assertEqual(res["status"], "success")
self.assertEqual(config_module.conf().get("bot_type"), f"custom:{self.res_b['id']}")
def test_set_active_unknown(self):
res = self.h.call(action="set_active_custom_provider", id="ghost")
self.assertEqual(res["status"], "error")
# bot_type unchanged
self.assertEqual(config_module.conf().get("bot_type"), f"custom:{self.res_a['id']}")
def test_activation_syncs_model_to_global(self):
"""Activating a provider must write its model into global model field."""
set_conf({"bot_type": "custom", "custom_providers": [], "model": "gpt-4o"})
h = _HandlerHarness()
res = h.call(action="set_custom_provider", name="sf",
api_base="https://sf/v1", api_key="k", model="deepseek-v3",
make_active=True)
# Global model field should now be the provider's model.
self.assertEqual(config_module.conf().get("model"), "deepseek-v3")
self.assertEqual(config_module.conf().get("bot_type"), f"custom:{res['id']}")
def test_activation_without_model_keeps_global_model(self):
"""Activating a provider with no model must not overwrite global model."""
set_conf({"bot_type": "custom", "custom_providers": [], "model": "gpt-4o"})
h = _HandlerHarness()
h.call(action="set_custom_provider", name="local",
api_base="http://localhost:11434/v1", api_key="",
make_active=True)
# Global model field should remain unchanged.
self.assertEqual(config_module.conf().get("model"), "gpt-4o")
class TestProviderOverviewExpansion(unittest.TestCase):
"""_provider_overview / _custom_provider_cards should expand the list."""
def test_no_custom_providers_keeps_single_card(self):
set_conf({"bot_type": "custom", "custom_providers": []})
cards = ModelsHandler._custom_provider_cards(config_module.conf())
self.assertEqual(cards, [])
overview = ModelsHandler._provider_overview()
custom_cards = [c for c in overview if c.get("id") == "custom"]
# Legacy single custom card remains present.
self.assertEqual(len(custom_cards), 1)
def test_multi_providers_expand_into_cards(self):
set_conf({
"bot_type": "custom:id_b",
"custom_providers": [
{"id": "id_a", "name": "a", "api_key": "ak", "api_base": "https://a/v1"},
{"id": "id_b", "name": "b", "api_key": "bk", "api_base": "https://b/v1", "model": "m"},
],
})
overview = ModelsHandler._provider_overview()
custom_cards = [c for c in overview if c.get("is_custom")]
self.assertEqual(len(custom_cards), 2)
by_id = {c["custom_id"]: c for c in custom_cards}
self.assertEqual(by_id["id_a"]["id"], "custom:id_a")
self.assertFalse(by_id["id_a"]["active"])
self.assertTrue(by_id["id_b"]["active"])
self.assertEqual(by_id["id_b"]["model"], "m")
# No single legacy "custom" card when expanded.
self.assertFalse(any(c.get("id") == "custom" for c in overview))
def test_no_active_shows_none_active(self):
"""When bot_type is plain 'custom', no card is marked active."""
set_conf({
"bot_type": "custom",
"custom_providers": [
{"id": "id_a", "name": "a", "api_key": "ak", "api_base": "https://a/v1"},
{"id": "id_b", "name": "b", "api_key": "bk", "api_base": "https://b/v1"},
],
})
cards = ModelsHandler._custom_provider_cards(config_module.conf())
active_cards = [c for c in cards if c.get("active")]
self.assertEqual(len(active_cards), 0)
if __name__ == "__main__":
unittest.main()