89 lines
3.2 KiB
Python
89 lines
3.2 KiB
Python
"""GovDoc 与 Video-Tree 最小接入冒烟(2026-07-20 拍板: 两个项目都做)。
|
|
|
|
复刻两项目的真实调用点形态,对真实网关跑一次治理调用,证明"调用点零改动
|
|
迁移"成立;并验证 VT 现有平铺键名(LLM_TIMEOUT 等)可直接装配。
|
|
reference/ 只读——本文件只 import 其 Protocol,绝不修改。
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from dotenv import dotenv_values
|
|
|
|
from polygateway import GatewayClient
|
|
|
|
_REPO = Path(__file__).resolve().parents[2]
|
|
_ENV = {k: v for k, v in {**dotenv_values(".env"), **os.environ}.items() if v is not None}
|
|
_HAS_SOURCE = any(k.split("__")[0] == "LLM" and k.endswith("__API_KEY") for k in _ENV)
|
|
|
|
pytestmark = pytest.mark.skipif(
|
|
not _HAS_SOURCE, reason="需真实网关凭据: 在 .env 配置 LLM__{PROVIDER}__1__*"
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
async def client():
|
|
c = GatewayClient.from_env("LLM", env=_ENV)
|
|
yield c
|
|
await c.aclose()
|
|
|
|
|
|
class TestGovDocOnboarding:
|
|
"""GovDoc agent/loop.py:377 调用形态: session_id + parent_call_id。"""
|
|
|
|
async def test_call_site_shape_runs_governed(self, client):
|
|
response = await client.chat(
|
|
[{"role": "user", "content": "Reply with exactly: govdoc-ok"}],
|
|
session_id="govdoc-e2e",
|
|
parent_call_id="step-1",
|
|
)
|
|
assert response.content.strip()
|
|
assert response.call_id # GovernedLLMClient 契约字段全在
|
|
|
|
async def test_structural_protocol_match(self, client):
|
|
sys.path.insert(0, str(_REPO / "reference/GovDoc-SaaS/packages/docagent-core/src"))
|
|
try:
|
|
from docagent_core.protocols import LLMProvider
|
|
except ImportError:
|
|
pytest.skip("GovDoc protocols 依赖不可导入(结构断言已由单测兜底覆盖)")
|
|
finally:
|
|
sys.path.pop(0)
|
|
assert isinstance(client, LLMProvider)
|
|
|
|
|
|
class TestVideoTreeOnboarding:
|
|
"""VT loop.py:336 调用形态: session_id + cache_salt(跨 epoch 重采样)。"""
|
|
|
|
async def test_call_site_shape_with_cache_salt(self, client):
|
|
response = await client.chat(
|
|
[{"role": "user", "content": "Reply with exactly: vt-ok"}],
|
|
session_id="vt-e2e",
|
|
cache_salt="epoch-1",
|
|
)
|
|
assert response.content.strip()
|
|
|
|
async def test_flat_legacy_keys_assemble(self):
|
|
"""VT 现有键名(LLM_TIMEOUT/LLM_MAX_RETRIES 等)零改名装配成功。"""
|
|
source_keys = {k: v for k, v in _ENV.items() if k.split("__")[0] == "LLM" and "__" in k}
|
|
flat_env = {
|
|
**source_keys,
|
|
"LLM_TIMEOUT": "120",
|
|
"LLM_MAX_RETRIES": "3",
|
|
"LLM_RETRY_BASE_DELAY": "2.0",
|
|
"LLM_RETRY_MAX_DELAY": "30.0",
|
|
"LLM_CIRCUIT_BREAKER_THRESHOLD": "5",
|
|
"LLM_CIRCUIT_BREAKER_COOLDOWN": "60",
|
|
"LLM_TTFT_TIMEOUT": "30",
|
|
"LLM_INTER_TOKEN_TIMEOUT": "15",
|
|
"PGW_CACHE_BACKEND": "none",
|
|
"PGW_TELEMETRY_BACKEND": "none",
|
|
}
|
|
client = GatewayClient.from_env("LLM", env=flat_env)
|
|
try:
|
|
resp = await client.chat([{"role": "user", "content": "Reply: flat-ok"}])
|
|
assert resp.content.strip()
|
|
finally:
|
|
await client.aclose()
|