Files
PolyGateway/tests/e2e/test_embed_probe.py
iomgaa 620b426ede test: keep gateway-dependent e2e out of the commit gate
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.
2026-08-20 05:49:28 -04:00

90 lines
3.1 KiB
Python

"""真实网关 /embeddings 端点探测(M2 设计 §11.6;人类默认口径: 实现时探测)。
对 .env 的 LLM 源网关发一次真实 embeddings 请求: 支持则记录向量证据,
不支持(404/翻译为领域错误)则 skip 并把响应记录进 tests/outputs/
(降级证据)。无 EMBED scope 配置时复用 LLM 源的 base_url/api_key。
"""
from __future__ import annotations
import dataclasses
import os
from datetime import datetime
from pathlib import Path
import pytest
from dotenv import dotenv_values
from polygateway.errors import PolyGatewayError
from polygateway.transports.openai_compat import OpenAICompatTransport
from polygateway.types import SourceConfig
_ENV = {k: v for k, v in {**dotenv_values(".env"), **os.environ}.items() if v is not None}
# 真实网关调用: 与 test_thinking_live.py 同待遇标 slow(pytest addopts 默认排除,
# 显式 `pytest -m slow` 运行)。理由见 test_compat_projects.py 同处注释。
pytestmark = [
pytest.mark.slow,
pytest.mark.skipif(
"LLM__MINIMAX__1__BASE_URL" not in _ENV,
reason="缺真实网关配置(.env)",
),
]
_OUT = Path("tests/outputs/embedding")
def _record(name: str, lines: list[str]) -> Path:
_OUT.mkdir(parents=True, exist_ok=True)
path = _OUT / f"{name}_{datetime.now():%Y%m%d_%H%M%S}.md"
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
return path
async def test_probe_real_gateway_embeddings():
source = SourceConfig(
name="probe_1",
provider="minimax",
base_url=_ENV["LLM__MINIMAX__1__BASE_URL"],
api_key=_ENV["LLM__MINIMAX__1__API_KEY"],
model=_ENV.get("PGW_EMBED_PROBE_MODEL", "text-embedding-v1"),
timeout_s=30.0,
est_tokens=8,
)
transport = OpenAICompatTransport()
try:
result = await transport.embed(
texts=["polygateway embedding probe"], source=source, call_id="probe"
)
except PolyGatewayError as exc:
path = _record(
"probe_unsupported",
[
"# Embedding 端点探测: 网关不支持",
f"- base_url: {source.base_url}",
f"- model: {source.model}",
f"- 错误分类: {type(exc).__name__}",
f"- status_code: {exc.status_code}",
f"- 详情: {exc}",
"",
"结论: e2e 按设计 §11.6 降级,embedding 行为由 unit 全覆盖。",
],
)
await transport.aclose()
pytest.skip(f"网关不支持 embeddings({type(exc).__name__}),证据: {path}")
else:
await transport.aclose()
assert result.dim > 0 and len(result.vectors) == 1
_record(
"probe_supported",
[
"# Embedding 端点探测: 网关支持",
f"- base_url: {source.base_url}",
f"- model: {source.model}",
f"- dim: {result.dim}",
f"- usage: {result.prompt_tokens}({result.usage_source})",
f"- 向量前 5 维: {result.vectors[0][:5]}",
f"- raw: {dataclasses.asdict(result)['raw']}",
],
)