diff --git a/backend/crates/gateway-api/tests/admin/accounts/mod.rs b/backend/crates/gateway-api/tests/admin/accounts/mod.rs index 6e1e9372f..b633ecda6 100644 --- a/backend/crates/gateway-api/tests/admin/accounts/mod.rs +++ b/backend/crates/gateway-api/tests/admin/accounts/mod.rs @@ -181,7 +181,13 @@ mod query { #[test] fn account_query_should_reject_unknown_fields() { - assert!(serde_json::from_value::(json!({"id": "cred_1"})).is_err()); + let mut query = json!({"page": 1, "pageSize": 20}); + serde_json::from_value::(query.clone()) + .unwrap() + .validate() + .expect("valid account query"); + query["unknownField"] = json!(true); + assert!(serde_json::from_value::(query).is_err()); } } diff --git a/backend/crates/gateway-api/tests/admin/settings.rs b/backend/crates/gateway-api/tests/admin/settings.rs index a81f7326a..decffb8c5 100644 --- a/backend/crates/gateway-api/tests/admin/settings.rs +++ b/backend/crates/gateway-api/tests/admin/settings.rs @@ -472,16 +472,15 @@ fn settings_response_should_cover_the_full_runtime_settings_contract() { } #[test] -fn settings_service_contract_stays_flat_and_rejects_unknown_fields() { +fn settings_service_contract_flattens_values_and_rejects_unknown_fields() { use gateway_admin::model::settings::{ReplaceRuntimeSettings, RuntimeSettings}; let current = super::test_runtime_settings(); - assert_eq!( - serde_json::from_slice::(&serde_json::to_vec(¤t).unwrap()).unwrap(), - current - ); let mut record = serde_json::to_value(¤t).unwrap(); - assert!(record.get("values").is_none()); + assert_eq!( + record["refresh_margin_seconds"], + current.values.refresh_margin_seconds + ); assert_eq!( serde_json::from_value::(record.clone()).unwrap(), current @@ -490,13 +489,11 @@ fn settings_service_contract_stays_flat_and_rejects_unknown_fields() { assert!(serde_json::from_value::(record).is_err()); let command = ReplaceRuntimeSettings::from(current); - assert_eq!( - serde_json::from_slice::(&serde_json::to_vec(&command).unwrap()) - .unwrap(), - command - ); let mut wire = serde_json::to_value(&command).unwrap(); - assert!(wire.get("values").is_none()); + assert_eq!( + wire["refresh_margin_seconds"], + command.values.refresh_margin_seconds + ); assert_eq!( serde_json::from_value::(wire.clone()).unwrap(), command @@ -1430,10 +1427,10 @@ async fn guardian_reservation_round_trips_and_rejects_invalid_values() { #[tokio::test] async fn account_affinity_and_rotation_budget_round_trip_and_reject_invalid_values() { + let fixture = AdminTestFixture::new().await; + fixture.auth.insert_session("valid-session"); + let mut body = update_body(); for (mode, rotations, ttl) in [("relaxed", 0, 1), ("preferred", 3, 24), ("strict", 31, 720)] { - let fixture = AdminTestFixture::new().await; - fixture.auth.insert_session("valid-session"); - let mut body = update_body(); body["openaiAccountAffinity"] = json!(mode); body["maxAccountRotations"] = json!(rotations); body["openaiSessionAffinityTtlHours"] = json!(ttl); @@ -1441,7 +1438,7 @@ async fn account_affinity_and_rotation_budget_round_trip_and_reject_invalid_valu .oneshot(request( Method::POST, "/api/admin/settings/update", - Some(body), + Some(body.clone()), )) .await .unwrap(); @@ -1450,7 +1447,9 @@ async fn account_affinity_and_rotation_budget_round_trip_and_reject_invalid_valu assert_eq!(response["data"]["openaiAccountAffinity"], mode); assert_eq!(response["data"]["maxAccountRotations"], rotations); assert_eq!(response["data"]["openaiSessionAffinityTtlHours"], ttl); + body["configRevision"] = response["data"]["configRevision"].clone(); } + let saved = fixture.settings.settings.lock().unwrap().clone(); // 类型错误沿用 AdminJson 的 422,合法类型越界由字段校验返回 400 for (field, value, expected_status) in [ ( @@ -1485,20 +1484,19 @@ async fn account_affinity_and_rotation_budget_round_trip_and_reject_invalid_valu StatusCode::UNPROCESSABLE_ENTITY, ), ] { - let fixture = AdminTestFixture::new().await; - fixture.auth.insert_session("valid-session"); - let mut body = update_body(); - body[field] = value; + let mut invalid = body.clone(); + invalid[field] = value; let response = app(fixture.state()) .oneshot(request( Method::POST, "/api/admin/settings/update", - Some(body), + Some(invalid), )) .await .unwrap(); assert_eq!(response.status(), expected_status, "{field}"); } + assert_eq!(*fixture.settings.settings.lock().unwrap(), saved); } #[test] diff --git a/backend/crates/gateway-api/tests/admin/wire.rs b/backend/crates/gateway-api/tests/admin/wire.rs index 9c60147ce..89d8e0e5a 100644 --- a/backend/crates/gateway-api/tests/admin/wire.rs +++ b/backend/crates/gateway-api/tests/admin/wire.rs @@ -1,29 +1,7 @@ //! 验证管理响应封装、分页结构与公开错误码保持稳定 -mod common { - use gateway_api::admin::PageMeta; - use serde_json::json; - - #[test] - fn page_meta_should_keep_the_existing_camel_case_wire_shape() { - let value = serde_json::to_value(PageMeta::new(2, 10, 21, 3)) - .expect("page metadata should serialize"); - - assert_eq!( - value, - json!({ - "page": 2, - "pageSize": 10, - "total": 21, - "totalPages": 3 - }) - ); - } -} - mod error { - use gateway_api::admin::{AdminErrorBody, AdminErrorCode}; - use serde_json::json; + use gateway_api::admin::AdminErrorCode; #[test] fn error_codes_should_keep_all_published_numeric_values() { @@ -56,24 +34,6 @@ mod error { ] ); } - - #[test] - fn invalid_admin_api_key_should_keep_code_and_null_data() { - let value = serde_json::to_value(AdminErrorBody::new( - AdminErrorCode::INVALID_API_KEY, - "管理 API Key 无效", - )) - .expect("admin error body should serialize"); - - assert_eq!( - value, - json!({ - "code": 40103, - "message": "管理 API Key 无效", - "data": null - }) - ); - } } mod response { @@ -85,21 +45,6 @@ mod response { use gateway_api::admin::{AdminEnvelope, AdminError, AdminPageData, AdminResponse, PageMeta}; use serde_json::json; - #[test] - fn success_envelope_should_keep_the_existing_wire_shape() { - let value = serde_json::to_value(AdminEnvelope::ok(json!({ "id": "resource_1" }))) - .expect("admin success envelope should serialize"); - - assert_eq!( - value, - json!({ - "code": 200, - "message": "OK", - "data": { "id": "resource_1" } - }) - ); - } - #[test] fn page_envelope_should_keep_items_and_page_nested_under_data() { let data = AdminPageData::new( @@ -145,8 +90,15 @@ mod response { let body = response_json(response).await; assert_eq!( - (status, body["code"].as_u64()), - (StatusCode::CREATED, Some(200)) + (status, body), + ( + StatusCode::CREATED, + json!({ + "code": 200, + "message": "OK", + "data": { "id": "resource_created" } + }) + ) ); } diff --git a/backend/crates/gateway-api/tests/openai/responses/http.rs b/backend/crates/gateway-api/tests/openai/responses/http.rs index fac5c710d..48405478c 100644 --- a/backend/crates/gateway-api/tests/openai/responses/http.rs +++ b/backend/crates/gateway-api/tests/openai/responses/http.rs @@ -3214,12 +3214,12 @@ async fn compressed_http_request_above_default_limit_should_reach_execution_afte #[tokio::test] async fn streaming_middleware_preserves_heartbeats_before_and_after_rewriting() { + let admin = crate::admin::AdminTestFixture::new().await.services; for opaque in [ b": ping\n\n".as_slice(), b"id: next\nretry: 1000\n\n".as_slice(), b"event: vendor.future\ndata: future non-JSON payload\n\n".as_slice(), ] { - let admin = crate::admin::AdminTestFixture::new().await.services; let trace = Arc::new(Trace::default()); let heartbeat = Bytes::copy_from_slice(opaque); let future = Bytes::from_static( diff --git a/backend/crates/gateway-api/tests/openai/responses/websocket/connection/middleware.rs b/backend/crates/gateway-api/tests/openai/responses/websocket/connection/middleware.rs index e6601e2f1..d5d2ebfc4 100644 --- a/backend/crates/gateway-api/tests/openai/responses/websocket/connection/middleware.rs +++ b/backend/crates/gateway-api/tests/openai/responses/websocket/connection/middleware.rs @@ -442,8 +442,17 @@ async fn input_plugin_runs_while_a_real_socket_write_is_blocked() { let (trace, mut socket, server) = start_active_response_with_middleware(Some(frozen), "/custom/websocket").await; trace.release_terminal.notify_one(); - // 客户端不读取 64 MiB 输出,真实 TCP 写入将受接收窗口背压 - tokio::time::sleep(Duration::from_millis(100)).await; + // 仅窥视帧头以确认写入开始,不消费 64 MiB 输出,保留真实 TCP 接收窗口背压 + let MaybeTlsStream::Plain(transport) = socket.get_ref() else { + panic!("test server must use plain TCP") + }; + let mut frame_header = [0; 1]; + let received = tokio::time::timeout(Duration::from_secs(2), transport.peek(&mut frame_header)) + .await + .expect("terminal output must start") + .expect("peek terminal frame"); + assert_eq!(received, 1); + assert_eq!(frame_header[0] & 0x0f, 2); assert!(!trace.finalized.load(Ordering::SeqCst)); socket .send(ClientMessage::Text("signal".into())) diff --git a/backend/crates/gateway-core/tests/engine/coordinator.rs b/backend/crates/gateway-core/tests/engine/coordinator.rs index 8335196c6..c6ae3a7cb 100644 --- a/backend/crates/gateway-core/tests/engine/coordinator.rs +++ b/backend/crates/gateway-core/tests/engine/coordinator.rs @@ -4660,15 +4660,10 @@ fn transient_backoff_can_be_cancelled_before_the_next_attempt() { cancellation.clone(), )) .expect("start execution"); - let cancel = std::thread::spawn(move || { - std::thread::sleep(Duration::from_millis(50)); - cancellation.cancel(); - }); - assert!(matches!( - block_on(session.collect_uncommitted()), - Err(EngineError::Cancelled) - )); - cancel.join().expect("cancel task"); + let mut collection = Box::pin(session.collect_uncommitted()); + assert!(block_on(async { futures::poll!(&mut collection) }).is_pending()); + cancellation.cancel(); + assert!(matches!(block_on(collection), Err(EngineError::Cancelled))); assert_eq!(provider.contexts.lock().expect("contexts").len(), 1); } } @@ -6085,63 +6080,69 @@ fn connection_recovery_rotates_only_after_retrying_the_first_account() { #[test] fn connection_recovery_keeps_one_account_and_http_transport_until_success() { - for unavailable in [ + let cases = [ ProviderErrorKind::NoEligibleAccount, ProviderErrorKind::AccountCapacityUnavailable, ProviderErrorKind::QuotaExhausted, ProviderErrorKind::ConcurrencyQueueFull, ProviderErrorKind::ConcurrencyQueueTimeout, - ] { - let operation = generate_operation(); - let route_plan = plan(&operation); - let failure = || Script::Stream { - account_id: "acct_only", - items: vec![Err(ProviderError::new( - ProviderErrorKind::Transport, - UpstreamSendState::NotSent, - ) - .with_connection_retry(AttemptTransport::Fallback))], - }; - let (coordinator, store, provider) = coordinator(vec![ - failure(), - failure(), - Script::Error(ProviderError::new(unavailable, UpstreamSendState::NotSent)), - failure(), - Script::Stream { + ]; + // 各场景独立持有存储与 Provider,并发等待各自的真实连接退避 + block_on(futures::future::join_all(cases.into_iter().map( + |unavailable| async move { + let operation = generate_operation(); + let route_plan = plan(&operation); + let failure = || Script::Stream { account_id: "acct_only", - items: complete_stream(None), - }, - ]); - let mut session = block_on(coordinator.start( - model_request(&operation, SystemTime::now() + Duration::from_secs(30)), - operation, - route_plan, - None, - None, - CancellationToken::new(), - )) - .unwrap(); - block_on(session.collect_uncommitted()).unwrap(); - block_on(session.commit_downstream(Some(200))).unwrap(); - let contexts = provider.contexts.lock().unwrap(); - assert_eq!(contexts.len(), 5); - let account = ProviderAccountId::new("acct_only").unwrap(); - assert!(contexts[2].required_account().is_none()); - assert!(contexts[2].excluded_accounts().contains(&account)); - assert_eq!(contexts[2].transport(), AttemptTransport::Fallback); - for index in [1, 3, 4] { - let context = &contexts[index]; - assert_eq!(context.required_account(), Some(&account)); - assert_eq!(context.transport(), AttemptTransport::Fallback); - assert!(!context.excluded_accounts().contains(&account)); - } - let state = store.state.lock().unwrap(); - assert_eq!(state.intermediate_failures, 3); - assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Succeeded); - assert_eq!(provider.released_leases.load(Ordering::SeqCst), 3); - drop(session); - assert_eq!(provider.released_leases.load(Ordering::SeqCst), 4); - } + items: vec![Err(ProviderError::new( + ProviderErrorKind::Transport, + UpstreamSendState::NotSent, + ) + .with_connection_retry(AttemptTransport::Fallback))], + }; + let (coordinator, store, provider) = coordinator(vec![ + failure(), + failure(), + Script::Error(ProviderError::new(unavailable, UpstreamSendState::NotSent)), + failure(), + Script::Stream { + account_id: "acct_only", + items: complete_stream(None), + }, + ]); + let mut session = coordinator + .start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation, + route_plan, + None, + None, + CancellationToken::new(), + ) + .await + .unwrap(); + session.collect_uncommitted().await.unwrap(); + session.commit_downstream(Some(200)).await.unwrap(); + let contexts = provider.contexts.lock().unwrap(); + assert_eq!(contexts.len(), 5); + let account = ProviderAccountId::new("acct_only").unwrap(); + assert!(contexts[2].required_account().is_none()); + assert!(contexts[2].excluded_accounts().contains(&account)); + assert_eq!(contexts[2].transport(), AttemptTransport::Fallback); + for index in [1, 3, 4] { + let context = &contexts[index]; + assert_eq!(context.required_account(), Some(&account)); + assert_eq!(context.transport(), AttemptTransport::Fallback); + assert!(!context.excluded_accounts().contains(&account)); + } + let state = store.state.lock().unwrap(); + assert_eq!(state.intermediate_failures, 3); + assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Succeeded); + assert_eq!(provider.released_leases.load(Ordering::SeqCst), 3); + drop(session); + assert_eq!(provider.released_leases.load(Ordering::SeqCst), 4); + }, + ))); } #[test] @@ -6385,15 +6386,10 @@ fn connection_backoff_cancellation_releases_account_without_another_attempt() { cancellation.clone(), )) .expect("start execution"); - let cancel = std::thread::spawn(move || { - std::thread::sleep(Duration::from_millis(50)); - cancellation.cancel(); - }); - assert!(matches!( - block_on(session.collect_uncommitted()), - Err(EngineError::Cancelled) - )); - cancel.join().expect("cancel task"); + let mut collection = Box::pin(session.collect_uncommitted()); + assert!(block_on(async { futures::poll!(&mut collection) }).is_pending()); + cancellation.cancel(); + assert!(matches!(block_on(collection), Err(EngineError::Cancelled))); assert_eq!(provider.contexts.lock().expect("contexts").len(), 1); assert_eq!(provider.released_leases.load(Ordering::SeqCst), 1); } diff --git a/backend/crates/gateway-host/tests/client_distribution.rs b/backend/crates/gateway-host/tests/client_distribution.rs index dff71ed81..ee2182795 100644 --- a/backend/crates/gateway-host/tests/client_distribution.rs +++ b/backend/crates/gateway-host/tests/client_distribution.rs @@ -7,6 +7,7 @@ use gateway_admin::model::client_distribution::{ use gateway_host::client_distribution::{ ClientDistributionResolutionErrorKind, complete_with_official_fallback, parse_store_packages, }; +use std::time::Duration; #[test] fn parser_selects_latest_valid_package_for_each_architecture() { @@ -114,10 +115,6 @@ async fn official_package_size_failure_keeps_metadata_optional() { for (route, response) in [ ("/missing.msix", ResponseTemplate::new(404)), ("/empty.msix", ResponseTemplate::new(200)), - ( - "/slow.msix", - ResponseTemplate::new(200).set_delay(std::time::Duration::from_secs(4)), - ), ] { Mock::given(method("HEAD")) .and(path(route)) @@ -127,19 +124,57 @@ async fn official_package_size_failure_keeps_metadata_optional() { .await; } let client = reqwest::Client::builder().no_proxy().build().unwrap(); - for route in ["/missing.msix", "/empty.msix", "/slow.msix"] { + for route in ["/missing.msix", "/empty.msix"] { assert_eq!( - tokio::time::timeout( - std::time::Duration::from_millis(3500), - gateway_host::client_distribution::official_package_size( - &client, - &format!("{}{route}", server.uri()), - ), + gateway_host::client_distribution::official_package_size( + &client, + &format!("{}{route}", server.uri()), ) - .await - .expect("metadata timeout must not wait for the slow response"), + .await, None ); } server.verify().await; } + +#[tokio::test] +async fn official_package_size_timeout_keeps_metadata_optional() { + use tokio::io::AsyncReadExt as _; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}/slow.msix", listener.local_addr().unwrap()); + let client = reqwest::Client::builder().no_proxy().build().unwrap(); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let mut request = tokio::spawn(async move { + started_tx.send(tokio::time::Instant::now()).unwrap(); + gateway_host::client_distribution::official_package_size(&client, &url).await + }); + let _stream = tokio::time::timeout(Duration::from_secs(2), async { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut head = Vec::new(); + while !head.ends_with(b"\r\n\r\n") { + head.push(stream.read_u8().await.unwrap()); + } + assert!(head.starts_with(b"HEAD /slow.msix HTTP/1.1\r\n")); + stream + }) + .await + .expect("metadata request must reach the endpoint"); + let started = started_rx.await.unwrap(); + + // 先完成真实连接再推进超时,避免暂停时钟抢跑网络 I/O + tokio::time::pause(); + assert!( + tokio::time::timeout_at(started + Duration::from_secs(2), &mut request) + .await + .is_err(), + "the three-second budget is still available" + ); + assert_eq!( + tokio::time::timeout_at(started + Duration::from_millis(3_100), request) + .await + .expect("metadata request must stop at its three-second deadline") + .unwrap(), + None + ); +} diff --git a/backend/crates/gateway-host/tests/config.rs b/backend/crates/gateway-host/tests/config.rs index 3918a5a05..83b2a2a8c 100644 --- a/backend/crates/gateway-host/tests/config.rs +++ b/backend/crates/gateway-host/tests/config.rs @@ -5,24 +5,6 @@ use std::path::PathBuf; use gateway_host::config::{FileLoggingConfig, HostConfig, ListenConfig, LoggingConfig}; use gateway_host::system_update::SystemUpdateConfig; -#[test] -fn system_update_defaults_should_use_host_build_metadata_and_official_repository() { - let config = SystemUpdateConfig::default(); - - assert_eq!( - ( - config.version.as_str(), - config.build_type.as_str(), - config.update_repository.as_deref(), - ), - ( - env!("CPR_VERSION"), - env!("CPR_BUILD_TYPE"), - Some("zyycn/codex-proxy-rs"), - ) - ); -} - #[test] fn host_config_should_default_update_assets_to_the_resolved_api_directory() { let mut config = valid_config(); diff --git a/backend/crates/gateway-host/tests/logging/writer.rs b/backend/crates/gateway-host/tests/logging/writer.rs index 8272af6af..272b4ce1f 100644 --- a/backend/crates/gateway-host/tests/logging/writer.rs +++ b/backend/crates/gateway-host/tests/logging/writer.rs @@ -239,7 +239,8 @@ fn request_dump_rotates_small_files_when_the_time_window_elapses() { if env::var_os(CHILD_PROCESS_ENV).is_some() { let directory = PathBuf::from(env::var_os(LOG_DIRECTORY_ENV).unwrap()); let mut config = logging_config(directory.clone(), true); - config.logging.request_dump_retention_days = 1.0 / 86_400.0; + let retention = std::time::Duration::from_millis(100); + config.logging.request_dump_retention_days = retention.as_secs_f64() / 86_400.0; with_logging(config, || { tracing::info!(target: REQUEST_DUMP_LOG_TARGET, "before retention window"); let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); @@ -254,7 +255,7 @@ fn request_dump_rotates_small_files_when_the_time_window_elapses() { ); std::thread::sleep(std::time::Duration::from_millis(10)); } - std::thread::sleep(std::time::Duration::from_millis(1100)); + std::thread::sleep(retention + std::time::Duration::from_millis(10)); tracing::info!(target: REQUEST_DUMP_LOG_TARGET, "after retention window"); }); return; diff --git a/backend/crates/gateway-plugin/runtime/tests/package/compatibility.rs b/backend/crates/gateway-plugin/runtime/tests/package/compatibility.rs index 5e59fa96f..a3c955601 100644 --- a/backend/crates/gateway-plugin/runtime/tests/package/compatibility.rs +++ b/backend/crates/gateway-plugin/runtime/tests/package/compatibility.rs @@ -36,23 +36,6 @@ fn host_compatibility_only_advertises_known_contracts() { } } -#[test] -fn host_compatibility_requires_the_trusted_middleware_contract() { - let compatibility: PluginHostCompatibility = - serde_json::from_str(include_str!("../../plugin-host-compatibility.json")) - .expect("compatibility JSON"); - assert!(!compatibility.supports_capability("middleware", 1)); - assert!(!compatibility.supports_capability("middleware", 2)); - assert!(!compatibility.supports_capability("middleware", 3)); - assert!(!compatibility.supports_capability("middleware", 4)); - assert!(compatibility.supports_capability("middleware", 5)); - assert!(!compatibility.supports_capability("upstream_adapter", 1)); - assert!(!compatibility.supports_capability("upstream_adapter", 2)); - assert!(!compatibility.supports_capability("upstream_adapter", 3)); - assert!(compatibility.supports_capability("upstream_adapter", 4)); - assert!(!compatibility.supports_capability("openai", 1)); -} - #[tokio::test] async fn package_inspector_returns_static_requirements_without_starting_the_plugin() { let archive = crate::support::package_with_contributions( diff --git a/backend/crates/gateway-plugin/runtime/tests/support/environment.rs b/backend/crates/gateway-plugin/runtime/tests/support/environment.rs index 3147aa83d..1a7c30463 100644 --- a/backend/crates/gateway-plugin/runtime/tests/support/environment.rs +++ b/backend/crates/gateway-plugin/runtime/tests/support/environment.rs @@ -715,6 +715,17 @@ impl Environment { database .query_pairs_mut() .append_pair("options", &format!("-csearch_path={schema}")); + // 每个夹具独占 UUID schema,先并行建表,避免完整 DDL 争用数据库级迁移锁 + // 原 Store 初始化仍会加锁核验同一套迁移记录,保留生产启动屏障 + let migration_pool = PgPoolOptions::new() + .max_connections(1) + .connect(database.as_str()) + .await + .unwrap(); + let mut migrator = sqlx::migrate!("../../../migrations"); + migrator.set_locking(false); + migrator.run(&migration_pool).await.unwrap(); + migration_pool.close().await; let connection = |mut url: url::Url| { let password = url.password().expect("test password").to_owned(); url.set_password(None).unwrap(); diff --git a/backend/crates/gateway-plugin/sdk/tests/client/session/credit.rs b/backend/crates/gateway-plugin/sdk/tests/client/session/credit.rs index 72af4f63e..3a288bc52 100644 --- a/backend/crates/gateway-plugin/sdk/tests/client/session/credit.rs +++ b/backend/crates/gateway-plugin/sdk/tests/client/session/credit.rs @@ -2,7 +2,7 @@ use super::*; -#[tokio::test] +#[tokio::test(start_paused = true)] async fn stream_waits_for_credit_and_rejects_uncreditable_buffered_chunk_before_result() { let (mut host, task) = start_session(TestHandler::default()).await; send_call(&mut host, 1, "stream", json!({}), Vec::new()).await; @@ -217,7 +217,7 @@ async fn dynamic_stream_ends_with_one_error_when_a_later_chunk_cannot_fit_credit shutdown(&mut host, task).await; } -#[tokio::test] +#[tokio::test(start_paused = true)] async fn resource_stream_keeps_callbacks_and_credit_alive_after_the_initial_deadline() { let (mut host, task) = start_session(TestHandler::default()).await; let mut context = context(1, Duration::from_millis(500)); diff --git a/backend/crates/gateway-plugin/sdk/tests/client/session/driver.rs b/backend/crates/gateway-plugin/sdk/tests/client/session/driver.rs index 2fa299514..5627c648b 100644 --- a/backend/crates/gateway-plugin/sdk/tests/client/session/driver.rs +++ b/backend/crates/gateway-plugin/sdk/tests/client/session/driver.rs @@ -144,7 +144,7 @@ async fn middleware_plugin_rejects_invalid_calls_before_business_dispatch() { shutdown(&mut host, task).await; } -#[tokio::test] +#[tokio::test(start_paused = true)] async fn independent_calls_complete_concurrently() { let (mut host, task) = start_session(TestHandler::default()).await; send_call(&mut host, 1, "slow", json!("slow"), vec![1]).await; @@ -170,7 +170,7 @@ async fn independent_calls_complete_concurrently() { shutdown(&mut host, task).await; } -#[tokio::test] +#[tokio::test(start_paused = true)] async fn completed_call_does_not_discard_a_partially_read_next_frame() { let next = Frame { message: Message::Call { diff --git a/backend/crates/gateway-store/tests/postgres/client_keys.rs b/backend/crates/gateway-store/tests/postgres/client_keys.rs index 3d44c77e0..0793fe393 100644 --- a/backend/crates/gateway-store/tests/postgres/client_keys.rs +++ b/backend/crates/gateway-store/tests/postgres/client_keys.rs @@ -26,18 +26,18 @@ use gateway_store::postgres::{ use super::TestDatabase; #[tokio::test] -async fn migrated_keys_persist_exactly_and_keep_short_keys_masked() { +async fn custom_keys_persist_exactly_and_keep_short_keys_masked() { use gateway_admin::model::{MutationActor, MutationContext, client_keys::NewClientKey}; use gateway_core::policy::RateLimits; use gateway_store::postgres::{PgRuntimeSnapshotRepository, RuntimeSnapshotRepository}; - let Some(database) = TestDatabase::create("migrated_keys").await else { + let Some(database) = TestDatabase::create("custom_keys").await else { return; }; let store = PgAdminClientKeyStore::new(database.pool.clone()); let context = MutationContext { actor: MutationActor::System, - request_id: "migration-test".to_owned(), + request_id: "custom-key-test".to_owned(), }; let long_key: String = (0..160) .map(|_| uuid::Uuid::new_v4().simple().to_string()) @@ -45,17 +45,17 @@ async fn migrated_keys_persist_exactly_and_keep_short_keys_masked() { let values = [ "q".to_owned(), "abc".to_owned(), - "sk-legacy/key+value=:!@\\".to_owned(), + "sk-custom/key+value=:!@\\".to_owned(), long_key, ]; for (index, key) in values.iter().enumerate() { - let id = ClientApiKeyId::new(format!("key_migrated_{index}")).unwrap(); + let id = ClientApiKeyId::new(format!("key_custom_{index}")).unwrap(); let (_, record) = store .create_client_key( NewClientKey { request_profile_overrides: Default::default(), id: id.clone(), - name: format!("Migrated {index}"), + name: format!("Custom {index}"), label: None, group_ids: Vec::new(), limits: RateLimits { @@ -91,7 +91,7 @@ async fn migrated_keys_persist_exactly_and_keep_short_keys_masked() { } #[tokio::test] -async fn duplicate_migrated_keys_conflict_atomically_without_extra_audits() { +async fn duplicate_custom_keys_conflict_atomically_without_extra_audits() { use gateway_admin::{ model::{MutationActor, MutationContext, client_keys::NewClientKey}, ports::store::AdminStoreErrorKind, @@ -106,7 +106,7 @@ async fn duplicate_migrated_keys_conflict_atomically_without_extra_audits() { actor: MutationActor::System, request_id: "duplicate-test".to_owned(), }; - let key = "legacy-key-case-sensitive!"; + let key = "custom-key-case-sensitive!"; let command = |id: &str| NewClientKey { request_profile_overrides: Default::default(), id: ClientApiKeyId::new(id).unwrap(), diff --git a/backend/crates/gateway-store/tests/postgres/mod.rs b/backend/crates/gateway-store/tests/postgres/mod.rs index d94529eab..788f81b9c 100644 --- a/backend/crates/gateway-store/tests/postgres/mod.rs +++ b/backend/crates/gateway-store/tests/postgres/mod.rs @@ -42,7 +42,13 @@ mod snapshots; mod usage_data; mod usage_paging; -static TEST_MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("../../migrations"); +static TEST_MIGRATOR: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + let mut migrator = sqlx::migrate!("../../migrations"); + // 每个夹具独占 UUID schema,迁移及其记录不共享对象,无需数据库级迁移锁 + migrator.set_locking(false); + migrator + }); pub(super) struct TestDatabase { admin: PgPool, @@ -70,7 +76,8 @@ pub(super) fn admin_account_store(pool: &PgPool) -> PgAdminAccountStore { impl TestDatabase { pub(super) async fn create(label: &str) -> Option { let database_url = crate::support::test_env("CPR_TEST_DATABASE_URL")?; - let schema = format!("cpr_store_{label}_{}", Uuid::new_v4().simple()); + // UUID 放在前面,长标签被 PostgreSQL 截断时仍保留完整隔离标识 + let schema = format!("cpr_store_{}_{label}", Uuid::new_v4().simple()); // 临时 schema 验证事务可见性与回滚,不模拟 PostgreSQL 掉电恢复 // 仅这些测试连接异步刷 WAL,保持服务端配置和生产连接行为不变 let options = PgConnectOptions::from_str(&database_url) diff --git a/backend/crates/gateway-store/tests/postgres/plugins/state.rs b/backend/crates/gateway-store/tests/postgres/plugins/state.rs index 0ec133225..3fc488621 100644 --- a/backend/crates/gateway-store/tests/postgres/plugins/state.rs +++ b/backend/crates/gateway-store/tests/postgres/plugins/state.rs @@ -11,7 +11,7 @@ use gateway_admin::{ state::{ ApplyPluginStateMigration, DeletePluginState, PluginStateCommit, PluginStateConfiguration, PluginStateMigrationAction, PluginStateMigrationChange, - PluginStateOwnerRequest, PluginStateSchema, PutPluginState, + PluginStateOwnerRequest, PluginStateSchema, PluginStateTransition, PutPluginState, }, }, }, @@ -164,6 +164,31 @@ async fn transition_fixture( ) } +async fn begin_transition_after_release( + store: &PgPluginStore, + instance: &PluginInstance, + digest: &str, + target: PluginStateConfiguration, +) -> PluginStateTransition { + // PostgreSQL 异步处理断连,所有者释放后仍需等待服务端解除会话锁 + tokio::time::timeout(Duration::from_secs(3), async { + loop { + match store + .begin_transition(&instance.id, instance.revision, digest, target.clone()) + .await + { + Ok(transition) => break transition, + Err(error) if error.kind() == PluginStateStoreErrorKind::Conflict => { + tokio::time::sleep(Duration::from_millis(10)).await; + } + Err(error) => panic!("unexpected retry failure: {error:?}"), + } + } + }) + .await + .expect("released migration owner must relinquish its session lock") +} + #[tokio::test] async fn cancelled_migration_releases_ownership_and_can_be_retried() { let Some(database) = TestDatabase::create("plugin_state_cancel").await else { @@ -191,23 +216,7 @@ async fn cancelled_migration_releases_ownership_and_can_be_retried() { let abandoned_id = migration.await.unwrap(); task.abort(); assert!(task.await.unwrap_err().is_cancelled()); - // 关闭会话由 PostgreSQL 异步处理,重试只等待所有权释放,不调用清理接口补偿 - let retried = tokio::time::timeout(Duration::from_secs(3), async { - loop { - match store - .begin_transition(&instance.id, instance.revision, &digest, target.clone()) - .await - { - Ok(transition) => break transition, - Err(error) if error.kind() == PluginStateStoreErrorKind::Conflict => { - tokio::time::sleep(Duration::from_millis(10)).await; - } - Err(error) => panic!("unexpected retry failure: {error:?}"), - } - } - }) - .await - .expect("cancelled migration must release its session lock"); + let retried = begin_transition_after_release(&store, &instance, &digest, target).await; assert_ne!(retried.id, abandoned_id); let retained: bool = sqlx::query_scalar( "select exists(select 1 from plugin_state_generations where transition_id=$1)", @@ -1050,15 +1059,13 @@ async fn migration_promotes_atomically_and_incomplete_staging_cannot_replace_act drop(transition); let third_state = configuration(schema(3, vec![2], 4)); - let incomplete = store - .begin_transition( - &upgraded.instance.id, - upgraded.instance.revision, - &third_artifact.artifact.metadata.sha256, - third_state.clone(), - ) - .await - .unwrap(); + let incomplete = begin_transition_after_release( + &store, + &upgraded.instance, + &third_artifact.artifact.metadata.sha256, + third_state.clone(), + ) + .await; let mut rejected = upgraded.instance.clone(); rejected.artifact_sha256 = third_artifact.artifact.metadata.sha256; let error = store diff --git a/backend/crates/gateway-store/tests/redis/auth.rs b/backend/crates/gateway-store/tests/redis/auth.rs index 966228bd6..8361d3858 100644 --- a/backend/crates/gateway-store/tests/redis/auth.rs +++ b/backend/crates/gateway-store/tests/redis/auth.rs @@ -46,12 +46,7 @@ async fn client_session_should_use_fixed_ttl_without_storing_session_or_api_key_ let key = &keys[0]; assert!(!key.contains(session_id)); assert!(!key.contains(raw_api_key)); - let ttl = redis::cmd("PTTL") - .arg(key) - .query_async::(&mut connection) - .await - .expect("read client session ttl"); - assert!((1..=60_000).contains(&ttl)); + assert_session_expiry(&mut connection, key, session.expires_at).await; let payload = redis::cmd("GET") .arg(key) .query_async::(&mut connection) @@ -151,6 +146,27 @@ async fn auth_repository() -> Option<(RedisAuthStateRepository, ConnectionManage Some((repository, connection, full_namespace)) } +async fn assert_session_expiry( + connection: &mut ConnectionManager, + key: &str, + expires_at: chrono::DateTime, +) { + // PXAT 固定绝对到期时间;PTTL 使用 Redis 时钟,不能与客户端相对时长直接比较 + let (stored_expiry, ttl): (i64, i64) = redis::pipe() + .cmd("PEXPIRETIME") + .arg(key) + .cmd("PTTL") + .arg(key) + .query_async(connection) + .await + .expect("read session expiry and ttl"); + assert_eq!(stored_expiry, expires_at.timestamp_millis()); + assert!( + ttl > 0, + "session must have a finite positive TTL, got {ttl}" + ); +} + #[test] fn admin_auth_state_rejects_invalid_ttl_boundaries() { let session = AuthSessionRecord { @@ -211,12 +227,7 @@ async fn admin_auth_state_keeps_fixed_ttl_and_opaque_keys() { assert_eq!(keys.len(), 1); for key in &keys { assert!(!key.contains(session_id)); - let ttl = redis::cmd("PTTL") - .arg(key) - .query_async::(&mut connection) - .await - .expect("read key ttl"); - assert!((1..=60_000).contains(&ttl)); + assert_session_expiry(&mut connection, key, session.expires_at).await; let value = redis::cmd("GET") .arg(key) .query_async::(&mut connection) @@ -396,12 +407,12 @@ async fn redis_retains_refresh_after_access_expiry_but_never_resurrects_expired_ .query_async(&mut connection) .await .unwrap(); - let ttl: i64 = redis::cmd("PTTL") - .arg(&keys[0]) - .query_async(&mut connection) - .await - .unwrap(); - assert!((240_000..=300_000).contains(&ttl)); + assert_session_expiry( + &mut connection, + &keys[0], + original.renewal.as_ref().unwrap().expires_at, + ) + .await; assert_eq!( repository.load_session("expiry-race").await.unwrap(), Some(original.clone()) @@ -415,12 +426,12 @@ async fn redis_retains_refresh_after_access_expiry_but_never_resurrects_expired_ .unwrap(), Some(replacement.clone()) ); - let ttl: i64 = redis::cmd("PTTL") - .arg(&keys[0]) - .query_async(&mut connection) - .await - .unwrap(); - assert!((300_000..=360_000).contains(&ttl)); + assert_session_expiry( + &mut connection, + &keys[0], + replacement.renewal.as_ref().unwrap().expires_at, + ) + .await; redis::cmd("PEXPIRE") .arg(&keys[0]) .arg(1) diff --git a/backend/crates/gateway-store/tests/redis/credential_leases/mod.rs b/backend/crates/gateway-store/tests/redis/credential_leases/mod.rs index 2eed26692..abf91bffd 100644 --- a/backend/crates/gateway-store/tests/redis/credential_leases/mod.rs +++ b/backend/crates/gateway-store/tests/redis/credential_leases/mod.rs @@ -189,10 +189,12 @@ async fn scheduling_lease_ignores_future_last_started_signal() { #[tokio::test] async fn maintained_scheduling_lease_survives_initial_ttl_and_releases_on_drop() { - for scope in [ + let scopes = [ CredentialLeaseScope::ProviderAccount, CredentialLeaseScope::ProviderAccountReserved, - ] { + ]; + // 每个场景创建独立 namespace,两个真实 TTL 窗口可以同时观察 + futures::future::join_all(scopes.into_iter().map(|scope| async move { let Some((repository, _, _)) = repository().await else { return; }; @@ -234,7 +236,8 @@ async fn maintained_scheduling_lease_survives_initial_ttl_and_releases_on_drop() }) .await .expect("dropped request releases its slot"); - } + })) + .await; } #[tokio::test] diff --git a/backend/crates/providers/openai/tests/config.rs b/backend/crates/providers/openai/tests/config.rs index 6f569e968..18632cdfd 100644 --- a/backend/crates/providers/openai/tests/config.rs +++ b/backend/crates/providers/openai/tests/config.rs @@ -1,8 +1,8 @@ -//! 验证 OpenAI 配置默认值、路径派生与上游地址约束 +//! 验证 OpenAI 配置解析、路径派生与上游地址约束 use std::path::Path; -use provider_openai::config::{DEFAULT_STREAM_MAX_RETRIES, MAX_STREAM_MAX_RETRIES, OpenAiConfig}; +use provider_openai::config::{MAX_STREAM_MAX_RETRIES, OpenAiConfig}; #[test] fn openai_config_keeps_residency_separate_from_client_identity() { @@ -59,39 +59,6 @@ fn openai_config_restricts_upstream_base_url_to_https_or_loopback_http() { } } -#[test] -fn openai_config_defaults_to_the_provider_owned_operating_values() { - let config = OpenAiConfig::default(); - assert_eq!(DEFAULT_STREAM_MAX_RETRIES, 5); - - assert_eq!( - ( - config.api.base_url.as_str(), - config.ws_pool.enabled, - config.ws_pool.max_age_ms, - config.ws_pool.max_connecting, - config.ws_pool.stream_idle_timeout_ms, - config.quota.refresh_interval_minutes, - config.auth.refresh_enabled, - config.auth.oauth_client_id.as_str(), - config.auth.oauth_token_endpoint.as_str(), - config.stream_max_retries(), - ), - ( - "https://chatgpt.com/backend-api", - true, - 3_300_000, - 8, - 300_000, - 15, - true, - "app_EMoamEEZ73f0CkXaXp7hrann", - "https://auth.openai.com/oauth/token", - u32::try_from(DEFAULT_STREAM_MAX_RETRIES).expect("default retry budget fits u32"), - ) - ); -} - #[test] fn openai_stream_retry_budget_uses_the_official_hard_cap() { let mut config = OpenAiConfig::default(); diff --git a/backend/crates/providers/openai/tests/credential/admin.rs b/backend/crates/providers/openai/tests/credential/admin.rs index 296eb0556..e17455c23 100644 --- a/backend/crates/providers/openai/tests/credential/admin.rs +++ b/backend/crates/providers/openai/tests/credential/admin.rs @@ -962,27 +962,28 @@ async fn oauth_import_does_not_use_top_level_identity_fields() { } #[tokio::test] -async fn oauth_import_rejects_legacy_bare_token_field() { +async fn oauth_import_requires_access_or_refresh_token() { let service = CodexCredentialAdminService::new( Arc::new(UnusedRefresher), Arc::new(TestLeaseCoordinator::default()), runtime_policy(), Arc::new(crate::RecordingDiagnostics::default()), ); - let error = service - .prepare_import_document_with_proxy( - serde_json::json!({ - "token": "header.payload.signature" - }), - None, - ) - .await - .expect_err("ambiguous token field must not be accepted"); + for document in [ + serde_json::json!({}), + serde_json::json!({"accessToken": null, "refreshToken": null}), + serde_json::json!({"idToken": "header.payload.signature"}), + ] { + let error = service + .prepare_import_document_with_proxy(document, None) + .await + .expect_err("access or refresh token is required"); - assert_eq!( - error, - provider_openai::credential::CodexCredentialAdminError::InvalidCredential - ); + assert_eq!( + error, + provider_openai::credential::CodexCredentialAdminError::InvalidCredential + ); + } } #[tokio::test] diff --git a/backend/crates/providers/openai/tests/credential/quota/initial_sync.rs b/backend/crates/providers/openai/tests/credential/quota/initial_sync.rs index c37cf7018..c5aab0413 100644 --- a/backend/crates/providers/openai/tests/credential/quota/initial_sync.rs +++ b/backend/crates/providers/openai/tests/credential/quota/initial_sync.rs @@ -82,7 +82,7 @@ async fn zero_initial_delay_keeps_immediate_first_observation() { assert_eq!( service.synchronize().await.expect("first cycle").updated, 1, - "零延迟必须保持首轮即查的旧行为" + "零延迟必须在首轮查询额度" ); // 已产生快照后退出首查路径,正常账号也不满足周期复核条件。 service.synchronize().await.expect("second cycle"); diff --git a/backend/crates/providers/openai/tests/credential/quota/mod.rs b/backend/crates/providers/openai/tests/credential/quota/mod.rs index 3d740cb31..6d5dc9dec 100644 --- a/backend/crates/providers/openai/tests/credential/quota/mod.rs +++ b/backend/crates/providers/openai/tests/credential/quota/mod.rs @@ -69,8 +69,8 @@ pub(super) fn quota_service_with_base_url( http: reqwest::Client, base_url: String, ) -> CodexCredentialQuotaService { - // 首查随机启动延迟只在 initial_sync 专项测试中注入非零值; - // 其余测试固定为零,保持「首轮即查」的确定性旧行为。 + // 首查随机启动延迟只在 initial_sync 专项测试中注入非零值 + // 其余测试固定为零,避免计时抖动影响业务断言 quota_service_with_initial_delays(store, http, base_url, Arc::new(|| Duration::ZERO)) } diff --git a/backend/crates/providers/openai/tests/credential/types.rs b/backend/crates/providers/openai/tests/credential/types.rs index f6635cfd9..e4ab2b537 100644 --- a/backend/crates/providers/openai/tests/credential/types.rs +++ b/backend/crates/providers/openai/tests/credential/types.rs @@ -72,22 +72,25 @@ fn plaintext_provider_schema_round_trips_dynamic_cookie_data() { } #[test] -fn provider_schema_rejects_unknown_public_layer_fields() { - let value = serde_json::json!({ +fn provider_schema_requires_installation_id_and_rejects_unknown_fields() { + let valid = serde_json::json!({ "schema_version": 1, "principal": {"oauth_subject": "subject-private", "poid": null}, "installation_id": "00000000-0000-4000-8000-000000000001", "access_token": "at", - "cookies": [], - "unknown_field": 9 + "cookies": [] }); - assert!(serde_json::from_value::(value).is_err()); - assert!( - serde_json::from_value::(serde_json::json!({ - "schema_version": 1, - "access_token": "at", - "cookies": [] - })) - .is_err() - ); + serde_json::from_value::(valid.clone()) + .expect("valid credential baseline"); + + let mut unknown = valid.clone(); + unknown["unknown_field"] = serde_json::json!(9); + assert!(serde_json::from_value::(unknown).is_err()); + + let mut missing_installation = valid; + missing_installation + .as_object_mut() + .unwrap() + .remove("installation_id"); + assert!(serde_json::from_value::(missing_installation).is_err()); } diff --git a/backend/crates/providers/openai/tests/provider/contract/mod.rs b/backend/crates/providers/openai/tests/provider/contract/mod.rs index 61d26605c..c39f7b35c 100644 --- a/backend/crates/providers/openai/tests/provider/contract/mod.rs +++ b/backend/crates/providers/openai/tests/provider/contract/mod.rs @@ -4257,6 +4257,7 @@ async fn websocket_midstream_error_frame_surfaces_upstream_message_after_deliver create_account(&store, ACCOUNT_ID).await; let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let base_url = format!("http://{}", listener.local_addr().unwrap()); + let (release_error, error_released) = oneshot::channel(); let server = tokio::spawn(async move { let (stream, _) = listener.accept().await.expect("accept WebSocket"); let mut websocket = accept_codex_test_websocket(stream).await; @@ -4288,7 +4289,9 @@ async fn websocket_midstream_error_frame_surfaces_upstream_message_after_deliver )) .await .expect("send response.in_progress"); - tokio::time::sleep(Duration::from_millis(1_600)).await; + error_released + .await + .expect("prelude delivered to the client"); websocket .send(Message::Text( json!({ @@ -4317,6 +4320,16 @@ async fn websocket_midstream_error_frame_surfaces_upstream_message_after_deliver ) .await .expect("prepare midstream-failure stream"); + for kind in ["response.created", "response.in_progress"] { + let event = timeout( + Duration::from_secs(2), + precommit::next_client_event(&mut stream), + ) + .await + .expect("upstream prelude must reach the client"); + assert_eq!(event.wire_event().unwrap().event_type(), Some(kind)); + } + release_error.send(()).unwrap(); let mut failure = loop { match stream.next().await { Some(Ok(_)) => {} @@ -5892,7 +5905,6 @@ async fn websocket_opening_account_rejection_keeps_replay_safe_without_transport "before-payload rejection is replay safe" ); // 回放安全的账号级拒绝必须把换号决策留给 Core,不得钉死同账号传输重试 - // (旧行为会携带小时级 retry-after 的同账号重试标记,请求必然超时) assert_eq!(error.pre_delivery_retry(), None); } diff --git a/backend/crates/providers/openai/tests/provider/contract/precommit.rs b/backend/crates/providers/openai/tests/provider/contract/precommit.rs index 5e9a5acf6..629092173 100644 --- a/backend/crates/providers/openai/tests/provider/contract/precommit.rs +++ b/backend/crates/providers/openai/tests/provider/contract/precommit.rs @@ -1,4 +1,4 @@ -//! 验证事件即时交付、长时间进度与取消对重试边界的影响 +//! 验证事件即时交付、前导事件与取消对重试边界的影响 use gateway_core::diagnostics::TraceContext; use gateway_core::engine::provider::ProviderStream; @@ -145,62 +145,6 @@ async fn large_preludes_are_forwarded_before_failure_on_http_and_websocket() { } } -#[tokio::test] -async fn late_overload_after_multiple_structural_events_remains_replayable() { - let store = Arc::new(MemoryAccountStore::default()); - create_account(&store, "acct_provider_contract").await; - let created = sse(&structural_event("response.created", 96 * 1024)); - let progress = sse(&structural_event("response.in_progress", 96 * 1024)); - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let base_url = format!("http://{}", listener.local_addr().unwrap()); - let (release, released) = oneshot::channel(); - let server = tokio::spawn(async move { - let (mut socket, _) = listener.accept().await.unwrap(); - read_http_request(&mut socket).await; - socket.write_all(b"HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\ntransfer-encoding: chunked\r\n\r\n").await.unwrap(); - write_http_chunk(&mut socket, &created).await; - tokio::time::sleep(Duration::from_millis(1_500)).await; - write_http_chunk(&mut socket, &progress).await; - released.await.unwrap(); - write_http_chunk(&mut socket, &sse(&overload())).await; - socket.write_all(b"0\r\n\r\n").await.unwrap(); - }); - let trace = TraceContext::new("req_precommit"); - let mut stream = provider_with_base_url(&store, base_url) - .execute( - planned_request("openai", http_generate_operation()), - traced_context(&trace, CancellationToken::new()), - ) - .await - .unwrap(); - for kind in ["response.created", "response.in_progress"] { - let event = timeout(Duration::from_secs(2), next_client_event(&mut stream)) - .await - .unwrap(); - assert_eq!(event.wire_event().unwrap().event_type(), Some(kind)); - assert!(event.is_replayable_prelude()); - } - release.send(()).unwrap(); - let (events, mut error) = timeout(Duration::from_secs(1), stream_failure(&mut stream)) - .await - .unwrap(); - assert!(events.iter().all(|event| !event.has_client_event())); - assert!(error.replay_is_safe()); - assert!(matches!( - error.pre_delivery_retry(), - Some(PreDeliveryRetry::SameAccountTransientRetry { .. }) - )); - assert_eq!( - error - .take_atomic_client_events() - .iter() - .filter_map(|event| event.wire_event()?.event_type()) - .collect::>(), - ["error"] - ); - server.await.unwrap(); -} - #[tokio::test] async fn immediate_delivery_preserves_wire_through_output_terminal_and_eof() { let semantic = json!({"type": "response.output_text.delta", "output_index": 0, "content_index": 0, "delta": "hello"}); diff --git a/backend/crates/providers/openai/tests/transport/catalog.rs b/backend/crates/providers/openai/tests/transport/catalog.rs index e15d630e9..1c21f4500 100644 --- a/backend/crates/providers/openai/tests/transport/catalog.rs +++ b/backend/crates/providers/openai/tests/transport/catalog.rs @@ -229,13 +229,17 @@ fn missing_capability_fields_should_remain_unknown() { } #[test] -fn legacy_data_shape_should_fail_the_entire_snapshot() { - let result = parse_codex_model_catalog( - br#"{"data":[{"slug":"gpt-5.4","display_name":"GPT-5.4"}]}"#, - None, - ); - - assert!(matches!(result, Err(CodexModelCatalogError::InvalidWire))); +fn model_array_is_required_and_must_have_array_shape() { + for body in [ + br#"{}"#.as_slice(), + br#"{"models":null}"#, + br#"{"models":{}}"#, + ] { + assert!(matches!( + parse_codex_model_catalog(body, None), + Err(CodexModelCatalogError::InvalidWire) + )); + } } #[test] diff --git a/backend/crates/providers/openai/tests/transport/subscription.rs b/backend/crates/providers/openai/tests/transport/subscription.rs index b9afd1af5..87248d3cf 100644 --- a/backend/crates/providers/openai/tests/transport/subscription.rs +++ b/backend/crates/providers/openai/tests/transport/subscription.rs @@ -1,8 +1,12 @@ //! 验证订阅查询的账号绑定、安全投影与未知状态保留 -use super::{CodexBackendClient, CodexRequestContext, test_wire_profile}; +use super::{CodexBackendClient, CodexRequestContext, read_http_request, test_wire_profile}; use serde_json::json; use std::time::Duration; +use tokio::{ + io::AsyncWriteExt as _, + net::{TcpListener, TcpStream}, +}; use wiremock::{ Mock, MockServer, ResponseTemplate, matchers::{header, method, path, query_param}, @@ -71,7 +75,6 @@ async fn subscription_failures_are_unknown_without_retry() { .set_body_json(json!({"active_until":"invalid","will_renew":false})), ResponseTemplate::new(200).set_body_json(json!({"will_renew":true})), ResponseTemplate::new(200).set_body_string("x".repeat(65537)), - ResponseTemplate::new(200).set_delay(std::time::Duration::from_secs(6)), ] { let server = MockServer::start().await; let official_server = MockServer::start().await; @@ -98,6 +101,54 @@ async fn subscription_failures_are_unknown_without_retry() { } } +#[tokio::test] +async fn subscription_timeout_is_unknown_without_retry() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let official_server = MockServer::start().await; + let test_client = client(&format!("http://{}", listener.local_addr().unwrap())) + .with_official_base_url(official_server.uri()); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let mut request = tokio::spawn(async move { + started_tx.send(tokio::time::Instant::now()).unwrap(); + test_client + .fetch_subscription( + CodexRequestContext::auxiliary("Bearer fixture", Some("account"), "req", None), + "account", + ) + .await + }); + let _stream = accept_subscription_request(&listener).await; + let started = started_rx.await.unwrap(); + + // 连接已建立且上游已收到请求后才暂停时钟,避免虚拟时间抢跑网络 I/O + tokio::time::pause(); + assert!( + tokio::time::timeout_at(started + Duration::from_secs(4), &mut request) + .await + .is_err(), + "the five-second budget is still available" + ); + let result = tokio::time::timeout_at(started + Duration::from_millis(5_100), request) + .await + .expect("subscription must stop at its five-second deadline") + .unwrap(); + assert!(result.is_none()); + tokio::time::resume(); + assert!( + tokio::time::timeout(Duration::from_millis(1), listener.accept()) + .await + .is_err(), + "timeout must not retry the custom endpoint" + ); + assert!( + official_server + .received_requests() + .await + .unwrap() + .is_empty() + ); +} + #[tokio::test] async fn subscription_unknown_renewal_is_not_false_and_empty_account_does_not_request() { let server = MockServer::start().await; @@ -218,32 +269,51 @@ async fn subscription_official_not_found_should_not_repeat_the_request() { #[tokio::test] async fn subscription_fallback_should_share_the_five_second_budget() { - let custom_server = MockServer::start().await; - let official_server = MockServer::start().await; - Mock::given(method("GET")) - .respond_with(ResponseTemplate::new(404).set_delay(Duration::from_secs(3))) - .expect(1) - .mount(&custom_server) - .await; - Mock::given(method("GET")) - .respond_with( - ResponseTemplate::new(200) - .set_body_json(json!({"active_until": "2026-10-01T00:00:00Z"})) - .set_delay(Duration::from_secs(3)), - ) - .expect(1) - .mount(&official_server) - .await; + let custom_server = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let official_server = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let test_client = client(&format!("http://{}", custom_server.local_addr().unwrap())) + .with_official_base_url(format!("http://{}", official_server.local_addr().unwrap())); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let mut request = tokio::spawn(async move { + started_tx.send(tokio::time::Instant::now()).unwrap(); + test_client + .fetch_subscription( + CodexRequestContext::auxiliary("Bearer fixture", Some("account"), "req", None), + "account", + ) + .await + }); + let mut custom_stream = accept_subscription_request(&custom_server).await; + let started = started_rx.await.unwrap(); + tokio::time::pause(); + assert!( + tokio::time::timeout_at(started + Duration::from_secs(3), &mut request) + .await + .is_err() + ); - let test_client = client(&custom_server.uri()).with_official_base_url(official_server.uri()); - let result = tokio::time::timeout( - Duration::from_secs(6), - test_client.fetch_subscription( - CodexRequestContext::auxiliary("Bearer fixture", Some("account"), "req", None), - "account", - ), - ) - .await - .expect("the original request and fallback must share a five-second deadline"); + // 404 之后恢复真实时钟完成第二次连接,回退请求收到后仍等待同一总期限 + tokio::time::resume(); + custom_stream + .write_all(b"HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + .await + .unwrap(); + let _official_stream = accept_subscription_request(&official_server).await; + tokio::time::pause(); + let result = tokio::time::timeout_at(started + Duration::from_millis(5_100), request) + .await + .expect("the original request and fallback must share a five-second deadline") + .unwrap(); assert!(result.is_none()); } + +async fn accept_subscription_request(listener: &TcpListener) -> TcpStream { + tokio::time::timeout(Duration::from_secs(2), async { + let (mut stream, _) = listener.accept().await.unwrap(); + let request = read_http_request(&mut stream).await; + assert!(request.starts_with("GET /backend-api/subscriptions?account_id=account ")); + stream + }) + .await + .expect("subscription request must reach its endpoint") +} diff --git a/backend/crates/providers/xai/tests/admin.rs b/backend/crates/providers/xai/tests/admin.rs index 4d844f52c..a08ea4a26 100644 --- a/backend/crates/providers/xai/tests/admin.rs +++ b/backend/crates/providers/xai/tests/admin.rs @@ -1184,21 +1184,18 @@ mod errors { async fn manual_refresh_keeps_expired_and_changed_credentials_distinct_without_network() { for expired in [true, false] { let store = MemoryProviderAccountStore::shared(); - let mut input = create_input("refresh_error", "synthetic-subject"); + let input = create_input("refresh_error", "synthetic-subject"); + let mut prepared = crate::support::prepare_input(&input).unwrap(); if expired { - let now = Utc::now(); - input.account.access_token_expires_at = now + chrono::Duration::seconds(1); - input.account.refresh_token_expires_at = Some(now + chrono::Duration::seconds(2)); - } - seed_input(&store, &input).await.unwrap(); - if expired { - // 新导入不接受已过期凭据;等待合法存入的短期凭据自然过期,验证历史账号刷新 - let remaining = input.account.refresh_token_expires_at.unwrap() - Utc::now(); - tokio::time::sleep( - remaining.to_std().unwrap_or_default() + Duration::from_millis(10), - ) - .await; + // 此处验证已存账号的刷新失败,直接设置持久化期限以免等待真实时间 + let mut credential = prepared.credential.expose_to_provider().clone(); + credential.insert( + "refresh_token_expires_at".to_owned(), + json!(Utc::now() - chrono::Duration::minutes(1)), + ); + prepared.credential = gateway_core::account::PlaintextCredential::new(credential); } + store.create_account(prepared).await.unwrap(); let before = store.account(&input.account_id).unwrap(); let mut account = account_record(&before); if !expired { diff --git a/backend/crates/providers/xai/tests/transport/canonical.rs b/backend/crates/providers/xai/tests/transport/canonical.rs index 2d17db7ab..d710d20f8 100644 --- a/backend/crates/providers/xai/tests/transport/canonical.rs +++ b/backend/crates/providers/xai/tests/transport/canonical.rs @@ -653,7 +653,7 @@ fn decoder_should_fail_closed_for_non_integer_provider_cost() { #[test] fn decoder_should_tolerate_content_field_error_without_breaking_stream() { // 内容事件的字段校验失败(此处空 delta)不应打断已开始的客户端流 - // 旧行为在此断整条流;修复后跳过该事件的 canonical 提取、wire 原样转发,并继续解出终态 + // 跳过该事件的 canonical 提取、wire 原样转发,并继续解出终态 let body = concat!( "event: response.created\n", "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_a3\",\"model\":\"grok-code-test\"}}\n\n", diff --git a/backend/crates/providers/xai/tests/transport/catalog.rs b/backend/crates/providers/xai/tests/transport/catalog.rs index bdebe5bb4..1e80fcb42 100644 --- a/backend/crates/providers/xai/tests/transport/catalog.rs +++ b/backend/crates/providers/xai/tests/transport/catalog.rs @@ -656,10 +656,13 @@ fn explicit_api_disable_and_non_responses_backend_should_remain_unsupported() { } #[test] -fn list_discriminator_should_be_required_and_exact() { +fn catalog_envelope_requires_list_discriminator_and_model_array() { for body in [ br#"{"data":[{"id":"grok-4","model":"grok-4"}]}"#.as_slice(), br#"{"object":"collection","data":[{"id":"grok-4","model":"grok-4"}]}"#.as_slice(), + br#"{"object":"list"}"#.as_slice(), + br#"{"object":"list","data":null}"#.as_slice(), + br#"{"object":"list","data":{}}"#.as_slice(), ] { assert!(matches!( parse_grok_model_catalog(body, None), @@ -668,13 +671,6 @@ fn list_discriminator_should_be_required_and_exact() { } } -#[test] -fn legacy_models_shape_should_fail_the_entire_snapshot() { - let result = parse_grok_model_catalog(br#"{"object":"list","models":[{"id":"grok-4"}]}"#, None); - - assert!(matches!(result, Err(GrokModelCatalogError::InvalidWire))); -} - #[test] fn empty_data_should_fail_the_entire_snapshot() { let result = parse_grok_model_catalog(br#"{"object":"list","data":[]}"#, None); diff --git a/backend/crates/providers/xai/tests/transport/config.rs b/backend/crates/providers/xai/tests/transport/config.rs deleted file mode 100644 index 2d85c02ed..000000000 --- a/backend/crates/providers/xai/tests/transport/config.rs +++ /dev/null @@ -1,13 +0,0 @@ -//! 验证 Grok 推理端点限定为配置的官方 CLI 来源 - -use provider_xai::GROK_CLI_BASE_URL; -use provider_xai::transport::GROK_RESPONSES_URL; - -#[test] -fn official_grok_build_endpoints_are_fixed_to_the_cli_origin() { - assert_eq!(GROK_CLI_BASE_URL, "https://cli-chat-proxy.grok.com/v1"); - assert_eq!( - GROK_RESPONSES_URL, - "https://cli-chat-proxy.grok.com/v1/responses" - ); -} diff --git a/backend/crates/providers/xai/tests/transport/mod.rs b/backend/crates/providers/xai/tests/transport/mod.rs index 9ed9c4516..2523f0ad6 100644 --- a/backend/crates/providers/xai/tests/transport/mod.rs +++ b/backend/crates/providers/xai/tests/transport/mod.rs @@ -4,7 +4,6 @@ mod canonical; mod catalog; mod client_profile; mod compaction; -mod config; mod egress; mod headers; mod network; diff --git a/docs/architecture.md b/docs/architecture.md index 2522179d2..c70b4b3a5 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1156,9 +1156,16 @@ PostgreSQL 备份恢复属于人工维护操作。当前没有部署级维护模 ```bash cargo +1.99.1 fmt --all --manifest-path backend/Cargo.toml -- --check RUST_MIN_STACK=16777216 cargo +1.99.1 clippy --manifest-path backend/Cargo.toml --all-targets --all-features --locked -- -D warnings -RUST_MIN_STACK=16777216 cargo +1.99.1 test --manifest-path backend/Cargo.toml --test main --locked +RUST_MIN_STACK=16777216 cargo +1.99.1 nextest run --manifest-path backend/Cargo.toml --test main --locked --no-tests fail ``` +测试使用 [nextest](https://nexte.st/docs/installation/pre-built-binaries/),版本与[质量工作流](../.github/workflows/_quality.yml)一致 +各 crate 的 `main` 测试入口统一构建,用例按可用 CPU 数跨 crate 并发执行 +局部验证可追加 `-E 'package(=gateway-store) & test(/^redis::/)'` 等过滤条件;资源受限时用 +`--test-threads 4` 限制同时运行的用例数,编译并发另由 `--build-jobs` 控制 +未安装 nextest 时可使用 `cargo +1.99.1 test --manifest-path backend/Cargo.toml --test main --locked`, +同样设置 `RUST_MIN_STACK=16777216`;该入口只在单个测试二进制内并发,跨 crate 依次运行 + 线程栈设置与当前 CI 一致。Store 与插件 Runtime 的 PostgreSQL/Redis 集成测试需按 [本地测试库](../backend/migrations/README.md#本地测试库) 配置专用服务;未设置环境变量时,本地相关测试会跳过。 其他检查与界面验证按 [贡献与审查](../CONTRIBUTING.md#验证) 执行