From b4a5210e4486971a613e5dd347ba90abbdc6f8dd Mon Sep 17 00:00:00 2001 From: Zyy Date: Thu, 22 Oct 2026 23:12:35 +0800 Subject: [PATCH] test: streamline behavior coverage and parallel execution Remove redundant contract cases and express fixtures in terms of current behavior. Reduce isolated database migration contention, run independent scenarios concurrently, and replace fixed waits with clock and I/O synchronization. Check absolute Redis session expiry and wait for PostgreSQL migration session locks to be released. Document nextest for local test execution. Validated 4060 tests, 70 Redis ACL tests, and five repeat runs of 26 focused tests. The 10 plugin state tests also pass 50 repeat runs with four workers. Rustfmt and strict Clippy pass. Local full-suite execution with 16 workers falls from 162.760s to 89.072s, excluding compilation. Co-authored-by: Codex --- .../gateway-api/tests/admin/accounts/mod.rs | 8 +- .../gateway-api/tests/admin/settings.rs | 40 +++--- .../crates/gateway-api/tests/admin/wire.rs | 68 ++------- .../tests/openai/responses/http.rs | 2 +- .../websocket/connection/middleware.rs | 13 +- .../gateway-core/tests/engine/coordinator.rs | 134 +++++++++--------- .../gateway-host/tests/client_distribution.rs | 61 ++++++-- backend/crates/gateway-host/tests/config.rs | 18 --- .../gateway-host/tests/logging/writer.rs | 5 +- .../runtime/tests/package/compatibility.rs | 17 --- .../runtime/tests/support/environment.rs | 11 ++ .../sdk/tests/client/session/credit.rs | 4 +- .../sdk/tests/client/session/driver.rs | 4 +- .../tests/postgres/client_keys.rs | 16 +-- .../gateway-store/tests/postgres/mod.rs | 11 +- .../tests/postgres/plugins/state.rs | 61 ++++---- .../crates/gateway-store/tests/redis/auth.rs | 59 ++++---- .../tests/redis/credential_leases/mod.rs | 9 +- .../crates/providers/openai/tests/config.rs | 37 +---- .../openai/tests/credential/admin.rs | 29 ++-- .../tests/credential/quota/initial_sync.rs | 2 +- .../openai/tests/credential/quota/mod.rs | 4 +- .../openai/tests/credential/types.rs | 29 ++-- .../openai/tests/provider/contract/mod.rs | 16 ++- .../tests/provider/contract/precommit.rs | 58 +------- .../openai/tests/transport/catalog.rs | 18 ++- .../openai/tests/transport/subscription.rs | 126 ++++++++++++---- backend/crates/providers/xai/tests/admin.rs | 23 ++- .../xai/tests/transport/canonical.rs | 2 +- .../providers/xai/tests/transport/catalog.rs | 12 +- .../providers/xai/tests/transport/config.rs | 13 -- .../providers/xai/tests/transport/mod.rs | 1 - docs/architecture.md | 9 +- 33 files changed, 454 insertions(+), 466 deletions(-) delete mode 100644 backend/crates/providers/xai/tests/transport/config.rs 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#验证) 执行