liumangmang
6044b00685
- 上游 Key 命名改为 {prefix}-{upstream.id}-{safe_group_name}-{group_id}
- 唯一约束 (upstream_id, group_id, managed_prefix) 加 managed_prefix 列
- 上游检测成功时同步 Key 状态，远端已删/分组已删自动清理
- 重复分组导入跳过，目标网站已存在同名分组返回 exists
- 账号导入平台自动识别（auto/manual 模式）
- 全选可导入 Key 按钮 + 目标分组自动匹配
- 导入幂等：已导入过的 Key 校验远端账号，不存在则重建
- 新增同步接口 POST /sync-imported-upstream-keys
- account_exists() 通过拉取账号列表判断，避免 404 误判
- credentials.base_url 注入来源上游地址，避免 401
- 前端导入弹窗自动同步+刷新按钮+并发/优先级设置
- 新增 12 个测试覆盖同步、幂等、远端删除、校验失败路径
235 lines
7.5 KiB
Python
235 lines
7.5 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from app.services.website_client import (
|
|
WebsiteError,
|
|
_friendly_connection_error,
|
|
_friendly_http_error,
|
|
normalize_groups,
|
|
)
|
|
|
|
|
|
# ——— normalize_groups ———
|
|
|
|
|
|
def test_normalize_groups_unwraps_sub2api_paginated_response():
|
|
groups = normalize_groups({
|
|
"code": 0,
|
|
"message": "success",
|
|
"data": {
|
|
"items": [
|
|
{"id": "codex-free", "name": "codex-free", "rate_multiplier": 1},
|
|
{"id": "my-plus", "name": "my-plus", "rate_multiplier": "1.5"},
|
|
{"id": "deepseek", "name": "deepseek"},
|
|
],
|
|
"total": 3,
|
|
"page": 1,
|
|
},
|
|
})
|
|
|
|
assert [group["id"] for group in groups] == ["codex-free", "my-plus", "deepseek"]
|
|
assert groups[0]["rate_multiplier"] == "1"
|
|
assert groups[1]["rate_multiplier"] == "1.5"
|
|
assert groups[2]["rate_multiplier"] is None
|
|
|
|
|
|
def test_normalize_groups_unwraps_wrapped_list_response():
|
|
groups = normalize_groups({
|
|
"code": 0,
|
|
"message": "success",
|
|
"data": [
|
|
{"id": "default", "name": "Default", "rateMultiplier": "2.0"},
|
|
],
|
|
})
|
|
|
|
assert groups == [{
|
|
"id": "default",
|
|
"name": "Default",
|
|
"rate_multiplier": "2",
|
|
"raw": {"id": "default", "name": "Default", "rateMultiplier": "2.0"},
|
|
}]
|
|
|
|
|
|
def test_normalize_groups_unwraps_groups_key_response():
|
|
groups = normalize_groups({
|
|
"data": {
|
|
"groups": [
|
|
{"group_id": "vip", "group_name": "VIP", "ratio": "0.75"},
|
|
],
|
|
},
|
|
})
|
|
|
|
assert groups[0]["id"] == "vip"
|
|
assert groups[0]["name"] == "VIP"
|
|
assert groups[0]["rate_multiplier"] == "0.75"
|
|
|
|
|
|
def test_normalize_groups_keeps_string_list_compatibility():
|
|
groups = normalize_groups(["free", "paid"])
|
|
|
|
assert [group["id"] for group in groups] == ["free", "paid"]
|
|
assert groups[0]["raw"] == {"id": "free", "name": "free"}
|
|
|
|
|
|
def test_normalize_groups_keeps_plain_dict_mapping_compatibility():
|
|
groups = normalize_groups({
|
|
"free": {"id": "free", "name": "Free", "rate_multiplier": "1.00"},
|
|
"paid": {"id": "paid", "name": "Paid"},
|
|
})
|
|
|
|
assert [group["id"] for group in groups] == ["free", "paid"]
|
|
assert groups[0]["rate_multiplier"] == "1"
|
|
|
|
|
|
# ——— _get_account_ids / account_exists ———
|
|
|
|
|
|
def test_get_account_ids_flat_list():
|
|
from app.services.website_client import Sub2ApiWebsiteClient
|
|
ids = Sub2ApiWebsiteClient._unwrap_list([
|
|
{"id": 1, "name": "a"}, {"id": 2, "name": "b"},
|
|
])
|
|
assert ids == [{"id": 1, "name": "a"}, {"id": 2, "name": "b"}]
|
|
|
|
|
|
def test_get_account_ids_top_level_items():
|
|
from app.services.website_client import Sub2ApiWebsiteClient
|
|
ids = Sub2ApiWebsiteClient._unwrap_list({"items": [
|
|
{"id": "k1"}, {"id": "k2"},
|
|
]})
|
|
assert ids == [{"id": "k1"}, {"id": "k2"}]
|
|
|
|
|
|
def test_get_account_ids_nested_data_items():
|
|
from app.services.website_client import Sub2ApiWebsiteClient
|
|
ids = Sub2ApiWebsiteClient._unwrap_list({"data": {"items": [
|
|
{"id": "a1", "name": "Alpha"},
|
|
{"id": "a2", "name": "Beta"},
|
|
]}})
|
|
assert ids == [{"id": "a1", "name": "Alpha"}, {"id": "a2", "name": "Beta"}]
|
|
|
|
|
|
def test_get_account_ids_nested_data_empty():
|
|
from app.services.website_client import Sub2ApiWebsiteClient
|
|
ids = Sub2ApiWebsiteClient._unwrap_list({"data": {"items": []}})
|
|
assert ids == []
|
|
|
|
|
|
def test_get_account_ids_unexpected_format():
|
|
from app.services.website_client import Sub2ApiWebsiteClient
|
|
ids = Sub2ApiWebsiteClient._unwrap_list({"error": "not found"})
|
|
assert ids is None
|
|
|
|
|
|
# ——— 友好错误提示 ———
|
|
|
|
def _make_response(status_code: int, path: str = "/groups") -> httpx.Response:
|
|
"""创建模拟的 httpx.Response 用于错误测试。"""
|
|
req = httpx.Request("GET", f"http://target.local/api/v1{path}")
|
|
resp = httpx.Response(status_code, request=req)
|
|
return resp
|
|
|
|
|
|
def test_friendly_http_401():
|
|
resp = _make_response(401)
|
|
exc = httpx.HTTPStatusError("401", request=resp.request, response=resp)
|
|
msg = _friendly_http_error(exc)
|
|
assert "认证失败" in msg
|
|
assert "API Key" in msg
|
|
assert "http://" not in msg
|
|
|
|
|
|
def test_friendly_http_403():
|
|
resp = _make_response(403)
|
|
exc = httpx.HTTPStatusError("403", request=resp.request, response=resp)
|
|
msg = _friendly_http_error(exc)
|
|
assert "权限不足" in msg
|
|
|
|
|
|
def test_friendly_http_404():
|
|
resp = _make_response(404, path="/wrong-path")
|
|
exc = httpx.HTTPStatusError("404", request=resp.request, response=resp)
|
|
msg = _friendly_http_error(exc)
|
|
assert "接口不存在" in msg
|
|
assert "/wrong-path" in msg
|
|
# 不包含完整 URL / MDN 链接
|
|
assert "http://" not in msg
|
|
assert "MDN" not in msg
|
|
|
|
|
|
def test_friendly_http_500():
|
|
resp = _make_response(502)
|
|
exc = httpx.HTTPStatusError("502", request=resp.request, response=resp)
|
|
msg = _friendly_http_error(exc)
|
|
assert "服务异常" in msg
|
|
|
|
|
|
def test_friendly_connect_error():
|
|
exc = httpx.ConnectError("Connection refused")
|
|
msg = _friendly_connection_error(exc)
|
|
assert "无法连接" in msg
|
|
|
|
|
|
def test_friendly_timeout_error():
|
|
exc = httpx.TimeoutException("Timed out")
|
|
msg = _friendly_connection_error(exc)
|
|
assert "请求超时" in msg
|
|
|
|
|
|
# ——— get_groups fallback（通过 mock httpx client 触发真实 _request 错误转换） ———
|
|
|
|
def _mock_httpx_request(status_code: int, path: str = "/groups"):
|
|
"""返回一个 mock 的 httpx.Client.request，直接抛 HTTPStatusError。"""
|
|
def request(self, method, url, **kwargs):
|
|
resp = _make_response(status_code, path=path)
|
|
raise httpx.HTTPStatusError(f"{status_code} {path}", request=resp.request, response=resp)
|
|
return request
|
|
|
|
|
|
def test_get_groups_401_returns_friendly_auth_error(monkeypatch):
|
|
from app.services.website_client import Sub2ApiWebsiteClient
|
|
|
|
monkeypatch.setattr(httpx.Client, "request", _mock_httpx_request(401))
|
|
client = Sub2ApiWebsiteClient("http://target.local", "api/v1", "api_key", {"key": "bad"})
|
|
with pytest.raises(WebsiteError) as excinfo:
|
|
client.get_groups("/groups")
|
|
msg = str(excinfo.value)
|
|
assert "认证失败" in msg
|
|
assert "http://" not in msg
|
|
assert "MDN" not in msg
|
|
|
|
|
|
def test_get_groups_404_fallback_succeeds(monkeypatch):
|
|
from app.services.website_client import Sub2ApiWebsiteClient
|
|
|
|
call_count = 0
|
|
|
|
def request_fallback(self, method, url, **kwargs):
|
|
nonlocal call_count
|
|
call_count += 1
|
|
if call_count == 1:
|
|
resp = _make_response(404)
|
|
raise httpx.HTTPStatusError("404", request=resp.request, response=resp)
|
|
req = httpx.Request("GET", url)
|
|
return httpx.Response(200, json={"data": [{"id": "default", "name": "Default", "rate_multiplier": "1"}]}, request=req)
|
|
|
|
monkeypatch.setattr(httpx.Client, "request", request_fallback)
|
|
client = Sub2ApiWebsiteClient("http://target.local", "api/v1", "api_key", {"key": "ok"})
|
|
groups = client.get_groups("/groups")
|
|
assert len(groups) == 1
|
|
assert groups[0]["id"] == "default"
|
|
assert call_count == 2
|
|
|
|
|
|
def test_get_groups_all_404_no_raw_url(monkeypatch):
|
|
from app.services.website_client import Sub2ApiWebsiteClient
|
|
|
|
monkeypatch.setattr(httpx.Client, "request", _mock_httpx_request(404))
|
|
client = Sub2ApiWebsiteClient("http://target.local", "api/v1", "api_key", {"key": "ok"})
|
|
with pytest.raises(WebsiteError) as excinfo:
|
|
client.get_groups("/groups")
|
|
msg = str(excinfo.value)
|
|
assert "接口不存在" in msg
|
|
assert "http://" not in msg
|
|
assert "MDN" not in msg
|