test: add real-gateway and dual-project onboarding smoke scaffolding
This commit is contained in:
@@ -0,0 +1,88 @@
|
||||
"""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()
|
||||
@@ -0,0 +1,117 @@
|
||||
"""真实网关端到端冒烟(M1 验收第 7 步)。
|
||||
|
||||
前置: `.env` 配置至少一个 `LLM__{PROVIDER}__1__*` 真实源 + 韧性键。
|
||||
缺配置时 skip(验收前必须真跑)。输出结构化 Markdown 落
|
||||
`tests/outputs/e2e/`(CLAUDE.md §4.6,不提交 git)。
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from dotenv import dotenv_values
|
||||
from pydantic import BaseModel
|
||||
|
||||
from polygateway import GatewayClient
|
||||
|
||||
_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__*(M1 验收前必须真跑)"
|
||||
)
|
||||
|
||||
_OUT_DIR = Path("tests/outputs/e2e")
|
||||
|
||||
|
||||
class MiniAnswer(BaseModel):
|
||||
answer: int
|
||||
reason: str
|
||||
|
||||
|
||||
def _report(name: str, sections: list[tuple[str, str]]) -> Path:
|
||||
_OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
path = _OUT_DIR / f"{name}_{ts}.md"
|
||||
body = [f"# e2e 冒烟: {name}", ""]
|
||||
for title, content in sections:
|
||||
body += [f"## {title}", "", "```", content, "```", ""]
|
||||
path.write_text("\n".join(body), encoding="utf-8")
|
||||
return path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def client():
|
||||
c = GatewayClient.from_env("LLM", env=_ENV)
|
||||
yield c
|
||||
await c.aclose()
|
||||
|
||||
|
||||
class TestRealGatewaySmoke:
|
||||
async def test_stream_chat(self, client):
|
||||
resp = await client.chat(
|
||||
[{"role": "user", "content": "Reply with exactly: pong"}], session_id="e2e-smoke"
|
||||
)
|
||||
path = _report(
|
||||
"stream_chat",
|
||||
[
|
||||
("响应", resp.content),
|
||||
(
|
||||
"元数据",
|
||||
json.dumps(
|
||||
{
|
||||
"model": resp.model,
|
||||
"source": resp.source_name,
|
||||
"usage_source": resp.usage_source,
|
||||
"prompt_tokens": resp.prompt_tokens,
|
||||
"completion_tokens": resp.completion_tokens,
|
||||
"latency_ms": resp.latency_ms,
|
||||
"ttft_ms": resp.ttft_ms,
|
||||
},
|
||||
ensure_ascii=False,
|
||||
indent=2,
|
||||
),
|
||||
),
|
||||
],
|
||||
)
|
||||
assert resp.content.strip()
|
||||
assert resp.ttft_ms is not None and resp.latency_ms > 0
|
||||
print(f"输出: {path}")
|
||||
|
||||
async def test_non_stream_fast_path(self, client):
|
||||
resp = await client.chat(
|
||||
[{"role": "user", "content": "Reply with exactly: pong"}], stream=False
|
||||
)
|
||||
_report("non_stream", [("响应", resp.content)])
|
||||
assert resp.content.strip() and resp.ttft_ms is None
|
||||
|
||||
async def test_structured_json_tier(self, client):
|
||||
resp = await client.chat(
|
||||
[{"role": "user", "content": 'Reply ONLY with JSON: {"ok": true}'}],
|
||||
structured="json",
|
||||
)
|
||||
_report("structured_json", [("解析产物", repr(resp.structured_data))])
|
||||
assert isinstance(resp.structured_data, dict | list)
|
||||
|
||||
async def test_structured_model_ladder(self, client):
|
||||
resp = await client.chat(
|
||||
[
|
||||
{
|
||||
"role": "user",
|
||||
"content": "What is 2+3? Reply ONLY with JSON matching "
|
||||
'{"answer": <int>, "reason": <short string>}',
|
||||
}
|
||||
],
|
||||
structured=MiniAnswer,
|
||||
)
|
||||
_report(
|
||||
"structured_model",
|
||||
[
|
||||
("原始响应", resp.content),
|
||||
("校验产物", resp.structured_data.model_dump_json()),
|
||||
],
|
||||
)
|
||||
assert isinstance(resp.structured_data, MiniAnswer)
|
||||
assert resp.structured_data.answer == 5
|
||||
Reference in New Issue
Block a user