test(openai): stabilize native TLS and WebSocket transport checks

Keep the Linux ClientHello fixture and compare system-native TLS on other platforms. Let the loopback WebSocket handshake finish before advancing idle time, and accept a peer reset on cancellation with unread frames.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-11-10 22:52:56 +08:00
co-authored by Codex
parent 725de6ead6
commit 13edec0b07
2 changed files with 88 additions and 52 deletions
@@ -33,31 +33,23 @@ fn http_client_hellos_should_match_official_tls_transports() {
.enable_all()
.build()
.unwrap();
let hello = runtime.block_on(async {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = format!(
"https://localhost:{}/",
listener.local_addr().unwrap().port()
);
runtime.block_on(async {
let client = provider_openai::transport::build_reqwest_client().unwrap();
let (hello, response) = timeout(Duration::from_secs(10), async {
tokio::join!(read_client_hello(listener), client.get(url).send())
})
.await
.expect("HTTP ClientHello within timeout");
assert!(
response.is_err(),
"capture endpoint rejects TLS after ClientHello"
);
hello
let mut hello = capture_http_hello(client).await;
if case == "custom-ca" {
hello.extensions.sort_unstable();
assert_eq!(hello, official_http_rustls_hello());
} else {
#[cfg(target_os = "linux")]
let expected = official_linux_http_hello();
// 官方无自定义 CA 时使用 reqwest 默认的原生 TLS;系统后端的握手随操作系统而变
#[cfg(not(target_os = "linux"))]
let expected =
capture_http_hello(reqwest::Client::builder().no_proxy().build().unwrap())
.await;
assert_eq!(hello, expected);
}
});
if case == "custom-ca" {
let mut normalized = hello;
normalized.extensions.sort_unstable();
assert_eq!(normalized, official_http_rustls_hello());
} else {
assert_eq!(hello, official_http_hello());
}
println!("\n{CASE_COMPLETED}");
return;
}
@@ -99,6 +91,24 @@ fn http_client_hellos_should_match_official_tls_transports() {
}
}
async fn capture_http_hello(client: reqwest::Client) -> ClientHello {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = format!(
"https://localhost:{}/",
listener.local_addr().unwrap().port()
);
let (hello, response) = timeout(Duration::from_secs(10), async {
tokio::join!(read_client_hello(listener), client.get(url).send())
})
.await
.expect("HTTP ClientHello within timeout");
assert!(
response.is_err(),
"capture endpoint rejects TLS after ClientHello"
);
hello
}
#[tokio::test]
async fn websocket_client_hello_should_match_official_rustls_transport() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
@@ -124,7 +134,8 @@ async fn websocket_client_hello_should_match_official_rustls_transport() {
assert_eq!(normalized, official_websocket_hello());
}
fn official_http_hello() -> ClientHello {
#[cfg(target_os = "linux")]
fn official_linux_http_hello() -> ClientHello {
ClientHello {
legacy_version: 771,
supported_versions: vec![772, 771],
@@ -203,16 +214,27 @@ async fn read_client_hello(listener: TcpListener) -> ClientHello {
assert!(values.insert(kind, take_vector(&mut extensions)).is_none());
}
let mut groups = values[&10].clone();
let groups = u16_values(take_vector(&mut groups));
let mut signatures = values[&13].clone();
let signature_algorithms = u16_values(take_vector(&mut signatures));
let mut shares = values[&51].clone();
let mut shares = take_vector(&mut shares);
// 原生后端可能只发送 TLS 1.2 扩展;扩展是否存在仍由完整的类型列表校验
let u16_list = |kind| {
values
.get(&kind)
.cloned()
.map(|mut data| u16_values(take_vector(&mut data)))
.unwrap_or_default()
};
let byte_list = |kind| {
values
.get(&kind)
.map(|data| data.slice(1..))
.unwrap_or_default()
};
let mut key_shares = Vec::new();
while shares.has_remaining() {
let group = shares.get_u16();
key_shares.push((group, take_vector(&mut shares).len()));
if let Some(mut data) = values.get(&51).cloned() {
let mut shares = take_vector(&mut data);
while shares.has_remaining() {
let group = shares.get_u16();
key_shares.push((group, take_vector(&mut shares).len()));
}
}
let mut alpn = Vec::new();
if let Some(protocols) = values.get(&16) {
@@ -225,13 +247,13 @@ async fn read_client_hello(listener: TcpListener) -> ClientHello {
}
ClientHello {
legacy_version,
supported_versions: u16_values(values[&43].slice(1..)),
point_formats: values[&11].slice(1..).to_vec(),
psk_modes: values[&45].slice(1..).to_vec(),
supported_versions: u16_values(byte_list(43)),
point_formats: byte_list(11).to_vec(),
psk_modes: byte_list(45).to_vec(),
cipher_suites,
extensions: extension_order,
groups,
signature_algorithms,
groups: u16_list(10),
signature_algorithms: u16_list(13),
key_shares,
alpn,
}
@@ -1451,18 +1451,16 @@ async fn websocket_execute_response_create_request_should_return_error_terminal_
assert!(response.body.contains("No tool output found"));
}
#[tokio::test(start_paused = true)]
#[tokio::test]
async fn codex_backend_client_should_timeout_when_upstream_is_silent() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let server = tokio::spawn(async move {
let mut websockets = Vec::new();
for _ in 0..2 {
let (stream, _) = listener.accept().await.unwrap();
let mut websocket = accept_codex_test_websocket(stream).await;
let _message = websocket.next().await.unwrap().unwrap();
websockets.push(websocket);
}
let (stream, _) = listener.accept().await.unwrap();
let mut websocket = accept_codex_test_websocket(stream).await;
let _message = websocket.next().await.unwrap().unwrap();
// 握手和首条请求使用实时时钟,随后让虚拟时钟推进到客户端的空闲期限
tokio::time::pause();
futures::future::pending::<()>().await;
});
let backend = CodexBackendClient::new(
@@ -1700,12 +1698,13 @@ async fn codex_backend_client_stream_should_reject_binary_websocket_event() {
);
}
#[tokio::test(start_paused = true)]
#[tokio::test]
async fn codex_backend_client_stream_should_keep_socket_after_structural_activity() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let accepted_connections = Arc::new(AtomicUsize::new(0));
let accepted_connections_for_server = Arc::clone(&accepted_connections);
let (output_tx, output_rx) = tokio::sync::oneshot::channel();
let server = tokio::spawn(async move {
let (first_stream, _) = listener.accept().await.unwrap();
accepted_connections_for_server.fetch_add(1, Ordering::SeqCst);
@@ -1726,7 +1725,7 @@ async fn codex_backend_client_stream_should_keep_socket_after_structural_activit
.await
.unwrap();
tokio::time::sleep(Duration::from_secs(30)).await;
output_rx.await.unwrap();
first_websocket
.send(Message::Text(
json!({
@@ -1768,7 +1767,13 @@ async fn codex_backend_client_stream_should_keep_socket_after_structural_activit
.expect("structural activity should keep the websocket stream open");
let decision = response.websocket_pool_decision;
let mut stream = response.body;
let mut body = String::new();
let first = stream.next().await.expect("structural frame").unwrap();
let mut body = std::str::from_utf8(&first).unwrap().to_owned();
assert!(body.contains("response.created"));
tokio::time::pause();
tokio::time::advance(Duration::from_secs(30)).await;
tokio::time::resume();
output_tx.send(()).unwrap();
while let Some(chunk) = stream.next().await {
let chunk = chunk.unwrap_or_else(|error| {
panic!("delayed websocket stream chunk should be valid: {error}")
@@ -2015,6 +2020,12 @@ async fn codex_backend_client_stream_should_cancel_while_inbound_is_backpressure
| Some(Err(tungstenite::Error::Protocol(
tungstenite::error::ProtocolError::ResetWithoutClosingHandshake,
))) => return,
// 取消时若 TCP 仍有未读帧,对端可直接收到连接重置
Some(Err(tungstenite::Error::Io(error)))
if error.kind() == std::io::ErrorKind::ConnectionReset =>
{
return;
}
Some(Ok(_)) => continue,
Some(Err(error)) => panic!("unexpected websocket cleanup error: {error}"),
}
@@ -2046,10 +2057,11 @@ async fn codex_backend_client_stream_should_cancel_while_inbound_is_backpressure
server.await.unwrap();
}
#[tokio::test(start_paused = true)]
#[tokio::test]
async fn codex_backend_client_stream_should_wait_for_terminal_after_active_websocket_gap() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let (finish_tx, finish_rx) = tokio::sync::oneshot::channel();
let server = tokio::spawn(async move {
let (stream, _) = listener.accept().await.unwrap();
let mut websocket = accept_codex_test_websocket(stream).await;
@@ -2065,7 +2077,7 @@ async fn codex_backend_client_stream_should_wait_for_terminal_after_active_webso
))
.await
.unwrap();
tokio::time::sleep(Duration::from_secs(30)).await;
finish_rx.await.unwrap();
websocket
.send(Message::Text(
completed_websocket_response("resp_after_active_gap", 3, 1).into(),
@@ -2098,8 +2110,10 @@ async fn codex_backend_client_stream_should_wait_for_terminal_after_active_webso
.expect("stream should yield partial frame")
.expect("partial frame should be valid");
assert!(std::str::from_utf8(&first).unwrap().contains("partial"));
tokio::time::pause();
tokio::time::advance(Duration::from_secs(30)).await;
tokio::task::yield_now().await;
tokio::time::resume();
finish_tx.send(()).unwrap();
let terminal = response
.body
.next()