iomgaa
620b426ede
The pre-commit hook runs the whole suite, and tests/e2e/ talks to a real LLM gateway, so whether a commit is allowed depended on how fast that gateway happened to be. During the issue 14 work it blocked two commits on two different cases; both passed when rerun alone, and the suite went from 165s to 336s that hour. The wasted minutes are not the real cost. Retrying on red teaches you to read "test failed" as "gateway was slow", and a genuinely flaky bug then gets retried away too. An alarm that cries wolf stops being an alarm. test_thinking_live.py already carried the slow marker; the other three files now match it, and the release checklist gains an explicit `pytest -m slow` step so they still run where a human is watching -- without that step this change would just delete the coverage. Also raises test_flat_legacy_keys_assemble's LLM_TIMEOUT from 120 to 300, matching .env. At 120 the case allowed half of what production allows, on a gateway that needs the full 300 -- it measured 116s in a solo run. The assertion is that the flat key name parses into SourceConfig.timeout_s; the value itself was never under test.
124 lines
4.1 KiB
Python
124 lines
4.1 KiB
Python
"""真实网关端到端冒烟(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)
|
|
|
|
# 真实网关调用: 与 test_thinking_live.py 同待遇标 slow(pytest addopts 默认排除,
|
|
# 显式 `pytest -m slow` 运行)。理由见 test_compat_projects.py 同处注释。
|
|
pytestmark = [
|
|
pytest.mark.slow,
|
|
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
|