fix(cli): 统一 CLI handler pending usage 的 api_format 来源

CLI stream/sync mixin 中创建 pending usage 记录时,api_format 取值
来源不一致(部分用 FORMAT_ID,部分用 allowed_api_formats[0])。
新增 primary_api_format 属性并统一使用,确保 pending 记录的格式
与实际客户端请求格式一致。
This commit is contained in:
fawney19
2026-03-02 15:13:28 +08:00
parent 310355bcc1
commit 5f1c74aca0
4 changed files with 80 additions and 4 deletions

View File

@@ -0,0 +1,74 @@
from __future__ import annotations
import pytest
from src.api.handlers.base.cli_stream_mixin import CliStreamMixin
from src.api.handlers.base.cli_sync_mixin import CliSyncMixin
class _StopExecution(Exception):
pass
class _DummySyncHandler(CliSyncMixin):
FORMAT_ID = "openai:cli"
def __init__(self) -> None:
self.allowed_api_formats = ["openai:compact"]
self.primary_api_format = "openai:compact"
self.pending_calls: list[dict[str, object]] = []
def extract_model_from_request(
self, request_body: dict[str, object], path_params: dict[str, object] | None
) -> str:
return str(request_body.get("model") or "unknown")
def _create_pending_usage(self, **kwargs: object) -> None:
self.pending_calls.append(kwargs)
raise _StopExecution()
class _DummyStreamHandler(CliStreamMixin):
FORMAT_ID = "openai:cli"
def __init__(self) -> None:
self.allowed_api_formats = ["openai:compact"]
self.primary_api_format = "openai:compact"
self.pending_calls: list[dict[str, object]] = []
def extract_model_from_request(
self, request_body: dict[str, object], path_params: dict[str, object] | None
) -> str:
return str(request_body.get("model") or "unknown")
def _create_pending_usage(self, **kwargs: object) -> None:
self.pending_calls.append(kwargs)
raise _StopExecution()
@pytest.mark.asyncio
async def test_sync_pending_usage_uses_primary_api_format() -> None:
handler = _DummySyncHandler()
with pytest.raises(_StopExecution):
await handler.process_sync( # type: ignore[misc]
original_request_body={"model": "gpt-5.3-codex"},
original_headers={},
)
assert handler.pending_calls
assert handler.pending_calls[0]["api_format"] == "openai:compact"
@pytest.mark.asyncio
async def test_stream_pending_usage_uses_primary_api_format() -> None:
handler = _DummyStreamHandler()
with pytest.raises(_StopExecution):
await handler.process_stream( # type: ignore[misc]
original_request_body={"model": "gpt-5.3-codex"},
original_headers={},
)
assert handler.pending_calls
assert handler.pending_calls[0]["api_format"] == "openai:compact"