From 137f1ffa364e75ce5041b05266c1e9b99c87a0aa Mon Sep 17 00:00:00 2001 From: iomgaa Date: Mon, 20 Jul 2026 07:56:09 -0400 Subject: [PATCH] test: add real-gateway and dual-project onboarding smoke scaffolding --- tests/e2e/test_compat_projects.py | 88 ++++++++++++++++++++++ tests/e2e/test_smoke_gateway.py | 117 ++++++++++++++++++++++++++++++ 2 files changed, 205 insertions(+) create mode 100644 tests/e2e/test_compat_projects.py create mode 100644 tests/e2e/test_smoke_gateway.py diff --git a/tests/e2e/test_compat_projects.py b/tests/e2e/test_compat_projects.py new file mode 100644 index 0000000..d5a761d --- /dev/null +++ b/tests/e2e/test_compat_projects.py @@ -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() diff --git a/tests/e2e/test_smoke_gateway.py b/tests/e2e/test_smoke_gateway.py new file mode 100644 index 0000000..1524771 --- /dev/null +++ b/tests/e2e/test_smoke_gateway.py @@ -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": , "reason": }', + } + ], + 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