test: remove synthetic stream goldens

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-06-15 18:52:21 +08:00
co-authored by Codex
parent c588fe2447
commit 86180b99d4
9 changed files with 123 additions and 92 deletions
+2 -2
View File
@@ -387,7 +387,7 @@ Rust 证据:
- 本轮修正后,HTTP SSE streaming 在上游正常 EOF 但缺少 `response.completed`/`response.failed`/`error` 时,会给客户端追加原版格式的 `response.failed`,错误码为 `stream_disconnected`;`tests/codex_serving/upstream_errors.rs::v1_responses_stream_should_synthesize_response_failed_when_http_sse_closes_before_terminal` 覆盖该行为。
- 本轮新增 `tests/fixtures/responses/golden/stream_premature_close.sse`,并在 `tests/codex_serving/upstream_errors.rs::v1_responses_stream_should_synthesize_response_failed_when_http_sse_closes_before_terminal` 中对 HTTP SSE premature close 的完整下游 SSE 输出做 exact golden 断言;动态 `resp_proxy_*` id 在测试侧 redaction 后比较。
- 本轮新增 `tests/fixtures/responses/golden/cloudflare_challenge_no_fallback.json`,并在 `tests/codex_serving/upstream_fallback.rs::v1_responses_should_return_502_when_cloudflare_challenge_has_no_fallback` 中对 Cloudflare challenge 无备用账号的 Responses error shell 做 exact golden 断言,锁定 `server_error/codex_api_error`、502 语义和 account exhaustion message。
- 本轮修正后,streaming `/v1/responses` 在上游请求建立前 fallback exhausted 时,不再返回 OpenAI JSON error 外壳,而是按原版 `streamErrorResponse()`/`PASSTHROUGH_FORMAT.formatStreamError()` 语义返回 HTTP 200 `text/event-stream` 的 `response.failed` SSE;`tests/fixtures/responses/golden/token_invalid_no_fallback_stream.sse`、`quota_exhausted_no_fallback_stream.sse`、`model_unsupported_no_fallback_stream.sse`、`path_block_no_fallback_stream.sse` 与 `tests/codex_serving/upstream_fallback.rs` 精确锁定 HTTP SSE streaming 401/402/model unsupported/path-block 单账号无 fallback 的 SSE 外壳,`tests/codex_serving/responses_websocket.rs` 复用同一组 golden 覆盖 WebSocket streaming 402/model unsupported/path-block no-fallback,并由 `websocket_rate_limit_fallback_exhausted_stream.sse` 额外锁定 WebSocket 429 fallback exhausted 的 `rate_limit_error/rate_limit_exceeded` SSE 外壳;动态 `resp_proxy_*` id 在测试侧 redaction 后比较。
- 本轮修正后,streaming `/v1/responses` 在上游请求建立前 fallback exhausted 时,不再返回 OpenAI JSON error 外壳,而是按原版 `streamErrorResponse()`/`PASSTHROUGH_FORMAT.formatStreamError()` 语义返回 HTTP 200 `text/event-stream` 的 `response.failed` SSE;HTTP SSE 401/402/model unsupported/path-block 单账号无 fallback,以及 WebSocket 429/402/model unsupported/path-block fallback exhausted 已由 `tests/codex_serving/upstream_fallback.rs` 和 `tests/codex_serving/responses_websocket.rs` 做结构断言,锁定 `event: response.failed`、动态 `resp_proxy_*` id、`response.error == error`、错误 `type/code` 和 account exhaustion/message 关键语义。这里不再保存每个 mock 组合的 exact stream golden,因为这些是代理生成的下游错误流组合,不是原版或上游提供的返回流样本。
- 本轮修正后,合成 `response.failed` 的 JSON 字段顺序已按原版 `JSON.stringify()` 插入顺序输出:外层 `type,response,error`,error 对象 `type,code,message`;`stream_premature_close.sse` golden 同步覆盖该顺序,避免 Rust `serde_json::Value` 默认 key 排序导致 wire 格式回退。
- 本轮修正后,WebSocket streaming 在首个可见 frame 已发送后遇到 `ClosedBeforeTerminal` 或其它未见 terminal 的 body stream 错误时,也会追加同样的 `response.failed/stream_disconnected`,不再把 body stream error 直接暴露给客户端;`tests/codex_serving/responses_websocket.rs::v1_responses_websocket_stream_should_synthesize_response_failed_when_closed_before_terminal` 覆盖该行为。
- 本轮修正后,HTTP SSE 与 WebSocket streaming 的 heartbeat 首次触发延后到 15 秒,不会因为 `tokio::time::interval()` 的立即 tick 抢在上游首个可见 frame 前面;`tests/codex_serving/responses_websocket.rs::v1_responses_websocket_should_stream_first_frame_before_terminal_event` 覆盖首个 WebSocket delta 可先于 terminal 返回给客户端。
@@ -404,7 +404,7 @@ Rust 证据:
- retry/recovery 分类优先级已集中为 `UpstreamRecoveryAction`:request recovery 优先于 account retry,model unsupported 只允许一次,再进入最终错误响应。
- HTTP SSE non-streaming collect 阶段的 `previous_response_not_found` 与 unanswered function call 已有端到端覆盖,确认 mid-SSE error event 会复用同账号 strip-and-retry。
- streaming premature close 已按原版对齐:HTTP SSE 正常 EOF、HTTP SSE body stream error、WebSocket body stream error 在缺少 terminal event 时都会合成 `event: response.failed`,错误码 `stream_disconnected`,并进入现有 stream 审计失败记录;合成事件的字段顺序也已按原版固定。
- streaming fallback exhausted 外壳已按原版对齐:上游请求建立前的账号级错误在无备用账号时返回 HTTP 200 SSE `response.failed`,而不是 OpenAI JSON error;401 token invalid、402 quota exhausted、model unsupported、Cloudflare path-block 的 HTTP SSE 单账号无 fallback,以及 WebSocket 429/402/model unsupported/path-block fallback exhausted 均已有 exact golden 覆盖。
- streaming fallback exhausted 外壳已按原版对齐:上游请求建立前的账号级错误在无备用账号时返回 HTTP 200 SSE `response.failed`,而不是 OpenAI JSON error;401 token invalid、402 quota exhausted、model unsupported、Cloudflare path-block 的 HTTP SSE 单账号无 fallback,以及 WebSocket 429/402/model unsupported/path-block fallback exhausted 均已有结构断言覆盖。该类测试只锁定原版稳定语义,不再把 mock account exhaustion 场景组合成 exact golden 返回流。
- HTTP SSE premature close 已有 exact golden 覆盖,锁定先透传已收到 delta、再追加 `response.failed/stream_disconnected`、并以 SSE 双换行结束的完整输出格式。
- 单个未解析 SSE event buffer 已按原版限制为 64 MiB,避免异常上游响应在解析阶段无界增长。
- OpenAI Responses 与 OpenAI Chat 的 tuple schema request conversion/reconvert 已对齐原版,不涉及非 OpenAI translator;Responses non-streaming 与 HTTP SSE streaming tuple reconvert 已有 exact golden 覆盖。
+39 -38
View File
@@ -22,7 +22,7 @@ use tower::ServiceExt;
use codex_proxy_rs::runtime::{build_router, state::AppState};
use crate::support::{
response_json, response_text,
assert_response_failed_stream, response_json, response_text,
upstream::{
build_imported_app, build_imported_app_with_accounts,
build_imported_app_with_accounts_and_config, ImportAccount,
@@ -53,14 +53,6 @@ const WEBSOCKET_COMPLETED_WITH_REASONING_REPLAY: &str =
include_str!("../fixtures/responses/websocket/completed_with_reasoning_replay.json");
const REASONING_REPLAY_REQUEST_GOLDEN: &str =
include_str!("../fixtures/responses/golden/reasoning_replay_request.json");
const WEBSOCKET_RATE_LIMIT_FALLBACK_EXHAUSTED_STREAM_GOLDEN: &str =
include_str!("../fixtures/responses/golden/websocket_rate_limit_fallback_exhausted_stream.sse");
const QUOTA_EXHAUSTED_NO_FALLBACK_STREAM_GOLDEN: &str =
include_str!("../fixtures/responses/golden/quota_exhausted_no_fallback_stream.sse");
const MODEL_UNSUPPORTED_NO_FALLBACK_STREAM_GOLDEN: &str =
include_str!("../fixtures/responses/golden/model_unsupported_no_fallback_stream.sse");
const PATH_BLOCK_NO_FALLBACK_STREAM_GOLDEN: &str =
include_str!("../fixtures/responses/golden/path_block_no_fallback_stream.sse");
#[expect(
clippy::result_large_err,
@@ -2248,9 +2240,16 @@ async fn v1_responses_websocket_without_history_should_return_429_when_fallback_
.unwrap_or_default()
.to_string();
assert!(content_type.starts_with("text/event-stream"));
let body = redact_proxy_response_ids(&response_text(response).await);
let expected = format!("{WEBSOCKET_RATE_LIMIT_FALLBACK_EXHAUSTED_STREAM_GOLDEN}\n");
assert_eq!(body, expected);
let body = response_text(response).await;
assert_response_failed_stream(
&body,
"rate_limit_error",
"rate_limit_exceeded",
&[
"All accounts exhausted (2 rate-limited)",
"second account limited",
],
);
server.await.unwrap();
let usage_a: (i64,) =
sqlx::query_as("select request_count from account_usage where account_id = ?")
@@ -2322,9 +2321,16 @@ async fn v1_responses_websocket_without_history_should_return_stream_error_when_
.unwrap_or_default()
.to_string();
assert!(content_type.starts_with("text/event-stream"));
let body = redact_proxy_response_ids(&response_text(response).await);
let expected = format!("{QUOTA_EXHAUSTED_NO_FALLBACK_STREAM_GOLDEN}\n");
assert_eq!(body, expected);
let body = response_text(response).await;
assert_response_failed_stream(
&body,
"invalid_request_error",
"codex_api_error",
&[
"All accounts exhausted (1 quota-exhausted)",
"quota reached",
],
);
server.await.unwrap();
let account_status: (String,) = sqlx::query_as("select status from accounts where id = ?")
.bind("acct_ws_402_single")
@@ -2388,9 +2394,17 @@ async fn v1_responses_websocket_without_history_should_return_stream_error_when_
.unwrap_or_default()
.to_string();
assert!(content_type.starts_with("text/event-stream"));
let body = redact_proxy_response_ids(&response_text(response).await);
let expected = format!("{MODEL_UNSUPPORTED_NO_FALLBACK_STREAM_GOLDEN}\n");
assert_eq!(body, expected);
let body = response_text(response).await;
assert_response_failed_stream(
&body,
"invalid_request_error",
"codex_api_error",
&[
"No accounts available",
"model_not_available",
"not available",
],
);
server.await.unwrap();
let account_status: (String,) = sqlx::query_as("select status from accounts where id = ?")
.bind("acct_ws_model_single")
@@ -2447,9 +2461,13 @@ async fn v1_responses_websocket_with_history_should_return_stream_error_when_pat
.unwrap_or_default()
.to_string();
assert!(content_type.starts_with("text/event-stream"));
let body = redact_proxy_response_ids(&response_text(response).await);
let expected = format!("{PATH_BLOCK_NO_FALLBACK_STREAM_GOLDEN}\n");
assert_eq!(body, expected);
let body = response_text(response).await;
assert_response_failed_stream(
&body,
"server_error",
"codex_api_error",
&["No accounts available", "Cloudflare path-block"],
);
server.await.unwrap();
}
@@ -2541,23 +2559,6 @@ fn websocket_completed_function_call_response(response_id: &str, call_id: &str)
value.to_string()
}
fn redact_proxy_response_ids(body: &str) -> String {
let mut redacted = String::with_capacity(body.len());
let mut rest = body;
while let Some(index) = rest.find("resp_proxy_") {
redacted.push_str(&rest[..index]);
redacted.push_str("resp_proxy_REDACTED");
rest = &rest[index + "resp_proxy_".len()..];
let id_len = rest
.bytes()
.take_while(|byte| byte.is_ascii_alphanumeric())
.count();
rest = &rest[id_len..];
}
redacted.push_str(rest);
redacted
}
async fn read_http_upgrade_request(stream: &mut TcpStream) -> String {
let mut request = Vec::new();
let mut buffer = [0u8; 1024];
+36 -42
View File
@@ -14,7 +14,7 @@ use wiremock::{
use codex_proxy_rs::codex::accounts::cookies::repository::CookieRepository;
use crate::support::{
response_json, response_text,
assert_response_failed_stream, response_json, response_text,
upstream::{build_imported_app_with_accounts, ImportAccount},
};
@@ -51,22 +51,6 @@ const CLOUDFLARE_CHALLENGE_NO_FALLBACK_GOLDEN: &str =
include_str!("../fixtures/responses/golden/cloudflare_challenge_no_fallback.json");
const TOKEN_INVALID_NO_FALLBACK_GOLDEN: &str =
include_str!("../fixtures/responses/golden/token_invalid_no_fallback.json");
const TOKEN_INVALID_NO_FALLBACK_STREAM_GOLDEN: &str = concat!(
include_str!("../fixtures/responses/golden/token_invalid_no_fallback_stream.sse"),
"\n"
);
const QUOTA_EXHAUSTED_NO_FALLBACK_STREAM_GOLDEN: &str = concat!(
include_str!("../fixtures/responses/golden/quota_exhausted_no_fallback_stream.sse"),
"\n"
);
const MODEL_UNSUPPORTED_NO_FALLBACK_STREAM_GOLDEN: &str = concat!(
include_str!("../fixtures/responses/golden/model_unsupported_no_fallback_stream.sse"),
"\n"
);
const PATH_BLOCK_NO_FALLBACK_STREAM_GOLDEN: &str = concat!(
include_str!("../fixtures/responses/golden/path_block_no_fallback_stream.sse"),
"\n"
);
#[tokio::test]
async fn v1_responses_should_retry_next_account_after_429_retry_after() {
@@ -1756,8 +1740,13 @@ async fn v1_responses_should_mark_expired_and_return_401_when_401_has_no_fallbac
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = redact_proxy_response_ids(&response_text(response).await);
assert_eq!(body, TOKEN_INVALID_NO_FALLBACK_STREAM_GOLDEN);
let body = response_text(response).await;
assert_response_failed_stream(
&body,
"invalid_request_error",
"authentication_error",
&["All accounts exhausted (1 expired)", "token_revoked"],
);
let account_status: (String,) = sqlx::query_as("select status from accounts where id = ?")
.bind("acct_401_single")
.fetch_one(&imported.pool)
@@ -1816,8 +1805,16 @@ async fn v1_responses_should_mark_quota_exhausted_and_return_stream_error_when_4
.unwrap_or_default()
.to_string();
assert!(content_type.starts_with("text/event-stream"));
let body = redact_proxy_response_ids(&response_text(response).await);
assert_eq!(body, QUOTA_EXHAUSTED_NO_FALLBACK_STREAM_GOLDEN);
let body = response_text(response).await;
assert_response_failed_stream(
&body,
"invalid_request_error",
"codex_api_error",
&[
"All accounts exhausted (1 quota-exhausted)",
"quota reached",
],
);
let account_status: (String,) = sqlx::query_as("select status from accounts where id = ?")
.bind("acct_402_single")
.fetch_one(&imported.pool)
@@ -1878,8 +1875,17 @@ async fn v1_responses_should_return_stream_error_when_model_unsupported_has_no_f
.unwrap_or_default()
.to_string();
assert!(content_type.starts_with("text/event-stream"));
let body = redact_proxy_response_ids(&response_text(response).await);
assert_eq!(body, MODEL_UNSUPPORTED_NO_FALLBACK_STREAM_GOLDEN);
let body = response_text(response).await;
assert_response_failed_stream(
&body,
"invalid_request_error",
"codex_api_error",
&[
"No accounts available",
"model_not_available",
"not available",
],
);
let account_status: (String,) = sqlx::query_as("select status from accounts where id = ?")
.bind("acct_model_single")
.fetch_one(&imported.pool)
@@ -1940,8 +1946,13 @@ async fn v1_responses_should_return_stream_error_when_cloudflare_path_block_has_
.unwrap_or_default()
.to_string();
assert!(content_type.starts_with("text/event-stream"));
let body = redact_proxy_response_ids(&response_text(response).await);
assert_eq!(body, PATH_BLOCK_NO_FALLBACK_STREAM_GOLDEN);
let body = response_text(response).await;
assert_response_failed_stream(
&body,
"server_error",
"codex_api_error",
&["No accounts available", "Cloudflare path-block"],
);
assert_eq!(
cookie_repo
.cookie_header("acct_path_block_single", "chatgpt.com")
@@ -2055,20 +2066,3 @@ async fn v1_responses_should_mark_banned_when_401_says_account_deactivated() {
.unwrap();
assert_eq!(account_status.0, "banned");
}
fn redact_proxy_response_ids(body: &str) -> String {
let mut redacted = String::with_capacity(body.len());
let mut rest = body;
while let Some(index) = rest.find("resp_proxy_") {
redacted.push_str(&rest[..index]);
redacted.push_str("resp_proxy_REDACTED");
rest = &rest[index + "resp_proxy_".len()..];
let id_len = rest
.bytes()
.take_while(|byte| byte.is_ascii_alphanumeric())
.count();
rest = &rest[id_len..];
}
redacted.push_str(rest);
redacted
}
@@ -1,2 +0,0 @@
event: response.failed
data: {"type":"response.failed","response":{"id":"resp_proxy_REDACTED","status":"failed","error":{"type":"invalid_request_error","code":"codex_api_error","message":"No accounts available. Codex upstream error: {\"error\":{\"code\":\"model_not_available\",\"message\":\"Model gpt-5.5 is not available on this account plan\"}}"}},"error":{"type":"invalid_request_error","code":"codex_api_error","message":"No accounts available. Codex upstream error: {\"error\":{\"code\":\"model_not_available\",\"message\":\"Model gpt-5.5 is not available on this account plan\"}}"}}
@@ -1,2 +0,0 @@
event: response.failed
data: {"type":"response.failed","response":{"id":"resp_proxy_REDACTED","status":"failed","error":{"type":"server_error","code":"codex_api_error","message":"No accounts available. Upstream blocked the request (Cloudflare path-block)"}},"error":{"type":"server_error","code":"codex_api_error","message":"No accounts available. Upstream blocked the request (Cloudflare path-block)"}}
@@ -1,2 +0,0 @@
event: response.failed
data: {"type":"response.failed","response":{"id":"resp_proxy_REDACTED","status":"failed","error":{"type":"invalid_request_error","code":"codex_api_error","message":"All accounts exhausted (1 quota-exhausted). Codex upstream error: {\"error\":{\"message\":\"quota reached\"}}"}},"error":{"type":"invalid_request_error","code":"codex_api_error","message":"All accounts exhausted (1 quota-exhausted). Codex upstream error: {\"error\":{\"message\":\"quota reached\"}}"}}
@@ -1,2 +0,0 @@
event: response.failed
data: {"type":"response.failed","response":{"id":"resp_proxy_REDACTED","status":"failed","error":{"type":"invalid_request_error","code":"authentication_error","message":"All accounts exhausted (1 expired). Codex upstream error: {\"error\":{\"code\":\"token_revoked\",\"message\":\"token revoked\"}}"}},"error":{"type":"invalid_request_error","code":"authentication_error","message":"All accounts exhausted (1 expired). Codex upstream error: {\"error\":{\"code\":\"token_revoked\",\"message\":\"token revoked\"}}"}}
@@ -1,2 +0,0 @@
event: response.failed
data: {"type":"response.failed","response":{"id":"resp_proxy_REDACTED","status":"failed","error":{"type":"rate_limit_error","code":"rate_limit_exceeded","message":"All accounts exhausted (2 rate-limited). Codex upstream error: {\"error\":{\"message\":\"second account limited\"}}\n"}},"error":{"type":"rate_limit_error","code":"rate_limit_exceeded","message":"All accounts exhausted (2 rate-limited). Codex upstream error: {\"error\":{\"message\":\"second account limited\"}}\n"}}
+46
View File
@@ -50,3 +50,49 @@ pub async fn response_sse_data(response: axum::response::Response) -> Vec<String
.filter_map(|line| line.strip_prefix("data: ").map(ToString::to_string))
.collect()
}
pub fn assert_response_failed_stream(
body: &str,
expected_type: &str,
expected_code: &str,
expected_message_parts: &[&str],
) {
assert!(
body.ends_with("\n\n"),
"stream should end with an SSE boundary"
);
let mut event = None;
let mut data = None;
for line in body.lines() {
if let Some(value) = line.strip_prefix("event: ") {
event = Some(value);
} else if let Some(value) = line.strip_prefix("data: ") {
data = Some(value);
break;
}
}
assert_eq!(event, Some("response.failed"));
let value: Value =
serde_json::from_str(data.expect("response.failed stream should include data")).unwrap();
assert_eq!(value["type"], "response.failed");
let response = &value["response"];
assert!(response["id"]
.as_str()
.is_some_and(|id| id.starts_with("resp_proxy_")));
assert_eq!(response["status"], "failed");
assert_eq!(response["error"], value["error"]);
let error = &value["error"];
assert_eq!(error["type"], expected_type);
assert_eq!(error["code"], expected_code);
let message = error["message"]
.as_str()
.expect("response.failed error should include message");
for expected in expected_message_parts {
assert!(
message.contains(expected),
"expected stream error message to contain {expected:?}, got {message:?}"
);
}
}