118 lines
3.9 KiB
Python
118 lines
3.9 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)
|
|
|
|
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
|