765dbdfbd4
M4（文风）: style-auditor 双轨（提取指纹/漂移第四审）+ jobs 长任务框架(zombie reaper) + 回炉 refine + GET /style read-back。 M5（生成+扩展）: worldbuilder/character-gen（入库 continuity 409 gate + partition_writes 白名单 + schema→JSONB 形变）； 网关多 provider 回退链/熔断/能力降级（Anthropic/Gemini 适配器）；Skill registry + 表权限沙箱 + 规则； 前端 角色生成器/世界观/Codex/规则页/技能库/⌘K 命令面板。 K1（Kimi Code 订阅接入）: OAuth device-flow（kimi-code）+ 静态 Console key（kimi-code-key）两路径； coding 端点 KimiCLI 伪造头（实测 UA allow-list 门禁，缺则 403）+ JSON 模式结构化（thinking ⊥ tool_choice）。 本地联调修复: CORS 中间件；assemble 注入 premise+「写第N章」指令（修空 prompt 400）； GET /outline·/draft read-back + 大纲/工作台/审稿页重载；写页 client/server 常量边界 + notFound 健壮化； 字数 toLocaleString locale 水合；审稿页终稿从已存草稿 seed（修 accept 422）。 门禁: backend ruff/mypy(157)/alembic 无漂移/pytest 451 · frontend lint/tsc/vitest/build。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
140 lines
4.7 KiB
Python
140 lines
4.7 KiB
Python
"""Kimi Code 适配器单测（K1.2）：断完整 7 头伪造头集 + base_url + bearer（不联网）。
|
||
|
||
验证点：构造出的 `AsyncOpenAI` 客户端带正确 base_url + default_headers（opencode 规范的
|
||
完整 7 头：UA `KimiCLI/1.37.0` + 6 个 `X-Msh-*`，真源 `ooojustin/opencode-kimi`），
|
||
access_token 作为 bearer，且适配器 provider 字段为 `kimi-code`。device-id 经环境变量
|
||
`KIMI_DEVICE_ID` 注入以保证 CI 确定性（不触碰真实文件/HOME）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
|
||
import pytest
|
||
from ww_llm_gateway.adapters.kimi_code import (
|
||
KIMI_CLI_VERSION,
|
||
KIMI_CODE_BASE_URL,
|
||
KIMI_CODE_PLATFORM,
|
||
KIMI_CODE_USER_AGENT,
|
||
KimiCodeAdapter,
|
||
build_kimi_code_client,
|
||
kimi_code_headers,
|
||
kimi_device_id,
|
||
)
|
||
|
||
_ACCESS_TOKEN = "kimi-oauth-access-token-not-real"
|
||
_TEST_DEVICE_ID = "0123456789abcdef0123456789abcdef"
|
||
|
||
#: opencode 发送的全部 7 个头键。
|
||
_EXPECTED_HEADER_KEYS = {
|
||
"User-Agent",
|
||
"X-Msh-Platform",
|
||
"X-Msh-Version",
|
||
"X-Msh-Device-Name",
|
||
"X-Msh-Device-Model",
|
||
"X-Msh-Device-Id",
|
||
"X-Msh-Os-Version",
|
||
}
|
||
|
||
_HEX32_RE = re.compile(r"^[0-9a-f]{32}$")
|
||
_ASCII_RE = re.compile(r"^[\x20-\x7e]+$")
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _stable_device_id(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""注入固定 device-id，使测试确定且不依赖真实 ~/.kimi 文件。"""
|
||
monkeypatch.setenv("KIMI_DEVICE_ID", _TEST_DEVICE_ID)
|
||
|
||
|
||
def test_user_agent_matches_kimi_cli_pattern() -> None:
|
||
assert KIMI_CODE_USER_AGENT == "KimiCLI/1.37.0"
|
||
assert KIMI_CODE_USER_AGENT == f"KimiCLI/{KIMI_CLI_VERSION}"
|
||
assert KIMI_CODE_USER_AGENT.startswith("KimiCLI/")
|
||
|
||
|
||
def test_headers_contain_all_seven_keys() -> None:
|
||
headers = kimi_code_headers()
|
||
|
||
assert set(headers) == _EXPECTED_HEADER_KEYS
|
||
|
||
|
||
def test_headers_fixed_literal_values() -> None:
|
||
headers = kimi_code_headers()
|
||
|
||
assert headers["User-Agent"] == "KimiCLI/1.37.0"
|
||
assert headers["X-Msh-Platform"] == KIMI_CODE_PLATFORM == "kimi_cli"
|
||
assert headers["X-Msh-Version"] == "1.37.0"
|
||
# UA 版本必须与 X-Msh-Version 一致。
|
||
assert headers["X-Msh-Version"] == headers["User-Agent"].removeprefix("KimiCLI/")
|
||
|
||
|
||
def test_device_id_is_stable_32_char_lowercase_hex() -> None:
|
||
headers_a = kimi_code_headers()
|
||
headers_b = kimi_code_headers()
|
||
|
||
device_id = headers_a["X-Msh-Device-Id"]
|
||
assert _HEX32_RE.match(device_id) # 32 位小写 hex，无连字符
|
||
assert "-" not in device_id
|
||
# 跨两次调用稳定。
|
||
assert headers_a["X-Msh-Device-Id"] == headers_b["X-Msh-Device-Id"]
|
||
assert device_id == _TEST_DEVICE_ID
|
||
|
||
|
||
def test_kimi_device_id_helper_stable_and_matches_env() -> None:
|
||
assert kimi_device_id() == _TEST_DEVICE_ID
|
||
assert kimi_device_id() == kimi_device_id()
|
||
|
||
|
||
def test_host_derived_headers_present_nonempty_ascii() -> None:
|
||
headers = kimi_code_headers()
|
||
|
||
for key in ("X-Msh-Device-Name", "X-Msh-Device-Model", "X-Msh-Os-Version"):
|
||
value = headers[key]
|
||
assert value, f"{key} must be non-empty"
|
||
assert _ASCII_RE.match(value), f"{key} must be printable ASCII"
|
||
|
||
|
||
def test_build_client_sets_base_url_and_full_headers() -> None:
|
||
client = build_kimi_code_client(_ACCESS_TOKEN)
|
||
|
||
assert str(client.base_url).rstrip("/") == KIMI_CODE_BASE_URL.rstrip("/")
|
||
assert client.default_headers["User-Agent"] == KIMI_CODE_USER_AGENT
|
||
assert _EXPECTED_HEADER_KEYS.issubset(set(client.default_headers))
|
||
|
||
|
||
def test_build_client_uses_access_token_as_bearer() -> None:
|
||
client = build_kimi_code_client(_ACCESS_TOKEN)
|
||
|
||
# access token 作为 OpenAI client 的 api_key → SDK 自动发 Authorization: Bearer。
|
||
assert client.api_key == _ACCESS_TOKEN
|
||
|
||
|
||
def test_build_client_honors_explicit_base_url_override() -> None:
|
||
custom = "https://example.test/coding/v1"
|
||
client = build_kimi_code_client(_ACCESS_TOKEN, base_url=custom)
|
||
|
||
assert str(client.base_url).rstrip("/") == custom.rstrip("/")
|
||
|
||
|
||
def test_adapter_provider_is_kimi_code() -> None:
|
||
adapter = KimiCodeAdapter(build_kimi_code_client(_ACCESS_TOKEN))
|
||
|
||
assert adapter.provider == "kimi-code"
|
||
# 复用 OpenAI 兼容能力（结构化输出 + 前缀缓存）。
|
||
caps = adapter.capabilities()
|
||
assert caps.structured_output is True
|
||
|
||
|
||
def test_structured_client_uses_instructor_json_mode() -> None:
|
||
"""kimi-for-coding 开启 thinking，与强制 tool_choice 互斥（live 400）。
|
||
|
||
因此 KimiCodeAdapter 的结构化客户端必须走 instructor JSON 模式（发
|
||
`response_format`，**不**发 `tool_choice`），而非默认的 TOOLS 模式。
|
||
"""
|
||
import instructor
|
||
|
||
adapter = KimiCodeAdapter(build_kimi_code_client(_ACCESS_TOKEN))
|
||
|
||
structured = adapter._structured()
|
||
assert getattr(structured, "mode", None) is instructor.Mode.JSON
|