1274 lines
46 KiB
Python
1274 lines
46 KiB
Python
"""live_evidence 与测试侧取证装配的日常离线反例。"""
|
||
|
||
import asyncio
|
||
import inspect
|
||
import json
|
||
from dataclasses import replace
|
||
from typing import Any
|
||
|
||
import httpx
|
||
import pytest
|
||
|
||
from polygateway.errors import RequestRejectedError, SourceDeadError, TransientError
|
||
from polygateway.ports import EmbeddingTransport, Transport
|
||
from polygateway.transports.openai_compat import OpenAICompatTransport
|
||
from polygateway.types import SourceConfig
|
||
from polygateway.types import ThinkingObservation as O
|
||
from tests.e2e.conftest import LiveCapture, ObservedTransport
|
||
from tests.live_evidence import (
|
||
AttemptEvidence,
|
||
HttpEvidence,
|
||
LiveVerdict,
|
||
assess_expected_rejection,
|
||
assess_model_identity,
|
||
assess_thinking_coverage,
|
||
classify_live_failure,
|
||
messages_digest,
|
||
request_is_valid,
|
||
safe_attempts,
|
||
summarize_verdicts,
|
||
write_live_round,
|
||
)
|
||
|
||
_CHECKS = tuple(
|
||
(key, True)
|
||
for key in (
|
||
"method",
|
||
"origin",
|
||
"path",
|
||
"model",
|
||
"stream",
|
||
"authorization",
|
||
"control",
|
||
"messages_digest",
|
||
)
|
||
)
|
||
_BODY = b'{"error":{"type":"model_not_found"}}'
|
||
_SECRET = "fake-unique-credential-sentinel"
|
||
_PROMPT = "fake-private-prompt-sentinel"
|
||
_MESSAGES = [{"role": "user", "content": _PROMPT}]
|
||
|
||
|
||
def _failure(*, body=_BODY, status=404, checks=_CHECKS):
|
||
"""完整的唯一错误证据，供逐维破坏。"""
|
||
error = RequestRejectedError("safe", status_code=status)
|
||
event = HttpEvidence("a", checks, status, body, (False, None))
|
||
return error, (AttemptEvidence("a", (event,), error),)
|
||
|
||
|
||
def test_exact_model_not_found_is_uncovered():
|
||
error, attempts = _failure()
|
||
assert classify_live_failure(error, attempts).status == "UNCOVERED"
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"body",
|
||
[
|
||
None,
|
||
b"",
|
||
b"{",
|
||
b"[]",
|
||
b'{"error":[]}',
|
||
b'{"error":{"type":7}}',
|
||
b'{"error":{"message":"model_not_found"}}',
|
||
b'{"error":{"type":"MODEL_NOT_FOUND"}}',
|
||
b'{"error":{"type":"wrong","type":"model_not_found"}}',
|
||
b'{"error":{},"error":{"type":"model_not_found"}}',
|
||
b"\xff",
|
||
_BODY + b" " * 65536,
|
||
],
|
||
)
|
||
def test_incomplete_or_ambiguous_error_body_fails(body):
|
||
assert classify_live_failure(*_failure(body=body)).status == "FAIL"
|
||
|
||
|
||
@pytest.mark.parametrize("status", [400, 401, 403, 429, 500, 503])
|
||
def test_status_classes_never_automatically_skip(status):
|
||
assert classify_live_failure(*_failure(status=status)).status == "FAIL"
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"error",
|
||
[TransientError("safe"), SourceDeadError("safe"), ValueError("safe"), AssertionError("safe")],
|
||
)
|
||
def test_whole_exception_class_skip_is_forbidden(error):
|
||
assert classify_live_failure(error, ()).status == "FAIL"
|
||
|
||
|
||
@pytest.mark.parametrize("key", [key for key, _ in _CHECKS])
|
||
@pytest.mark.parametrize("mode", ["missing", "false", "duplicate"])
|
||
def test_all_request_checks_are_required(key, mode):
|
||
checks = tuple(
|
||
(k, False if k == key and mode == "false" else v)
|
||
for k, v in _CHECKS
|
||
if mode != "missing" or k != key
|
||
)
|
||
if mode == "duplicate":
|
||
checks += ((key, True),)
|
||
assert classify_live_failure(*_failure(checks=checks)).status == "FAIL"
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"mode",
|
||
["empty", "second_attempt", "second_http", "wrong_id", "different_error", "status_mismatch"],
|
||
)
|
||
def test_attempt_pairing_cannot_be_guessed(mode):
|
||
error, attempts = _failure()
|
||
first = attempts[0]
|
||
if mode == "empty":
|
||
attempts = (replace(first, http=()),)
|
||
elif mode == "second_attempt":
|
||
attempts += (AttemptEvidence("b", (), TransientError("safe")),)
|
||
elif mode == "second_http":
|
||
attempts = (replace(first, http=first.http * 2),)
|
||
elif mode == "wrong_id":
|
||
attempts = (replace(first, call_id="other"),)
|
||
elif mode == "different_error":
|
||
attempts = (replace(first, error=RequestRejectedError("safe", status_code=404)),)
|
||
else:
|
||
attempts = (replace(first, http=(replace(first.http[0], status_code=400),)),)
|
||
assert classify_live_failure(error, attempts).status == "FAIL"
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("reported", "raw", "expected"),
|
||
[
|
||
("m", (False, None), "PASS"),
|
||
("alias", (True, "alias"), "PASS"),
|
||
(None, (False, None), "FAIL"),
|
||
("other", (False, None), "FAIL"),
|
||
(None, (True, None), "UNCOVERED"),
|
||
("other", (True, "other"), "UNCOVERED"),
|
||
(None, (True, "m"), "FAIL"),
|
||
("wrong", (True, "m"), "FAIL"),
|
||
("m", (True, None), "FAIL"),
|
||
],
|
||
)
|
||
def test_identity_requires_independent_raw_evidence(reported, raw, expected):
|
||
assert (
|
||
assess_model_identity(
|
||
requested="m",
|
||
aliases=frozenset({"alias"}),
|
||
reported=reported,
|
||
raw_identity=raw,
|
||
request_valid=True,
|
||
).status
|
||
== expected
|
||
)
|
||
assert (
|
||
assess_model_identity(
|
||
requested="m",
|
||
aliases=frozenset(),
|
||
reported=reported,
|
||
raw_identity=raw,
|
||
request_valid=False,
|
||
).status
|
||
== "FAIL"
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("proposition", "observations", "expected"),
|
||
[
|
||
("enabled", [O.OBSERVED, O.OBSERVED, O.UNKNOWN], "PASS"),
|
||
("enabled", [O.UNKNOWN] * 3, "UNCOVERED"),
|
||
("enabled", [O.ABSENT] * 3, "FAIL"),
|
||
("disabled", [O.ABSENT] * 3, "PASS"),
|
||
("disabled", [O.ABSENT, O.UNKNOWN, O.ABSENT], "UNCOVERED"),
|
||
("disabled", [O.OBSERVED, O.UNKNOWN, O.ABSENT], "FAIL"),
|
||
("cannot_disable", [O.OBSERVED, O.UNKNOWN, O.ABSENT], "PASS"),
|
||
("cannot_disable", [O.ABSENT] * 3, "FAIL"),
|
||
("cannot_disable", [O.UNKNOWN] * 3, "UNCOVERED"),
|
||
],
|
||
)
|
||
def test_coverage_is_proposition_specific(proposition, observations, expected):
|
||
assert (
|
||
assess_thinking_coverage(observations, planned_rounds=3, proposition=proposition).status
|
||
== expected
|
||
)
|
||
|
||
|
||
def test_missing_round_never_reduces_denominator():
|
||
assert (
|
||
assess_thinking_coverage([O.OBSERVED] * 2, planned_rounds=3, proposition="enabled").status
|
||
== "FAIL"
|
||
)
|
||
assert summarize_verdicts(
|
||
[LiveVerdict("PASS", "ok"), LiveVerdict("FAIL", "bad")], planned_rounds=3
|
||
) == {"PASS": 1, "FAIL": 1, "UNCOVERED": 0, "missing": 1}
|
||
|
||
|
||
def test_expected_400_is_a_separate_negative_proposition():
|
||
error, attempts = _failure(status=400, body=b'{"error":{"type":"unsupported_value"}}')
|
||
assert classify_live_failure(error, attempts).status == "FAIL"
|
||
assert (
|
||
assess_expected_rejection(
|
||
error, attempts, status_code=400, machine_type="unsupported_value"
|
||
).status
|
||
== "PASS"
|
||
)
|
||
assert (
|
||
assess_expected_rejection(error, attempts, status_code=400, machine_type="other").status
|
||
== "FAIL"
|
||
)
|
||
|
||
|
||
def _source():
|
||
"""假凭据与非默认超时，不接真实网络。"""
|
||
return SourceConfig(
|
||
name="source",
|
||
provider="openai",
|
||
model="gpt-5.5",
|
||
base_url="https://example.test/v1",
|
||
api_key=_SECRET,
|
||
timeout_s=137,
|
||
trust_env=False,
|
||
)
|
||
|
||
|
||
def _capture(*, stream=False, embedding=False, **changes):
|
||
"""预期独立于生产 payload。"""
|
||
expected = {
|
||
"model": "gpt-5.5",
|
||
"origin": "https://example.test",
|
||
"path": "/v1/embeddings" if embedding else "/v1/chat/completions",
|
||
"control": {},
|
||
"messages_digest": messages_digest([_PROMPT] if embedding else _MESSAGES),
|
||
}
|
||
expected.update({"input_shape": 1} if embedding else {"stream": stream})
|
||
expected.update(changes)
|
||
return LiveCapture(expectations={"source": expected})
|
||
|
||
|
||
def _response(model="gpt-5.5"):
|
||
"""OpenAI 非流式完整响应形态。"""
|
||
result = {
|
||
"id": "test",
|
||
"choices": [
|
||
{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}
|
||
],
|
||
"usage": {"prompt_tokens": 2, "completion_tokens": 1, "total_tokens": 3},
|
||
}
|
||
if model != "missing":
|
||
result["model"] = model
|
||
return result
|
||
|
||
|
||
def _real_transport(capture, handler, *, tamper=None):
|
||
"""仅替换 HTTP 边界，保留真实 hooks 与 production transport。"""
|
||
|
||
def factory(source):
|
||
"""在网络边界换成 MockTransport。"""
|
||
client = capture.client_factory(source)
|
||
client._transport = httpx.MockTransport(handler)
|
||
if tamper is not None:
|
||
client.event_hooks["request"].insert(0, tamper)
|
||
return client
|
||
|
||
return OpenAICompatTransport(client_factory=factory)
|
||
|
||
|
||
async def _complete(observed, *, call_id="a", stream=False):
|
||
"""沿原参数调用薄委托器。"""
|
||
return await observed.complete(
|
||
messages=_MESSAGES,
|
||
source=_source(),
|
||
stream=stream,
|
||
overlay={},
|
||
call_id=call_id,
|
||
reasoning_effort=None,
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize("model", ["gpt-5.5", "missing", None])
|
||
async def test_raw_identity_snapshot_reaches_round_consumer(model):
|
||
capture = _capture()
|
||
real = _real_transport(capture, lambda request: httpx.Response(200, json=_response(model)))
|
||
try:
|
||
with capture.round_context(session_id="s", parent_call_id="p"):
|
||
result = await _complete(ObservedTransport(real, capture))
|
||
raw = capture.raw_identity(session_id="s", parent_call_id="p", call_id="a")
|
||
assert raw == (True, None if model == "missing" else model)
|
||
assert request_is_valid(capture.attempts(session_id="s", parent_call_id="p")[0].http[0])
|
||
verdict = assess_model_identity(
|
||
requested="gpt-5.5",
|
||
aliases=frozenset(),
|
||
reported=result.model_reported,
|
||
raw_identity=raw,
|
||
request_valid=True,
|
||
)
|
||
assert verdict.status == ("PASS" if model == "gpt-5.5" else "UNCOVERED")
|
||
if model == "gpt-5.5":
|
||
assert (
|
||
assess_model_identity(
|
||
requested="gpt-5.5",
|
||
aliases=frozenset(),
|
||
reported=None,
|
||
raw_identity=raw,
|
||
request_valid=True,
|
||
).status
|
||
== "FAIL"
|
||
)
|
||
finally:
|
||
await real.aclose()
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"body", [b"[]", b"{", b'{"model":"a","model":"b"}', b'{"model":7}', b"\xff"]
|
||
)
|
||
async def test_invalid_raw_json_is_not_missing_identity(body):
|
||
capture = _capture()
|
||
real = _real_transport(capture, lambda request: httpx.Response(200, content=body))
|
||
try:
|
||
with (
|
||
capture.round_context(session_id="s", parent_call_id="p"),
|
||
pytest.raises((ValueError, AttributeError, KeyError, TransientError)),
|
||
):
|
||
await _complete(ObservedTransport(real, capture))
|
||
assert capture.raw_identity(session_id="s", parent_call_id="p", call_id="a") == (
|
||
False,
|
||
None,
|
||
)
|
||
assert capture.notes(session_id="s", parent_call_id="p")
|
||
finally:
|
||
await real.aclose()
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"field", ["model", "authorization", "path", "stream", "control", "messages_digest"]
|
||
)
|
||
async def test_hooks_detect_corrupted_request_without_repair(field):
|
||
capture = _capture()
|
||
|
||
async def tamper(request):
|
||
"""故意改错实发请求，而不是改预期。"""
|
||
if field == "authorization":
|
||
request.headers["Authorization"] = "Bearer wrong"
|
||
elif field == "path":
|
||
request.url = request.url.copy_with(path="/wrong")
|
||
else:
|
||
body = json.loads(request.content)
|
||
key, value = {
|
||
"model": ("model", "wrong"),
|
||
"stream": ("stream", True),
|
||
"control": ("reasoning_effort", "high"),
|
||
"messages_digest": ("messages", []),
|
||
}[field]
|
||
body[key] = value
|
||
request._content = json.dumps(body).encode()
|
||
|
||
real = _real_transport(
|
||
capture, lambda request: httpx.Response(404, content=_BODY), tamper=tamper
|
||
)
|
||
try:
|
||
with (
|
||
capture.round_context(session_id="s", parent_call_id="p"),
|
||
pytest.raises(RequestRejectedError) as caught,
|
||
):
|
||
await _complete(ObservedTransport(real, capture))
|
||
attempts = capture.attempts(session_id="s", parent_call_id="p")
|
||
assert dict(attempts[0].http[0].request_checks)[field] is False
|
||
assert classify_live_failure(caught.value, attempts).status == "FAIL"
|
||
finally:
|
||
await real.aclose()
|
||
|
||
|
||
async def test_embed_uses_same_capture_and_source_factory_contract():
|
||
capture = _capture(embedding=True)
|
||
real = _real_transport(
|
||
capture,
|
||
lambda request: httpx.Response(
|
||
200,
|
||
json={"data": [{"index": 0, "embedding": [0.1, 0.2]}], "usage": {"prompt_tokens": 1}},
|
||
),
|
||
)
|
||
try:
|
||
with capture.round_context(session_id="s", parent_call_id="p"):
|
||
result = await ObservedTransport(real, capture).embed(
|
||
texts=[_PROMPT], source=_source(), call_id="e"
|
||
)
|
||
assert result.dim == 2
|
||
assert request_is_valid(capture.attempts(session_id="s", parent_call_id="p")[0].http[0])
|
||
client = real._clients["source"]
|
||
assert client.timeout.read == 137 and client.trust_env is False
|
||
assert client.headers["Authorization"] == f"Bearer {_SECRET}"
|
||
finally:
|
||
await real.aclose()
|
||
assert real._clients == {}
|
||
|
||
|
||
class _SSE(httpx.AsyncByteStream):
|
||
"""可检查是否被 hook 提前消费的 SSE。"""
|
||
|
||
def __init__(self):
|
||
self.reads = 0
|
||
self.closed = False
|
||
|
||
async def __aiter__(self):
|
||
self.reads += 1
|
||
yield b'data: {"model":"gpt-5.5","choices":[{"delta":{"content":"ok"},"finish_reason":null}]}\n\n'
|
||
yield b'data: {"choices":[],"usage":{"prompt_tokens":2,"completion_tokens":1,"total_tokens":3}}\n\ndata: [DONE]\n\n'
|
||
|
||
async def aclose(self):
|
||
self.closed = True
|
||
|
||
|
||
async def test_success_sse_is_not_preconsumed_and_has_no_raw_snapshot():
|
||
capture = _capture(stream=True)
|
||
stream = _SSE()
|
||
|
||
def handler(request):
|
||
return httpx.Response(200, stream=stream)
|
||
|
||
def factory(source):
|
||
client = capture.client_factory(source)
|
||
|
||
async def after_hook(response):
|
||
assert stream.reads == 0
|
||
|
||
client.event_hooks["response"].append(after_hook)
|
||
client._transport = httpx.MockTransport(handler)
|
||
return client
|
||
|
||
real = OpenAICompatTransport(client_factory=factory)
|
||
try:
|
||
with capture.round_context(session_id="s", parent_call_id="p"):
|
||
result = await _complete(ObservedTransport(real, capture), stream=True)
|
||
assert result.content == "ok" and stream.reads == 1 and stream.closed
|
||
raw = capture.raw_identity(session_id="s", parent_call_id="p", call_id="a")
|
||
assert raw == (False, None)
|
||
assert (
|
||
assess_model_identity(
|
||
requested="gpt-5.5",
|
||
aliases=frozenset(),
|
||
reported=None,
|
||
raw_identity=raw,
|
||
request_valid=True,
|
||
).status
|
||
== "FAIL"
|
||
)
|
||
finally:
|
||
await real.aclose()
|
||
|
||
|
||
async def test_concurrent_rounds_retry_and_cancellation_reset_context():
|
||
capture = _capture()
|
||
reached = asyncio.Event()
|
||
|
||
async def handler(request):
|
||
await asyncio.sleep(0)
|
||
return httpx.Response(200, json=_response())
|
||
|
||
real = _real_transport(capture, handler)
|
||
observed = ObservedTransport(real, capture)
|
||
|
||
async def one(parent):
|
||
with capture.round_context(session_id="s", parent_call_id=parent):
|
||
# 零 HTTP 的前次失败与成功 call_id 不得混淆。
|
||
with pytest.raises(ValueError), capture.attempt_context(parent + "-failed"):
|
||
raise ValueError("safe")
|
||
await _complete(observed, call_id=parent + "-ok")
|
||
return capture.raw_identity(session_id="s", parent_call_id=parent, call_id=parent + "-ok")
|
||
|
||
async def cancelled():
|
||
with (
|
||
capture.round_context(session_id="s", parent_call_id="cancel"),
|
||
capture.attempt_context("cancelled"),
|
||
):
|
||
reached.set()
|
||
await asyncio.Future()
|
||
|
||
try:
|
||
assert await asyncio.gather(one("p1"), one("p2")) == [(True, "gpt-5.5")] * 2
|
||
for parent in ("p1", "p2"):
|
||
attempts = capture.attempts(session_id="s", parent_call_id=parent)
|
||
assert len(attempts) == 2 and len(attempts[0].http) == 0 and len(attempts[1].http) == 1
|
||
with pytest.raises(ValueError, match="跨轮"):
|
||
capture.raw_identity(session_id="s", parent_call_id="p1", call_id="p2-ok")
|
||
task = asyncio.create_task(cancelled())
|
||
await reached.wait()
|
||
task.cancel()
|
||
with pytest.raises(asyncio.CancelledError):
|
||
await task
|
||
assert len(capture.attempts(session_id="s", parent_call_id="cancel")) == 1
|
||
with pytest.raises(LookupError):
|
||
capture._round.get()
|
||
with pytest.raises(LookupError):
|
||
capture._attempt.get()
|
||
with (
|
||
capture.round_context(session_id="s", parent_call_id="new"),
|
||
pytest.raises(ValueError, match="重复"),
|
||
capture.attempt_context("p1-ok"),
|
||
):
|
||
pytest.fail("不应进入")
|
||
finally:
|
||
await real.aclose()
|
||
|
||
|
||
def test_delegator_signatures_match_ports():
|
||
assert inspect.signature(ObservedTransport.complete) == inspect.signature(Transport.complete)
|
||
assert inspect.signature(ObservedTransport.embed) == inspect.signature(EmbeddingTransport.embed)
|
||
|
||
|
||
def test_missing_expectations_fail_before_network():
|
||
with pytest.raises(ValueError):
|
||
LiveCapture(expectations={"s": {"model": "m"}})
|
||
|
||
|
||
def test_round_reports_preserve_prior_round_and_redact_sentinels(tmp_path):
|
||
error, attempts = _failure(
|
||
body=(f'{{"error":{{"type":"model_not_found","message":"{_SECRET} {_PROMPT}"}}}}').encode()
|
||
)
|
||
error.args = (f"{_SECRET} {_PROMPT}",)
|
||
first = write_live_round(
|
||
tmp_path,
|
||
run_id="run",
|
||
matrix_id="matrix",
|
||
round_index=1,
|
||
safe_fields={"status": "PASS", "attempts": safe_attempts(attempts)},
|
||
)
|
||
second = write_live_round(
|
||
tmp_path,
|
||
run_id="run",
|
||
matrix_id="matrix",
|
||
round_index=2,
|
||
safe_fields={"status": "FAIL", "attempts": safe_attempts(attempts)},
|
||
)
|
||
assert first != second and first.exists()
|
||
for path in tmp_path.rglob("*.md"):
|
||
text = path.read_text()
|
||
assert _SECRET not in text and _PROMPT not in text
|
||
assert '"status": "PASS"' in first.read_text()
|
||
assert '"status": "FAIL"' in second.read_text()
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"fields",
|
||
[
|
||
{"raw": _BODY},
|
||
{"error_type": ValueError("secret")},
|
||
{"attempts": [AttemptEvidence("a", (), None)]},
|
||
{"attempts": [{"call_id": "a", "error_type": None, "http": [{"Authorization": "secret"}]}]},
|
||
],
|
||
)
|
||
def test_report_refuses_raw_objects_and_unknown_fields(tmp_path, fields):
|
||
with pytest.raises((ValueError, TypeError)):
|
||
write_live_round(
|
||
tmp_path, run_id="run", matrix_id="matrix", round_index=1, safe_fields=fields
|
||
)
|
||
|
||
|
||
def test_report_write_failure_is_not_skip(tmp_path):
|
||
blocked = tmp_path / "file"
|
||
blocked.write_text("not a directory")
|
||
with pytest.raises(OSError):
|
||
write_live_round(
|
||
blocked,
|
||
run_id="run",
|
||
matrix_id="matrix",
|
||
round_index=1,
|
||
safe_fields={"status": "UNCOVERED"},
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"body", [b"", _BODY, _BODY + b" " * (65536 - len(_BODY)), _BODY + b" " * 65536]
|
||
)
|
||
async def test_error_snapshot_keeps_only_complete_bounded_body(body):
|
||
"""真实 hook 的边界，不把生产摘要当完整错误 JSON。"""
|
||
capture = _capture()
|
||
real = _real_transport(capture, lambda request: httpx.Response(404, content=body))
|
||
try:
|
||
with (
|
||
capture.round_context(session_id="s", parent_call_id="p"),
|
||
pytest.raises(RequestRejectedError) as caught,
|
||
):
|
||
await _complete(ObservedTransport(real, capture))
|
||
attempts = capture.attempts(session_id="s", parent_call_id="p")
|
||
assert attempts[0].http[0].error_body == (body if len(body) <= 65536 else None)
|
||
expected = "UNCOVERED" if 0 < len(body) <= 65536 else "FAIL"
|
||
assert classify_live_failure(caught.value, attempts).status == expected
|
||
finally:
|
||
await real.aclose()
|
||
|
||
|
||
async def test_multiple_success_http_candidates_fail_identity_accessor():
|
||
"""一个 attempt 多 HTTP 不能猜最后一条。"""
|
||
capture = _capture()
|
||
client = capture.client_factory(_source())
|
||
client._transport = httpx.MockTransport(lambda request: httpx.Response(200, json=_response()))
|
||
try:
|
||
with (
|
||
capture.round_context(session_id="s", parent_call_id="p"),
|
||
capture.attempt_context("a"),
|
||
):
|
||
for _ in range(2):
|
||
await client.post(
|
||
"https://example.test/v1/chat/completions",
|
||
json={"model": "gpt-5.5", "stream": False, "messages": _MESSAGES},
|
||
)
|
||
with pytest.raises(ValueError, match="多个成功"):
|
||
capture.raw_identity(session_id="s", parent_call_id="p", call_id="a")
|
||
finally:
|
||
await client.aclose()
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"body", [b"data: not-json\n\n", b'data: {"choices":[]}\n\ndata: [DONE]\n\n']
|
||
)
|
||
async def test_sse_parser_failure_is_not_external_uncovered(body):
|
||
capture = _capture(stream=True)
|
||
real = _real_transport(capture, lambda request: httpx.Response(200, content=body))
|
||
try:
|
||
with (
|
||
capture.round_context(session_id="s", parent_call_id="p"),
|
||
pytest.raises(TransientError) as caught,
|
||
):
|
||
await _complete(ObservedTransport(real, capture), stream=True)
|
||
assert (
|
||
classify_live_failure(
|
||
caught.value, capture.attempts(session_id="s", parent_call_id="p")
|
||
).status
|
||
== "FAIL"
|
||
)
|
||
finally:
|
||
await real.aclose()
|
||
|
||
|
||
async def test_round_consumer_keeps_first_success_when_second_assertion_fails(tmp_path):
|
||
"""真实 GatewayClient 与报告出口连通；断言异常不得漏报告。"""
|
||
from polygateway import GatewaySettings
|
||
from tests.e2e.conftest import captured_chat_round, observed_client
|
||
from tests.unit.test_config import _BASE_ENV
|
||
|
||
settings = GatewaySettings.from_env("LLM", env=_BASE_ENV)
|
||
settings = replace(settings, sources=(_source(),))
|
||
capture = _capture()
|
||
original_factory = capture.client_factory
|
||
clients = []
|
||
|
||
def factory(source):
|
||
client = original_factory(source)
|
||
client._transport = httpx.MockTransport(
|
||
lambda request: httpx.Response(200, json=_response())
|
||
)
|
||
clients.append(client)
|
||
return client
|
||
|
||
capture.client_factory = factory
|
||
|
||
def fail(response):
|
||
raise AssertionError(_SECRET + _PROMPT)
|
||
|
||
async with observed_client(settings, capture) as client:
|
||
for index, validate in ((1, None), (2, fail)):
|
||
_, verdict = await captured_chat_round(
|
||
client,
|
||
capture,
|
||
run_id="run",
|
||
matrix_id="matrix",
|
||
round_index=index,
|
||
output_dir=tmp_path,
|
||
messages=_MESSAGES,
|
||
models={"source": "gpt-5.5"},
|
||
aliases={},
|
||
stream=False,
|
||
validate=validate,
|
||
)
|
||
assert verdict.status == ("PASS" if index == 1 else "FAIL")
|
||
paths = list(tmp_path.rglob("*.md"))
|
||
assert len(paths) == 2
|
||
text = "".join(path.read_text() for path in paths)
|
||
assert '"status": "PASS"' in text and '"status": "FAIL"' in text
|
||
assert _SECRET not in text and _PROMPT not in text
|
||
assert clients and all(client.is_closed for client in clients)
|
||
|
||
|
||
@pytest.mark.parametrize("model", ["embedding-override-a", "embedding-override-b"])
|
||
@pytest.mark.parametrize(
|
||
("outcome", "status", "error_type"),
|
||
[
|
||
("success", "PASS", None),
|
||
("503", "FAIL", "TransientError"),
|
||
("404", "UNCOVERED", "RequestRejectedError"),
|
||
("request_error", "FAIL", "TransientError"),
|
||
("cancelled", "FAIL", None),
|
||
],
|
||
)
|
||
async def test_embed_probe_report_keeps_actual_source_and_round_identity(
|
||
tmp_path, monkeypatch, model, outcome, status, error_type
|
||
):
|
||
"""真实探测消费者在成功与失败均留完整关联；只替换外部 HTTP。"""
|
||
import ast
|
||
from pathlib import Path
|
||
|
||
from tests.unit.test_config import _BASE_ENV
|
||
|
||
path = Path(__file__).parents[1] / "e2e/test_embed_probe.py"
|
||
tree = ast.parse(path.read_text())
|
||
tree.body = [
|
||
node
|
||
for node in tree.body
|
||
if not (
|
||
isinstance(node, ast.Assign)
|
||
and any(
|
||
isinstance(target, ast.Name) and target.id in {"_ENV", "pytestmark"}
|
||
for target in node.targets
|
||
)
|
||
)
|
||
]
|
||
env = {
|
||
**_BASE_ENV,
|
||
"LLM__MINIMAX__1__BASE_URL": "https://example.test/v1",
|
||
"LLM__MINIMAX__1__API_KEY": _SECRET,
|
||
"LLM__MINIMAX__1__MODEL": "configured-chat-model",
|
||
"LLM__MINIMAX__1__TIMEOUT_S": "137",
|
||
"LLM__MINIMAX__1__TRUST_ENV": "false",
|
||
"PGW_EMBED_PROBE_MODEL": model,
|
||
}
|
||
namespace: dict[str, Any] = {"_ENV": env}
|
||
exec(compile(tree, str(path), "exec"), namespace)
|
||
monkeypatch.chdir(tmp_path)
|
||
original_factory = LiveCapture.client_factory
|
||
clients = []
|
||
calls = []
|
||
reached = asyncio.Event()
|
||
|
||
def factory(capture, source):
|
||
"""保留真实取证 hooks、源与逻辑 ID，只隔离网络出口。"""
|
||
client = original_factory(capture, source)
|
||
|
||
async def handler(request):
|
||
"""提供完整成功／错误样本，敏感回显不得进入报告。"""
|
||
payload = json.loads(request.content)
|
||
assert payload["model"] == model == source.model
|
||
assert source.model != env["LLM__MINIMAX__1__MODEL"]
|
||
assert request.headers["Authorization"] == f"Bearer {_SECRET}"
|
||
calls.append((source, capture._round.get(), capture._attempt.get().call_id))
|
||
if outcome == "cancelled":
|
||
reached.set()
|
||
await asyncio.Future()
|
||
if outcome == "request_error":
|
||
raise httpx.ConnectError(_SECRET + _PROMPT, request=request)
|
||
if outcome == "success":
|
||
return httpx.Response(
|
||
200,
|
||
json={
|
||
"data": [{"index": 0, "embedding": [0.1, 0.2]}],
|
||
"usage": {"prompt_tokens": 1},
|
||
"model": _SECRET + _PROMPT,
|
||
},
|
||
)
|
||
return httpx.Response(
|
||
int(outcome),
|
||
json={"error": {"type": "model_not_found", "message": _SECRET + _PROMPT}},
|
||
)
|
||
|
||
client._transport = httpx.MockTransport(handler)
|
||
clients.append(client)
|
||
return client
|
||
|
||
monkeypatch.setattr(LiveCapture, "client_factory", factory)
|
||
probe = namespace["test_probe_real_gateway_embeddings"]
|
||
if outcome == "cancelled":
|
||
task = asyncio.create_task(probe())
|
||
try:
|
||
async with asyncio.timeout(5):
|
||
await reached.wait()
|
||
task.cancel()
|
||
with pytest.raises(asyncio.CancelledError):
|
||
await task
|
||
assert task.cancelled()
|
||
finally:
|
||
task.cancel()
|
||
await asyncio.gather(task, return_exceptions=True)
|
||
elif status == "PASS":
|
||
await probe()
|
||
elif status == "UNCOVERED":
|
||
with pytest.raises(pytest.skip.Exception):
|
||
await probe()
|
||
else:
|
||
with pytest.raises(AssertionError):
|
||
await probe()
|
||
|
||
assert len(calls) == len(clients) == 1
|
||
assert all(client.is_closed for client in clients)
|
||
source, (session_id, parent_call_id), call_id = calls[0]
|
||
paths = list((tmp_path / "tests/outputs/134/live").rglob("*.md"))
|
||
assert len(paths) == 1
|
||
text = paths[0].read_text()
|
||
assert _SECRET not in text and _PROMPT not in text
|
||
row = json.loads(text.split("```json\n")[1].split("\n```")[0])
|
||
assert row["status"] == status
|
||
assert row["requested_model"] == source.model == model
|
||
assert row["provider"] == source.provider == "minimax"
|
||
assert row["planned_rounds"] == 1
|
||
assert row["completed_rounds"] == (0 if outcome == "cancelled" else 1)
|
||
if outcome == "cancelled":
|
||
assert row["reason"] == "轮次未完成"
|
||
assert row["session_id"] == paths[0].parent.name == session_id
|
||
assert row["parent_call_id"] == parent_call_id
|
||
assert len({session_id, parent_call_id, call_id}) == 3
|
||
assert paths[0].name.startswith("embedding-1-")
|
||
assert text.startswith("# embedding · 轮次 1\n")
|
||
assert len(row["attempts"]) == 1
|
||
attempt = row["attempts"][0]
|
||
assert attempt["call_id"] == call_id and attempt["error_type"] == error_type
|
||
assert len(attempt["http"]) == 1
|
||
event = attempt["http"][0]
|
||
assert all(event["request_checks"].values())
|
||
assert (
|
||
event["status_code"]
|
||
== {
|
||
"success": 200,
|
||
"503": 503,
|
||
"404": 404,
|
||
"request_error": 0,
|
||
"cancelled": 0,
|
||
}[outcome]
|
||
)
|
||
if outcome in {"503", "404"}:
|
||
assert event["error_body_complete"] is True
|
||
assert event["machine_type"] == "model_not_found"
|
||
elif outcome in {"request_error", "cancelled"}:
|
||
assert "无可配对响应" in row["evidence_notes"]
|
||
|
||
|
||
def test_partial_uncovered_and_failed_rounds_never_become_model_pass():
|
||
from tests.live_evidence import combine_live_verdicts, qualify_live_rounds
|
||
|
||
passed = LiveVerdict("PASS", "safe")
|
||
uncovered = LiveVerdict("UNCOVERED", "safe")
|
||
failed = LiveVerdict("FAIL", "safe")
|
||
assert combine_live_verdicts([passed, uncovered]).status == "UNCOVERED"
|
||
assert combine_live_verdicts([passed, uncovered, failed]).status == "FAIL"
|
||
assert qualify_live_rounds([passed, passed], planned_rounds=3).status == "FAIL"
|
||
assert qualify_live_rounds([passed, failed, passed], planned_rounds=3).status == "FAIL"
|
||
|
||
|
||
async def test_real_retry_success_call_id_selects_exact_raw_identity(tmp_path):
|
||
"""真实 RetryMW 先503再成功，两个并发 parent 的身份不串线。"""
|
||
from polygateway import GatewaySettings
|
||
from tests.e2e.conftest import captured_chat_round, observed_client
|
||
from tests.unit.test_config import _BASE_ENV
|
||
|
||
settings = GatewaySettings.from_env("LLM", env=_BASE_ENV)
|
||
settings = replace(
|
||
settings,
|
||
sources=(_source(),),
|
||
retry=replace(settings.retry, backoff_base_s=0.001, backoff_max_s=0.002),
|
||
)
|
||
capture = _capture()
|
||
original_factory = capture.client_factory
|
||
counts = {}
|
||
|
||
async def handler(request):
|
||
key = capture._round.get()
|
||
counts[key] = counts.get(key, 0) + 1
|
||
await asyncio.sleep(0)
|
||
if counts[key] == 1:
|
||
return httpx.Response(503, json={"error": {"type": "temporary"}, "model": "wrong"})
|
||
return httpx.Response(200, json=_response())
|
||
|
||
def factory(source):
|
||
client = original_factory(source)
|
||
client._transport = httpx.MockTransport(handler)
|
||
return client
|
||
|
||
capture.client_factory = factory
|
||
async with observed_client(settings, capture) as client:
|
||
|
||
async def one(index):
|
||
return await captured_chat_round(
|
||
client,
|
||
capture,
|
||
run_id="retry",
|
||
matrix_id="retry",
|
||
round_index=index,
|
||
output_dir=tmp_path,
|
||
messages=_MESSAGES,
|
||
models={"source": "gpt-5.5"},
|
||
aliases={},
|
||
stream=False,
|
||
)
|
||
|
||
results = await asyncio.gather(one(1), one(2))
|
||
assert all(verdict.status == "PASS" for _, verdict in results)
|
||
assert len(counts) == 2 and set(counts.values()) == {2}
|
||
for key in counts:
|
||
attempts = capture.attempts(session_id=key[0], parent_call_id=key[1])
|
||
assert len(attempts) == 2
|
||
assert attempts[0].http[0].raw_identity == (False, None)
|
||
assert capture.raw_identity(
|
||
session_id=key[0], parent_call_id=key[1], call_id=attempts[1].call_id
|
||
) == (True, "gpt-5.5")
|
||
|
||
|
||
def test_live_model_provider_mapping_covers_default_capabilities_without_env_import():
|
||
"""L8 映射装配守卫离线化，只读取显式字面矩阵，不执行 .env 模块。"""
|
||
import ast
|
||
from pathlib import Path
|
||
|
||
from polygateway.thinking import DEFAULT_CAPABILITIES
|
||
|
||
tree = ast.parse((Path(__file__).parents[1] / "e2e/test_thinking_live.py").read_text())
|
||
mapping = next(
|
||
ast.literal_eval(node.value)
|
||
for node in tree.body
|
||
if isinstance(node, ast.Assign)
|
||
and any(
|
||
isinstance(target, ast.Name) and target.id == "_MODEL_PROVIDER"
|
||
for target in node.targets
|
||
)
|
||
)
|
||
assert not set(DEFAULT_CAPABILITIES) - mapping.keys()
|
||
|
||
|
||
async def test_unbuffered_error_is_insufficient_evidence():
|
||
"""response hook 不预读错误流；未完成缓冲不能借片段归因。"""
|
||
capture = _capture()
|
||
stream = _SSE()
|
||
client = capture.client_factory(_source())
|
||
client._transport = httpx.MockTransport(lambda request: httpx.Response(404, stream=stream))
|
||
error = RequestRejectedError("safe", status_code=404)
|
||
try:
|
||
with (
|
||
pytest.raises(RequestRejectedError),
|
||
capture.round_context(session_id="s", parent_call_id="p"),
|
||
capture.attempt_context("a"),
|
||
):
|
||
async with client.stream(
|
||
"POST",
|
||
"https://example.test/v1/chat/completions",
|
||
json={"model": "gpt-5.5", "messages": _MESSAGES, "stream": False},
|
||
):
|
||
raise error
|
||
attempts = capture.attempts(session_id="s", parent_call_id="p")
|
||
assert attempts[0].http[0].error_body is None and stream.reads == 0
|
||
assert classify_live_failure(error, attempts).status == "FAIL"
|
||
finally:
|
||
await client.aclose()
|
||
|
||
|
||
def test_pytest_report_hook_preserves_report_and_uses_safe_fallback(tmp_path, monkeypatch):
|
||
"""pytest wrapper generator 的 return 值是协议必需，不可按错误 LSP 建议删除。"""
|
||
from types import SimpleNamespace
|
||
|
||
from tests.e2e.conftest import pytest_runtest_makereport
|
||
|
||
monkeypatch.chdir(tmp_path)
|
||
report = SimpleNamespace(skipped=False, failed=True, when="call")
|
||
hook = pytest_runtest_makereport(SimpleNamespace(nodeid="safe-node"), None)
|
||
assert next(hook) is None
|
||
with pytest.raises(StopIteration) as finished:
|
||
hook.send(report)
|
||
assert finished.value.value is report
|
||
paths = list(tmp_path.rglob("*.md"))
|
||
assert len(paths) == 1 and '"status": "FAIL"' in paths[0].read_text()
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"damage", [None, "prefix", "role", "content", "unpaired", "budget", "wire"]
|
||
)
|
||
async def test_structured_reask_preserves_message_contract(tmp_path, damage, monkeypatch):
|
||
"""真实 StructuredMW 缺字段后重问成功；前缀、反馈结构及 wire 破坏均失败。"""
|
||
from pydantic import BaseModel
|
||
|
||
from polygateway import GatewaySettings
|
||
from polygateway.middleware.structured import StructuredMW
|
||
from tests.e2e.conftest import captured_chat_round, observed_client
|
||
from tests.unit.test_config import _BASE_ENV
|
||
|
||
class Answer(BaseModel):
|
||
"""离线最小结构化契约。"""
|
||
|
||
answer: int
|
||
reason: str
|
||
|
||
settings = replace(
|
||
GatewaySettings.from_env("LLM", env=_BASE_ENV),
|
||
sources=(_source(),),
|
||
structured_max_retries=1,
|
||
)
|
||
capture = _capture(messages_prefix_length=1, structured_max_retries=1)
|
||
original_feedback = StructuredMW._with_feedback
|
||
|
||
def feedback(self, *args):
|
||
"""只破坏重问产物，不替代生产阶梯或解析。"""
|
||
request = original_feedback(self, *args)
|
||
messages = [dict(message) for message in request.messages]
|
||
if damage == "prefix":
|
||
messages[0]["content"] = "changed"
|
||
elif damage == "role":
|
||
messages[-1]["role"] = "assistant"
|
||
elif damage == "content":
|
||
messages[-1]["content"] = ["wrong-type"]
|
||
elif damage == "unpaired":
|
||
messages.pop()
|
||
elif damage == "budget":
|
||
messages.extend(messages[-2:])
|
||
return replace(request, messages=messages)
|
||
|
||
monkeypatch.setattr(StructuredMW, "_with_feedback", feedback)
|
||
requests = []
|
||
|
||
def handler(request):
|
||
requests.append(json.loads(request.content))
|
||
body = _response()
|
||
body["choices"][0]["message"]["content"] = (
|
||
'{"answer":5}' if len(requests) == 1 else '{"answer":5,"reason":"sum"}'
|
||
)
|
||
return httpx.Response(200, json=body)
|
||
|
||
original_factory = capture.client_factory
|
||
|
||
def factory(source):
|
||
client = original_factory(source)
|
||
client._transport = httpx.MockTransport(handler)
|
||
if damage == "wire":
|
||
|
||
async def corrupt(request):
|
||
payload = json.loads(request.content)
|
||
if len(payload["messages"]) > 1:
|
||
payload["messages"][-1]["content"] = "well-shaped-but-corrupted"
|
||
request._content = json.dumps(payload).encode()
|
||
|
||
client.event_hooks["request"].insert(0, corrupt)
|
||
return client
|
||
|
||
capture.client_factory = factory
|
||
async with observed_client(settings, capture) as client:
|
||
response, verdict = await captured_chat_round(
|
||
client,
|
||
capture,
|
||
run_id="structured",
|
||
matrix_id="reask",
|
||
round_index=1,
|
||
output_dir=tmp_path,
|
||
messages=_MESSAGES,
|
||
models={"source": "gpt-5.5"},
|
||
aliases={},
|
||
stream=False,
|
||
structured=Answer,
|
||
)
|
||
assert response.structured_data.answer == 5
|
||
assert len(requests) == 2
|
||
assert verdict.status == ("PASS" if damage is None else "FAIL")
|
||
text = "".join(path.read_text() for path in tmp_path.rglob("*.md"))
|
||
assert _SECRET not in text and _PROMPT not in text
|
||
|
||
|
||
def _thinking_consumer():
|
||
"""仅加载 live 消费者定义与字面矩阵，跳过所有环境读取语句。"""
|
||
import ast
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
|
||
path = Path(__file__).parents[1] / "e2e/test_thinking_live.py"
|
||
tree = ast.parse(path.read_text())
|
||
excluded = {
|
||
"_ENV",
|
||
"_HAS_SOURCE",
|
||
"pytestmark",
|
||
"_ROUNDS",
|
||
"_TIER_ROUNDS",
|
||
"_TIER_LONG_ROUNDS",
|
||
"_TIER_CONCURRENCY",
|
||
}
|
||
tree.body = [
|
||
node
|
||
for node in tree.body
|
||
if not (
|
||
isinstance(node, ast.Assign)
|
||
and any(
|
||
isinstance(target, ast.Name) and target.id in excluded for target in node.targets
|
||
)
|
||
)
|
||
]
|
||
namespace: dict[str, Any] = {
|
||
"_ROUNDS": 3,
|
||
"_TIER_ROUNDS": 2,
|
||
"_TIER_LONG_ROUNDS": 1,
|
||
"_TIER_CONCURRENCY": 1,
|
||
}
|
||
exec(compile(tree, str(path), "exec"), namespace)
|
||
return SimpleNamespace(**namespace), namespace
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("observation", "qualification", "expected"),
|
||
[
|
||
(O.ABSENT, "PASS", "UNCOVERED"),
|
||
(O.OBSERVED, "PASS", "UNCOVERED"),
|
||
(O.UNKNOWN, "PASS", "UNCOVERED"),
|
||
(O.ABSENT, "FAIL", "FAIL"),
|
||
],
|
||
)
|
||
async def test_unregistered_candidate_has_no_disable_declaration(
|
||
observation, qualification, expected
|
||
):
|
||
"""执行真实 T10 消费者，未登记不等于不可关闭，资格失败仍红。"""
|
||
from types import SimpleNamespace
|
||
|
||
live, namespace = _thinking_consumer()
|
||
conclusions = []
|
||
calls = []
|
||
|
||
async def probe(model, effort, *, rounds, **kwargs):
|
||
calls.append(rounds)
|
||
return [
|
||
{
|
||
"verdict": LiveVerdict(qualification, "safe"),
|
||
"response": SimpleNamespace(thinking_observation=observation),
|
||
}
|
||
for _ in range(rounds)
|
||
]
|
||
|
||
namespace["_probe_effort"] = probe
|
||
namespace["_conclude"] = lambda matrix, verdict, **kwargs: conclusions.append(verdict)
|
||
await live.TestTierProbe().test_t10_none_direction_matches_declaration("claude-haiku-5")
|
||
assert conclusions[0].status == expected
|
||
assert calls == ([2, 1] if observation is not O.OBSERVED and qualification == "PASS" else [2])
|
||
|
||
|
||
@pytest.mark.parametrize("case", ["none", "tiers", "L8"])
|
||
async def test_capability_conclusions_link_models_and_all_subruns(tmp_path, case):
|
||
"""两个型号的 PASS／UNCOVERED 结论必须关联原件和完整短长／档位轮数。"""
|
||
from contextlib import asynccontextmanager
|
||
from types import SimpleNamespace
|
||
|
||
from polygateway import GatewaySettings
|
||
from tests.unit.test_config import _BASE_ENV
|
||
|
||
live, namespace = _thinking_consumer()
|
||
settings = GatewaySettings.from_env("LLM", env=_BASE_ENV)
|
||
namespace["_OUT_DIR"] = tmp_path
|
||
namespace["_tier_settings"] = lambda model: replace(
|
||
settings, sources=(replace(_source(), model=model),)
|
||
)
|
||
namespace["_settings"] = lambda **kwargs: replace(
|
||
settings, sources=(replace(_source(), **kwargs),)
|
||
)
|
||
namespace["enforce_verdict"] = lambda verdict: None
|
||
|
||
@asynccontextmanager
|
||
async def client(*args, **kwargs):
|
||
yield None
|
||
|
||
async def round_call(client, capture, **kwargs):
|
||
model = kwargs["models"]["source"]
|
||
observation = (
|
||
O.UNKNOWN if model == "gpt-5.4" else O.OBSERVED if case == "tiers" else O.ABSENT
|
||
)
|
||
write_live_round(
|
||
kwargs["output_dir"],
|
||
run_id=kwargs["run_id"],
|
||
matrix_id=kwargs["matrix_id"],
|
||
round_index=kwargs["round_index"],
|
||
safe_fields={
|
||
"requested_model": model,
|
||
"session_id": kwargs["run_id"],
|
||
"status": "PASS",
|
||
"thinking_observation": observation,
|
||
},
|
||
)
|
||
return SimpleNamespace(thinking_observation=observation), LiveVerdict("PASS", "safe")
|
||
|
||
namespace["observed_client"] = client
|
||
namespace["captured_chat_round"] = round_call
|
||
for model in ("gpt-5.5", "gpt-5.4"):
|
||
if case == "none":
|
||
await live.TestTierProbe().test_t10_none_direction_matches_declaration(model)
|
||
elif case == "tiers":
|
||
await live.TestTierProbe().test_t10_declared_tiers_actually_reason(model)
|
||
else:
|
||
await live.TestCapabilityDrift().test_declared_capability_matches_reality(model)
|
||
matrix = {"none": "T10-none", "tiers": "T10-tiers", "L8": "L8"}[case]
|
||
finals = list(tmp_path.rglob(f"{matrix}-0-*.md"))
|
||
assert len(finals) == 2
|
||
seen = set()
|
||
for path in finals:
|
||
row = json.loads(path.read_text().split("```json\n")[1].split("\n```")[0])
|
||
assert "requested_model" in row, "结论缺型号，无法关联逐轮原件"
|
||
model = row["requested_model"]
|
||
seen.add(model)
|
||
assert row["session_id"] == path.parent.name
|
||
assert row["status"] == ("PASS" if model == "gpt-5.5" else "UNCOVERED")
|
||
assert row["proposition"]
|
||
subruns = row["subruns"]
|
||
expected_groups = (
|
||
2
|
||
if case == "none"
|
||
else len(
|
||
[
|
||
effort
|
||
for effort in live.DEFAULT_CAPABILITIES[model].supported_efforts
|
||
if effort is not live.Effort.NONE
|
||
]
|
||
)
|
||
if case == "tiers"
|
||
else 1
|
||
)
|
||
assert len(subruns) == expected_groups
|
||
assert row["planned_rounds"] == sum(subrun["planned_rounds"] for subrun in subruns)
|
||
assert row["completed_rounds"] == row["planned_rounds"]
|
||
for subrun in subruns:
|
||
originals = [
|
||
p
|
||
for p in path.parent.glob(f"{subrun['matrix_id']}-*.md")
|
||
if p.name[len(subrun["matrix_id"]) + 1 :].split("-", 1)[0].isdigit()
|
||
and not p.name.startswith(subrun["matrix_id"] + "-0-")
|
||
]
|
||
assert len(originals) == subrun["planned_rounds"] == subrun["completed_rounds"]
|
||
for original in originals:
|
||
data = json.loads(original.read_text().split("```json\n")[1].split("\n```")[0])
|
||
assert data["requested_model"] == model
|
||
assert data["session_id"] == row["session_id"]
|
||
assert seen == {"gpt-5.5", "gpt-5.4"}
|
||
|
||
|
||
@pytest.mark.parametrize("machine_type", ["model_not_found", _SECRET, "arbitrary-upstream-text"])
|
||
def test_safe_machine_type_retains_only_known_enum(tmp_path, machine_type):
|
||
"""认可机器枚举可复核；任意机器正文与 sentinel 不得落盘。"""
|
||
_, attempts = _failure(body=json.dumps({"error": {"type": machine_type}}).encode())
|
||
fields = {"attempts": safe_attempts(attempts)}
|
||
event = fields["attempts"][0]["http"][0]
|
||
assert event.get("machine_type") == (
|
||
"model_not_found" if machine_type == "model_not_found" else "omitted"
|
||
)
|
||
path = write_live_round(
|
||
tmp_path, run_id="safe", matrix_id="machine", round_index=1, safe_fields=fields
|
||
)
|
||
text = path.read_text()
|
||
assert _SECRET not in text and "arbitrary-upstream-text" not in text
|
||
if machine_type != "model_not_found":
|
||
event["machine_type"] = machine_type
|
||
with pytest.raises(ValueError, match="机器"):
|
||
write_live_round(
|
||
tmp_path, run_id="unsafe", matrix_id="machine", round_index=1, safe_fields=fields
|
||
)
|
||
|
||
|
||
async def test_structured_first_attempt_requires_exact_initial_messages():
|
||
"""结构化窄规则不能允许首轮凭空带入一对反馈。"""
|
||
capture = _capture(messages_prefix_length=1, structured_max_retries=1)
|
||
real = _real_transport(capture, lambda request: httpx.Response(200, json=_response()))
|
||
try:
|
||
with capture.round_context(session_id="first", parent_call_id="parent"):
|
||
await ObservedTransport(real, capture).complete(
|
||
messages=[
|
||
*_MESSAGES,
|
||
{"role": "assistant", "content": "old"},
|
||
{"role": "user", "content": "retry"},
|
||
],
|
||
source=_source(),
|
||
stream=False,
|
||
overlay={},
|
||
call_id="first",
|
||
reasoning_effort=None,
|
||
)
|
||
event = capture.attempts(session_id="first", parent_call_id="parent")[0].http[0]
|
||
assert not request_is_valid(event)
|
||
finally:
|
||
await real.aclose()
|