test: align backend coverage with current behavior

Consolidate overlapping scenarios and remove obsolete assertions and
migration-specific SQL tests. Keep persistence fixtures on the current schema
and document the test maintenance rules in the development skill.

Validation: 3,997 tests passed across 12 crates, plus rustfmt, strict Clippy,
and skill validation. All 33 migration SQL files passed temporary isolated
execution, and the opt-in xAI official network test passed.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-10-19 11:36:06 +08:00
co-authored by Codex
parent 406256bcc0
commit f6538d433d
41 changed files with 414 additions and 1993 deletions
+11
View File
@@ -15,6 +15,7 @@ description: Codex Proxy RS 前后端开发、排障、代码审查和文档修
| 修改服务端接口、路由、调度、账号与用量规则、持久化、后台任务或协议行为;目标是 `backend/` 中的实现、配置、迁移,或 Rust 源码 | [Workspace 边界](../../../docs/architecture.md#3-workspace-边界)、所属模块、[后端自审](../../../docs/architecture.md#后端自审);涉及 Rust 时使用 `$rust-best-practices` 和下方后端注释要求,验证时读[命令](../../../docs/architecture.md#验证命令) |
| 修改页面接口调用,或开发、排查、审查面向管理端或模型客户端的 HTTP/WS API 合同 | [API](../../../docs/api.md)对应章节;优先复用已有接口,核对调用方 → 服务端解析 → 实际查询或执行,不能只凭前端类型认定参数生效;按实际改动叠加前后端规范 |
| 模块或调用链变更 | [Workspace 边界](../../../docs/architecture.md#3-workspace-边界)及所属模块章节,不通读整份架构 |
| 新增、调整、清理或审查测试 | 下方[测试维护](#测试维护),按当前行为、职责归属与已有覆盖判断保留、合并或删除 |
| 修改部署行为、`deploy/` 配置或数据库迁移 | 按范围读取[部署](../../../deploy/README.md)、[迁移](../../../backend/migrations/README.md)对应章节 |
| 开发、排查或审查宿主 SDK、Runtime、扩展接口或独立插件;目标在 `backend/crates/gateway-plugin/` | [插件职责边界](references/plugin-boundaries.md)及所用 [SDK](../../../backend/crates/gateway-plugin/sdk/README.md) 类型与能力合同;宿主守住自身合同、资源与稳定性,不猜测或补偿插件业务错误;仅独立插件创作、排查另用 [cpr-plugin-dev](../cpr-plugin-dev/SKILL.md) |
| 修改或审查文档 | [文档检查](references/documentation.md)及目标章节,不默认读取代码开发、页面验收或发布流程 |
@@ -35,6 +36,16 @@ description: Codex Proxy RS 前后端开发、排障、代码审查和文档修
4. 按变更范围执行[验证](../../../CONTRIBUTING.md#验证)及其停止条件,沿用本轮有效证据与用户已确定的验收分工,区分通过、失败、跳过与未执行;提交时补读[提交约定](../../../CONTRIBUTING.md#项目约定)
5. 交付实际结果与验证缺口;前端按[前端流程](references/frontend.md)核对最终差异和运行证据,后端按上表检查生命周期,不把构建通过当作行为验收
## 测试维护
- 测试随当前功能维护,名称、夹具和断言描述当前输入、行为与结果;功能变更时优先更新所属模块的现有用例,不为每次改动叠加一组记录变更经过的测试
- 清理只证明旧字段、旧路由或旧实现已经消失的用例与附带断言;未知字段校验使用合法请求基线加无业务含义的未知字段,避免因缺少必填项等其他错误而误通过
- 同一层对同一行为的重复覆盖合并为清晰的场景矩阵,复用相邻测试夹具;跨层测试确有不同职责或失败路径时保留,不为统一而扩大测试辅助抽象
- 断言可观察的行为与业务不变量,避免复述结构体赋值、罗列表列或字段清单、匹配 SQL 文本与源码布局;项目明确要求的架构边界检查继续维护
- 当前协议兼容、数据读取兼容、安全边界、并发和资源生命周期仍需覆盖,不按 `old`、`legacy` 等名称机械删除,也不能用改名代替无效用例清理
- 不在仓库中新增或保留迁移 SQL 专项测试,包括迁移文本、schema 演变与指定版本升级用例;需要验证迁移执行、数据保留或回滚时,在临时目录编写一次性验证并使用隔离数据库,测试通过即可,临时脚本不纳入维护测试集
- 持久化功能测试使用当前完整 schema,验证现行读写、事务与数据约束;插件运行时状态迁移等业务能力按当前功能测试,不归为迁移 SQL 专项测试
## 后端注释
- 每个 Rust 文件顶层使用简短的中文 `//!` 注释说明文件职责,覆盖生产源码、测试与构建脚本;测试文件说明测试范围或辅助用途,生成文件同步维护生成器
+3 -23
View File
@@ -155,14 +155,6 @@ fn config_loader_should_ignore_unknown_fields_in_configuration_sections() {
}
}
#[test]
fn config_loader_should_ignore_removed_tls_and_fingerprint_sections() {
let mut document = valid_config_document();
document["openai"]["tls"] = serde_json::json!({});
document["openai"]["fingerprint"] = serde_json::json!({"browser": "removed"});
parse_config(&document.to_string()).expect("removed provider settings are ignored");
}
#[test]
fn config_loader_should_reject_invalid_known_fields_alongside_unknown_fields() {
for port in [serde_json::json!(0), serde_json::json!("not-a-port")] {
@@ -192,7 +184,7 @@ fn config_loader_should_report_startup_configuration_diagnostics() {
] {
let mut document = valid_config_document();
if case == "unused" {
document["openai"]["wire_profile"]["location"] = serde_json::json!(UNUSED_SECRET);
document["openai"]["unused_setting"] = serde_json::json!(UNUSED_SECRET);
} else if case == "missing" {
document["host"]["listen"]
.as_object_mut()
@@ -232,7 +224,7 @@ fn config_loader_should_report_startup_configuration_diagnostics() {
);
match case {
"unused" => assert!(
stderr.contains("openai.wire_profile") && stderr.contains("已忽略"),
stderr.contains("openai.unused_setting") && stderr.contains("已忽略"),
"{stderr}"
),
"missing" => assert!(stderr.contains("host.listen.port"), "{stderr}"),
@@ -256,19 +248,7 @@ fn config_loader_should_allow_missing_provider_configuration() {
let mut document = valid_config_document();
document.as_object_mut().unwrap().remove("openai");
document.as_object_mut().unwrap().remove("xai");
parse_config(&document.to_string()).expect("provider defaults require no YAML identity");
}
#[test]
fn config_loader_should_ignore_removed_provider_identity_sections() {
let mut document = valid_config_document();
document["openai"]["wire_profile"] = serde_json::json!({"codex_version":"legacy-invalid"});
document["xai"] = serde_json::json!({"wire_profile": {"client_identifier": "legacy-client"}});
let (config, _directory) =
parse_config(&document.to_string()).expect("removed identities ignored");
let debug = format!("{config:?}");
assert!(!debug.contains("legacy-invalid"));
assert!(!debug.contains("legacy-client"));
parse_config(&document.to_string()).expect("providers use defaults when omitted");
}
#[test]
+9 -27
View File
@@ -5,20 +5,16 @@ use std::str::FromStr as _;
use chrono::{TimeDelta, Utc};
use gateway_core::account::OpaqueProviderData;
use gateway_admin::{
model::{
PageSize, Revision,
accounts::AccountUsage,
auth::LoginCommand,
client_keys::ClientKeyPageSize,
observability::{DecimalAmount, RequestOutcome, TimeRange},
provider_credentials::{
CredentialCommitGuard, ProviderDocument, ProviderQuota, ProviderQuotaWindow,
QuotaLocalUsageAttribution,
},
settings::AdminApiKey,
use gateway_admin::model::{
PageSize, Revision,
accounts::AccountUsage,
auth::LoginCommand,
client_keys::ClientKeyPageSize,
observability::{DecimalAmount, RequestOutcome, TimeRange},
provider_credentials::{
ProviderDocument, ProviderQuota, ProviderQuotaWindow, QuotaLocalUsageAttribution,
},
ports::store::{AccountStore, AuthStore, ClientKeyStore, ObservabilityStore, SettingsStore},
settings::AdminApiKey,
};
mod backup;
@@ -99,20 +95,6 @@ fn login_command_debug_should_redact_password() {
assert!(!format!("{command:?}").contains("secret-password"));
}
#[test]
fn all_store_capability_traits_should_be_object_safe() {
fn assert_object_safe<T: ?Sized + Send + Sync>() {}
assert_object_safe::<dyn AccountStore>();
assert_object_safe::<dyn AuthStore>();
assert_object_safe::<dyn ClientKeyStore>();
assert_object_safe::<dyn ObservabilityStore>();
assert_object_safe::<dyn SettingsStore>();
fn assert_send_object_safe<T: ?Sized + Send>() {}
assert_send_object_safe::<dyn CredentialCommitGuard>();
}
#[test]
fn provider_document_debug_should_not_expose_opaque_material() {
let document = ProviderDocument::new(OpaqueProviderData::new(Default::default()));
@@ -21,7 +21,7 @@ fn counters_preserve_integer_precision_and_reject_json_numbers() {
}
}
#[tokio::test]
async fn usage_mutations_require_trusted_actor_and_old_workflow_operations_are_absent() {
async fn usage_mutations_require_a_trusted_actor() {
let admin = super::AdminHarness::new().build().await;
let registry = admin.public_services();
let denied = registry
@@ -38,21 +38,6 @@ async fn usage_mutations_require_trusted_actor_and_old_workflow_operations_are_a
.await
.is_err()
);
let ops = registry
.operations()
.filter(|op| op.starts_with("usage.") || *op == "pricing.evaluate")
.collect::<std::collections::BTreeSet<_>>();
assert_eq!(
ops,
std::collections::BTreeSet::from([
"usage.records.get",
"usage.records.list",
"usage.records.write",
"usage.aggregates.list",
"usage.aggregates.write",
"pricing.evaluate"
])
);
}
#[test]
fn sdk_wire_types_match_host_contracts() {
@@ -46,7 +46,6 @@ async fn list_route_should_keep_camel_case_group_and_page_wire() {
value["data"]["items"][0]["usage"]["retainedTotalUsd"],
"5.5"
);
assert!(value["data"]["items"][0].get("member_count").is_none());
}
#[tokio::test]
@@ -149,7 +149,7 @@ fn account_page_item(
}
#[tokio::test]
async fn account_usage_returns_distinct_cache_read_and_write_tokens_without_aliases() {
async fn account_usage_preserves_cache_read_and_write_counts_and_missing_values() {
use chrono::Utc;
use gateway_admin::model::accounts::AccountUsage;
use gateway_core::account::{AccountStatusFacts, CredentialState, QuotaState};
@@ -215,37 +215,9 @@ async fn account_usage_returns_distinct_cache_read_and_write_tokens_without_alia
usage["cacheWriteInputTokensDisplay"],
if write.is_some() { "7" } else { "—" }
);
for alias in [
"createdTokens",
"createdTokensDisplay",
"readTokens",
"readTokensDisplay",
] {
assert!(usage.get(alias).is_none(), "obsolete token alias: {alias}");
}
}
}
#[tokio::test]
async fn standalone_credential_rotation_route_is_not_exposed() {
let fixture = AdminTestFixture::new().await;
fixture.auth.insert_session("valid-session");
let response = admin::router::<AdminTestState>()
.with_state(fixture.state())
.oneshot(
Request::builder()
.method("POST")
.uri("/api/admin/accounts/rotate")
.header(header::COOKIE, "cpr_session=valid-session")
.header(header::CONTENT_TYPE, "application/json")
.body(Body::from("{}"))
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::NOT_FOUND);
}
#[tokio::test]
async fn connection_update_requires_admin_and_validates_before_calling_the_service() {
let fixture = AdminTestFixture::new().await;
@@ -248,9 +248,8 @@ mod profile_statistics {
value["activityInsights"]["invocations"][0]["usageCount"],
29
);
assert!(value.get("cycle").is_none());
assert!(value.get("models").is_none());
assert!(value.get("estimatedCost").is_none());
assert_eq!(value["activityInsights"]["fastModePercent"], 0.0);
assert!(value["activityInsights"]["invocations"][0]["skillId"].is_null());
}
#[test]
@@ -375,7 +374,7 @@ mod batch_update {
assert!(request.concurrency_limit.is_none());
assert!(
serde_json::from_value::<BatchUpdateAccountsRequest>(json!({
"accountIds": ["acct_test"], "enabled": true, "legacy": true
"accountIds": ["acct_test"], "enabled": true, "unknownField": true
}))
.is_err()
);
@@ -583,12 +582,12 @@ mod actions {
}
#[test]
fn account_actions_should_require_frozen_account_ids_and_reject_unknown_revision() {
fn account_actions_should_validate_account_ids_and_reject_unknown_fields() {
let id: AccountIdQuery =
serde_json::from_value(json!({ "accountId": "acct_1" })).expect("decode ID query");
assert!(id.validate().is_ok());
let action: AccountActionRequest =
serde_json::from_value(json!({ "accountId": "legacy-id" })).expect("decode action");
serde_json::from_value(json!({ "accountId": "invalid-id" })).expect("decode action");
assert_eq!(action.validate().unwrap_err().field(), "accountId");
let refresh: AccountRefreshRequest = serde_json::from_value(json!({
"accountId": "acct_1"
@@ -598,7 +597,7 @@ mod actions {
assert!(
serde_json::from_value::<AccountRefreshRequest>(json!({
"accountId": "acct_1",
"expectedConfigRevision": 0
"unknownField": true
}))
.is_err()
);
@@ -772,15 +771,14 @@ mod actions {
assert_eq!(invalid.validate().unwrap_err().field(), "data");
assert!(
serde_json::from_value::<AccountImportRequest>(json!({
"provider": "openai",
"document": {}
"provider": "openai"
}))
.is_err()
);
assert!(
serde_json::from_value::<AccountImportRequest>(json!({
"provider": "openai",
"expectedConfigRevision": 7,
"unknownField": true,
"data": {}
}))
.is_err()
@@ -11,7 +11,7 @@ use gateway_api::admin::accounts::{
use gateway_core::operation::RawJsonPayload;
#[test]
fn quota_forecast_projection_only_exposes_capacity_and_preserves_null_zero() {
fn quota_forecast_projection_formats_capacity_and_preserves_null_zero() {
let now = "2026-09-12T00:00:00Z".parse().unwrap();
let forecast = AccountQuotaForecast {
period: AccountUsagePeriod::Weekly,
@@ -54,16 +54,6 @@ fn quota_forecast_projection_only_exposes_capacity_and_preserves_null_zero() {
assert_eq!(week["estimatedTokensDisplay"], "1M");
assert!(week["estimatedUsd"].is_null());
assert_eq!(week["estimatedUsdDisplay"], "—");
for forecast in value["forecasts"].as_array().unwrap() {
for field in [
"remainingTokens",
"remainingTokensDisplay",
"remainingUsd",
"remainingUsdDisplay",
] {
assert!(forecast.get(field).is_none(), "unexpected field: {field}");
}
}
assert_eq!(
week["source"],
serde_json::json!({
@@ -77,9 +67,6 @@ fn quota_forecast_projection_only_exposes_capacity_and_preserves_null_zero() {
"usdDisplay": "$0.1234"
})
);
assert!(week.get("method").is_none());
assert!(week.get("methodDisplay").is_none());
assert!(value.get("generatedAtDisplay").is_none());
assert_eq!(value["forecasts"][1]["estimatedTokens"], 0);
assert_eq!(value["forecasts"][1]["estimatedTokensDisplay"], "0");
assert_eq!(value["forecasts"][1]["estimatedUsd"], 0.0);
@@ -87,7 +74,6 @@ fn quota_forecast_projection_only_exposes_capacity_and_preserves_null_zero() {
assert_eq!(value["forecasts"][1]["period"], "monthly");
assert_eq!(value["forecasts"][1]["targetDays"], 30.0);
assert_eq!(value["forecasts"][1]["extrapolated"], true);
assert!(value.get("account").is_none());
}
#[test]
@@ -375,18 +375,17 @@ fn client_key_mutations_should_validate_text_limits_and_unknown_fields() {
assert!(
serde_json::from_value::<UpdateClientKeyRequest>(json!({
"id": "key_1",
"expectedConfigRevision": 1,
"name": "key",
"groupIds": [],
"maxConcurrency": 0,
"requestsPerMinute": 0,
"tokensPerMinute": 0
"unknownField": true
}))
.is_err()
);
assert!(
serde_json::from_value::<CreateClientKeyRequest>(json!({
"expectedConfigRevision": 7,
"unknownField": true,
"name": "terminal key",
"groupIds": [],
"maxConcurrency": 2,
@@ -458,7 +457,6 @@ fn client_key_responses_should_keep_shape_and_redact_creation_debug() {
));
let list =
serde_json::to_value(ClientKeyListData::new(vec![view], None, 1)).expect("serialize list");
assert!(list.get("configRevision").is_none());
assert_eq!(list["total"], 1);
assert_eq!(list["items"][0]["id"], "key_visible");
assert_eq!(
@@ -468,12 +466,9 @@ fn client_key_responses_should_keep_shape_and_redact_creation_debug() {
assert_eq!(list["items"][0]["routingScope"], "all");
assert_eq!(list["items"][0]["maxConcurrency"], 2);
assert_eq!(list["items"][0]["requestsPerMinute"], 60);
assert!(list["items"][0].get("tokensPerMinute").is_none());
assert!(list["items"][0].get("policyJson").is_none());
assert_eq!(list["items"][0]["dailyLimitUsd"], "0");
assert_eq!(list["items"][0]["weeklyLimitUsd"], "0");
assert_eq!(list["items"][0]["dailyUsedUsd"], "0");
assert!(list["items"][0].get("unresolvedRequests").is_none());
assert_eq!(
DateTime::parse_from_rfc3339(
list["items"][0]["lastUsedAt"]
@@ -7,7 +7,7 @@ use gateway_api::admin::observability::{
use serde_json::json;
#[test]
fn dashboard_query_should_parse_terminal_trend_kinds() {
fn dashboard_query_should_parse_error_trends() {
let query: DashboardQuery = serde_json::from_value(json!({"kind": "errors"})).unwrap();
assert_eq!(query.trend_kind().unwrap(), TrendKind::Errors);
}
@@ -19,7 +19,7 @@ fn dashboard_query_should_reject_unknown_trend_kind() {
}
#[test]
fn usage_query_should_accept_element_plus_pagination() {
fn usage_query_should_accept_page_number_and_size() {
let query: UsageQuery = serde_json::from_value(json!({
"currentPage": 129,
"pageSize": 10
@@ -29,32 +29,20 @@ fn usage_query_should_accept_element_plus_pagination() {
}
#[test]
fn usage_query_should_reject_removed_total_toggle() {
assert!(serde_json::from_value::<UsageQuery>(json!({"includeTotal": true})).is_err());
fn observability_queries_should_reject_unknown_fields() {
let query = json!({"unknownFilter": true});
assert!(serde_json::from_value::<DashboardQuery>(query.clone()).is_err());
assert!(serde_json::from_value::<UsageQuery>(query.clone()).is_err());
assert!(serde_json::from_value::<OpsQuery>(query.clone()).is_err());
assert!(serde_json::from_value::<DiagnosticsQuery>(query).is_err());
}
#[test]
fn usage_query_should_reject_removed_cursor_contract() {
assert!(serde_json::from_value::<UsageQuery>(json!({"cursor": "opaque"})).is_err());
}
#[test]
fn usage_query_should_reject_nonstandard_page_name() {
assert!(serde_json::from_value::<UsageQuery>(json!({"page": 1})).is_err());
}
#[test]
fn ops_query_should_reject_page_size_above_terminal_limit() {
fn ops_query_should_reject_page_size_above_maximum() {
let query: OpsQuery = serde_json::from_value(json!({"pageSize": 101})).unwrap();
assert_eq!(query.validate_pagination().unwrap_err().field(), "pageSize");
}
#[test]
fn ops_query_should_reject_removed_filters() {
assert!(serde_json::from_value::<OpsQuery>(json!({"route": "/v1/responses"})).is_err());
assert!(serde_json::from_value::<OpsQuery>(json!({"failureClass": "rate_limited"})).is_err());
}
#[test]
fn diagnostics_query_should_keep_wire_dimension_name() {
let query: DiagnosticsQuery =
@@ -1,4 +1,4 @@
//! 响应 DTO 与固定 wire 形状的合同测试
//! 验证观测响应的指标、展示投影与查询时间边界
use chrono::Utc;
use serde_json::json;
@@ -358,13 +358,6 @@ async fn usage_detail_should_keep_attempt_snapshot_contract() {
value["data"]["latencyDetails"]["upstreamEngineMs"],
json!(6400.0)
);
assert!(value["data"].get("firstTokenMs").is_none());
assert!(value["data"].get("firstTokenLatencyMsDisplay").is_none());
assert!(
value["data"]["latencyDetails"]
.get("firstTokenMs")
.is_none()
);
assert_eq!(
serde_json::json!({
"accountId": value["data"]["accountId"],
@@ -470,8 +463,6 @@ async fn zero_attempt_failure_detail_omits_trace_and_keeps_missing_upstream_fact
assert_eq!(data["clientStatusCode"], 503);
assert_eq!(data["message"], "no upstream account is available");
assert_eq!(data["model"], "coding");
assert!(data.get("upstreamStatusCode").is_none());
assert!(data.get("httpVersion").is_none());
for field in [
"provider",
"accountId",
@@ -622,7 +613,6 @@ async fn diagnostics_should_keep_stable_key_and_display_name_contract() {
.await
.expect("diagnostics body");
let value: serde_json::Value = serde_json::from_slice(&body).expect("diagnostics JSON");
assert!(value["data"]["items"][0].get("impactScore").is_none());
assert_eq!(value["data"]["items"][0]["requestShare"], 0.5);
assert_eq!(value["data"]["items"][0]["retryCount"], 3);
assert_eq!(value["data"]["items"][0]["retryRate"], 0.5);
@@ -35,7 +35,7 @@ use std::{collections::BTreeMap, sync::Arc};
mod management;
#[test]
fn artifact_metadata_wire_exposes_v2_identity_and_contributions() {
fn artifact_metadata_wire_preserves_identity_icons_and_contributions() {
use gateway_admin::model::plugins::{
PluginArtifactIcon, PluginArtifactIconVariants, PluginArtifactMetadata, PluginContribution,
};
@@ -70,7 +70,7 @@ fn artifact_metadata_wire_exposes_v2_identity_and_contributions() {
secret_fields: vec![],
state_namespaces: vec![],
};
let value = serde_json::to_value(metadata).expect("v2 artifact metadata JSON");
let value = serde_json::to_value(metadata).expect("artifact metadata JSON");
assert_eq!(value["displayName"], "Example");
assert_eq!(value["publisher"], "test");
assert!(value["author"].is_null());
@@ -80,33 +80,29 @@ fn artifact_metadata_wire_exposes_v2_identity_and_contributions() {
value["contributes"]["middleware"]["id"],
"test.example.middleware"
);
assert!(value.get("capabilities").is_none());
assert!(value.get("permissionDescriptions").is_none());
}
#[test]
fn instance_binding_wire_uses_contribution_without_capability_alias() {
fn instance_binding_requires_a_contribution_and_rejects_unknown_fields() {
use gateway_admin::model::plugins::instances::PluginCapabilityBinding;
let binding = serde_json::from_value::<PluginCapabilityBinding>(serde_json::json!({
let input = serde_json::json!({
"contribution": "test.example.usage",
"stage": "observation",
"order": 0,
"failurePolicy": "observe"
}))
.expect("v2 contribution binding");
});
let binding = serde_json::from_value::<PluginCapabilityBinding>(input.clone())
.expect("contribution binding");
assert_eq!(binding.contribution, "test.example.usage");
assert!(
serde_json::from_value::<PluginCapabilityBinding>(serde_json::json!({
"capability": "usage",
"stage": "observation",
"order": 0,
"failurePolicy": "observe"
}))
.is_err(),
"legacy capability binding must be rejected"
);
let mut missing = input.clone();
missing.as_object_mut().unwrap().remove("contribution");
assert!(serde_json::from_value::<PluginCapabilityBinding>(missing).is_err());
let mut unknown = input;
unknown["unknownField"] = serde_json::json!(true);
assert!(serde_json::from_value::<PluginCapabilityBinding>(unknown).is_err());
}
#[tokio::test]
@@ -381,7 +377,7 @@ async fn plugin_upload_endpoints_require_admin_and_bound_raw_bodies() {
}
#[tokio::test]
async fn artifact_acceptance_returns_the_flat_install_contract() {
async fn artifact_acceptance_returns_installed_artifact_and_configuration_status() {
use axum::{
body::{Body, to_bytes},
http::{Request, StatusCode},
@@ -418,22 +414,11 @@ async fn artifact_acceptance_returns_the_flat_install_contract() {
assert_eq!(result["defaultInstanceId"], serde_json::Value::Null);
assert_eq!(result["configurationRequired"], false);
assert_eq!(result["artifact"]["metadata"]["sha256"], "a".repeat(64));
assert!(
result["artifact"]["metadata"]
.get("permissionDescriptions")
.is_none()
);
assert!(
result["artifact"]["metadata"]
.get("requestedPermissions")
.is_none()
);
assert!(result["artifact"]["acceptedAt"].is_string());
assert!(result.get("mutation").is_none());
}
#[tokio::test]
async fn removed_instance_policy_fields_and_extra_accept_fields_are_rejected() {
async fn plugin_mutations_reject_unknown_fields() {
use axum::{
body::Body,
http::{Request, StatusCode},
@@ -442,24 +427,25 @@ async fn removed_instance_policy_fields_and_extra_accept_fields_are_rejected() {
let fixture = super::AdminTestFixture::new().await;
fixture.auth.insert_session("valid-session");
for (path, body) in [
(
"/api/admin/plugins/instances",
serde_json::json!({
"name":"Example",
"artifactSha256":"a".repeat(64),
"enabled":false,
"trustedProcess":true,
"configuration":{},
"grants":[],
"bindings":[]
}),
),
let instance = serde_json::json!({
"name":"Example",
"artifactSha256":"a".repeat(64),
"enabled":false,
"configuration":{},
"bindings":[]
});
serde_json::from_value::<gateway_admin::model::plugins::instances::ConfigurePluginInstance>(
instance.clone(),
)
.expect("valid instance request");
for (path, mut body) in [
("/api/admin/plugins/instances", instance),
(
"/api/admin/plugins/artifacts/accept",
serde_json::json!({"sha256":"a".repeat(64),"trusted":true}),
serde_json::json!({"sha256":"a".repeat(64)}),
),
] {
body["unknownField"] = serde_json::json!(true);
let response = gateway_api::admin::router::<super::AdminTestState>()
.with_state(fixture.state())
.oneshot(
@@ -181,7 +181,7 @@ fn privacy_policy_is_required_and_rejects_unknown_contract_fields() {
.remove("codexPrivacyPolicy");
assert!(serde_json::from_value::<UpdateRuntimeSettingsRequest>(missing).is_err());
let mut unknown = update_body();
unknown["codexPrivacyPolicy"]["protectedFields"] = json!([]);
unknown["codexPrivacyPolicy"]["unknownField"] = json!(true);
assert!(serde_json::from_value::<UpdateRuntimeSettingsRequest>(unknown).is_err());
}
@@ -471,110 +471,6 @@ fn settings_response_should_cover_the_full_runtime_settings_contract() {
);
}
#[test]
fn settings_request_and_response_fields_should_stay_in_lockstep() {
use std::collections::{BTreeMap, BTreeSet};
use gateway_admin::model::Revision;
use gateway_admin::model::settings::RuntimeSettings;
use gateway_api::admin::settings::RuntimeSettingsView;
use gateway_core::account::RotationStrategy;
use gateway_core::routing::{PublicModelId, UpstreamModelId};
let request: UpdateRuntimeSettingsRequest =
serde_json::from_value(update_body()).expect("decode settings");
request.validate().expect("fixture settings must validate");
let request_fields: BTreeSet<String> = update_body()
.as_object()
.expect("request body object")
.keys()
.cloned()
.collect();
let settings = RuntimeSettings {
request_profiles: Default::default(),
config_revision: Revision::new(7).expect("revision"),
model_mappings: request
.model_mappings
.iter()
.map(|(public, upstream)| {
Ok((
PublicModelId::new(public.clone())?,
UpstreamModelId::new(upstream.clone())?,
))
})
.collect::<Result<BTreeMap<_, _>, gateway_core::error::IdentifierError>>()
.expect("valid model mappings"),
rotation_strategy: RotationStrategy::parse(&request.rotation_strategy)
.expect("fixture rotation strategy"),
updated_at: chrono::Utc::now(),
values: gateway_admin::model::settings::RuntimeSettingsValues {
codex_privacy_policy: Default::default(),
request_location_enabled: false,
request_location: Default::default(),
refresh_margin_seconds: request.values.refresh_margin_seconds,
refresh_concurrency: u32::try_from(request.values.refresh_concurrency).expect("u32"),
max_concurrent_per_account: u32::try_from(request.values.max_concurrent_per_account)
.expect("u32"),
max_waiting_per_key: 0,
max_waiting_per_account: 0,
concurrency_wait_timeout_seconds: 30,
openai_guardian_reserved_concurrency: 0,
openai_account_affinity: gateway_core::account::AccountAffinity::Relaxed,
max_account_rotations: 3,
openai_session_affinity_ttl_hours: 24,
responses_max_decompressed_body_bytes: 64 * 1024 * 1024,
smart_scheduling: request.values.smart_scheduling,
min_codex_desktop_version: request.values.min_codex_desktop_version,
min_codex_cli_version: request.values.min_codex_cli_version,
usage_retention_days: u32::try_from(request.values.usage_retention_days).expect("u32"),
ops_event_retention_days: u32::try_from(request.values.ops_event_retention_days)
.expect("u32"),
audit_retention_days: u32::try_from(request.values.audit_retention_days).expect("u32"),
account_auto_freeze_enabled: true,
account_auto_freeze_threshold: 12,
account_auto_freeze_window_seconds: 600,
account_auto_freeze_duration_seconds: 7_200,
account_auto_freeze_probe_enabled: true,
account_auto_freeze_probe_model: None,
account_auto_freeze_adaptive_concurrency: true,
account_warmup_enabled: false,
account_warmup_schedule_time: "08:00".to_owned(),
account_warmup_model: None,
reset_credit_lead_minutes: 10,
reset_credit_max_weekly_remaining_percent: None,
},
};
let response_fields: BTreeSet<String> = serde_json::to_value(RuntimeSettingsView::from((
settings,
gateway_api::TimePresenter::new(Default::default()),
)))
.expect("serialize view")
.as_object()
.expect("view object")
.keys()
.cloned()
.collect();
let mut expected_fields = request_fields;
expected_fields.insert("providerRequestProfiles".to_owned());
expected_fields.insert("openaiClientProfile".to_owned());
expected_fields.insert("xaiClientProfile".to_owned());
expected_fields.insert("updatedAt".to_owned());
expected_fields.insert("updatedAtDisplay".to_owned());
expected_fields.insert("smartSchedulingDefaults".to_owned());
assert_eq!(response_fields, expected_fields);
}
#[test]
fn settings_request_accepts_json_bytes_with_flattened_fields() {
serde_json::from_slice::<UpdateRuntimeSettingsRequest>(
&serde_json::to_vec(&update_body()).unwrap(),
)
.expect("settings JSON bytes must decode like the HTTP request body");
}
#[test]
fn settings_service_contract_stays_flat_and_rejects_unknown_fields() {
use gateway_admin::model::settings::{ReplaceRuntimeSettings, RuntimeSettings};
@@ -610,17 +506,9 @@ fn settings_service_contract_stays_flat_and_rejects_unknown_fields() {
}
#[test]
fn settings_request_should_reject_unknown_revision_field() {
fn settings_request_should_reject_unknown_fields() {
let mut body = update_body();
body["expectedConfigRevision"] = json!(7);
assert!(serde_json::from_value::<UpdateRuntimeSettingsRequest>(body).is_err());
}
#[test]
fn settings_request_should_reject_removed_bucket_retention() {
let mut body = update_body();
body["bucketRetentionDays"] = json!(365);
body["unknownSetting"] = json!(true);
assert!(serde_json::from_value::<UpdateRuntimeSettingsRequest>(body).is_err());
}
@@ -960,36 +848,23 @@ fn decompression_setting_should_reject_invalid_values() {
}
#[tokio::test]
async fn settings_reject_removed_global_fast_policy() {
async fn settings_update_rejects_unknown_fields_without_writing() {
let fixture = AdminTestFixture::new().await;
fixture.auth.insert_session("valid-session");
let app = app(fixture.state());
let initial = fixture.settings.settings.lock().unwrap().clone();
for value in [json!(true), json!(false), Value::Null] {
let mut body = update_body();
body["disableFast"] = value;
let response = app
.clone()
.oneshot(request(
Method::POST,
"/api/admin/settings/update",
Some(body),
))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY);
assert_eq!(*fixture.settings.settings.lock().unwrap(), initial);
}
let mut body = update_body();
body["unknownSetting"] = json!(true);
let response = app
.oneshot(request(Method::GET, "/api/admin/settings", None))
.oneshot(request(
Method::POST,
"/api/admin/settings/update",
Some(body),
))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert!(
response_json(response).await["data"]
.get("disableFast")
.is_none()
);
assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY);
assert_eq!(*fixture.settings.settings.lock().unwrap(), initial);
}
#[test]
@@ -86,7 +86,7 @@ async fn update_event_stream_should_preserve_event_id_and_download_progress_perc
}
#[tokio::test]
async fn update_should_return_accepted_and_report_restart_only_through_status() {
async fn update_should_return_accepted_and_expose_restart_status() {
let fixture = AdminTestFixture::with_system(Arc::new(ProgressSystem::default())).await;
fixture.auth.insert_session("valid-session");
let router = app(fixture.state());
@@ -110,7 +110,6 @@ async fn update_should_return_accepted_and_report_restart_only_through_status()
.expect("JSON");
assert_eq!(body["data"]["operationId"], "update-1");
assert_eq!(body["data"]["targetVersion"], "0.2.0");
assert!(body["data"].get("needRestart").is_none());
let response = router
.oneshot(
Request::builder()
+22 -22
View File
@@ -297,15 +297,15 @@ async fn unified_login_returns_server_identity_and_rotates_the_previous_session(
}
#[tokio::test]
async fn login_rejects_legacy_type_without_replacing_the_current_session() {
async fn login_rejects_missing_mode_and_unknown_fields_without_replacing_the_session() {
let (app, _) = auth_app().await;
let cookie =
session_cookie(&login(&app, json!({"mode": "key", "apiKey": RAW_KEY}), None).await);
for body in [
json!({"type": "key", "apiKey": RAW_KEY}),
json!({"type": "admin", "password": "strong-admin-password"}),
json!({"mode": "key", "type": "key", "apiKey": RAW_KEY}),
json!({"mode": "admin", "type": "admin", "password": "strong-admin-password"}),
json!({"apiKey": RAW_KEY}),
json!({"password": "strong-admin-password"}),
json!({"mode": "key", "unknownField": true, "apiKey": RAW_KEY}),
json!({"mode": "admin", "unknownField": true, "password": "strong-admin-password"}),
] {
let response = login(&app, body, Some(&cookie)).await;
assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY);
@@ -458,26 +458,26 @@ async fn key_store_outage_is_unavailable_not_expired_or_authenticated() {
}
#[tokio::test]
async fn removed_auth_routes_do_not_fall_back_to_spa_or_accept_old_cookies() {
async fn unknown_auth_routes_return_json_not_found() {
let (app, _) = auth_app().await;
let response = app
.oneshot(empty_request(Method::GET, "/api/auth/unknown"))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::NOT_FOUND);
assert_eq!(response_json(response).await["code"], 40401);
}
#[tokio::test]
async fn session_authentication_requires_the_configured_cookie_name() {
let (app, _) = auth_app().await;
for path in ["/api/admin/auth/login", "/api/auth/unknown"] {
let response = app
.clone()
.oneshot(empty_request(Method::GET, path))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::NOT_FOUND);
assert_eq!(response_json(response).await["code"], 40401);
}
let cookie =
session_cookie(&login(&app, json!({"mode": "key", "apiKey": RAW_KEY}), None).await);
for old_name in ["cpr_admin_session", "cpr_client_session"] {
let old_cookie = cookie.replace("cpr_session", old_name);
assert_eq!(
response_json(get(&app, "/api/auth/status", &old_cookie).await).await["data"]["authenticated"],
false
);
}
let unrelated_cookie = cookie.replace("cpr_session", "unrelated_session");
assert_eq!(
response_json(get(&app, "/api/auth/status", &unrelated_cookie).await).await["data"]["authenticated"],
false
);
}
#[test]
@@ -799,7 +799,7 @@ async fn websocket_failure_before_first_event_should_return_observed_metadata_th
}
#[tokio::test]
async fn websocket_response_create_should_not_have_a_private_16_mib_frame_limit() {
async fn large_websocket_response_create_frames_should_reach_execution() {
let trace = Arc::new(AtomicFailureTrace::default());
let execution = Arc::new(AtomicFailureExecution {
client: authenticated_client("sk_ws_atomic"),
@@ -829,15 +829,14 @@ async fn websocket_response_create_should_not_have_a_private_16_mib_frame_limit(
let payload = json!({
"type": "response.create",
"model": "model-a",
"input": "x".repeat(16 * 1024 * 1024 + 1),
"input": "x".repeat(17 * 1024 * 1024),
})
.to_string();
assert!(payload.len() > 16 * 1024 * 1024);
socket
.send(ClientMessage::Text(payload.into()))
.await
.expect("send response.create above the former frame limit");
.expect("send large response.create frame");
let mut saw_upstream_event = false;
for _ in 0..3 {
@@ -2,14 +2,14 @@
use axum::{
body::Body,
http::{Method, Request, StatusCode, header::AUTHORIZATION},
http::{Method, Request, StatusCode},
};
use tower::ServiceExt;
use super::api_router_with_origins;
use super::models::ModelsExecution;
const REMOVED_BODY_LIMIT_BYTES: usize = 16 * 1024 * 1024;
const LARGE_BODY_BYTES: usize = 17 * 1024 * 1024;
#[tokio::test]
async fn browser_origin_controls_http_admin_sessions_without_configuration() {
@@ -117,64 +117,16 @@ async fn browser_origin_controls_http_admin_sessions_without_configuration() {
}
#[tokio::test]
async fn removed_responses_review_route_should_not_reach_the_responses_handler() {
let response = api_router_with_origins(ModelsExecution::new(), Vec::new())
.await
.oneshot(
Request::post("/v1/responses/review")
.header(AUTHORIZATION, "Bearer sk_models_test")
.body(Body::empty())
.expect("build removed review request"),
)
.await
.expect("route removed review request");
assert_eq!(response.status(), StatusCode::METHOD_NOT_ALLOWED);
}
#[tokio::test]
async fn removed_models_catalog_extension_should_use_standard_model_lookup() {
let response = api_router_with_origins(ModelsExecution::new(), Vec::new())
.await
.oneshot(
Request::get("/v1/models/catalog")
.header(AUTHORIZATION, "Bearer sk_models_test")
.body(Body::empty())
.expect("build removed model catalog request"),
)
.await
.expect("route removed model catalog request");
assert_eq!(response.status(), StatusCode::NOT_FOUND);
}
#[tokio::test]
async fn removed_model_info_route_should_return_not_found() {
let response = api_router_with_origins(ModelsExecution::new(), Vec::new())
.await
.oneshot(
Request::get("/v1/models/model-a/info")
.header(AUTHORIZATION, "Bearer sk_models_test")
.body(Body::empty())
.expect("build removed model info request"),
)
.await
.expect("route removed model info request");
assert_eq!(response.status(), StatusCode::NOT_FOUND);
}
#[tokio::test]
async fn responses_body_should_accept_payload_above_the_removed_private_limit() {
async fn large_responses_bodies_should_reach_authentication() {
let router = api_router_with_origins(ModelsExecution::new(), Vec::new()).await;
let response = router
.oneshot(
Request::post("/v1/responses")
.body(Body::from(vec![b'a'; REMOVED_BODY_LIMIT_BYTES + 1]))
.expect("build request above the removed limit"),
.body(Body::from(vec![b'a'; LARGE_BODY_BYTES]))
.expect("build large request"),
)
.await
.expect("route request above the removed limit");
.expect("route large request");
// 请求进入处理器后应因缺少凭据而被拒绝
// 若提前施加正文大小限制,则会在认证前返回 413
+1 -3
View File
@@ -10,9 +10,7 @@ use gateway_core::upstream::UpstreamTransport;
use serde_json::json;
#[test]
fn provider_event_keeps_large_wire_and_observation_payloads_behind_one_indirection() {
assert!(std::mem::size_of::<ProviderEvent>() <= 64);
fn provider_event_should_distinguish_client_delivery_from_observation() {
let wire = ProviderEvent::wire(
ProtocolWireEvent::json("openai", None, json!({"type": "response.created"}))
.expect("wire event"),
+3 -8
View File
@@ -162,20 +162,15 @@ fn logging_rejects_zero_retention_windows() {
}
#[test]
fn logging_defaults_to_time_retention_and_ignores_removed_count_retention() {
let mut value = serde_json::json!({
fn logging_defaults_to_time_retention() {
let value = serde_json::json!({
"level": "info", "stdout": true,
"file": { "enabled": true, "directory": "logs", "max_file_size_mb": 20 }
});
let config: LoggingConfig = serde_json::from_value(value.clone()).unwrap();
let config: LoggingConfig = serde_json::from_value(value).unwrap();
assert_eq!(config.request_dump_retention_days, 1.0);
assert_eq!(config.file.retention_days, 7);
assert!(!config.oauth_recovery);
value["file"]["max_files"] = serde_json::json!(20);
assert_eq!(
serde_json::from_value::<LoggingConfig>(value).unwrap(),
config
);
}
#[test]
@@ -103,10 +103,24 @@ async fn middleware_preserves_three_state_fast_and_sources() {
Some(90_000)
);
let sources = settings.inspect();
assert!(sources["execution"].get("disable_fast").is_none());
if mode != FastMode::Default && mode == expected {
assert_eq!(sources["execution"]["fast_mode"]["instance_id"], "seed");
let fast_source = &sources["execution"]["fast_mode"];
if mode == FastMode::Default && expected == FastMode::Default {
assert!(fast_source.is_null());
} else {
assert_eq!(
fast_source["instance_id"],
if mode == expected {
"seed"
} else {
"fast-settings"
}
);
assert_eq!(fast_source["value"], expected.as_str());
}
assert_eq!(
sources["execution"]["timeout_ms"]["instance_id"],
"fast-settings"
);
next.run(request).await
})
}),
@@ -278,7 +278,7 @@ fn icon_content_must_fully_decode_match_its_type_and_fit_dimensions() {
}
#[test]
fn obsolete_manifest_fields_are_rejected_without_aliases() {
fn manifest_rejects_unknown_fields_before_package_validation() {
let manifest = serde_json::json!({
"manifestVersion":2,
"name": "example",
@@ -290,7 +290,8 @@ fn obsolete_manifest_fields_are_rejected_without_aliases() {
"engines": {"codex-proxy-rs": ">=1.0.0, <2.0.0"},
"main": "bin/worker",
"runtime": "trustedProcess",
"capabilities": [{"capability":"request_normalizer"}]
"contributes": {},
"unknownField": true
});
let archive = crate::support::archive(BTreeMap::from([(
"plugin.json".to_owned(),
@@ -132,52 +132,40 @@ fn installation_checks_protocol_engine_and_target() {
#[test]
fn capability_contract_versions_are_checked_independently_of_rpc_version() {
let mut manifest = packaged_manifest();
assert!(manifest.validate().is_ok());
for version in [0, 1, 2, 3, 4, 6, u32::MAX] {
manifest
.contributes
.get_mut(&Capability::Middleware)
.unwrap()
.version = version;
assert_eq!(manifest.validate(), Err(ManifestError::Invalid));
for (capability, stage, supported, needs_formats) in [
(Capability::Middleware, Stage::Request, 5, true),
(Capability::Management, Stage::Management, 2, false),
(Capability::UpstreamAdapter, Stage::Upstream, 3, true),
(Capability::Scheduler, Stage::Scheduling, 3, false),
] {
for mut manifest in [source_manifest(), packaged_manifest()] {
let formats = if needs_formats {
vec!["openai".into()]
} else {
vec![]
};
manifest.contributes.clear();
manifest.contributes.insert(
capability,
ContributionDeclaration {
id: "9acme.request-tags.contract".into(),
version: supported,
stages: vec![stage],
input_formats: formats.clone(),
output_formats: formats,
},
);
assert!(manifest.validate().is_ok(), "{capability:?} v{supported}");
for version in (0..supported).chain([supported + 1, u32::MAX]) {
manifest.contributes.get_mut(&capability).unwrap().version = version;
assert_eq!(
manifest.validate(),
Err(ManifestError::Invalid),
"{capability:?} v{version}"
);
}
}
}
manifest.contributes.clear();
manifest.contributes.insert(
Capability::Management,
ContributionDeclaration {
id: "9acme.request-tags.management".into(),
version: 3,
stages: vec![Stage::Management],
input_formats: vec![],
output_formats: vec![],
},
);
assert_eq!(manifest.validate(), Err(ManifestError::Invalid));
}
#[test]
fn retired_upstream_contract_requires_an_author_upgrade() {
let mut manifest = source_manifest();
manifest.contributes.clear();
manifest.contributes.insert(
Capability::UpstreamAdapter,
ContributionDeclaration {
id: "9acme.request-tags.upstream".into(),
version: 3,
stages: vec![Stage::Upstream],
input_formats: vec!["openai".into()],
output_formats: vec!["openai".into()],
},
);
assert!(manifest.validate().is_ok());
manifest
.contributes
.get_mut(&Capability::UpstreamAdapter)
.unwrap()
.version = 1;
assert_eq!(manifest.validate(), Err(ManifestError::Invalid));
}
#[test]
@@ -371,18 +359,19 @@ fn state_namespaces_are_base_infrastructure_with_bounded_versions() {
}
#[test]
fn public_manifest_fields_use_only_the_v3_camel_case_shape() {
let value = serde_json::to_value(source_manifest()).unwrap();
assert!(value.get("manifestVersion").is_some());
assert!(value.get("displayName").is_some());
assert!(value.get("configurationSchema").is_some());
assert!(value.get("schema_version").is_none());
fn manifest_round_trips_public_fields_and_rejects_unknown_fields() {
let manifest = source_manifest();
let mut value = serde_json::to_value(&manifest).unwrap();
assert_eq!(value["manifestVersion"], MANIFEST_VERSION);
assert_eq!(value["displayName"], manifest.display_name);
assert_eq!(value["configurationSchema"], manifest.configuration_schema);
assert_eq!(
serde_json::from_value::<Manifest>(value.clone()).unwrap(),
manifest
);
let mut old = value;
old.as_object_mut()
.unwrap()
.insert("schema_version".into(), json!(3));
assert!(serde_json::from_value::<Manifest>(old).is_err());
value["unknownField"] = json!(true);
assert!(serde_json::from_value::<Manifest>(value).is_err());
}
#[test]
@@ -439,26 +428,3 @@ fn contribution_declaration_uses_camel_case_description_fields() {
})
);
}
#[test]
fn scheduler_requires_the_current_contract() {
let mut manifest = packaged_manifest();
manifest.contributes.clear();
manifest.contributes.insert(
Capability::Scheduler,
ContributionDeclaration {
id: "9acme.request-tags.scheduler".into(),
version: 3,
stages: vec![Stage::Scheduling],
input_formats: vec![],
output_formats: vec![],
},
);
assert!(manifest.validate().is_ok());
manifest
.contributes
.get_mut(&Capability::Scheduler)
.unwrap()
.version = 1;
assert_eq!(manifest.validate(), Err(ManifestError::Invalid));
}
+1 -8
View File
@@ -20,7 +20,7 @@ fn store_config_should_derive_backup_staging_from_runtime_data_dir() {
}
#[test]
fn store_pool_config_ignores_unknown_options_and_keeps_connection_pool_limits() {
fn store_pool_config_defaults_reserve_capacity_for_observability() {
let pool: StorePoolConfig =
serde_json::from_value(serde_json::json!({})).expect("default pool configuration");
@@ -37,13 +37,6 @@ fn store_pool_config_ignores_unknown_options_and_keeps_connection_pool_limits()
.observability_max_connections(),
40
);
assert_eq!(
serde_json::from_value::<StorePoolConfig>(serde_json::json!({
"statement_timeout_seconds": 60,
}))
.unwrap(),
pool,
);
}
#[test]
@@ -27,36 +27,6 @@ use super::TestDatabase;
const MIXED_GROUP: &str = "grp_00000000000000000000000000000001";
const EMPTY_GROUP: &str = "grp_00000000000000000000000000000002";
#[tokio::test]
async fn fast_mode_migration_preserves_existing_group_choices() {
let Some(database) = TestDatabase::create_through("group_fast_mode_upgrade", 21).await else {
return;
};
for (id, disabled) in [(MIXED_GROUP, true), (EMPTY_GROUP, false)] {
sqlx::query("insert into account_groups (id, name, color, disable_fast, created_at, updated_at) values ($1, $1, '#2563EBFF', $2, now(), now())")
.bind(id)
.bind(disabled)
.execute(&database.pool)
.await
.unwrap();
}
super::TEST_MIGRATOR.run(&database.pool).await.unwrap();
let groups = sqlx::query_as::<_, (String, String)>(
"select id, fast_mode from account_groups order by id",
)
.fetch_all(&database.pool)
.await
.unwrap();
assert_eq!(
groups,
vec![
(MIXED_GROUP.to_owned(), "disabled".to_owned()),
(EMPTY_GROUP.to_owned(), "default".to_owned())
]
);
database.close().await;
}
#[tokio::test]
async fn groups_aggregate_cross_provider_members_and_key_bindings_without_multiplication() {
let Some(database) = TestDatabase::create("account_group_aggregate").await else {
@@ -840,20 +810,19 @@ async fn image_generation_group_updates_preserve_omitted_values_and_publish_snap
}
#[tokio::test]
async fn image_generation_migration_allows_existing_and_default_groups() {
let Some(database) = TestDatabase::create_through("group_image_upgrade", 31).await else {
async fn image_generation_defaults_to_allowed_for_new_groups() {
let Some(database) = TestDatabase::create("group_image_default").await else {
return;
};
sqlx::query("insert into account_groups (id, name, color, created_at, updated_at) values ($1, $1, '#2563EBFF', now(), now())")
.bind(MIXED_GROUP).execute(&database.pool).await.unwrap();
super::TEST_MIGRATOR.run(&database.pool).await.unwrap();
sqlx::query("insert into account_groups (id, name, color, created_at, updated_at) values ($1, $1, '#2563EBFF', now(), now())")
.bind(EMPTY_GROUP).execute(&database.pool).await.unwrap();
let allowed =
sqlx::query_scalar::<_, bool>("select allow_image_generation from account_groups")
.fetch_all(&database.pool)
.await
.unwrap();
assert_eq!(allowed, vec![true, true]);
let allowed = sqlx::query_scalar::<_, bool>(
"select allow_image_generation from account_groups where id = $1",
)
.bind(EMPTY_GROUP)
.fetch_one(&database.pool)
.await
.unwrap();
assert!(allowed);
database.close().await;
}
@@ -1,87 +0,0 @@
//! 验证统计升级保留历史事实与回填进度,并重建已有时间桶
use chrono::{DateTime, Utc};
use serde_json::Value;
use super::{TEST_MIGRATOR, TestDatabase};
#[tokio::test]
async fn analytics_upgrade_preserves_facts_and_backfill_progress_and_requeues_buckets() {
let Some(database) = TestDatabase::create_through("analytics_upgrade", 29).await else {
return;
};
sqlx::raw_sql(
"insert into request_analytics_facts(id,started_at,source_version) values
('first','2026-10-01 10:01:00+00',1),
('same_bucket','2026-10-01 10:02:00+00',2),
('next_bucket','2026-10-01 10:16:00+00',3);
insert into request_analytics_pending_buckets(bucket_start,oldest_pending_at)
values ('2026-10-01 10:00:00+00','2026-10-01 09:00:00+00');
update request_analytics_build_state set
backfill_complete=true, backfill_cursor_id='next_bucket',
backfill_cursor_started_at='2026-10-01 10:16:00+00',
coverage_start='2026-10-01 10:00:00+00',
coverage_end='2026-10-01 10:30:00+00', generation=7;",
)
.execute(&database.pool)
.await
.unwrap();
let facts_before: Vec<Value> =
sqlx::query_scalar("select to_jsonb(f) from request_analytics_facts f order by id")
.fetch_all(&database.pool)
.await
.unwrap();
let progress_before: Value =
sqlx::query_scalar("select to_jsonb(s) from request_analytics_build_state s")
.fetch_one(&database.pool)
.await
.unwrap();
let pending_before: (i64, DateTime<Utc>) = sqlx::query_as(
"select change_version,oldest_pending_at from request_analytics_pending_buckets",
)
.fetch_one(&database.pool)
.await
.unwrap();
TEST_MIGRATOR.run(&database.pool).await.unwrap();
let facts_after: Vec<Value> =
sqlx::query_scalar("select to_jsonb(f) from request_analytics_facts f order by id")
.fetch_all(&database.pool)
.await
.unwrap();
let progress_after: Value =
sqlx::query_scalar("select to_jsonb(s) from request_analytics_build_state s")
.fetch_one(&database.pool)
.await
.unwrap();
assert_eq!(facts_after, facts_before);
assert_eq!(progress_after, progress_before);
let pending: Vec<(DateTime<Utc>, i64, DateTime<Utc>)> = sqlx::query_as(
"select bucket_start,change_version,oldest_pending_at
from request_analytics_pending_buckets order by bucket_start",
)
.fetch_all(&database.pool)
.await
.unwrap();
let expected_buckets = ["2026-10-01T10:00:00Z", "2026-10-01T10:15:00Z"]
.map(|value| value.parse::<DateTime<Utc>>().unwrap());
assert_eq!(
pending.iter().map(|row| row.0).collect::<Vec<_>>(),
expected_buckets
);
assert!(pending[0].1 > pending_before.0);
assert_eq!(pending[0].2, pending_before.1);
// 重启不会再次入队,也不会覆盖最早待处理时间
TEST_MIGRATOR.run(&database.pool).await.unwrap();
let pending_after_restart: Vec<(DateTime<Utc>, i64, DateTime<Utc>)> = sqlx::query_as(
"select bucket_start,change_version,oldest_pending_at
from request_analytics_pending_buckets order by bucket_start",
)
.fetch_all(&database.pool)
.await
.unwrap();
assert_eq!(pending_after_restart, pending);
database.close().await;
}
@@ -1,4 +1,4 @@
//! PostgreSQL 连接错误、迁移应用与事务归属测试
//! PostgreSQL 连接初始化、会话设置与事务取消测试
use std::error::Error as _;
@@ -109,32 +109,32 @@ async fn invalid_postgres_options_preserve_the_sqlx_source() {
}
#[tokio::test]
async fn connect_and_migrate_should_apply_all_migrations_once_and_reopen_cleanly() {
async fn connection_initialization_preserves_settings_when_reopened() {
let Some(database_url) = crate::support::test_env("CPR_TEST_DATABASE_URL") else {
return;
};
let database = format!("cpr_store_migrator_{}", Uuid::new_v4().simple());
let database = format!("cpr_store_connection_{}", Uuid::new_v4().simple());
let admin = PgPoolOptions::new()
.max_connections(1)
.connect(&database_url)
.await
.expect("connect migration test PostgreSQL");
.expect("connect test PostgreSQL");
sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
"create database \"{database}\""
)))
.execute(&admin)
.await
.expect("create migration test database");
.expect("create connection test database");
let isolated_url = PgConnectOptions::from_str(&database_url)
.expect("parse migration test PostgreSQL URL")
.expect("parse test PostgreSQL URL")
.database(&database)
.to_url_lossy()
.to_string();
let pool_config = gateway_store::StorePoolConfig::default();
let first = connect_and_migrate(&isolated_url, pool_config)
.await
.expect("apply migrations through production migrator");
.expect("initialize production connection");
let session_settings = sqlx::query_as::<_, (String, i64, i64, i64)>(
"select current_setting('application_name'),
extract(epoch from current_setting('statement_timeout')::interval)::bigint,
@@ -154,70 +154,20 @@ async fn connect_and_migrate_should_apply_all_migrations_once_and_reopen_cleanly
timezone, "UTC",
"runtime sessions use an explicit technical time basis"
);
let first_tables = sqlx::query_scalar::<_, String>(
"select table_name
from information_schema.tables
where table_schema = 'public'
order by table_name",
)
.fetch_all(&first)
.await
.expect("load migrated tables");
sqlx::query("update runtime_settings set refresh_margin_seconds = 1800 where id = 1")
.execute(&first)
.await
.expect("save custom runtime setting");
first.close().await;
let second = connect_and_migrate(&isolated_url, pool_config)
.await
.expect("reopen database through production migrator");
let migration_count =
sqlx::query_scalar::<_, i64>("select count(*) from _sqlx_migrations where success")
.expect("reopen production connection");
let refresh_margin: i64 =
sqlx::query_scalar("select refresh_margin_seconds from runtime_settings where id = 1")
.fetch_one(&second)
.await
.expect("count successful migrations");
let response_id_types = sqlx::query_scalar::<_, String>(
"select data_type
from information_schema.columns
where table_schema = 'public'
and table_name = 'model_requests'
and column_name in ('client_response_id', 'upstream_response_id')
order by column_name",
)
.fetch_all(&second)
.await
.expect("load opaque response ID column types");
let raw_response_id_index_exists = sqlx::query_scalar::<_, bool>(
"select exists (
select 1
from pg_indexes
where schemaname = 'public'
and indexname = 'model_requests_client_response_uq'
)",
)
.fetch_one(&second)
.await
.expect("check removed raw response ID index");
let legacy_key_provider_column_exists = sqlx::query_scalar::<_, bool>(
"select exists (
select 1 from information_schema.columns
where table_schema = 'public'
and table_name = 'client_api_keys'
and column_name = 'provider_kind'
)",
)
.fetch_one(&second)
.await
.expect("check removed client key provider column");
let routing_history_columns = sqlx::query_scalar::<_, String>(
"select column_name from information_schema.columns
where table_schema = 'public'
and table_name = 'model_requests'
and column_name in (
'routing_scope', 'routing_group_refs', 'routing_group_names_snapshot'
)
order by column_name",
)
.fetch_all(&second)
.await
.expect("load routing history columns");
.expect("load saved runtime setting");
second.close().await;
sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
@@ -225,72 +175,9 @@ async fn connect_and_migrate_should_apply_all_migrations_once_and_reopen_cleanly
)))
.execute(&admin)
.await
.expect("drop migration test database");
.expect("drop connection test database");
admin.close().await;
assert_eq!(
first_tables,
[
"_sqlx_migrations",
"account_group_accounts",
"account_groups",
"admin_audit_events",
"admin_users",
"backup_records",
"backup_settings",
"client_api_key_budget_windows",
"client_api_key_charge_events",
"client_api_key_groups",
"client_api_keys",
"model_request_leases",
"model_request_observations",
"model_requests",
"ops_events",
"outbound_proxies",
"plugin_account_group_resources",
"plugin_artifact_credentials",
"plugin_artifact_platforms",
"plugin_artifacts",
"plugin_client_api_key_resources",
"plugin_instance_secrets",
"plugin_instances",
"plugin_source_credentials",
"plugin_state_generations",
"plugin_state_records",
"plugin_update_sources",
"plugin_version_configurations",
"provider_accounts",
"provider_authorization_receipts",
"request_analytics_15m",
"request_analytics_1d_counts",
"request_analytics_1h",
"request_analytics_build_state",
"request_analytics_facts",
"request_analytics_pending_buckets",
"request_analytics_pending_requests",
"runtime_settings",
]
);
assert_eq!(session_settings, ("codex-proxy-rs".to_owned(), 30, 5, 30));
assert_eq!(
migration_count,
i64::try_from(TEST_MIGRATOR.iter().count())
.expect("migration count fits PostgreSQL bigint")
);
assert_eq!(response_id_types, ["bytea", "bytea"]);
assert!(!raw_response_id_index_exists);
assert!(!legacy_key_provider_column_exists);
assert!(routing_history_columns.is_empty());
}
#[test]
fn migrations_should_leave_transaction_ownership_to_sqlx() {
let transaction_statements = TEST_MIGRATOR
.iter()
.flat_map(|migration| migration.sql.as_str().lines())
.map(str::trim)
.filter(|line| matches!(*line, "begin;" | "commit;"))
.count();
assert_eq!(transaction_statements, 0);
assert_eq!(refresh_margin, 1800, "reopening preserves custom settings");
}
@@ -18,7 +18,6 @@ mod admission_recovery;
mod analytics;
mod analytics_buckets;
mod analytics_reset;
mod analytics_upgrade;
mod backup;
mod client_budgets;
mod client_keys;
@@ -69,10 +68,6 @@ pub(super) fn admin_account_store(pool: &PgPool) -> PgAdminAccountStore {
impl TestDatabase {
pub(super) async fn create(label: &str) -> Option<Self> {
Self::create_through(label, i64::MAX).await
}
pub(super) async fn create_through(label: &str, migration_version: i64) -> Option<Self> {
let database_url = crate::support::test_env("CPR_TEST_DATABASE_URL")?;
let schema = format!("cpr_store_{label}_{}", Uuid::new_v4().simple());
// 临时 schema 验证事务可见性与回滚,不模拟 PostgreSQL 掉电恢复
@@ -95,7 +90,7 @@ impl TestDatabase {
.await
.expect("connect isolated test schema");
TEST_MIGRATOR
.run_to(migration_version, &pool)
.run(&pool)
.await
.expect("apply test migrations");
Some(Self {
@@ -1,58 +1,10 @@
//! 观测视图升级兼容性与普通聚合的覆盖索引规划回归测试
//! 验证观测视图普通聚合能够使用覆盖索引
use chrono::Utc;
use serde_json::Value;
use super::{TestDatabase, seed_observability_facts};
#[tokio::test]
async fn projection_upgrade_preserves_observation_values() {
let Some(db) = TestDatabase::create_through("observation_projection_upgrade", 26).await else {
return;
};
// 此测试固定覆盖 26 → 27 的冻结视图合同,不依赖最新版本的写入夹具
sqlx::query("insert into model_requests(id,client_api_key_ref,operation,client_transport,outcome,client_status_code,started_at,deadline_at,completed_at,request_observation_json) values('req_observe_success','historical_key','responses','http_sse','succeeded',200,now()-interval '1 hour',now(),now(),'{\"request\":{\"configRevision\":1,\"protocol\":\"openai\",\"endpoint\":\"/v1/responses\",\"compact\":false},\"routing\":{\"scope\":\"all\",\"groupRefs\":[],\"groupNamesSnapshot\":[]},\"timings\":{\"local\":{\"latencyMs\":800}}}'::jsonb)")
.execute(&db.pool).await.unwrap();
sqlx::query(
"update model_requests
set request_observation_json = jsonb_set(request_observation_json,
'{timings,upstream}', '{\"responseMs\":700,\"engineIapiTbtMs\":2.450638}')
where id = 'req_observe_success'",
)
.execute(&db.pool)
.await
.unwrap();
let before: Vec<Value> =
sqlx::query_scalar("select to_jsonb(mr) from model_request_observations mr order by id")
.fetch_all(&db.pool)
.await
.unwrap();
super::super::TEST_MIGRATOR
.run_to(27, &db.pool)
.await
.unwrap();
let after: Vec<Value> =
sqlx::query_scalar("select to_jsonb(mr) from model_request_observations mr order by id")
.fetch_all(&db.pool)
.await
.unwrap();
assert_eq!(
after, before,
"projection upgrade must preserve every field"
);
let updatable: String = sqlx::query_scalar(
"select is_updatable from information_schema.views
where table_schema = current_schema() and table_name = 'model_request_observations'",
)
.fetch_one(&db.pool)
.await
.unwrap();
assert_eq!(updatable, "NO", "observation projection remains read-only");
db.close().await;
}
#[tokio::test]
async fn plain_observation_aggregates_can_use_a_covering_index() {
let Some(db) = TestDatabase::create("observation_covering_index").await else {
@@ -1,4 +1,4 @@
//! 验证插件制品身份、信任接受、迁移与安装幂等性
//! 验证插件制品身份、信任接受与安装幂等性
use std::collections::BTreeMap;
@@ -83,12 +83,7 @@ async fn unknown_metadata_fields_do_not_change_artifact_identity() {
.fetch_one(&database.pool)
.await
.unwrap();
assert!(stored.get("permissionDescriptions").is_none());
assert!(stored.get("requestedPermissions").is_none());
let mut extended = stored;
extended["permissionDescriptions"] =
serde_json::json!([{"permission": "models", "label": "旧文案"}]);
extended["requestedPermissions"] = serde_json::json!(["models"]);
extended["futureMetadata"] = serde_json::json!({"enabled": true});
extended["contributes"]["middleware"]["futureField"] = serde_json::json!(true);
sqlx::query("update plugin_artifacts set metadata_json=$2 where sha256=$1")
@@ -136,94 +131,6 @@ async fn unknown_metadata_fields_do_not_change_artifact_identity() {
database.close().await;
}
#[tokio::test]
async fn trust_migration_removes_only_obsolete_metadata_fields() {
let Some(database) = TestDatabase::create_through("plugin_trust_upgrade", 19).await else {
return;
};
initialize_revision(&database).await;
let store = PgPluginStore::new(database.pool.clone());
for (digest, extra) in [
('a', serde_json::json!({"requestedPermissions": ["models"]})),
(
'b',
serde_json::json!({"permissionDescriptions": ["旧文案"]}),
),
(
'c',
serde_json::json!({"requestedPermissions": [], "permissionDescriptions": []}),
),
('d', serde_json::json!({})),
] {
// 每个制品使用独立平台,保留同一版本的平台不可变约束
let fixture = artifact(digest, &[&format!("test-{digest}")]);
let installed = store
.install_artifact(fixture.clone(), PluginSource::Upload, &context())
.await
.unwrap();
if digest == 'a' {
store
.accept_artifact(&fixture.metadata.sha256, &context())
.await
.unwrap();
}
sqlx::query("update plugin_artifacts set metadata_json=metadata_json || $2 || '{\"futureMetadata\":true}'::jsonb where sha256=$1")
.bind(&fixture.metadata.sha256).bind(extra).execute(&database.pool).await.unwrap();
sqlx::query("insert into plugin_instances(id,artifact_sha256,name,enabled,configuration_json,bindings_json,revision) values ($1,$2,'Example',false,'{}','[]',$3)")
.bind(uuid::Uuid::now_v7()).bind(&fixture.metadata.sha256)
.bind(i64::try_from(installed.config_revision.get()).unwrap())
.execute(&database.pool).await.unwrap();
}
let mut expected: Vec<serde_json::Value> =
sqlx::query_scalar("select to_jsonb(a) from plugin_artifacts a order by sha256")
.fetch_all(&database.pool)
.await
.unwrap();
for row in &mut expected {
let metadata = row["metadata_json"].as_object_mut().unwrap();
metadata.remove("requestedPermissions");
metadata.remove("permissionDescriptions");
}
let instances: Vec<serde_json::Value> =
sqlx::query_scalar("select to_jsonb(i) from plugin_instances i order by id")
.fetch_all(&database.pool)
.await
.unwrap();
let revision: i64 =
sqlx::query_scalar("select config_revision from runtime_settings where id=1")
.fetch_one(&database.pool)
.await
.unwrap();
for _ in 0..2 {
super::super::TEST_MIGRATOR
.run(&database.pool)
.await
.unwrap();
let actual: Vec<serde_json::Value> =
sqlx::query_scalar("select to_jsonb(a) from plugin_artifacts a order by sha256")
.fetch_all(&database.pool)
.await
.unwrap();
assert_eq!(
actual, expected,
"只清理两个旧字段,保留包体、接受事实与其他元数据"
);
let current_instances: Vec<serde_json::Value> =
sqlx::query_scalar("select to_jsonb(i) from plugin_instances i order by id")
.fetch_all(&database.pool)
.await
.unwrap();
assert_eq!(current_instances, instances);
let snapshot = store.load_instances().await.unwrap();
assert_eq!(snapshot.instances.len(), 4);
assert_eq!(
snapshot.config_revision.get(),
u64::try_from(revision).unwrap()
);
}
database.close().await;
}
#[tokio::test]
async fn artifact_acceptance_is_exact_and_idempotent() {
let Some(database) = TestDatabase::create("plugin_artifact_acceptance").await else {
@@ -262,51 +169,6 @@ async fn artifact_acceptance_is_exact_and_idempotent() {
database.close().await;
}
#[tokio::test]
async fn plugin_migration_preserves_main_settings_and_accepts_new_installations() {
let Some(database) = TestDatabase::create_through("plugin_upgrade_from_main", 17).await else {
return;
};
initialize_revision(&database).await;
// 0022 会把仍为旧默认 3600 的 margin 收敛到 300;预设自定义值,
// 让本测试继续验证「所有已自定义设置都原样保留」。
sqlx::query("update runtime_settings set max_concurrent_per_account=0, refresh_margin_seconds=1800 where id=1")
.execute(&database.pool)
.await
.unwrap();
let before: serde_json::Value =
sqlx::query_scalar("select to_jsonb(s) from runtime_settings s where id=1")
.fetch_one(&database.pool)
.await
.unwrap();
super::super::TEST_MIGRATOR
.run(&database.pool)
.await
.unwrap();
let after: serde_json::Value =
sqlx::query_scalar("select to_jsonb(s) from runtime_settings s where id=1")
.fetch_one(&database.pool)
.await
.unwrap();
for (key, value) in before.as_object().unwrap() {
if key == "request_interval_ms" {
assert!(after.get(key).is_none(), "removed setting {key}");
continue;
}
assert_eq!(after.get(key), Some(value), "existing setting {key}");
}
let store = PgPluginStore::new(database.pool.clone());
store
.install_artifact(
artifact('a', &["linux-x86_64"]),
PluginSource::Upload,
&context(),
)
.await
.unwrap();
database.close().await;
}
#[tokio::test]
async fn source_identity_requires_publisher_and_name() {
let Some(database) = TestDatabase::create("plugin_source_identity").await else {
@@ -1193,63 +1193,6 @@ async fn legacy_urls_join_one_catalog_entry_and_invalid_batch_rolls_back() {
database.close().await;
}
#[tokio::test]
async fn migration_backfills_shared_proxies_without_changing_credentials() {
let Some(database) = TestDatabase::create_through("proxy_backfill", 4).await else {
return;
};
// 从真实旧 schema 正向升级,不拆卸最新结构,避免遗漏后续增加的引用约束
for id in ["acct_one", "acct_two", "acct_direct"] {
let proxy_url = (id != "acct_direct").then_some("http://user:secret@127.0.0.1:8080/");
sqlx::query(
"insert into provider_accounts (
id,provider_kind,name,authentication_kind,provider_credentials_json,
has_refresh_token,credential_observed_at,created_at,updated_at,outbound_proxy_url
) values ($1,'openai',$1,'oauth','{}',false,now(),now(),now(),$2)",
)
.bind(id)
.bind(proxy_url)
.execute(&database.pool)
.await
.unwrap();
}
super::TEST_MIGRATOR.run(&database.pool).await.unwrap();
assert!(
sqlx::query_scalar::<_, bool>("select bool_and(enabled) from outbound_proxies")
.fetch_one(&database.pool)
.await
.unwrap()
);
let accounts = PgProviderAccountRepository::new(database.pool.clone());
let inherited = gateway_core::account::ProviderAccountStore::list_accounts(&accounts)
.await
.unwrap();
assert_eq!(inherited.len(), 3);
assert!(
inherited
.iter()
.all(|account| account.request_location().is_none())
);
assert_eq!(
sqlx::query_scalar::<_, i64>("select count(*) from outbound_proxies")
.fetch_one(&database.pool)
.await
.unwrap(),
1
);
assert_eq!(sqlx::query_scalar::<_, i64>("select count(*) from provider_accounts a join outbound_proxies p on a.outbound_proxy_id = p.id where a.outbound_proxy_url = p.proxy_url and a.credential_revision = 1").fetch_one(&database.pool).await.unwrap(), 2);
assert!(
sqlx::query_scalar::<_, Option<String>>(
"select outbound_proxy_id from provider_accounts where id = 'acct_direct'"
)
.fetch_one(&database.pool)
.await
.unwrap()
.is_none()
);
database.close().await;
}
#[tokio::test]
async fn automatic_proxy_location_projects_to_accounts_and_guards_refresh_results() {
use gateway_core::account::{ProviderAccountStore, RequestLocation};
@@ -1,4 +1,4 @@
//! 验证运行设置的数据库约束、升级保留与快照发布
//! 验证运行设置的读写、数据库约束与快照发布
use std::collections::BTreeMap;
@@ -61,56 +61,12 @@ fn runtime_settings_keep_account_rotation_global() {
}
#[tokio::test]
async fn removing_request_interval_preserves_settings_and_snapshot() {
use gateway_store::postgres::{PgRuntimeSnapshotRepository, RuntimeSnapshotRepository};
let Some(database) = TestDatabase::create_through("drop_interval", 29).await else {
return;
};
sqlx::query("update runtime_settings set request_interval_ms = 60000, max_concurrent_per_account = 7, refresh_concurrency = 12, refresh_margin_seconds = 1234, config_revision = 70 where id = 1")
.execute(&database.pool).await.unwrap();
super::TEST_MIGRATOR.run(&database.pool).await.unwrap();
let column_exists: bool = sqlx::query_scalar("select exists(select 1 from information_schema.columns where table_schema = current_schema() and table_name = 'runtime_settings' and column_name = 'request_interval_ms')")
.fetch_one(&database.pool).await.unwrap();
assert!(!column_exists);
let repository = PgRuntimeSettingsRepository::new(database.pool.clone());
let settings = repository.load_runtime_settings().await.unwrap();
assert_eq!(settings.values.max_concurrent_per_account, 7);
assert_eq!(settings.values.refresh_concurrency, 12);
assert_eq!(settings.values.refresh_margin_seconds, 1234);
let snapshot = PgRuntimeSnapshotRepository::new(database.pool.clone())
.load_runtime_snapshot()
.await
.unwrap();
assert!(
serde_json::to_value(snapshot.settings)
.unwrap()
.get("request_interval_ms")
.is_none()
);
repository
.update_runtime_settings(settings_with_margin(600))
.await
.unwrap();
assert_eq!(
repository
.load_runtime_settings()
.await
.unwrap()
.values
.refresh_margin_seconds,
600
);
database.close().await;
}
#[tokio::test]
async fn privacy_policy_upgrade_defaults_off_and_round_trips_in_snapshot() {
async fn privacy_policy_defaults_off_and_round_trips_in_snapshot() {
use gateway_core::settings::privacy::*;
use gateway_store::postgres::{PgRuntimeSnapshotRepository, RuntimeSnapshotRepository};
let Some(database) = TestDatabase::create_through("privacy_policy", 25).await else {
let Some(database) = TestDatabase::create("privacy_policy").await else {
return;
};
super::TEST_MIGRATOR.run(&database.pool).await.unwrap();
let repository = PgRuntimeSettingsRepository::new(database.pool.clone());
let before = repository.load_runtime_settings().await.unwrap();
assert_eq!(
@@ -151,167 +107,18 @@ async fn privacy_policy_upgrade_defaults_off_and_round_trips_in_snapshot() {
serde_json::to_value(snapshot.settings).unwrap()["codex_privacy_policy"],
serde_json::to_value(expected).unwrap()
);
assert_eq!(
before.values.codex_privacy_policy,
CodexPrivacyPolicy::default()
);
database.close().await;
}
#[tokio::test]
async fn privacy_steps_migration_preserves_rules_and_other_settings() {
let Some(database) = TestDatabase::create_through("privacy_steps", 30).await else {
return;
};
let legacy = serde_json::json!({
"enabled":true, "onError":"skip_rule", "rules":[
{"id":"first","name":"旧规则一","enabled":true,"scope":"request_header","selector":"x-private",
"action":"remove_field","pattern":null,"replacement":"","value":null,
"replaceAll":true,"caseInsensitive":false,"multiLine":false},
{"id":"second","name":"旧规则二","enabled":false,"scope":"request_body","selector":"$.owner",
"action":"regex_replace","pattern":"alice","replacement":"user","value":null,
"replaceAll":false,"caseInsensitive":true,"multiLine":true}
]
});
let mut expected = serde_json::json!({
"enabled":true, "onError":"skip_rule", "rules":[
{"id":"first","name":"旧规则一","enabled":true,"steps":[
{"scope":"request_header","selector":"x-private","action":"remove_field",
"pattern":null,"replacement":"","value":null,
"replaceAll":true,"caseInsensitive":false,"multiLine":false}]},
{"id":"second","name":"旧规则二","enabled":false,"steps":[
{"scope":"request_body","selector":"$.owner","action":"regex_replace",
"pattern":"alice","replacement":"user","value":null,
"replaceAll":false,"caseInsensitive":true,"multiLine":true}]}
]
});
// 同时包含聚合规则,迁移不能嵌套包装或拆散现有步骤
let canonical = serde_json::json!({
"id":"aggregate","name":"通用聚合规则","enabled":true,"steps":[
{"scope":"request_body","selector":"$.metadata","action":"set_value",
"pattern":null,"replacement":"","value":{"owner":"user","items":[1,2]},
"replaceAll":true,"caseInsensitive":false,"multiLine":false},
{"scope":"request_header","selector":"x-private","action":"remove_field",
"pattern":null,"replacement":"","value":null,
"replaceAll":true,"caseInsensitive":false,"multiLine":false}
]
});
let mut mixed = legacy;
mixed["rules"]
.as_array_mut()
.unwrap()
.insert(1, canonical.clone());
expected["rules"]
.as_array_mut()
.unwrap()
.insert(1, canonical);
sqlx::query("update runtime_settings set codex_privacy_policy_json = $1, config_revision = 70, refresh_margin_seconds = 1234 where id = 1")
.bind(&mixed).execute(&database.pool).await.unwrap();
let mut expected_row: serde_json::Value =
sqlx::query_scalar("select to_jsonb(runtime_settings) from runtime_settings where id = 1")
.fetch_one(&database.pool)
.await
.unwrap();
expected_row["codex_privacy_policy_json"] = expected.clone();
expected_row["reset_credit_lead_minutes"] = serde_json::json!(10);
expected_row["reset_credit_max_weekly_remaining_percent"] = serde_json::Value::Null;
// 完整升级还会清理自动回切字段,其余运行设置必须原样保留
expected_row["smart_scheduling_json"]
.as_object_mut()
.unwrap()
.remove("preferHigherWeight");
super::TEST_MIGRATOR.run(&database.pool).await.unwrap();
let stored: serde_json::Value =
sqlx::query_scalar("select to_jsonb(runtime_settings) from runtime_settings where id = 1")
.fetch_one(&database.pool)
.await
.unwrap();
assert_eq!(stored, expected_row);
let repository = PgRuntimeSettingsRepository::new(database.pool.clone());
let loaded = repository.load_runtime_settings().await.unwrap();
assert_eq!(
serde_json::to_value(&loaded.values.codex_privacy_policy).unwrap(),
expected
);
// 服务重启后再次检查迁移,不产生写入或额外嵌套
super::TEST_MIGRATOR.run(&database.pool).await.unwrap();
let stored: serde_json::Value =
sqlx::query_scalar("select to_jsonb(runtime_settings) from runtime_settings where id = 1")
.fetch_one(&database.pool)
.await
.unwrap();
assert_eq!(stored, expected_row);
let snapshot = gateway_store::postgres::RuntimeSnapshotRepository::load_runtime_snapshot(
&gateway_store::postgres::PgRuntimeSnapshotRepository::new(database.pool.clone()),
)
.await
.unwrap();
assert_eq!(
serde_json::to_value(snapshot.settings).unwrap()["codex_privacy_policy"],
expected
);
database.close().await;
}
#[tokio::test]
async fn privacy_steps_migration_rejects_non_object_without_losing_data() {
let Some(database) = TestDatabase::create_through("privacy_invalid", 30).await else {
return;
};
let invalid = serde_json::json!({
"enabled":false,"onError":"reject_request","rules":[{},42]
});
sqlx::query("update runtime_settings set codex_privacy_policy_json = $1 where id = 1")
.bind(&invalid)
.execute(&database.pool)
.await
.unwrap();
assert!(super::TEST_MIGRATOR.run(&database.pool).await.is_err());
let stored: serde_json::Value =
sqlx::query_scalar("select codex_privacy_policy_json from runtime_settings where id = 1")
.fetch_one(&database.pool)
.await
.unwrap();
assert_eq!(stored, invalid);
let applied: bool =
sqlx::query_scalar("select exists(select 1 from _sqlx_migrations where version = 31)")
.fetch_one(&database.pool)
.await
.unwrap();
assert!(!applied);
sqlx::query("update runtime_settings set codex_privacy_policy_json = jsonb_set(codex_privacy_policy_json, '{rules}', '[]') where id = 1")
.execute(&database.pool).await.unwrap();
super::TEST_MIGRATOR.run(&database.pool).await.unwrap();
let loaded = PgRuntimeSettingsRepository::new(database.pool.clone())
.load_runtime_settings()
.await
.unwrap();
assert!(!loaded.values.codex_privacy_policy.enabled);
assert!(loaded.values.codex_privacy_policy.rules.is_empty());
assert_eq!(
loaded.values.codex_privacy_policy.on_error,
gateway_core::settings::privacy::PrivacyFailureMode::RejectRequest
);
database.close().await;
}
#[tokio::test]
async fn smart_settings_upgrade_preserves_selection_and_publishes_custom_config() {
async fn smart_settings_round_trip_publish_and_restore_database_defaults() {
use gateway_core::account::SmartSchedulingConfig;
use gateway_store::postgres::{PgRuntimeSnapshotRepository, RuntimeSnapshotRepository};
let Some(database) = TestDatabase::create_through("smart_config", 18).await else {
let Some(database) = TestDatabase::create("smart_config").await else {
return;
};
// 升级前不能用包含新列的 Repository,直接写入旧版本已有字段
sqlx::query("update runtime_settings set rotation_strategy = 'sticky', refresh_margin_seconds = 3600 where id = 1")
.execute(&database.pool).await.unwrap();
super::TEST_MIGRATOR.run(&database.pool).await.unwrap();
let repository = PgRuntimeSettingsRepository::new(database.pool.clone());
let before = repository.load_runtime_settings().await.unwrap();
assert_eq!(before.rotation_strategy, "sticky");
// 0022 将仍为旧默认 3600 的行迁移到 300;管理员自定义值才会原样保留。
assert_eq!(before.values.refresh_margin_seconds, 300);
assert_eq!(
before.values.smart_scheduling,
SmartSchedulingConfig::default()
@@ -333,8 +140,17 @@ async fn smart_settings_upgrade_preserves_selection_and_publishes_custom_config(
serde_json::to_value(expected).unwrap()
);
assert!(snapshot.config_revision > before.config_revision);
sqlx::query("update runtime_settings set smart_scheduling_json = default where id = 1")
.execute(&database.pool)
.await
.unwrap();
assert_eq!(
before.values.smart_scheduling,
repository
.load_runtime_settings()
.await
.unwrap()
.values
.smart_scheduling,
SmartSchedulingConfig::default()
);
database.close().await;
@@ -434,52 +250,33 @@ async fn unlimited_default_account_concurrency_round_trips_without_relaxing_othe
}
#[tokio::test]
async fn migrations_should_narrow_the_default_refresh_margin_to_the_codex_baseline() {
async fn refresh_margin_defaults_apply_to_initial_settings_and_database_resets() {
let Some(database) = TestDatabase::create("refresh_margin_default").await else {
return;
};
let repository = PgRuntimeSettingsRepository::new(database.pool.clone());
// 新装实例:0001 种子行携带旧默认 3600,0022 数据迁移统一收敛到 300。
let settings = repository
.load_runtime_settings()
.await
.expect("load settings");
assert_eq!(settings.values.refresh_margin_seconds, 300);
// 列默认值同步收窄,重建行不会回退到旧值;目录查询限定本测试 schema,
// 同库并行测试里停留在旧迁移版本的 schema 仍保留 3600 旧默认,不能被读到。
let column_default: String = sqlx::query_scalar(
"select column_default from information_schema.columns \
where table_schema = current_schema() \
and table_name = 'runtime_settings' \
and column_name = 'refresh_margin_seconds'",
)
.fetch_one(&database.pool)
.await
.expect("read column default");
assert_eq!(column_default, "300");
database.close().await;
}
#[tokio::test]
async fn refresh_margin_migration_should_preserve_admin_customized_values() {
// 升级语义:0021 之前部署的实例经 0022 迁移后,仅旧默认 3600 收敛到 300,
// 管理员自定义值原样保留。
let Some(database) = TestDatabase::create_through("refresh_margin_custom", 21).await else {
return;
};
sqlx::query("update runtime_settings set refresh_margin_seconds = 1800 where id = 1")
.execute(&database.pool)
.await
.expect("persist customized margin");
super::TEST_MIGRATOR
.run(&database.pool)
.unwrap();
sqlx::query("update runtime_settings set refresh_margin_seconds = default where id = 1")
.execute(&database.pool)
.await
.expect("apply remaining migrations");
let settings = PgRuntimeSettingsRepository::new(database.pool.clone())
.load_runtime_settings()
.await
.expect("load settings");
assert_eq!(settings.values.refresh_margin_seconds, 1_800);
.expect("reset refresh margin to the database default");
assert_eq!(
repository
.load_runtime_settings()
.await
.unwrap()
.values
.refresh_margin_seconds,
300
);
database.close().await;
}
@@ -714,7 +511,9 @@ async fn concurrency_queue_settings_round_trip_into_the_runtime_snapshot() {
),
(0, 0, 30, 0)
);
let mut update = settings_with_margin(3600);
let mut update = settings_with_margin(1234);
update.values.max_concurrent_per_account = 7;
update.values.refresh_concurrency = 12;
update.values.max_waiting_per_key = 5;
update.values.max_waiting_per_account = 7;
update.values.concurrency_wait_timeout_seconds = 12;
@@ -723,12 +522,15 @@ async fn concurrency_queue_settings_round_trip_into_the_runtime_snapshot() {
let settings = repository.load_runtime_settings().await.unwrap();
assert_eq!(
(
settings.values.max_concurrent_per_account,
settings.values.refresh_concurrency,
settings.values.refresh_margin_seconds,
settings.values.max_waiting_per_key,
settings.values.max_waiting_per_account,
settings.values.concurrency_wait_timeout_seconds,
settings.values.openai_guardian_reserved_concurrency,
),
(5, 7, 12, 1)
(7, 12, 1234, 5, 7, 12, 1)
);
let snapshot = PgRuntimeSnapshotRepository::new(database.pool.clone())
.load_runtime_snapshot()
@@ -737,6 +539,9 @@ async fn concurrency_queue_settings_round_trip_into_the_runtime_snapshot() {
let snapshot_values = serde_json::to_value(&snapshot.settings).unwrap();
assert_eq!(
(
snapshot_values["max_concurrent_per_account"]
.as_u64()
.unwrap(),
snapshot_values["max_waiting_per_key"].as_u64().unwrap(),
snapshot_values["max_waiting_per_account"].as_u64().unwrap(),
snapshot_values["concurrency_wait_timeout_seconds"]
@@ -746,7 +551,7 @@ async fn concurrency_queue_settings_round_trip_into_the_runtime_snapshot() {
.as_u64()
.unwrap(),
),
(5, 7, 12, 1)
(7, 5, 7, 12, 1)
);
assert!(snapshot.config_revision > before.config_revision);
database.close().await;
@@ -1235,48 +1040,6 @@ async fn warmup_cursor_survives_restart_and_settings_updates_without_rewinding()
database.close().await;
}
#[tokio::test]
async fn runtime_scheduling_upgrade_preserves_settings_without_a_warmup_table() {
let Some(database) = TestDatabase::create_through("runtime_scheduling_upgrade", 20).await
else {
return;
};
sqlx::query(
"update runtime_settings set rotation_strategy = 'sticky',
account_warmup_enabled = true, account_warmup_schedule_time = '08:00,13:00',
account_warmup_model = 'test-model' where id = 1",
)
.execute(&database.pool)
.await
.unwrap();
super::TEST_MIGRATOR.run(&database.pool).await.unwrap();
let warmup_table: Option<String> =
sqlx::query_scalar("select to_regclass('account_warmup_slots')::text")
.fetch_one(&database.pool)
.await
.unwrap();
assert!(warmup_table.is_none());
let cursor: Option<DateTime<Utc>> =
sqlx::query_scalar("select account_warmup_cursor from runtime_settings where id = 1")
.fetch_one(&database.pool)
.await
.unwrap();
assert!(cursor.is_none());
let settings = PgRuntimeSettingsRepository::new(database.pool.clone())
.load_runtime_settings()
.await
.unwrap();
assert_eq!(settings.rotation_strategy, "sticky");
assert!(settings.values.account_warmup_enabled);
assert_eq!(settings.values.account_warmup_schedule_time, "08:00,13:00");
assert_eq!(
settings.values.account_warmup_model.as_deref(),
Some("test-model")
);
assert_eq!(settings.values.openai_guardian_reserved_concurrency, 0);
database.close().await;
}
#[tokio::test]
async fn warmup_cursor_resolves_dst_and_deduplicates_across_timezones() {
use gateway_core::{provider_ports::ProviderRuntimePolicyPort, time::DeploymentTimeZone};
@@ -1320,23 +1083,17 @@ async fn warmup_cursor_resolves_dst_and_deduplicates_across_timezones() {
}
#[tokio::test]
async fn account_affinity_upgrade_defaults_and_updates_reach_the_snapshot() {
async fn account_affinity_defaults_and_updates_reach_the_snapshot() {
use gateway_core::account::AccountAffinity;
use gateway_store::postgres::{PgRuntimeSnapshotRepository, RuntimeSnapshotRepository};
let Some(database) = TestDatabase::create_through("account_affinity_upgrade", 22).await else {
let Some(database) = TestDatabase::create("account_affinity").await else {
return;
};
sqlx::query("update runtime_settings set rotation_strategy = 'sticky', config_revision = config_revision + 1 where id = 1")
.execute(&database.pool)
.await
.unwrap();
super::TEST_MIGRATOR.run(&database.pool).await.unwrap();
let repository = PgRuntimeSettingsRepository::new(database.pool.clone());
let before = repository.load_runtime_settings().await.unwrap();
assert_eq!(before.rotation_strategy, "sticky");
assert_eq!(
before.values.openai_account_affinity,
AccountAffinity::Relaxed
AccountAffinity::Strict
);
assert_eq!(before.values.max_account_rotations, 3);
assert_eq!(before.values.openai_session_affinity_ttl_hours, 24);
@@ -1402,105 +1159,12 @@ async fn account_affinity_upgrade_defaults_and_updates_reach_the_snapshot() {
}
#[tokio::test]
async fn preferred_affinity_migration_defaults_to_strict_and_preserves_saved_modes() {
use gateway_core::account::AccountAffinity;
let Some(fresh) = TestDatabase::create("affinity_default").await else {
async fn reset_credit_weekly_threshold_defaults_unset_and_round_trips() {
let Some(database) = TestDatabase::create("reset_weekly_threshold").await else {
return;
};
let repository = PgRuntimeSettingsRepository::new(fresh.pool.clone());
assert_eq!(
repository
.load_runtime_settings()
.await
.unwrap()
.values
.openai_account_affinity,
AccountAffinity::Strict
);
fresh.close().await;
for mode in [AccountAffinity::Relaxed, AccountAffinity::Strict] {
let database = TestDatabase::create_through("affinity_preserve", 23)
.await
.unwrap();
let repository = PgRuntimeSettingsRepository::new(database.pool.clone());
// 升级前只写当时已有列,现行 Repository 依赖完整迁移后的 schema
sqlx::query("update runtime_settings set openai_account_affinity = $1, config_revision = config_revision + 1, updated_at = now() where id = 1")
.bind(match mode {
AccountAffinity::Relaxed => "relaxed",
_ => "strict",
})
.execute(&database.pool)
.await
.unwrap();
super::TEST_MIGRATOR.run(&database.pool).await.unwrap();
assert_eq!(
repository
.load_runtime_settings()
.await
.unwrap()
.values
.openai_account_affinity,
mode
);
database.close().await;
}
}
#[tokio::test]
async fn removing_switchback_preserves_custom_weights_and_replaces_database_default() {
let Some(database) = TestDatabase::create_through("remove_switchback", 31).await else {
return;
};
let expected =
gateway_core::account::SmartSchedulingConfig::new([0.0, 2.0, 1.0, 0.5, 1.2, 2.3]).unwrap();
let mut legacy = serde_json::to_value(expected).unwrap();
legacy["preferHigherWeight"] = serde_json::json!(true);
sqlx::query("update runtime_settings set smart_scheduling_json = $1 where id = 1")
.bind(legacy)
.execute(&database.pool)
.await
.unwrap();
super::TEST_MIGRATOR.run(&database.pool).await.unwrap();
let repository = PgRuntimeSettingsRepository::new(database.pool.clone());
assert_eq!(
repository
.load_runtime_settings()
.await
.unwrap()
.values
.smart_scheduling,
expected
);
sqlx::query("update runtime_settings set smart_scheduling_json = default where id = 1")
.execute(&database.pool)
.await
.unwrap();
assert_eq!(
repository
.load_runtime_settings()
.await
.unwrap()
.values
.smart_scheduling,
gateway_core::account::SmartSchedulingConfig::default()
);
database.close().await;
}
#[tokio::test]
async fn reset_credit_weekly_threshold_upgrade_defaults_unset_and_round_trips() {
let Some(database) = TestDatabase::create_through("reset_weekly_threshold", 34).await else {
return;
};
sqlx::query("update runtime_settings set reset_credit_lead_minutes = 20 where id = 1")
.execute(&database.pool)
.await
.unwrap();
super::TEST_MIGRATOR.run(&database.pool).await.unwrap();
let repository = PgRuntimeSettingsRepository::new(database.pool.clone());
let settings = repository.load_runtime_settings().await.unwrap();
assert_eq!(settings.values.reset_credit_lead_minutes, 20);
assert_eq!(
settings.values.reset_credit_max_weekly_remaining_percent,
None
@@ -2,59 +2,6 @@
use super::TestDatabase;
#[tokio::test]
async fn error_details_migration_preserves_existing_request_and_event_text() {
let Some(db) = TestDatabase::create_through("error_details_upgrade", 22).await else {
return;
};
seed_legacy_request(&db.pool).await;
let raw = "{ \"error\": {\"code\":\"Vendor.Unknown\",\"message\":\"原始错误\"} }";
sqlx::query("update model_requests set raw_upstream_error = $1 where id = 'req_integrity'")
.bind(raw)
.execute(&db.pool)
.await
.unwrap();
sqlx::query(
"insert into ops_events (id, level, component, operation, failure_kind, message, created_at, raw_upstream_error)
values ('error_details_upgrade', 'error', 'provider', 'generate', 'unavailable', 'safe message', now(), $1)",
).bind(raw).execute(&db.pool).await.unwrap();
super::TEST_MIGRATOR.run(&db.pool).await.unwrap();
let details: Vec<String> = sqlx::query_scalar(
"select error_details from model_requests where id = 'req_integrity'
union all select error_details from ops_events where id = 'error_details_upgrade'",
)
.fetch_all(&db.pool)
.await
.unwrap();
assert_eq!(details, vec![raw.to_owned(), raw.to_owned()]);
let old_columns: i64 = sqlx::query_scalar(
"select count(*) from information_schema.columns
where table_schema = current_schema() and table_name in ('model_requests', 'ops_events')
and column_name = 'raw_upstream_error'",
)
.fetch_one(&db.pool)
.await
.unwrap();
assert_eq!(old_columns, 0);
db.close().await;
}
async fn seed_legacy_request(pool: &sqlx::PgPool) {
sqlx::query(
"insert into model_requests (
id, client_api_key_ref, config_revision, protocol, operation, endpoint,
client_transport, started_at, deadline_at, outcome, completed_at, routing_scope
) values (
'req_integrity', 'deleted_key', 1, 'openai', 'responses', '/v1/responses',
'http_sse', now(), now() + interval '1 hour', 'failed', now(), 'all'
)",
)
.execute(pool)
.await
.expect("seed request without optional facts");
}
async fn seed_request(pool: &sqlx::PgPool) {
sqlx::query(
"insert into model_requests (
@@ -191,11 +138,6 @@ async fn historical_references_are_validated_without_live_foreign_keys() {
.await
.unwrap();
assert_eq!(references, "already_deleted_account");
let live_columns: i64 = sqlx::query_scalar("select count(*) from information_schema.columns where table_schema = current_schema()
and ((table_name = 'model_requests' and column_name in ('client_api_key_id','provider_account_id'))
or (table_name = 'ops_events' and column_name = 'provider_account_id'))")
.fetch_one(&db.pool).await.unwrap();
assert_eq!(live_columns, 0);
db.close().await;
}
@@ -252,89 +194,6 @@ async fn backup_completion_cannot_precede_its_start() {
db.close().await;
}
#[tokio::test]
async fn request_observation_migration_preserves_every_historical_field() {
let Some(db) = TestDatabase::create_through("request_observation_upgrade", 24).await else {
return;
};
seed_legacy_request(&db.pool).await;
sqlx::raw_sql(
"insert into model_requests
select (jsonb_populate_record(null::model_requests,
to_jsonb(mr) || jsonb_build_object('id', 'req_empty'))).*
from model_requests mr;
update model_requests set config_revision = 7,
provider_account_name_snapshot = 'historical account',
provider_account_email_snapshot = 'history@example.invalid',
provider_account_authentication_kind_snapshot = 'oauth',
routing_scope = 'groups', routing_group_refs = array['deleted_group'],
routing_group_names_snapshot = '[\"历史组\"]',
provider_error_code = 'rate_limit', error_message = 'safe error', retry_after_ms = 0,
client_ip = '192.0.2.10/24', user_agent = 'fixture', reasoning_effort = 'high',
reasoning_preset = 'balanced', subagent_kind = 'review', compact = true,
transport_decision_wait_ms = 0, connect_ms = 2, headers_ms = 3,
first_event_ms = 4, first_reasoning_ms = 5, first_text_ms = 6,
first_token_ms = 5, latency_ms = 1000, provider_processing_ms = 7,
admission_decision_ms = 0, account_selection_wait_ms = 8,
capacity_used_slots = 0, capacity_total_slots = 20,
http_version = 'HTTP/2', websocket_pool = 'reuse',
upstream_connection_id = 'connection', upstream_connection_exit_reason = 'peer_closed',
upstream_connection_age_ms = 10, upstream_connection_idle_ms = 0,
upstream_response_model = 'response-model', continuation_requested = true,
continuation_previous_response_id_hash = repeat('a', 64),
continuation_unavailable_reason = 'missing_response',
recovery_request_id = 'recovery', recovered_at = completed_at + interval '1 second',
recovery_attempt_count = 1, recovery_retry_delay_ms = 0, recovery_total_latency_ms = 1000,
input_tokens = 10, output_tokens = 20, total_tokens = 30,
cost_source = 'provider_reported', cost_amount = 0.0123456789, cost_currency = 'USD',
error_details = '{ \"opaque\": true }', provider_observation_json = '{\"vendor\":true}',
diagnostic_trace_json = '{\"attempts\":[]}'
where id = 'req_integrity'",
).execute(&db.pool).await.expect("populate all moved fields in the released schema");
let before: Vec<serde_json::Value> =
sqlx::query_scalar("select to_jsonb(mr) from model_requests mr order by id")
.fetch_all(&db.pool)
.await
.unwrap();
super::TEST_MIGRATOR.run_to(27, &db.pool).await.unwrap();
let after: Vec<serde_json::Value> = sqlx::query_scalar(
"select to_jsonb(mr) - 'request_observation_json' - 'upstream_response_ms' - 'upstream_api_overhead_ms' - 'upstream_engine_ms' - 'upstream_engine_iapi_ttft_ms' - 'upstream_engine_service_ttft_ms' - 'upstream_engine_iapi_tbt_ms' - 'upstream_engine_service_tbt_ms'
from model_request_observations mr order by id",
)
.fetch_all(&db.pool)
.await
.unwrap();
assert_eq!(
after, before,
"all 90 historical facts survive the migration"
);
let columns: i64 = sqlx::query_scalar(
"select count(*) from information_schema.columns
where table_schema = current_schema() and table_name = 'model_requests'",
)
.fetch_one(&db.pool)
.await
.unwrap();
assert_eq!(columns, 49);
let observation: serde_json::Value = sqlx::query_scalar(
"select request_observation_json from model_requests where id = 'req_integrity'",
)
.fetch_one(&db.pool)
.await
.unwrap();
assert_eq!(
observation["timings"]["local"]["transportDecisionWaitMs"],
0
);
assert_eq!(observation["scheduling"]["capacityUsedSlots"], 0);
assert!(
observation["timings"]["upstream"]
.get("responseMs")
.is_none()
);
db.close().await;
}
#[tokio::test]
async fn request_observation_reject_malformed_documents_and_keep_upstream_clock_independent() {
use serde_json::json;
@@ -8,7 +8,7 @@ use gateway_store::postgres::{
use super::TestDatabase;
#[test]
fn snapshot_client_policy_contains_only_common_limits() {
fn client_key_snapshot_debug_redacts_plaintext_key() {
let policy = ClientApiKeySnapshot {
name_snapshot: None,
request_profiles: Default::default(),
@@ -20,8 +20,6 @@ fn snapshot_client_policy_contains_only_common_limits() {
requests_per_minute: 60,
},
};
assert_eq!(policy.limits.max_concurrency, 3);
assert!(policy.group_ids.is_empty());
assert!(!format!("{policy:?}").contains("sk_snapshot_secret"));
}
@@ -50,6 +48,8 @@ async fn runtime_snapshot_loads_enabled_plaintext_key_without_debug_exposure() {
Some("snapshot")
);
assert!(snapshot.client_api_keys[0].group_ids.is_empty());
assert_eq!(snapshot.client_api_keys[0].limits.max_concurrency, 2);
assert_eq!(snapshot.client_api_keys[0].limits.requests_per_minute, 60);
assert_eq!(
snapshot.client_api_keys[0].plaintext_key.expose_for_auth(),
plaintext
@@ -4,14 +4,6 @@ use std::path::Path;
use provider_openai::config::{DEFAULT_STREAM_MAX_RETRIES, MAX_STREAM_MAX_RETRIES, OpenAiConfig};
#[test]
fn openai_config_ignores_removed_yaml_identity_fields() {
let config: OpenAiConfig = serde_json::from_value(serde_json::json!({
"wire_profile": {"originator": "legacy-client", "codex_version": "invalid", "residency": "us"}
})).unwrap();
assert_eq!(config, OpenAiConfig::default());
}
#[test]
fn openai_config_keeps_residency_separate_from_client_identity() {
use provider_openai::transport::profile::CodexResidency;
@@ -98,10 +90,6 @@ fn openai_config_defaults_to_the_provider_owned_operating_values() {
u32::try_from(DEFAULT_STREAM_MAX_RETRIES).expect("default retry budget fits u32"),
)
);
assert_eq!(
provider_openai::transport::profile::CodexWireProfile::default().user_agent(),
"Codex Desktop/0.153.4 (Mac OS 15.7.1; arm64) unknown (Codex Desktop; 26.901.51231)"
);
}
#[test]
@@ -119,24 +119,23 @@ fn raw_identity_rejects_injection_conflicting_headers_and_unknown_modes() {
}
#[test]
fn old_configuration_and_old_frozen_profile_keep_their_existing_identity() {
fn preset_identity_round_trip_preserves_request_headers_and_official_preview() {
let state = CodexWireProfileState::new(CodexWireProfile::default());
let legacy = ClientProfileSelection::default();
let resolved = legacy.resolve(&state).unwrap();
let configuration = ClientProfileSelection::default().document().unwrap();
let resolved = RequestProfileSelection::parse(&configuration)
.unwrap()
.resolve(&state)
.unwrap();
let restored: CodexWireProfile =
serde_json::from_value(serde_json::to_value(&resolved).unwrap()).unwrap();
assert_eq!(
RequestProfileSelection::parse(&legacy.document().unwrap())
.unwrap()
.resolve(&state)
.unwrap(),
resolved
build_codex_model_headers(&restored, "Bearer fixture", None).unwrap(),
build_codex_model_headers(&resolved, "Bearer fixture", None).unwrap()
);
let old_document = serde_json::to_value(&resolved).unwrap();
assert!(old_document.get("exact_user_agent").is_none());
let restored: CodexWireProfile = serde_json::from_value(old_document).unwrap();
assert_eq!(restored.user_agent(), resolved.user_agent());
assert_eq!(
state
.preview_selection(&legacy.document().unwrap())
.preview_selection(&configuration)
.unwrap()
.expose_to_provider()["versionSource"],
"official"
@@ -529,98 +529,30 @@ fn encoder_should_restore_conversation_fallback_after_body_and_context_values()
}
#[test]
fn encoder_should_accept_metadata_session_and_thread_ids_but_ignore_legacy_context_aliases() {
fn encoder_should_extract_websocket_context_from_client_metadata() {
let metadata = json!({
"session_id": "metadata-session",
"thread_id": "metadata-thread",
"x-codex-turn-state": "metadata-turn-state",
"turn_id": "metadata-turn-id",
"future_context": {"keep": true}
});
let request = request(Map::from_iter([
("model".to_owned(), json!("client-model")),
("input".to_owned(), json!("prompt")),
("turn_state".to_owned(), json!("legacy-turn-state")),
(
"x-codex-turn-state".to_owned(),
json!("legacy-header-turn-state"),
),
("turn_metadata".to_owned(), json!("legacy-turn-metadata")),
("beta_features".to_owned(), json!("legacy-beta")),
("include_timing_metrics".to_owned(), json!("legacy-timing")),
("codex_window_id".to_owned(), json!("legacy-window")),
(
"x-codex-window-id".to_owned(),
json!("legacy-header-window"),
),
("parent_thread_id".to_owned(), json!("legacy-parent")),
(
"x-codex-parent-thread-id".to_owned(),
json!("legacy-header-parent"),
),
("conversationId".to_owned(), json!("legacy-conversation")),
("sessionId".to_owned(), json!("legacy-session")),
("threadId".to_owned(), json!("legacy-thread")),
("client_request_id".to_owned(), json!("legacy-request")),
("clientRequestId".to_owned(), json!("legacy-camel-request")),
("turnId".to_owned(), json!("legacy-turn")),
("x-codex-turn-id".to_owned(), json!("legacy-header-turn")),
(
"client_metadata".to_owned(),
json!({
"turnState": "metadata-turn-state",
"turnMetadata": "metadata-turn-metadata",
"conversation_id": "metadata-conversation",
"session_id": "metadata-session",
"thread_id": "metadata-thread",
"x-codex-window-id": "metadata-window"
}),
),
("client_metadata".to_owned(), metadata.clone()),
]));
let encoded = encode_generate_request(&request, "gpt-test", None).expect("encode");
assert_eq!(
(
encoded.turn_state,
encoded.turn_metadata,
encoded.beta_features,
encoded.include_timing_metrics,
encoded.codex_window_id,
encoded.parent_thread_id,
encoded.client_conversation_id,
encoded.client_session_id,
encoded.client_thread_id,
encoded.client_request_id,
encoded.client_turn_id,
),
(
None,
None,
None,
None,
None,
None,
None,
Some("metadata-session".to_owned()),
Some("metadata-thread".to_owned()),
None,
None,
)
encoded.client_session_id.as_deref(),
Some("metadata-session")
);
}
#[test]
fn encoder_should_extract_official_websocket_context_projection() {
let request = request(Map::from_iter([
("model".to_owned(), json!("client-model")),
("input".to_owned(), json!("prompt")),
(
"client_metadata".to_owned(),
json!({
"x-codex-turn-state": "metadata-turn-state",
"turn_id": "metadata-turn-id"
}),
),
]));
let encoded = encode_generate_request(&request, "gpt-test", None).expect("encode");
assert_eq!(encoded.client_thread_id.as_deref(), Some("metadata-thread"));
assert_eq!(encoded.turn_state.as_deref(), Some("metadata-turn-state"));
assert_eq!(encoded.client_turn_id.as_deref(), Some("metadata-turn-id"));
assert_eq!(encoded.body().get("client_metadata"), Some(&metadata));
}
#[test]
@@ -365,8 +365,8 @@ async fn websocket_execute_response_create_request_should_collect_completed_sse(
}
#[tokio::test]
async fn websocket_exchange_should_accept_upstream_frame_above_removed_private_limit() {
const REMOVED_FRAME_LIMIT_BYTES: usize = 16 * 1024 * 1024;
async fn websocket_exchange_should_deliver_large_upstream_frames() {
const LARGE_DELTA_BYTES: usize = 17 * 1024 * 1024;
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
@@ -376,14 +376,13 @@ async fn websocket_exchange_should_accept_upstream_frame_above_removed_private_l
let mut websocket =
accept_codex_test_websocket_with(stream, |_request, _response| {}).await;
let _message = websocket.next().await.unwrap().unwrap();
let oversized_event = json!({
let large_event = json!({
"type": "response.output_text.delta",
"delta": "x".repeat(REMOVED_FRAME_LIMIT_BYTES + 1),
"delta": "x".repeat(LARGE_DELTA_BYTES),
})
.to_string();
assert!(oversized_event.len() > REMOVED_FRAME_LIMIT_BYTES);
websocket
.send(Message::Text(oversized_event.into()))
.send(Message::Text(large_event.into()))
.await
.unwrap();
websocket
@@ -391,7 +390,7 @@ async fn websocket_exchange_should_accept_upstream_frame_above_removed_private_l
json!({
"type": "response.completed",
"response": {
"id": "resp_after_oversized_frame",
"id": "resp_after_large_frame",
"object": "response",
"output": [],
"usage": {
@@ -421,11 +420,11 @@ async fn websocket_exchange_should_accept_upstream_frame_above_removed_private_l
let response = execute_response_create_request(&prepared)
.await
.expect("upstream frame above the former private limit should remain forwardable");
.expect("large upstream frame should remain forwardable");
server.await.unwrap();
assert!(response.body.len() > REMOVED_FRAME_LIMIT_BYTES);
assert!(response.body.contains("resp_after_oversized_frame"));
assert!(response.body.len() > LARGE_DELTA_BYTES);
assert!(response.body.contains("resp_after_large_frame"));
}
#[tokio::test]
@@ -228,39 +228,16 @@ async fn quota_refresh_persists_subscription_without_user_profile_and_reads_it_f
assert!(document.get("userId").is_none());
}
#[tokio::test]
async fn billing_percent_does_not_rewrite_quota_access_fact() {
let (store, repository) =
repository_with_accounts(&[("still-exhausted", "still-exhausted")]).await;
let id = account_id("still-exhausted");
store
.apply_quota_access(QuotaAccessChange {
account_id: id.clone(),
expected_revision: CredentialRevision::new(1).expect("revision"),
state: QuotaState::exhausted(QuotaEvidence::ProviderDenied, SystemTime::now(), None),
})
.await
.expect("mark quota exhausted");
crate::support::grok_quota_service(
repository,
QueueBillingTransport::success(br#"{"config":{"creditUsagePercent":100}}"#),
)
.refresh_account(&id)
.await
.expect("refresh exhausted quota");
assert_eq!(
store.account(&id).expect("account").quota().access(),
gateway_core::account::QuotaAccessState::Exhausted
);
}
#[tokio::test]
async fn billing_access_and_scheduling_use_current_quota_fields() {
use gateway_core::account::QuotaAccessState::{Allowed, Exhausted};
for (config, expected_percent, expected_access) in [
(
serde_json::json!({"creditUsagePercent": 0}),
Some(0.0),
Allowed,
),
(
serde_json::json!({"creditUsagePercent": 100}),
Some(100.0),
@@ -306,6 +283,10 @@ async fn billing_access_and_scheduling_use_current_quota_fields() {
let account = store.account(&id).expect("refreshed account");
assert_eq!(snapshot.billing().used_percent(), expected_percent);
assert!(
snapshot.billing().has_authoritative_quota(),
"config: {config}"
);
assert_eq!(
account.quota().access(),
expected_access,
@@ -526,66 +507,10 @@ async fn quota_refresh_rejects_a_concurrent_credential_revision() {
}
#[tokio::test]
async fn legacy_billing_fields_do_not_supply_quota_evidence_or_period() {
for (limit, used) in [
(
serde_json::json!({"val": 10000}),
serde_json::json!({"val": 2500}),
),
(
serde_json::json!("uninterpreted"),
serde_json::json!({"val": -1}),
),
] {
let (store, repository) = repository_with_accounts(&[("legacy", "legacy")]).await;
let id = account_id("legacy");
store
.apply_quota_access(QuotaAccessChange {
account_id: id.clone(),
expected_revision: CredentialRevision::new(1).expect("revision"),
state: QuotaState::exhausted(
QuotaEvidence::ProviderDenied,
SystemTime::now(),
None,
),
})
.await
.expect("mark quota exhausted");
let body = serde_json::to_vec(&serde_json::json!({"config": {
"monthlyLimit": limit, "used": used,
"billingPeriodStart": "2026-07-01T00:00:00Z",
"billingPeriodEnd": "2099-08-01T00:00:00Z"
}}))
.expect("billing document");
let service =
crate::support::grok_quota_service(repository, QueueBillingTransport::success(&body));
let snapshot = service
.refresh_account(&id)
.await
.expect("refresh opaque billing");
let account = store.account(&id).expect("account");
assert_eq!(snapshot.billing().used_percent(), None);
assert!(!snapshot.billing().has_authoritative_quota());
assert_eq!(
snapshot.billing().period_kind(),
provider_xai::GrokQuotaPeriodKind::Other
);
assert_eq!(snapshot.billing().period_start(), None);
assert_eq!(snapshot.billing().period_end(), None);
assert_eq!(service.scheduling_signals(&account), None);
assert_eq!(
account.quota().access(),
gateway_core::account::QuotaAccessState::Exhausted
);
}
}
#[tokio::test]
async fn current_quota_period_does_not_fill_missing_dates_from_legacy_fields() {
async fn quota_period_dates_remain_absent_when_not_reported() {
let (_, repository) = repository_with_accounts(&[("current-period", "current-period")]).await;
let transport = QueueBillingTransport::success(
br#"{"config":{"creditUsagePercent":25,"currentPeriod":{"type":"USAGE_PERIOD_TYPE_WEEKLY"},"billingPeriodStart":"2026-07-01T00:00:00Z","billingPeriodEnd":"2099-08-01T00:00:00Z"}}"#,
br#"{"config":{"creditUsagePercent":25,"currentPeriod":{"type":"USAGE_PERIOD_TYPE_WEEKLY"}}}"#,
);
let snapshot = crate::support::grok_quota_service(repository, transport)
.refresh_account(&account_id("current-period"))
@@ -658,61 +583,59 @@ async fn expired_billing_window_does_not_participate_in_scheduling_rank() {
}
#[tokio::test]
async fn non_authoritative_quota_refresh_does_not_clear_existing_quota_exhaustion() {
let (store, repository) =
repository_with_accounts(&[("free-quota", "subject-free-quota")]).await;
let id = account_id("free-quota");
store
.apply_quota_access(QuotaAccessChange {
account_id: id.clone(),
expected_revision: CredentialRevision::new(1).expect("revision"),
state: QuotaState::exhausted(QuotaEvidence::ProviderDenied, SystemTime::now(), None),
})
.await
.expect("mark quota exhausted");
let transport = QueueBillingTransport::success(
br#"{"config":{"currentPeriod":{"type":"USAGE_PERIOD_TYPE_WEEKLY","start":"2026-07-15T00:00:00Z","end":"2026-07-22T00:00:00Z"},"onDemandCap":{"val":0},"onDemandUsed":{"val":0},"prepaidBalance":{"val":0}}}"#,
);
let snapshot = crate::support::grok_quota_service(repository, transport)
.refresh_account(&id)
.await
.expect("refresh Free quota");
async fn non_authoritative_quota_refresh_preserves_exhaustion_without_scheduling_evidence() {
use provider_xai::GrokQuotaPeriodKind::{Other, Weekly};
assert!(!snapshot.billing().has_authoritative_quota());
assert_eq!(
store.account(&id).expect("account").quota().access(),
gateway_core::account::QuotaAccessState::Exhausted
);
}
for (config, expected_period) in [
(
serde_json::json!({"future_quota": {"remaining": 100}}),
Other,
),
(
serde_json::json!({
"currentPeriod": {"type": "USAGE_PERIOD_TYPE_WEEKLY"},
"onDemandCap": {"val": 0},
"onDemandUsed": {"val": 0},
"prepaidBalance": {"val": 0}
}),
Weekly,
),
] {
let (store, repository) =
repository_with_accounts(&[("free-quota", "subject-free-quota")]).await;
let id = account_id("free-quota");
store
.apply_quota_access(QuotaAccessChange {
account_id: id.clone(),
expected_revision: CredentialRevision::new(1).expect("revision"),
state: QuotaState::exhausted(
QuotaEvidence::ProviderDenied,
SystemTime::now(),
None,
),
})
.await
.expect("mark quota exhausted");
let body = serde_json::to_vec(&serde_json::json!({"config": config})).expect("billing");
let service =
crate::support::grok_quota_service(repository, QueueBillingTransport::success(&body));
let snapshot = service
.refresh_account(&id)
.await
.expect("refresh Free quota");
let account = store.account(&id).expect("account");
#[tokio::test]
async fn reported_zero_percent_is_authoritative_quota() {
let (_, repository) =
repository_with_accounts(&[("zero-percent", "subject-zero-percent")]).await;
let transport = QueueBillingTransport::success(
br#"{"config":{"creditUsagePercent":0,"currentPeriod":{"type":"USAGE_PERIOD_TYPE_WEEKLY","start":"2026-07-15T00:00:00Z","end":"2026-07-22T00:00:00Z"}}}"#,
);
let snapshot = crate::support::grok_quota_service(repository, transport)
.refresh_account(&account_id("zero-percent"))
.await
.expect("refresh reported quota");
assert!(snapshot.billing().has_authoritative_quota());
}
#[tokio::test]
async fn positive_prepaid_balance_is_authoritative_quota() {
let (_, repository) =
repository_with_accounts(&[("prepaid-quota", "subject-prepaid-quota")]).await;
let transport = QueueBillingTransport::success(
br#"{"config":{"currentPeriod":{"type":"USAGE_PERIOD_TYPE_WEEKLY","start":"2026-07-15T00:00:00Z","end":"2026-07-22T00:00:00Z"},"prepaidBalance":{"val":500}}}"#,
);
let snapshot = crate::support::grok_quota_service(repository, transport)
.refresh_account(&account_id("prepaid-quota"))
.await
.expect("refresh prepaid quota");
assert!(snapshot.billing().has_authoritative_quota());
assert!(!snapshot.billing().has_authoritative_quota());
assert_eq!(snapshot.billing().used_percent(), None);
assert_eq!(snapshot.billing().period_kind(), expected_period);
assert_eq!(snapshot.billing().period_start(), None);
assert_eq!(snapshot.billing().period_end(), None);
assert_eq!(service.scheduling_signals(&account), None);
assert_eq!(
account.quota().access(),
gateway_core::account::QuotaAccessState::Exhausted
);
}
}
#[tokio::test]
@@ -177,109 +177,40 @@ fn encoder_should_strip_all_client_metadata_before_grok_build() {
}
#[test]
fn encoder_should_apply_build_defaults_and_normalize_reasoning_effort() {
let request = raw_request(json!({
"model": "client-model",
"input": "hello",
"include": ["web_search_call.action.sources"],
"reasoning": {"effort": "XHIGH", "summary": "auto"}
}));
fn encoder_should_apply_build_defaults_and_preserve_requested_includes() {
for (fields, expected_include) in [
(json!({}), json!(["reasoning.encrypted_content"])),
(
json!({"store": null, "include": null}),
json!(["reasoning.encrypted_content"]),
),
(
json!({"include": ["web_search_call.action.sources"]}),
json!([
"web_search_call.action.sources",
"reasoning.encrypted_content"
]),
),
] {
let mut body = fields;
body["input"] = json!("hello");
let encoded = GrokResponsesRequest::encode(&raw_request(body), "grok-4.7", &client_key())
.expect("build defaults");
let encoded = GrokResponsesRequest::encode(&request, "grok-4.5", &client_key())
.expect("normalized request");
let body = Value::Object(encoded.body().clone());
assert_eq!(body.pointer("/store"), Some(&json!(false)));
assert_eq!(
body.pointer("/include"),
Some(&json!([
"web_search_call.action.sources",
"reasoning.encrypted_content"
]))
);
assert_eq!(body.pointer("/reasoning/effort"), Some(&json!("xhigh")));
assert_eq!(body.pointer("/reasoning/summary"), Some(&json!("auto")));
let request = raw_request(json!({
"model": "client-model",
"input": "hello",
"reasoning": {"effort": "xhigh"}
}));
let encoded = GrokResponsesRequest::encode(&request, "xai/grok-4.6-latest", &client_key())
.expect("routed model request");
let body = Value::Object(encoded.body().clone());
assert_eq!(body.pointer("/model"), Some(&json!("xai/grok-4.6-latest")));
assert_eq!(body.pointer("/reasoning/effort"), Some(&json!("xhigh")));
assert_eq!(body.pointer("/reasoning_effort"), None);
assert_eq!(body.pointer("/reasoningEffort"), None);
let request = raw_request(json!({
"model": "client-model",
"input": "hello",
"reasoning": {"effort": "high", "summary": "auto"}
}));
let encoded = GrokResponsesRequest::encode(&request, "grok-composer-2.5-fast", &client_key())
.expect("Composer request");
let body = Value::Object(encoded.body().clone());
assert_eq!(
body.pointer("/reasoning"),
Some(&json!({"effort":"high", "summary":"auto"}))
);
let request = raw_request(json!({
"model": "client-model",
"input": "hello",
"store": null,
"include": null,
"reasoning": {"effort": "max"}
}));
let encoded = GrokResponsesRequest::encode(&request, "grok-composer-2.5-fast", &client_key())
.expect("empty Composer reasoning");
let body = Value::Object(encoded.body().clone());
assert_eq!(body.pointer("/store"), Some(&json!(false)));
assert_eq!(
body.pointer("/include"),
Some(&json!(["reasoning.encrypted_content"]))
);
assert_eq!(body.pointer("/reasoning"), Some(&json!({"effort":"max"})));
let request = raw_request(json!({
"model": "client-model",
"input": "hello",
"reasoning": {}
}));
let encoded = GrokResponsesRequest::encode(&request, "grok-composer-2.5-fast", &client_key())
.expect("Composer reasoning without effort");
assert_eq!(
Value::Object(encoded.body().clone()).pointer("/reasoning"),
Some(&json!({}))
);
let request = raw_request(json!({
"model": "client-model",
"input": "hello",
"reasoning": {"effort": "max"}
}));
let encoded = GrokResponsesRequest::encode(&request, "tenant/foo/grok-4.6", &client_key())
.expect("unknown provider prefix");
assert_eq!(
Value::Object(encoded.body().clone()).pointer("/reasoning/effort"),
Some(&json!("max"))
);
assert_eq!(encoded.body()["store"], json!(false));
assert_eq!(encoded.body()["include"], expected_include);
}
}
#[test]
fn encoder_should_strip_codex_only_fields_without_model_guesses() {
fn encoder_should_strip_codex_only_fields_and_preserve_business_extensions() {
let request = raw_request(json!({
"model": "client-model",
"input": "hello",
"prompt_cache_retention": "24h",
"safety_identifier": "user-1",
"presence_penalty": 0.1,
"presencePenalty": 0.2,
"frequency_penalty": 0.3,
"frequencyPenalty": 0.4,
"stop": ["done"],
"provider_options": {"providers": {"xai": {"transport": "websocket"}}},
"future_official_field": {"keep": true},
"external_web_access": true,
"metadata": {"external_web_access": false},
"tools": [{
@@ -299,6 +230,7 @@ fn encoder_should_strip_codex_only_fields_without_model_guesses() {
for pointer in [
"/prompt_cache_retention",
"/safety_identifier",
"/provider_options",
"/external_web_access",
"/metadata/external_web_access",
"/tools/0/external_web_access",
@@ -306,18 +238,10 @@ fn encoder_should_strip_codex_only_fields_without_model_guesses() {
assert_eq!(body.pointer(pointer), None, "field survived at {pointer}");
}
let request = raw_request(json!({
"model": "client-model",
"input": "hello",
"logprobs": true,
"top_logprobs": 5
}));
let encoded = GrokResponsesRequest::encode(&request, "grok-4.20-reasoning", &client_key())
.expect("sanitized 4.20 request");
let body = Value::Object(encoded.body().clone());
assert_eq!(body.pointer("/model"), Some(&json!("grok-4.20-reasoning")));
assert_eq!(body.pointer("/logprobs"), Some(&json!(true)));
assert_eq!(body.pointer("/top_logprobs"), Some(&json!(5)));
assert_eq!(
body.pointer("/future_official_field"),
Some(&json!({"keep": true}))
);
}
#[test]
@@ -444,31 +368,6 @@ fn request_debug_should_not_expose_prompt_or_unknown_values() {
assert!(!debug.contains("must-not-leak"));
}
#[test]
fn encoder_should_remove_retired_gateway_provider_options_before_xai_conversion() {
let request = raw_request(Value::Object(Map::from_iter([
("model".to_owned(), json!("client")),
("input".to_owned(), json!("hello")),
(
"provider_options".to_owned(),
json!({
"version": "future-version",
"providers": {"xai": {"transport": "websocket", "turn_index": "7"}}
}),
),
("future_official_field".to_owned(), json!({"keep": true})),
])));
let encoded =
GrokResponsesRequest::encode(&request, "grok-routed", &client_key()).expect("request");
assert!(encoded.body().get("provider_options").is_none());
assert_eq!(
encoded.body().get("future_official_field"),
Some(&json!({"keep": true}))
);
}
#[test]
fn explicit_session_should_be_tenant_isolated_and_stable_across_turns() {
let first = raw_request(json!({
@@ -1,52 +1,32 @@
//! 路由已选定真实模型,xAI 编码器不再解释旧别名或 Provider 前缀
//! 验证路由模型标识与业务参数在 xAI 请求编码后保持原值
use super::*;
#[test]
fn routed_model_identity_is_preserved_without_provider_local_aliases() {
let request = raw_request(json!({"model": "client-model", "input": "hello"}));
fn routed_model_and_business_parameters_are_preserved() {
let fields = json!({
"presence_penalty": 0.1,
"frequency_penalty": 0.2,
"stop": ["done"],
"logprobs": true,
"top_logprobs": 5
});
let mut body = fields.clone();
body["model"] = json!("client-model");
body["input"] = json!("hello");
let request = raw_request(body);
for model in [
"grok",
"grok-latest",
"grok-build-latest",
"grok-4.5-latest",
"grok-4.6-latest",
"grok-4.3-latest",
"grok-build",
"grok-composer",
"composer-2.5",
"grok-4.20-reasoning",
"grok-4.20-non-reasoning",
"grok-4.20-multi-agent",
"grok-4.20-multi-agent-latest",
"grok-4.7",
"Grok-4.6",
"grok-latest",
"grok-4.20-reasoning",
"Grok-4.7",
"xai/grok-4.7",
"x-ai/grok-4.7",
"grok/grok-4.7",
"tenant/custom-model",
] {
let encoded = GrokResponsesRequest::encode(&request, model, &client_key())
.expect("routed upstream model");
assert_eq!(encoded.body()["model"], model);
}
}
#[test]
fn routed_model_identity_does_not_infer_unsupported_fields_from_old_slugs() {
let fields = json!({"presence_penalty": 0.1, "frequency_penalty": 0.2,
"stop": ["done"], "logprobs": true, "top_logprobs": 5});
let request = raw_request(fields.clone());
for model in [
"grok-4.5",
"grok-4.6",
"grok-4.3",
"grok-4.20-reasoning",
"tenant/grok-4.6",
"grok-4.7",
] {
let encoded =
GrokResponsesRequest::encode(&request, model, &client_key()).expect("request");
for (key, value) in fields.as_object().expect("fields") {
assert_eq!(encoded.body().get(key), Some(value), "{model} {key}");
}
@@ -4,14 +4,7 @@ use super::*;
#[test]
fn official_reasoning_efforts_are_preserved_for_current_and_future_models() {
for model in [
"grok-4.7",
"grok-future",
"grok-4.5",
"grok-4.6",
"grok-3-mini",
"grok-composer-2.5-fast",
] {
for model in ["grok-4.7", "grok-future", "tenant/custom-model"] {
for effort in ["none", "minimal", "low", "medium", "high", "xhigh", "max"] {
let request = raw_request(json!({
"input": "hello",