feat(endpoint): 端点默认 body_rules 机制与 Codex 规则回填

- EndpointDefinition 新增 default_body_rules 字段,openai:cli/compact 配置 Codex 默认规则
- 创建端点时若未指定 body_rules 则自动填充对应格式的默认值
- 新增 GET /defaults/{api_format}/body-rules 接口查询默认规则
- 前端 EndpointFormDialog 增加"重置请求体"按钮,支持一键恢复默认
- Alembic 迁移回填已有 Codex 端点的默认 body_rules
- 新增 metadata 和 endpoint 创建默认值的单元测试
This commit is contained in:
fawney19
2026-03-02 22:08:39 +08:00
parent 3384c6d666
commit 0bff15f964
8 changed files with 392 additions and 1 deletions

View File

@@ -0,0 +1,103 @@
from types import SimpleNamespace
import pytest
from src.api.admin.endpoints import routes
from src.api.admin.endpoints.routes import AdminCreateProviderEndpointAdapter
from src.models.database import Provider, ProviderEndpoint
from src.models.endpoint_models import ProviderEndpointCreate
class _FakeQuery:
def __init__(self, result: object | None) -> None:
self._result = result
def filter(self, *_args: object, **_kwargs: object) -> "_FakeQuery":
return self
def first(self) -> object | None:
return self._result
class _FakeDB:
def __init__(self, provider: object) -> None:
self.provider = provider
self.added: ProviderEndpoint | None = None
def query(self, model: object) -> _FakeQuery:
if model is Provider:
return _FakeQuery(self.provider)
if model is ProviderEndpoint:
return _FakeQuery(None)
raise AssertionError(f"unexpected model: {model}")
def add(self, obj: ProviderEndpoint) -> None:
self.added = obj
def commit(self) -> None:
return None
def refresh(self, _obj: ProviderEndpoint) -> None:
return None
async def _noop_invalidate_cache() -> None:
return None
@pytest.mark.asyncio
async def test_create_endpoint_injects_default_body_rules_when_missing(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(routes, "invalidate_models_list_cache", _noop_invalidate_cache)
monkeypatch.setattr(
routes,
"get_default_body_rules_for_endpoint",
lambda _fmt: [{"action": "drop", "path": "max_output_tokens"}],
)
db = _FakeDB(
provider=SimpleNamespace(id="p1", name="P1", provider_type="custom"),
)
adapter = AdminCreateProviderEndpointAdapter(
provider_id="p1",
endpoint_data=ProviderEndpointCreate(
provider_id="p1",
api_format="openai:cli",
base_url="https://api.example.com",
),
)
await adapter.handle(SimpleNamespace(db=db)) # type: ignore[arg-type]
assert db.added is not None
assert db.added.body_rules == [{"action": "drop", "path": "max_output_tokens"}]
@pytest.mark.asyncio
async def test_create_endpoint_keeps_user_body_rules_when_provided(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(routes, "invalidate_models_list_cache", _noop_invalidate_cache)
monkeypatch.setattr(
routes,
"get_default_body_rules_for_endpoint",
lambda _fmt: [{"action": "drop", "path": "max_output_tokens"}],
)
user_rules = [{"action": "set", "path": "metadata.source", "value": "user"}]
db = _FakeDB(
provider=SimpleNamespace(id="p1", name="P1", provider_type="custom"),
)
adapter = AdminCreateProviderEndpointAdapter(
provider_id="p1",
endpoint_data=ProviderEndpointCreate(
provider_id="p1",
api_format="openai:cli",
base_url="https://api.example.com",
body_rules=user_rules,
),
)
await adapter.handle(SimpleNamespace(db=db)) # type: ignore[arg-type]
assert db.added is not None
assert db.added.body_rules == user_rules

View File

@@ -0,0 +1,61 @@
import pytest
import src.core.api_format.metadata as metadata
from src.core.api_format.enums import ApiFamily, EndpointKind
from src.core.api_format.metadata import EndpointDefinition, get_default_body_rules_for_endpoint
def test_get_default_body_rules_for_endpoint_returns_empty_for_invalid() -> None:
assert get_default_body_rules_for_endpoint("not-a-valid-signature") == []
def test_get_default_body_rules_for_endpoint_returns_empty_for_no_rules() -> None:
"""claude:chat 没有配置 default_body_rules应返回空列表。"""
assert get_default_body_rules_for_endpoint("claude:chat") == []
def test_get_default_body_rules_for_endpoint_returns_codex_rules() -> None:
"""openai:cli 和 openai:compact 应返回 Codex 默认规则。"""
cli_rules = get_default_body_rules_for_endpoint("openai:cli")
assert len(cli_rules) == 5
actions = [r["action"] for r in cli_rules]
assert actions == ["drop", "drop", "drop", "set", "set"]
assert cli_rules[0]["path"] == "max_output_tokens"
assert cli_rules[1]["path"] == "temperature"
assert cli_rules[2]["path"] == "top_p"
assert cli_rules[3] == {"action": "set", "path": "store", "value": False}
assert cli_rules[4]["path"] == "instructions"
assert cli_rules[4]["condition"]["op"] == "not_exists"
compact_rules = get_default_body_rules_for_endpoint("openai:compact")
assert compact_rules == cli_rules
def test_get_default_body_rules_for_endpoint_returns_deep_copy(
monkeypatch: pytest.MonkeyPatch,
) -> None:
definition = EndpointDefinition(
api_family=ApiFamily.OPENAI,
endpoint_kind=EndpointKind.CLI,
default_body_rules=(
{
"action": "set",
"path": "metadata",
"value": {"source": "default"},
},
),
)
monkeypatch.setattr(metadata, "resolve_endpoint_definition", lambda _value: definition)
rules = get_default_body_rules_for_endpoint("openai:cli")
assert rules == [
{
"action": "set",
"path": "metadata",
"value": {"source": "default"},
}
]
rules[0]["value"]["source"] = "changed"
assert definition.default_body_rules[0]["value"]["source"] == "default"