142 lines
5.5 KiB
Python
142 lines
5.5 KiB
Python
"""PI provider 配置到 DSH pi-ai 接缝的合同测试。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import pathlib
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
ROOT = next(
|
|
parent
|
|
for parent in (pathlib.Path(__file__).resolve().parent, *pathlib.Path(__file__).resolve().parents)
|
|
if (parent / "AGENTS.md").is_file() and (parent / ".git").exists()
|
|
)
|
|
if str(ROOT) not in __import__("sys").path:
|
|
__import__("sys").path.insert(0, str(ROOT))
|
|
|
|
from framework.adapters.dsh.pi_ai import PiAiRouteConfigError, load_pi_ai_route # noqa: E402
|
|
from framework.adapters.dsh.runner import DshExecutionPolicy, build_dsh_patch # noqa: E402
|
|
from framework.primitives.execution import FrameworkExecutionRequest # noqa: E402
|
|
|
|
|
|
class DshPiRouteTest(unittest.TestCase):
|
|
def test_pi_models_config_becomes_secret_free_dsh_profile(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = pathlib.Path(directory) / "models.json"
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"providers": {
|
|
"catproxy-openai": {
|
|
"baseUrl": "https://gateway.example/v1",
|
|
"api": "openai-responses",
|
|
"models": [
|
|
{
|
|
"id": "gpt-test",
|
|
"name": "GPT Test",
|
|
"contextWindow": 600000,
|
|
"maxTokens": 128000,
|
|
"thinkingLevelMap": {"off": "none", "high": "high"},
|
|
}
|
|
],
|
|
}
|
|
}
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
route = load_pi_ai_route(
|
|
path,
|
|
provider="catproxy-openai",
|
|
model="gpt-test",
|
|
api_key_env="DEEPSEEK_API_KEY",
|
|
)
|
|
|
|
self.assertEqual(route.api, "openai-responses")
|
|
self.assertEqual(route.base_url, "https://gateway.example/v1")
|
|
self.assertEqual(route.model, "gpt-test")
|
|
self.assertEqual(route.api_key_env, "DEEPSEEK_API_KEY")
|
|
profile = route.as_profile()
|
|
self.assertEqual(profile["apiKeyEnv"], "DEEPSEEK_API_KEY")
|
|
self.assertNotIn("key", json.dumps(profile))
|
|
self.assertEqual(profile["models"][0]["contextWindow"], 600000)
|
|
self.assertNotIn("minimal", profile["models"][0]["reasoningEfforts"])
|
|
|
|
def test_malformed_pi_catalog_fails_with_config_error(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = pathlib.Path(directory) / "models.json"
|
|
path.write_text("[]", encoding="utf-8")
|
|
with self.assertRaises(PiAiRouteConfigError):
|
|
load_pi_ai_route(path, provider="p", model="m")
|
|
|
|
def test_runner_patch_routes_custom_provider_and_model(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = pathlib.Path(directory) / "models.json"
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"providers": {
|
|
"catproxy-openai": {
|
|
"baseUrl": "https://gateway.example/v1",
|
|
"api": "openai-responses",
|
|
"models": [{"id": "gpt-test"}],
|
|
}
|
|
}
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
route = load_pi_ai_route(path, provider="catproxy-openai", model="gpt-test")
|
|
policy = DshExecutionPolicy(
|
|
provider=route.provider,
|
|
model=route.model,
|
|
pi_ai_route=route,
|
|
)
|
|
patch = build_dsh_patch(
|
|
FrameworkExecutionRequest(system_prompt="s", user_content="u"),
|
|
policy,
|
|
session_root=directory,
|
|
)
|
|
|
|
rows = {row["id"]: row for row in patch}
|
|
self.assertEqual(
|
|
rows["agent-default-model"]["config"],
|
|
{"provider": "catproxy-openai", "model": "gpt-test"},
|
|
)
|
|
self.assertEqual(
|
|
rows["llm-pi-ai"]["config"]["providers"]["catproxy-openai"]["api"],
|
|
"openai-responses",
|
|
)
|
|
self.assertEqual(
|
|
rows["llm-pi-ai"]["config"]["providers"]["catproxy-openai"]["apiKeyEnv"],
|
|
"DEEPSEEK_API_KEY",
|
|
)
|
|
|
|
def test_route_and_policy_must_name_the_same_model(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = pathlib.Path(directory) / "models.json"
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"providers": {
|
|
"p": {
|
|
"baseUrl": "https://gateway.example/v1",
|
|
"api": "openai-responses",
|
|
"models": [{"id": "m"}],
|
|
}
|
|
}
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
route = load_pi_ai_route(path, provider="p", model="m")
|
|
with self.assertRaisesRegex(ValueError, "model"):
|
|
DshExecutionPolicy(provider="p", model="other", pi_ai_route=route)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|