fix: shorten New-API token names
This commit is contained in:
liumangmang
@@ -823,3 +823,399 @@ def test_generate_keys_allows_new_api_user_upstream(db_session, monkeypatch):
|
||||
assert response.success is True
|
||||
assert response.items[0].status == "created"
|
||||
assert response.items[0].key_value == "new-api-plain-key"
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────
|
||||
# Tests for build_new_api_token_name helper
|
||||
# ──────────────────────────────────────────────
|
||||
|
||||
def test_build_new_api_token_name_fits_50_bytes_with_long_chinese_group():
|
||||
"""长中文分组名 + 长 group_id 生成的 token name 不超过 50 UTF-8 字节。"""
|
||||
from app.services.upstream_client import build_new_api_token_name
|
||||
|
||||
name = build_new_api_token_name(
|
||||
prefix="SmartUp",
|
||||
upstream_id=7,
|
||||
group_id="claude-max-智能分组-special",
|
||||
group_name="ClaudeMax智能分组SuperLongNameThatWouldBreakOldFormat",
|
||||
)
|
||||
assert len(name.encode("utf-8")) <= 50, (
|
||||
f"token name too long: {len(name.encode('utf-8'))} bytes — '{name}'"
|
||||
)
|
||||
assert name.startswith("SmartUp-7-")
|
||||
|
||||
|
||||
def test_build_new_api_token_name_no_broken_multibyte():
|
||||
"""UTF-8 截断后不应出现乱码（decode 不报错）。"""
|
||||
from app.services.upstream_client import build_new_api_token_name
|
||||
|
||||
name = build_new_api_token_name(
|
||||
prefix="SmartUp",
|
||||
upstream_id=99,
|
||||
group_id="group-x",
|
||||
group_name="中文分组名称超长超长超长超长超长超长超长超长超长超长",
|
||||
)
|
||||
# 结果必须是有效 UTF-8 字符串，且 ≤ 50 字节
|
||||
encoded = name.encode("utf-8")
|
||||
assert len(encoded) <= 50
|
||||
# 没有乱码（可以解回去）
|
||||
assert encoded.decode("utf-8") == name
|
||||
|
||||
|
||||
def test_build_new_api_token_name_different_group_ids_produce_different_names():
|
||||
"""两个相同短分组名但不同 group_id 生成不同 token name。"""
|
||||
from app.services.upstream_client import build_new_api_token_name
|
||||
|
||||
name1 = build_new_api_token_name("SmartUp", 7, "group-a", "智能分组")
|
||||
name2 = build_new_api_token_name("SmartUp", 7, "group-b", "智能分组")
|
||||
assert name1 != name2, "相同分组名但不同 group_id 应生成不同 token name"
|
||||
|
||||
|
||||
def test_build_new_api_token_name_prefix_too_long_raises():
|
||||
"""prefix 过长导致骨架超 50 字节时应抛出 ValueError。"""
|
||||
from app.services.upstream_client import build_new_api_token_name
|
||||
|
||||
# 超长前缀：40 个字符 + upstream_id + separators + hash8 会超 50
|
||||
very_long_prefix = "A" * 40
|
||||
with pytest.raises(ValueError, match="过长"):
|
||||
build_new_api_token_name(very_long_prefix, 99, "grp", "name")
|
||||
|
||||
|
||||
def test_ensure_group_key_reuses_old_key_name_when_local_record_exists(db_session, monkeypatch):
|
||||
"""本地已有旧格式 key_name 记录时，应用旧 key_name 查远端，不重新创建。"""
|
||||
from app.routers.upstreams import _ensure_group_key
|
||||
from app.models.upstream_key import UpstreamGeneratedKey
|
||||
from app.schemas.upstream import GenerateKeysByGroupsRequest
|
||||
|
||||
# New-API 上游（auth_type=cookie + new_api_user → 使用新短名逻辑）
|
||||
upstream = Upstream(
|
||||
name="NewAPI",
|
||||
base_url="http://newapi.local",
|
||||
api_prefix="",
|
||||
auth_type="cookie",
|
||||
auth_config_json=json.dumps({"cookie_string": "s=x", "new_api_user": "7"}),
|
||||
groups_endpoint="/api/user/self/groups",
|
||||
rate_endpoint="/api/user/self/groups",
|
||||
)
|
||||
db_session.add(upstream)
|
||||
db_session.commit()
|
||||
db_session.refresh(upstream)
|
||||
|
||||
old_key_name = f"SmartUp-{upstream.id}-ClaudeMax智能分组-claude-max-智能分组"
|
||||
db_session.add(UpstreamGeneratedKey(
|
||||
upstream_id=upstream.id,
|
||||
group_id="claude-max-智能分组",
|
||||
group_name="ClaudeMax智能分组",
|
||||
key_name=old_key_name,
|
||||
key_value="sk-old",
|
||||
managed_prefix="SmartUp",
|
||||
key_id="remote-old-id",
|
||||
))
|
||||
db_session.commit()
|
||||
|
||||
find_calls: list[str] = []
|
||||
|
||||
class MockClient:
|
||||
def find_smartup_group_key(self, gid, name, prefix):
|
||||
find_calls.append(name)
|
||||
# 旧格式能找到
|
||||
if name == old_key_name:
|
||||
return {
|
||||
"id": "remote-old-id",
|
||||
"name": old_key_name,
|
||||
"key": "sk-old-plain",
|
||||
}
|
||||
return None
|
||||
|
||||
def create_api_key(self, *args, **kwargs):
|
||||
raise AssertionError("create_api_key should NOT be called when old remote key exists")
|
||||
|
||||
group = {"id": "claude-max-智能分组", "name": "ClaudeMax智能分组"}
|
||||
body = GenerateKeysByGroupsRequest(group_ids=["claude-max-智能分组"], name_prefix="SmartUp", quota=0)
|
||||
|
||||
result = _ensure_group_key(db_session, MockClient(), upstream, group, "SmartUp", body)
|
||||
|
||||
assert result.status == "exists"
|
||||
# 第一次调用用的应该是旧 key_name
|
||||
assert old_key_name in find_calls, f"expected old key_name in find calls, got {find_calls}"
|
||||
|
||||
|
||||
def test_ensure_group_key_new_api_uses_short_name_for_create(db_session, monkeypatch):
|
||||
"""New-API 上游新建时，发往远端的 token name 是新短名（≤50 字节）。"""
|
||||
from app.routers.upstreams import _ensure_group_key
|
||||
from app.schemas.upstream import GenerateKeysByGroupsRequest
|
||||
from app.services.upstream_client import build_new_api_token_name
|
||||
|
||||
upstream = Upstream(
|
||||
name="NewAPI",
|
||||
base_url="http://newapi.local",
|
||||
api_prefix="",
|
||||
auth_type="cookie",
|
||||
auth_config_json=json.dumps({"cookie_string": "s=x", "new_api_user": "7"}),
|
||||
groups_endpoint="/api/user/self/groups",
|
||||
rate_endpoint="/api/user/self/groups",
|
||||
)
|
||||
db_session.add(upstream)
|
||||
db_session.commit()
|
||||
db_session.refresh(upstream)
|
||||
|
||||
gid = "claude-max-智能分组"
|
||||
gname = "ClaudeMax智能分组"
|
||||
expected_name = build_new_api_token_name("SmartUp", upstream.id, gid, gname)
|
||||
|
||||
created_names: list[str] = []
|
||||
|
||||
class MockClient:
|
||||
def find_smartup_group_key(self, group_id, name, prefix):
|
||||
return None # 远端不存在
|
||||
|
||||
def create_api_key(self, name, group_id, **kwargs):
|
||||
created_names.append(name)
|
||||
return {"id": "new-123", "key": "sk-new-plain", "masked_key": "sk-****"}
|
||||
|
||||
group = {"id": gid, "name": gname}
|
||||
body = GenerateKeysByGroupsRequest(group_ids=[gid], name_prefix="SmartUp", quota=0)
|
||||
|
||||
result = _ensure_group_key(db_session, MockClient(), upstream, group, "SmartUp", body)
|
||||
|
||||
assert result.status == "created"
|
||||
assert len(created_names) == 1
|
||||
assert created_names[0] == expected_name, (
|
||||
f"expected short name '{expected_name}', got '{created_names[0]}'"
|
||||
)
|
||||
assert len(expected_name.encode("utf-8")) <= 50
|
||||
|
||||
|
||||
def test_ensure_group_key_prefix_too_long_returns_failed(db_session):
|
||||
"""prefix 过长时，_ensure_group_key 应返回 failed 状态，错误信息指向 prefix 过长。"""
|
||||
from app.routers.upstreams import _ensure_group_key
|
||||
from app.schemas.upstream import GenerateKeysByGroupsRequest
|
||||
|
||||
upstream = Upstream(
|
||||
name="NewAPI",
|
||||
base_url="http://newapi.local",
|
||||
api_prefix="",
|
||||
auth_type="cookie",
|
||||
auth_config_json=json.dumps({"cookie_string": "s=x", "new_api_user": "7"}),
|
||||
groups_endpoint="/api/user/self/groups",
|
||||
rate_endpoint="/api/user/self/groups",
|
||||
)
|
||||
db_session.add(upstream)
|
||||
db_session.commit()
|
||||
db_session.refresh(upstream)
|
||||
|
||||
very_long_prefix = "MyCompanyLongPrefixName" * 3 # >> 50 bytes
|
||||
|
||||
class MockClient:
|
||||
def find_smartup_group_key(self, *a, **kw):
|
||||
raise AssertionError("should not reach remote call")
|
||||
def create_api_key(self, *a, **kw):
|
||||
raise AssertionError("should not reach remote call")
|
||||
|
||||
group = {"id": "vip", "name": "VIP"}
|
||||
body = GenerateKeysByGroupsRequest(group_ids=["vip"], name_prefix=very_long_prefix, quota=0)
|
||||
|
||||
result = _ensure_group_key(db_session, MockClient(), upstream, group, very_long_prefix, body)
|
||||
|
||||
assert result.status == "failed"
|
||||
assert "过长" in (result.error or ""), f"expected '过长' in error, got: {result.error}"
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────
|
||||
# P1 Regression tests
|
||||
# ──────────────────────────────────────────────
|
||||
|
||||
def test_p1a_local_key_name_without_key_id_still_queries_remote(db_session):
|
||||
"""P1-A: 本地有旧 key_name 但 key_id 为空时，应用旧 key_name 查远端，不跳过。"""
|
||||
from app.routers.upstreams import _ensure_group_key
|
||||
from app.models.upstream_key import UpstreamGeneratedKey
|
||||
from app.schemas.upstream import GenerateKeysByGroupsRequest
|
||||
|
||||
upstream = Upstream(
|
||||
name="NewAPI",
|
||||
base_url="http://newapi.local",
|
||||
api_prefix="",
|
||||
auth_type="cookie",
|
||||
auth_config_json=json.dumps({"cookie_string": "s=x", "new_api_user": "7"}),
|
||||
groups_endpoint="/api/user/self/groups",
|
||||
rate_endpoint="/api/user/self/groups",
|
||||
)
|
||||
db_session.add(upstream)
|
||||
db_session.commit()
|
||||
db_session.refresh(upstream)
|
||||
|
||||
old_key_name = f"SmartUp-{upstream.id}-VIP-vip"
|
||||
# 本地记录：有 key_name，但 key_id 为 None
|
||||
db_session.add(UpstreamGeneratedKey(
|
||||
upstream_id=upstream.id,
|
||||
group_id="vip",
|
||||
group_name="VIP",
|
||||
key_name=old_key_name,
|
||||
key_value="", # 无明文
|
||||
masked_key="sk-old-masked",
|
||||
managed_prefix="SmartUp",
|
||||
key_id=None, # ← P1-A 的关键：key_id 为空
|
||||
))
|
||||
db_session.commit()
|
||||
|
||||
find_calls: list[str] = []
|
||||
|
||||
class MockClient:
|
||||
def find_smartup_group_key(self, gid, name, prefix):
|
||||
find_calls.append(name)
|
||||
if name == old_key_name:
|
||||
return {"id": "remote-777", "name": old_key_name, "key": "sk-plain-found"}
|
||||
return None
|
||||
|
||||
def create_api_key(self, *args, **kwargs):
|
||||
raise AssertionError("create_api_key should NOT be called when remote key found by old name")
|
||||
|
||||
group = {"id": "vip", "name": "VIP"}
|
||||
body = GenerateKeysByGroupsRequest(group_ids=["vip"], name_prefix="SmartUp", quota=0)
|
||||
result = _ensure_group_key(db_session, MockClient(), upstream, group, "SmartUp", body)
|
||||
|
||||
assert result.status == "exists", f"expected exists, got {result.status}"
|
||||
# 第一次查询必须用旧 key_name
|
||||
assert find_calls[0] == old_key_name, (
|
||||
f"expected first find call with old_key_name='{old_key_name}', got {find_calls}"
|
||||
)
|
||||
# 本地记录 key_id 已被回填
|
||||
row = db_session.query(UpstreamGeneratedKey).filter(
|
||||
UpstreamGeneratedKey.upstream_id == upstream.id,
|
||||
UpstreamGeneratedKey.group_id == "vip",
|
||||
).one()
|
||||
assert row.key_id == "remote-777"
|
||||
assert row.key_value == "sk-plain-found"
|
||||
|
||||
|
||||
def test_p1b_no_local_record_old_format_remote_token_imported_not_duplicated(db_session):
|
||||
"""P1-B: 无本地记录但远端已有旧格式 token 时，应导入旧 token，不创建新 token。"""
|
||||
from app.routers.upstreams import _ensure_group_key
|
||||
from app.models.upstream_key import UpstreamGeneratedKey
|
||||
from app.schemas.upstream import GenerateKeysByGroupsRequest
|
||||
from app.services.upstream_client import build_new_api_token_name
|
||||
|
||||
upstream = Upstream(
|
||||
name="NewAPI",
|
||||
base_url="http://newapi.local",
|
||||
api_prefix="",
|
||||
auth_type="cookie",
|
||||
auth_config_json=json.dumps({"cookie_string": "s=x", "new_api_user": "7"}),
|
||||
groups_endpoint="/api/user/self/groups",
|
||||
rate_endpoint="/api/user/self/groups",
|
||||
)
|
||||
db_session.add(upstream)
|
||||
db_session.commit()
|
||||
db_session.refresh(upstream)
|
||||
|
||||
gid = "vip"
|
||||
gname = "VIP"
|
||||
# 旧格式 token 名（未截断的完整 group_id 拼接）
|
||||
old_format_name = f"SmartUp-{upstream.id}-VIP-{gid}"
|
||||
# 新短名（不同于旧格式）
|
||||
new_short_name = build_new_api_token_name("SmartUp", upstream.id, gid, gname)
|
||||
assert old_format_name != new_short_name, "test setup: old and new names must differ"
|
||||
|
||||
find_calls: list[str] = []
|
||||
create_calls: list[str] = []
|
||||
|
||||
class MockClient:
|
||||
def find_smartup_group_key(self, group_id, name, prefix):
|
||||
find_calls.append(name)
|
||||
# 仅旧格式能找到
|
||||
if name == old_format_name:
|
||||
return {"id": "remote-old-888", "name": old_format_name, "key": "sk-old-plain"}
|
||||
return None
|
||||
|
||||
def create_api_key(self, name, group_id, **kwargs):
|
||||
create_calls.append(name)
|
||||
return {"id": "new-999", "key": "sk-new", "masked_key": "sk-****"}
|
||||
|
||||
group = {"id": gid, "name": gname}
|
||||
body = GenerateKeysByGroupsRequest(group_ids=[gid], name_prefix="SmartUp", quota=0)
|
||||
result = _ensure_group_key(db_session, MockClient(), upstream, group, "SmartUp", body)
|
||||
|
||||
assert result.status == "exists", f"expected exists (imported old token), got {result.status}"
|
||||
assert len(create_calls) == 0, f"create_api_key should not be called, got: {create_calls}"
|
||||
|
||||
# 本地应只有一条记录，key_name 保留的是旧格式（不覆写为新短名）
|
||||
rows = db_session.query(UpstreamGeneratedKey).filter(
|
||||
UpstreamGeneratedKey.upstream_id == upstream.id,
|
||||
UpstreamGeneratedKey.group_id == gid,
|
||||
).all()
|
||||
assert len(rows) == 1, f"expected exactly 1 local record, got {len(rows)}"
|
||||
assert rows[0].key_id == "remote-old-888"
|
||||
assert rows[0].key_value == "sk-old-plain"
|
||||
assert rows[0].key_name == old_format_name
|
||||
|
||||
|
||||
def test_p3_local_old_key_name_synced_to_new_short_name_when_remote_migrated(db_session):
|
||||
"""P3: 本地旧 key_name 在远端已消失，远端新短名存在时，本地 key_name 应同步更新为新短名。"""
|
||||
from app.routers.upstreams import _ensure_group_key
|
||||
from app.models.upstream_key import UpstreamGeneratedKey
|
||||
from app.schemas.upstream import GenerateKeysByGroupsRequest
|
||||
from app.services.upstream_client import build_new_api_token_name
|
||||
|
||||
upstream = Upstream(
|
||||
name="NewAPI",
|
||||
base_url="http://newapi.local",
|
||||
api_prefix="",
|
||||
auth_type="cookie",
|
||||
auth_config_json=json.dumps({"cookie_string": "s=x", "new_api_user": "7"}),
|
||||
groups_endpoint="/api/user/self/groups",
|
||||
rate_endpoint="/api/user/self/groups",
|
||||
)
|
||||
db_session.add(upstream)
|
||||
db_session.commit()
|
||||
db_session.refresh(upstream)
|
||||
|
||||
gid = "vip"
|
||||
gname = "VIP"
|
||||
old_key_name = f"SmartUp-{upstream.id}-VIP-vip" # 旧格式
|
||||
new_short_name = build_new_api_token_name("SmartUp", upstream.id, gid, gname)
|
||||
assert old_key_name != new_short_name, "test setup: old and new names must differ"
|
||||
|
||||
# 本地记录用旧名，有 key_id
|
||||
db_session.add(UpstreamGeneratedKey(
|
||||
upstream_id=upstream.id,
|
||||
group_id=gid,
|
||||
group_name=gname,
|
||||
key_name=old_key_name,
|
||||
key_value="sk-stale",
|
||||
managed_prefix="SmartUp",
|
||||
key_id="remote-old-id",
|
||||
))
|
||||
db_session.commit()
|
||||
|
||||
class MockClient:
|
||||
def find_smartup_group_key(self, group_id, name, prefix):
|
||||
if name == old_key_name:
|
||||
return None # 旧名在远端已消失
|
||||
if name == new_short_name:
|
||||
return { # 新短名在远端存在
|
||||
"id": "remote-new-id",
|
||||
"name": new_short_name,
|
||||
"key": "sk-new-plain",
|
||||
}
|
||||
return None
|
||||
|
||||
def create_api_key(self, *a, **kw):
|
||||
raise AssertionError("create_api_key should not be called")
|
||||
|
||||
group = {"id": gid, "name": gname}
|
||||
body = GenerateKeysByGroupsRequest(group_ids=[gid], name_prefix="SmartUp", quota=0)
|
||||
result = _ensure_group_key(db_session, MockClient(), upstream, group, "SmartUp", body)
|
||||
|
||||
# 远端新短名找到 → status=exists，不创建
|
||||
assert result.status == "exists", f"expected exists, got {result.status}"
|
||||
|
||||
# 本地 key_name 应已同步为新短名（P3 一致性）
|
||||
row = db_session.query(UpstreamGeneratedKey).filter(
|
||||
UpstreamGeneratedKey.upstream_id == upstream.id,
|
||||
UpstreamGeneratedKey.group_id == gid,
|
||||
).one()
|
||||
assert row.key_name == new_short_name, (
|
||||
f"expected key_name updated to '{new_short_name}', got '{row.key_name}'"
|
||||
)
|
||||
assert row.key_id == "remote-new-id"
|
||||
assert row.key_value == "sk-new-plain"
|
||||
|
||||
Reference in New Issue
Block a user