96 lines
3.7 KiB
Python
96 lines
3.7 KiB
Python
"""真实 embedding 探测；404 仅证明请求型号不可用，不外推端点能力。"""
|
|
|
|
import dataclasses
|
|
import os
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import httpx
|
|
import pytest
|
|
from dotenv import dotenv_values
|
|
|
|
from polygateway import GatewaySettings
|
|
from polygateway.transports.openai_compat import OpenAICompatTransport
|
|
from tests.e2e.conftest import LiveCapture, ObservedTransport, enforce_verdict
|
|
from tests.live_evidence import (
|
|
LiveVerdict,
|
|
classify_live_failure,
|
|
messages_digest,
|
|
request_is_valid,
|
|
safe_attempts,
|
|
write_live_round,
|
|
)
|
|
|
|
_ENV = {k: v for k, v in {**dotenv_values(".env"), **os.environ}.items() if v is not None}
|
|
pytestmark = [
|
|
pytest.mark.slow,
|
|
pytest.mark.skipif("LLM__MINIMAX__1__BASE_URL" not in _ENV, reason="缺少矩阵必需配置，未覆盖"),
|
|
]
|
|
|
|
|
|
async def test_probe_real_gateway_embeddings():
|
|
"""沿已校验源的 timeout/trust_env，所有路径 finally 关闭。"""
|
|
settings = GatewaySettings.from_env("LLM", env=_ENV)
|
|
configured = next(s for s in settings.sources if s.name == "minimax_1")
|
|
source = dataclasses.replace(
|
|
configured, model=_ENV.get("PGW_EMBED_PROBE_MODEL", "text-embedding-v1")
|
|
)
|
|
texts = ["polygateway embedding probe"]
|
|
url = httpx.URL(source.base_url)
|
|
capture = LiveCapture(
|
|
expectations={
|
|
source.name: {
|
|
"model": source.model,
|
|
"origin": str(url.copy_with(path="", query=None)).rstrip("/"),
|
|
"path": url.path.rstrip("/") + "/embeddings",
|
|
"input_shape": 1,
|
|
"control": {},
|
|
"messages_digest": messages_digest(texts),
|
|
}
|
|
}
|
|
)
|
|
real = OpenAICompatTransport(client_factory=capture.client_factory)
|
|
transport = ObservedTransport(real, capture)
|
|
run_id, parent, call_id = uuid4().hex, uuid4().hex, uuid4().hex
|
|
verdict = LiveVerdict("FAIL", "轮次未完成")
|
|
try:
|
|
with capture.round_context(session_id=run_id, parent_call_id=parent):
|
|
try:
|
|
result = await transport.embed(texts=texts, source=source, call_id=call_id)
|
|
events = [
|
|
e
|
|
for a in capture.attempts(session_id=run_id, parent_call_id=parent)
|
|
for e in a.http
|
|
]
|
|
assert len(events) == 1 and request_is_valid(events[0])
|
|
assert result.dim > 0 and len(result.vectors) == 1
|
|
verdict = LiveVerdict("PASS", "向量形状与实发请求合格")
|
|
except Exception as error:
|
|
verdict = classify_live_failure(
|
|
error, capture.attempts(session_id=run_id, parent_call_id=parent)
|
|
)
|
|
finally:
|
|
write_live_round(
|
|
Path("tests/outputs/134/live"),
|
|
run_id=run_id,
|
|
matrix_id="embedding",
|
|
round_index=1,
|
|
safe_fields={
|
|
"requested_model": source.model,
|
|
"provider": source.provider,
|
|
"planned_rounds": 1,
|
|
"completed_rounds": 1,
|
|
"status": verdict.status,
|
|
"reason": verdict.reason,
|
|
"session_id": run_id,
|
|
"parent_call_id": parent,
|
|
"attempts": safe_attempts(
|
|
capture.attempts(session_id=run_id, parent_call_id=parent)
|
|
),
|
|
"evidence_notes": capture.notes(session_id=run_id, parent_call_id=parent),
|
|
},
|
|
)
|
|
finally:
|
|
await real.aclose()
|
|
enforce_verdict(verdict)
|