test: guard reasoning-free telemetry through real client paths
This commit is contained in:
@@ -533,3 +533,70 @@ class TestEmbeddingSettings:
|
||||
s = EmbeddingSettings.from_env("EMBED", env=self._ENV)
|
||||
client = EmbeddingClient.from_settings(s)
|
||||
assert isinstance(client, EmbeddingClient)
|
||||
|
||||
|
||||
class TestReasonlessTelemetryContract:
|
||||
"""从真实客户端到落库,误配推理配置也不能产生推理档。"""
|
||||
|
||||
@pytest.mark.parametrize("config", [{"enable_thinking": True}, {"reasoning_effort": "high"}])
|
||||
@pytest.mark.parametrize("backend", ["memory", "sqlite"])
|
||||
async def test_failed_then_successful_attempts_have_null_effort(
|
||||
self, config, backend, tmp_path
|
||||
):
|
||||
import sqlite3
|
||||
|
||||
from polygateway.telemetry.sqlite import SQLiteRecorder
|
||||
|
||||
path = tmp_path / "embed.sqlite"
|
||||
recorder = (
|
||||
_MemoryRecorder() if backend == "memory" else SQLiteRecorder(path, auto_migrate=True)
|
||||
)
|
||||
client, _ = _embed_client(
|
||||
[_src(**config)], [TransientError("retry"), "ok"], telemetry=recorder
|
||||
)
|
||||
try:
|
||||
await client.embed(["text"], session_id="run", parent_call_id="embed")
|
||||
if backend == "memory":
|
||||
rows = [(r["error"], r["reasoning_effort"]) for r in recorder.rows]
|
||||
else:
|
||||
with sqlite3.connect(path) as db:
|
||||
rows = db.execute("SELECT error, reasoning_effort FROM llm_calls").fetchall()
|
||||
assert len(rows) == 2
|
||||
assert sum(bool(error) for error, _ in rows) == 1
|
||||
assert [tier for _, tier in rows] == [None, None]
|
||||
finally:
|
||||
await client.aclose()
|
||||
if backend == "sqlite":
|
||||
recorder.close()
|
||||
|
||||
@pytest.mark.parametrize("exhausted", [False, True])
|
||||
async def test_failed_attempts_still_have_null_effort(self, exhausted):
|
||||
from polygateway.errors import AllSourcesExhausted
|
||||
|
||||
script = (
|
||||
[TransientError("retry")] * 3 if exhausted else [RequestRejectedError("bad request")]
|
||||
)
|
||||
recorder = _MemoryRecorder()
|
||||
client, _ = _embed_client([_src(enable_thinking=True)], script, telemetry=recorder)
|
||||
with pytest.raises(AllSourcesExhausted if exhausted else RequestRejectedError):
|
||||
await client.embed(["text"])
|
||||
assert len(recorder.rows) == len(script)
|
||||
assert all(r["error"] and r["reasoning_effort"] is None for r in recorder.rows)
|
||||
|
||||
async def test_embedding_wire_ignores_reasoning_configuration(self):
|
||||
seen = []
|
||||
|
||||
def handler(request):
|
||||
seen.append(json.loads(request.content))
|
||||
return httpx.Response(200, json=_ok_body([[1.0]], usage={"prompt_tokens": 1}))
|
||||
|
||||
transport = _transport_with(handler)
|
||||
try:
|
||||
await transport.embed(
|
||||
texts=["text"],
|
||||
source=_src(enable_thinking=True, reasoning_effort="high"),
|
||||
call_id="wire",
|
||||
)
|
||||
assert seen == [{"model": "embed-1", "input": ["text"]}]
|
||||
finally:
|
||||
await transport.aclose()
|
||||
|
||||
@@ -405,3 +405,31 @@ class TestLifecycle:
|
||||
await t.check_health(source=_source())
|
||||
await t.aclose()
|
||||
await t.aclose()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("method", ["recognize_text", "parse_layout"])
|
||||
async def test_ocr_wire_does_not_send_reasoning_configuration(method):
|
||||
"""真实 multipart 与 ZIP 下载两段均不发送源级推理配置。"""
|
||||
sent = []
|
||||
|
||||
def handler(request):
|
||||
sent.append(request)
|
||||
if request.method == "GET":
|
||||
return httpx.Response(200, content=_zip_bytes())
|
||||
return httpx.Response(
|
||||
200, json=_text_body() if method == "recognize_text" else _parse_body()
|
||||
)
|
||||
|
||||
transport = _transport_for(handler)
|
||||
try:
|
||||
await getattr(transport, method)(
|
||||
image=b"image",
|
||||
source=_source(enable_thinking=True, reasoning_effort="high"),
|
||||
call_id="wire",
|
||||
)
|
||||
assert len(sent) == (1 if method == "recognize_text" else 2)
|
||||
for request in sent:
|
||||
for key in (b"reasoning_effort", b"enable_thinking", b"thinking_budget"):
|
||||
assert key not in request.content
|
||||
finally:
|
||||
await transport.aclose()
|
||||
|
||||
@@ -564,3 +564,54 @@ class TestAssembly:
|
||||
client = OcrClient.from_env("OCR", env=dict(self._ENV))
|
||||
await client.aclose()
|
||||
await client.aclose()
|
||||
|
||||
|
||||
class TestReasonlessTelemetryContract:
|
||||
"""text/layout 两个入口分别验证错误行不受源级推理配置污染。"""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"method,action", [("recognize_text", "text"), ("parse_layout", "layout")]
|
||||
)
|
||||
@pytest.mark.parametrize("config", [{"enable_thinking": True}, {"reasoning_effort": "high"}])
|
||||
@pytest.mark.parametrize("backend", ["memory", "sqlite"])
|
||||
async def test_failed_then_successful_attempts_have_null_effort(
|
||||
self, method, action, config, backend, tmp_path
|
||||
):
|
||||
import sqlite3
|
||||
|
||||
from polygateway.telemetry.sqlite import SQLiteRecorder
|
||||
|
||||
path = tmp_path / "ocr.sqlite"
|
||||
recorder = (
|
||||
_MemoryRecorder() if backend == "memory" else SQLiteRecorder(path, auto_migrate=True)
|
||||
)
|
||||
client, _, _ = _client(
|
||||
[_src(**config)], [TransientError("retry"), action], telemetry=recorder
|
||||
)
|
||||
try:
|
||||
await getattr(client, method)(b"image", session_id="run", parent_call_id=method)
|
||||
if backend == "memory":
|
||||
rows = [(r["error"], r["reasoning_effort"]) for r in recorder.rows]
|
||||
else:
|
||||
with sqlite3.connect(path) as db:
|
||||
rows = db.execute("SELECT error, reasoning_effort FROM llm_calls").fetchall()
|
||||
assert len(rows) == 2
|
||||
assert sum(bool(error) for error, _ in rows) == 1
|
||||
assert [tier for _, tier in rows] == [None, None]
|
||||
finally:
|
||||
await client.aclose()
|
||||
if backend == "sqlite":
|
||||
recorder.close()
|
||||
|
||||
@pytest.mark.parametrize("method", ["recognize_text", "parse_layout"])
|
||||
@pytest.mark.parametrize("exhausted", [False, True])
|
||||
async def test_failed_attempts_still_have_null_effort(self, method, exhausted):
|
||||
script = (
|
||||
[TransientError("retry")] * 3 if exhausted else [RequestRejectedError("bad request")]
|
||||
)
|
||||
recorder = _MemoryRecorder()
|
||||
client, _, _ = _client([_src(enable_thinking=True)], script, telemetry=recorder)
|
||||
with pytest.raises(AllSourcesExhausted if exhausted else RequestRejectedError):
|
||||
await getattr(client, method)(b"image")
|
||||
assert len(recorder.rows) == len(script)
|
||||
assert all(r["error"] and r["reasoning_effort"] is None for r in recorder.rows)
|
||||
|
||||
@@ -2839,3 +2839,90 @@ class TestPostgresFailureClassification:
|
||||
|
||||
assert [s for s in conn.statements if s.startswith("INSERT INTO llm_calls")]
|
||||
assert recorder.telemetry_status.degraded is False
|
||||
|
||||
|
||||
class TestConcurrentReasoningPathContracts:
|
||||
"""真实治理链路并发时,按逻辑调用归组且 attempts 不串档。"""
|
||||
|
||||
async def test_chat_and_reasonless_clients_keep_distinct_rows(self):
|
||||
from polygateway.errors import TransientError
|
||||
from tests.unit.test_client import _client as chat_client
|
||||
from tests.unit.test_client import _source, _sse
|
||||
from tests.unit.test_embedding import _embed_client
|
||||
from tests.unit.test_embedding import _src as embed_source
|
||||
from tests.unit.test_ocr_client import _client as ocr_client
|
||||
from tests.unit.test_ocr_client import _src as ocr_source
|
||||
|
||||
recorder = _MemoryRecorder()
|
||||
calls = 0
|
||||
|
||||
def handler(request):
|
||||
nonlocal calls
|
||||
calls += 1
|
||||
if calls == 1:
|
||||
import httpx
|
||||
|
||||
return httpx.Response(503, json={"error": {"message": "retry"}})
|
||||
return _sse()
|
||||
|
||||
chat = chat_client(
|
||||
sources=[_source(provider="zhipu", model="glm-5.3", effort_fallback="nearest")],
|
||||
handler=handler,
|
||||
telemetry=recorder,
|
||||
retry=RetryPolicy(3, 0.001, 0.01),
|
||||
)
|
||||
embed, _ = _embed_client(
|
||||
[embed_source(enable_thinking=True)],
|
||||
[TransientError("retry"), "ok"],
|
||||
telemetry=recorder,
|
||||
)
|
||||
ocr, _, _ = ocr_client(
|
||||
[ocr_source(enable_thinking=True)],
|
||||
[TransientError("retry"), "text"],
|
||||
telemetry=recorder,
|
||||
)
|
||||
try:
|
||||
await asyncio.gather(
|
||||
chat.chat([], reasoning_effort="medium", session_id="run", parent_call_id="chat"),
|
||||
embed.embed(["text"], session_id="run", parent_call_id="embed"),
|
||||
ocr.recognize_text(b"image", session_id="run", parent_call_id="ocr"),
|
||||
)
|
||||
groups = {
|
||||
name: [r for r in recorder.rows if r["parent_call_id"] == name]
|
||||
for name in ("chat", "embed", "ocr")
|
||||
}
|
||||
assert len(recorder.rows) == 6
|
||||
assert len({r["call_id"] for r in recorder.rows}) == 6
|
||||
assert all(r["session_id"] == "run" for r in recorder.rows)
|
||||
assert [r["reasoning_effort"] for r in groups["chat"]] == ["medium", "low"]
|
||||
for name in ("embed", "ocr"):
|
||||
assert len(groups[name]) == 2
|
||||
assert [r["reasoning_effort"] for r in groups[name]] == [None, None]
|
||||
finally:
|
||||
await chat._transport.aclose()
|
||||
await chat.aclose()
|
||||
await embed.aclose()
|
||||
await ocr.aclose()
|
||||
|
||||
async def test_chat_sugar_failure_records_auto_through_retry(self):
|
||||
import httpx
|
||||
|
||||
from polygateway.errors import AllSourcesExhausted
|
||||
from tests.unit.test_client import _client, _source
|
||||
|
||||
recorder = _MemoryRecorder()
|
||||
client = _client(
|
||||
sources=[_source(model="qwen3.7-plus", enable_thinking=True)],
|
||||
handler=lambda request: httpx.Response(503),
|
||||
telemetry=recorder,
|
||||
retry=RetryPolicy(1, 0.001, 0.01),
|
||||
)
|
||||
try:
|
||||
with pytest.raises(AllSourcesExhausted):
|
||||
await client.chat([])
|
||||
attempts = [r for r in recorder.rows if r["source_name"]]
|
||||
assert len(attempts) == 1
|
||||
assert attempts[0]["reasoning_effort"] == "auto"
|
||||
assert attempts[0]["error"]
|
||||
finally:
|
||||
await client._transport.aclose()
|
||||
|
||||
Reference in New Issue
Block a user