Files
Aether/apps/aether-gateway/src/tests/control/admin/endpoints/keys.rs
T

3013 lines
111 KiB
Rust

use std::sync::{Arc, Mutex};
use aether_contracts::ExecutionPlan;
use aether_crypto::{
decrypt_python_fernet_ciphertext, encrypt_python_fernet_plaintext, DEVELOPMENT_ENCRYPTION_KEY,
};
use aether_data::repository::provider_catalog::InMemoryProviderCatalogReadRepository;
use aether_data_contracts::repository::provider_catalog::{
ProviderCatalogKeyListQuery, ProviderCatalogReadRepository, StoredProviderCatalogEndpoint,
StoredProviderCatalogKey, StoredProviderCatalogKeyMaintenanceSummary,
StoredProviderCatalogKeyPage, StoredProviderCatalogKeyStats, StoredProviderCatalogProvider,
};
use aether_data_contracts::DataLayerError;
use axum::body::Body;
use axum::routing::any;
use axum::{extract::Request, Json, Router};
use http::StatusCode;
use serde_json::json;
use super::super::super::{
build_router_with_state, build_state_with_execution_runtime_override, sample_endpoint,
sample_key, sample_provider, start_server, AppState,
};
use crate::constants::{
GATEWAY_HEADER, TRUSTED_ADMIN_SESSION_ID_HEADER, TRUSTED_ADMIN_USER_ID_HEADER,
TRUSTED_ADMIN_USER_ROLE_HEADER,
};
use crate::data::GatewayDataState;
const PROVIDER_KEYS_TEST_STACK_BYTES: usize = 16 * 1024 * 1024;
fn run_provider_keys_test<F, Fut>(test_name: &'static str, make_future: F)
where
F: FnOnce() -> Fut + Send + 'static,
Fut: std::future::Future<Output = ()> + 'static,
{
let handle = std::thread::Builder::new()
.name(test_name.to_string())
.stack_size(PROVIDER_KEYS_TEST_STACK_BYTES)
.spawn(move || {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("test runtime should build");
runtime.block_on(make_future());
})
.expect("provider keys test thread should spawn");
if let Err(payload) = handle.join() {
std::panic::resume_unwind(payload);
}
}
struct SummaryNullingProviderCatalogReadRepository {
inner: InMemoryProviderCatalogReadRepository,
}
impl SummaryNullingProviderCatalogReadRepository {
fn seed(
providers: Vec<StoredProviderCatalogProvider>,
endpoints: Vec<StoredProviderCatalogEndpoint>,
keys: Vec<StoredProviderCatalogKey>,
) -> Self {
Self {
inner: InMemoryProviderCatalogReadRepository::seed(providers, endpoints, keys),
}
}
}
#[async_trait::async_trait]
impl ProviderCatalogReadRepository for SummaryNullingProviderCatalogReadRepository {
async fn list_providers(
&self,
active_only: bool,
) -> Result<Vec<StoredProviderCatalogProvider>, DataLayerError> {
self.inner.list_providers(active_only).await
}
async fn list_providers_by_ids(
&self,
provider_ids: &[String],
) -> Result<Vec<StoredProviderCatalogProvider>, DataLayerError> {
self.inner.list_providers_by_ids(provider_ids).await
}
async fn list_endpoints_by_ids(
&self,
endpoint_ids: &[String],
) -> Result<Vec<StoredProviderCatalogEndpoint>, DataLayerError> {
self.inner.list_endpoints_by_ids(endpoint_ids).await
}
async fn list_endpoints_by_provider_ids(
&self,
provider_ids: &[String],
) -> Result<Vec<StoredProviderCatalogEndpoint>, DataLayerError> {
self.inner
.list_endpoints_by_provider_ids(provider_ids)
.await
}
async fn list_keys_by_ids(
&self,
key_ids: &[String],
) -> Result<Vec<StoredProviderCatalogKey>, DataLayerError> {
self.inner.list_keys_by_ids(key_ids).await
}
async fn list_keys_by_provider_ids(
&self,
provider_ids: &[String],
) -> Result<Vec<StoredProviderCatalogKey>, DataLayerError> {
self.inner.list_keys_by_provider_ids(provider_ids).await
}
async fn list_key_summaries_by_provider_ids(
&self,
provider_ids: &[String],
) -> Result<Vec<StoredProviderCatalogKey>, DataLayerError> {
let mut keys = self.inner.list_keys_by_provider_ids(provider_ids).await?;
for key in &mut keys {
key.internal_priority = 50;
key.global_priority_by_format = None;
key.rate_multipliers = None;
key.request_count = None;
key.success_count = None;
key.error_count = None;
key.total_response_time_ms = None;
key.circuit_breaker_by_format = None;
}
Ok(keys)
}
async fn list_key_maintenance_summaries_by_provider_ids(
&self,
provider_ids: &[String],
) -> Result<Vec<StoredProviderCatalogKeyMaintenanceSummary>, DataLayerError> {
self.inner
.list_key_maintenance_summaries_by_provider_ids(provider_ids)
.await
}
async fn list_keys_page(
&self,
query: &ProviderCatalogKeyListQuery,
) -> Result<StoredProviderCatalogKeyPage, DataLayerError> {
self.inner.list_keys_page(query).await
}
async fn list_key_stats_by_provider_ids(
&self,
provider_ids: &[String],
) -> Result<Vec<StoredProviderCatalogKeyStats>, DataLayerError> {
self.inner
.list_key_stats_by_provider_ids(provider_ids)
.await
}
}
#[tokio::test]
async fn gateway_handles_admin_provider_keys_locally_with_trusted_admin_principal() {
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/endpoints/providers/provider-openai/keys",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let mut key_a = sample_key(
"key-openai-a",
"provider-openai",
"openai:chat",
"sk-test-a",
);
key_a.internal_priority = 10;
key_a.request_count = Some(12);
key_a.success_count = Some(9);
key_a.error_count = Some(3);
key_a.total_response_time_ms = Some(1800);
key_a.created_at_unix_ms = Some(1_711_000_000);
key_a.updated_at_unix_secs = Some(1_711_000_100);
key_a.last_used_at_unix_secs = Some(1_711_000_120);
key_a.note = Some("primary key".to_string());
key_a.status_snapshot = Some(json!({
"oauth": {"code": "none", "requires_reauth": false, "expiring_soon": false},
"account": {"code": "ok", "blocked": false, "recoverable": false},
"quota": {"code": "unknown", "exhausted": false}
}));
let mut key_b = sample_key(
"key-openai-b",
"provider-openai",
"openai:chat",
"sk-test-b",
);
key_b.internal_priority = 20;
key_b.request_count = Some(4);
key_b.success_count = Some(4);
key_b.error_count = Some(0);
key_b.total_response_time_ms = Some(400);
key_b.created_at_unix_ms = Some(1_711_100_000);
key_b.updated_at_unix_secs = Some(1_711_100_100);
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-openai", "openai", 10)],
vec![],
vec![key_a, key_b],
));
let (upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(GatewayDataState::with_provider_catalog_reader_for_tests(
provider_catalog_repository,
)),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.get(format!(
"{gateway_url}/api/admin/endpoints/providers/provider-openai/keys?skip=0&limit=50"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
let items = payload.as_array().expect("payload should be an array");
assert_eq!(items.len(), 2);
assert_eq!(items[0]["id"], "key-openai-a");
assert_eq!(items[0]["internal_priority"], 10);
assert_eq!(items[0]["request_count"], 12);
assert_eq!(items[0]["success_count"], 9);
assert_eq!(items[0]["error_count"], 3);
assert_eq!(items[0]["note"], "primary key");
assert_eq!(items[0]["api_key_masked"], "sk-test-a***");
assert_eq!(items[1]["id"], "key-openai-b");
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_provider_keys_expose_circuit_breaker_and_recover_clears_it() {
let key = sample_key("key-1", "provider-1", "openai:chat", "sk-test-a").with_health_fields(
Some(json!({"openai:chat": {
"health_score": 0.2,
"consecutive_failures": 8,
"last_failure_at": "2026-03-26T12:00:00+00:00"
}})),
Some(json!({"openai:chat": {
"open": true,
"open_at": "2026-03-26T12:00:00+00:00",
"reason": "consecutive_failures_8",
"next_probe_at": "2099-03-26T12:01:00+00:00",
"next_probe_at_unix_secs": 4078209660u64,
"probe_interval_minutes": 1,
"max_probe_interval_minutes": 32,
"half_open_until": null,
"half_open_successes": 0,
"half_open_failures": 0
}})),
);
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-1", "openai", 10)],
vec![sample_endpoint(
"endpoint-1",
"provider-1",
"openai:chat",
"https://example.com/v1",
)],
vec![key],
));
let gateway_state = AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_repository_for_tests(Arc::clone(
&provider_catalog_repository,
)),
);
let gateway = build_router_with_state(gateway_state);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let client = reqwest::Client::new();
let response = client
.get(format!(
"{gateway_url}/api/admin/endpoints/providers/provider-1/keys?skip=0&limit=50"
))
.header(GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload[0]["circuit_breaker_open"], true);
assert_eq!(
payload[0]["circuit_breaker_by_format"]["openai:chat"]["reason"],
"consecutive_failures_8"
);
assert_eq!(
payload[0]["circuit_breaker_by_format"]["openai:chat"]["probe_interval_minutes"],
1
);
assert!(
payload[0]["circuit_breaker_by_format"]["openai:chat"]["next_probe_at_unix_secs"]
.as_u64()
.is_some()
);
let recover_response = client
.patch(format!(
"{gateway_url}/api/admin/endpoints/health/keys/key-1"
))
.header(GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("recover request should succeed");
assert_eq!(recover_response.status(), StatusCode::OK);
let response = client
.get(format!(
"{gateway_url}/api/admin/endpoints/providers/provider-1/keys?skip=0&limit=50"
))
.header(GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload[0]["circuit_breaker_open"], false);
gateway_handle.abort();
}
#[tokio::test]
async fn gateway_handles_admin_provider_keys_page_locally_with_total() {
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/endpoints/providers/provider-openai/keys",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let mut key_a = sample_key(
"key-openai-a",
"provider-openai",
"openai:chat",
"sk-test-a",
);
key_a.internal_priority = 10;
key_a.created_at_unix_ms = Some(1_711_000_000);
let mut key_b = sample_key(
"key-openai-b",
"provider-openai",
"openai:chat",
"sk-test-b",
);
key_b.internal_priority = 20;
key_b.created_at_unix_ms = Some(1_711_100_000);
let mut key_c = sample_key(
"key-openai-c",
"provider-openai",
"openai:chat",
"sk-test-c",
);
key_c.internal_priority = 30;
key_c.created_at_unix_ms = Some(1_711_200_000);
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-openai", "openai", 10)],
vec![],
vec![key_a, key_b, key_c],
));
let (_upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(GatewayDataState::with_provider_catalog_reader_for_tests(
provider_catalog_repository,
)),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.get(format!(
"{gateway_url}/api/admin/endpoints/providers/provider-openai/keys?page=2&page_size=1"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["total"], 3);
assert_eq!(payload["page"], 2);
assert_eq!(payload["page_size"], 1);
let items = payload["keys"].as_array().expect("keys should be an array");
assert_eq!(items.len(), 1);
assert_eq!(items[0]["id"], "key-openai-b");
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_admin_provider_keys_prefers_upstream_plan_type_over_auth_config() {
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/endpoints/providers/provider-codex/keys",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let mut provider = sample_provider("provider-codex", "codex", 10);
provider.provider_type = "codex".to_string();
let mut key = sample_key(
"key-codex-oauth",
"provider-codex",
"openai:responses",
"oauth-placeholder",
);
key.auth_type = "oauth".to_string();
key.encrypted_auth_config = Some(
encrypt_python_fernet_plaintext(
DEVELOPMENT_ENCRYPTION_KEY,
&json!({
"plan_type": "free",
"account_id": "acct-codex-legacy"
})
.to_string(),
)
.expect("auth config should encrypt"),
);
key.upstream_metadata = Some(json!({
"codex": {
"plan_type": "plus",
"updated_at": 1_775_553_285u64
}
}));
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![provider],
vec![],
vec![key],
));
let (_upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_reader_for_tests(
provider_catalog_repository,
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.get(format!(
"{gateway_url}/api/admin/endpoints/providers/provider-codex/keys?skip=0&limit=50"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
let items = payload.as_array().expect("payload should be an array");
assert_eq!(items.len(), 1);
assert_eq!(items[0]["oauth_plan_type"], "plus");
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_admin_provider_keys_marks_oauth_header_auth() {
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/endpoints/providers/provider-codex/keys",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let mut provider = sample_provider("provider-codex", "codex", 10);
provider.provider_type = "codex".to_string();
let mut key = sample_key(
"key-codex-oauth-header",
"provider-codex",
"openai:responses",
"imported-session-token",
);
key.auth_type = "oauth".to_string();
key.encrypted_auth_config = Some(
encrypt_python_fernet_plaintext(
DEVELOPMENT_ENCRYPTION_KEY,
r#"{"provider_type":"codex","headers":{"authorization":"Bearer imported-session-token"}}"#,
)
.expect("auth config should encrypt"),
);
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![provider],
vec![],
vec![key],
));
let (_upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_reader_for_tests(
provider_catalog_repository,
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.get(format!(
"{gateway_url}/api/admin/endpoints/providers/provider-codex/keys?skip=0&limit=50"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
let items = payload.as_array().expect("payload should be an array");
assert_eq!(items.len(), 1);
assert_eq!(items[0]["api_key_masked"], "[OAuth Header]");
assert_eq!(items[0]["oauth_header_auth"], true);
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_creates_admin_provider_key_locally_with_trusted_admin_principal() {
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/endpoints/providers/provider-openai/keys",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-openai", "openai", 10)],
vec![],
vec![],
));
let (upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_repository_for_tests(
provider_catalog_repository.clone(),
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!(
"{gateway_url}/api/admin/endpoints/providers/provider-openai/keys"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.json(&json!({
"api_formats": ["openai:chat"],
"api_key": "sk-created-openai",
"name": "created key",
"internal_priority": 15,
"capabilities": {"cache_1h": true},
"note": "created from rust"
}))
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["provider_id"], "provider-openai");
assert_eq!(payload["name"], "created key");
assert_eq!(payload["internal_priority"], 15);
assert_eq!(payload["api_formats"], json!(["openai:chat"]));
assert_eq!(payload["api_key_masked"], "sk-creat***enai");
assert_eq!(payload["request_count"], 0);
assert_eq!(payload["success_count"], 0);
assert_eq!(payload["error_count"], 0);
assert_eq!(payload["note"], "created from rust");
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
let keys = provider_catalog_repository
.list_keys_by_provider_ids(&["provider-openai".to_string()])
.await
.expect("keys should read");
assert_eq!(keys.len(), 1);
assert_eq!(keys[0].name, "created key");
assert_eq!(keys[0].auth_type, "api_key");
assert_eq!(keys[0].internal_priority, 15);
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn generic_key_routes_reject_agent_identity_credential_writes() {
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-codex", "codex", 10)],
vec![],
vec![sample_key(
"key-codex-existing",
"provider-codex",
"openai:responses",
"existing-secret",
)],
));
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_repository_for_tests(
provider_catalog_repository.clone(),
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let agent_identity = json!({
"provider_type": "codex",
"auth_mode": "agentIdentity",
"agent_runtime_id": "runtime-bypass",
"agent_private_key": "private-key-must-use-dedicated-import"
});
let client = reqwest::Client::new();
let create_response = client
.post(format!(
"{gateway_url}/api/admin/endpoints/providers/provider-codex/keys"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.json(&json!({
"api_formats": ["openai:responses"],
"auth_type": "oauth",
"auth_config": agent_identity,
"name": "bypass create"
}))
.send()
.await
.expect("create request should complete");
assert_eq!(create_response.status(), StatusCode::BAD_REQUEST);
let create_payload: serde_json::Value = create_response
.json()
.await
.expect("create error should be JSON");
assert!(create_payload["detail"]
.as_str()
.is_some_and(|detail| detail.contains("专属创建或导入接口")));
let update_response = client
.put(format!(
"{gateway_url}/api/admin/endpoints/keys/key-codex-existing"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.json(&json!({
"auth_type": "oauth",
"auth_config": {
"provider_type": "codex",
"auth_mode": "agentIdentity",
"agent_runtime_id": "runtime-bypass-update",
"agent_private_key": "private-key-must-use-dedicated-import"
}
}))
.send()
.await
.expect("update request should complete");
assert_eq!(update_response.status(), StatusCode::BAD_REQUEST);
let update_payload: serde_json::Value = update_response
.json()
.await
.expect("update error should be JSON");
assert!(update_payload["detail"]
.as_str()
.is_some_and(|detail| detail.contains("专属创建或导入接口")));
let keys = provider_catalog_repository
.list_keys_by_provider_ids(&["provider-codex".to_string()])
.await
.expect("keys should read");
assert_eq!(keys.len(), 1);
assert_eq!(keys[0].id, "key-codex-existing");
assert_eq!(keys[0].auth_type, "api_key");
gateway_handle.abort();
}
#[tokio::test]
async fn generic_codex_key_credential_switch_rotates_generation_and_clears_quota() {
let mut existing_key = sample_key(
"key-codex-existing",
"provider-codex",
"openai:responses",
"old-oauth-access-token",
);
existing_key.auth_type = "oauth".to_string();
existing_key.encrypted_auth_config = Some(
encrypt_python_fernet_plaintext(
DEVELOPMENT_ENCRYPTION_KEY,
r#"{"provider_type":"codex","refresh_token":"old-refresh-token"}"#,
)
.expect("old auth config should encrypt"),
);
existing_key.upstream_metadata = Some(json!({
"codex": {
"credential_generation": "generation-before-switch",
"primary_used_percent": 80.0,
},
"unrelated": {"preserved": true},
}));
existing_key.status_snapshot = Some(json!({
"oauth": {"status": "valid"},
"quota": {"used_ratio": 0.8},
}));
let mut provider = sample_provider("provider-codex", "codex", 10);
provider.provider_type = "codex".to_string();
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![provider],
vec![],
vec![existing_key],
));
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_repository_for_tests(
provider_catalog_repository.clone(),
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.put(format!(
"{gateway_url}/api/admin/endpoints/keys/key-codex-existing"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.json(&json!({
"auth_type": "api_key",
"api_key": "new-codex-api-key"
}))
.send()
.await
.expect("credential switch should complete");
assert_eq!(response.status(), StatusCode::OK);
let reloaded = provider_catalog_repository
.list_keys_by_ids(&["key-codex-existing".to_string()])
.await
.expect("key should reload");
assert_eq!(reloaded.len(), 1);
let key = &reloaded[0];
assert_eq!(key.auth_type, "api_key");
let codex = key
.upstream_metadata
.as_ref()
.and_then(|metadata| metadata.get("codex"))
.and_then(serde_json::Value::as_object)
.expect("codex metadata should exist");
assert_eq!(codex.len(), 1, "unexpected Codex metadata: {codex:?}");
assert_ne!(
codex
.get(aether_admin::provider::quota::CODEX_CREDENTIAL_GENERATION_KEY)
.and_then(serde_json::Value::as_str),
Some("generation-before-switch")
);
assert_eq!(
key.upstream_metadata
.as_ref()
.and_then(|metadata| metadata.pointer("/unrelated/preserved")),
Some(&json!(true))
);
assert_eq!(
key.status_snapshot
.as_ref()
.and_then(|snapshot| snapshot.get("quota")),
Some(&serde_json::Value::Null)
);
gateway_handle.abort();
}
#[tokio::test]
async fn provider_key_concurrent_limit_create_and_list_responses() {
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-openai", "openai", 10)],
vec![],
vec![],
));
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_repository_for_tests(
provider_catalog_repository.clone(),
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let create_with_limit_response = reqwest::Client::new()
.post(format!(
"{gateway_url}/api/admin/endpoints/providers/provider-openai/keys"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.json(&json!({
"api_formats": ["openai:chat"],
"api_key": "sk-created-openai-concurrent",
"name": "created key with concurrency",
"rpm_limit": 60,
"concurrent_limit": 3
}))
.send()
.await
.expect("request should succeed");
assert_eq!(create_with_limit_response.status(), StatusCode::OK);
let create_payload: serde_json::Value = create_with_limit_response
.json()
.await
.expect("json body should parse");
let create_with_limit_id = create_payload["id"]
.as_str()
.expect("created key id should be returned")
.to_string();
assert_eq!(create_payload["rpm_limit"], 60);
assert_eq!(create_payload["concurrent_limit"], 3);
let create_null_response = reqwest::Client::new()
.post(format!(
"{gateway_url}/api/admin/endpoints/providers/provider-openai/keys"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.json(&json!({
"api_formats": ["openai:chat"],
"api_key": "sk-created-openai-concurrent-null",
"name": "created key with null concurrency",
"concurrent_limit": null
}))
.send()
.await
.expect("request should succeed");
assert_eq!(create_null_response.status(), StatusCode::OK);
let null_payload: serde_json::Value = create_null_response
.json()
.await
.expect("json body should parse");
let create_null_id = null_payload["id"]
.as_str()
.expect("created null-limit key id should be returned")
.to_string();
assert_eq!(null_payload["concurrent_limit"], serde_json::Value::Null);
let create_negative_response = reqwest::Client::new()
.post(format!(
"{gateway_url}/api/admin/endpoints/providers/provider-openai/keys"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.json(&json!({
"api_formats": ["openai:chat"],
"api_key": "sk-created-openai-concurrent-negative",
"name": "created key with negative concurrency",
"concurrent_limit": -1
}))
.send()
.await
.expect("request should succeed");
assert_eq!(create_negative_response.status(), StatusCode::BAD_REQUEST);
let negative_payload: serde_json::Value = create_negative_response
.json()
.await
.expect("json body should parse");
assert!(negative_payload["detail"]
.as_str()
.expect("detail should be string")
.contains("concurrent_limit"));
let list_response = reqwest::Client::new()
.get(format!(
"{gateway_url}/api/admin/endpoints/providers/provider-openai/keys?skip=0&limit=50"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(list_response.status(), StatusCode::OK);
let list_payload: serde_json::Value =
list_response.json().await.expect("json body should parse");
let items = list_payload.as_array().expect("payload should be an array");
assert_eq!(items.len(), 2);
let with_limit = items
.iter()
.find(|item| item["name"].as_str() == Some("created key with concurrency"))
.expect("created key with concurrency should be listed");
let with_null = items
.iter()
.find(|item| item["name"].as_str() == Some("created key with null concurrency"))
.expect("created key with null concurrency should be listed");
assert_eq!(with_limit["concurrent_limit"], 3);
assert_eq!(with_null["concurrent_limit"], serde_json::Value::Null);
let read_back = provider_catalog_repository
.list_keys_by_ids(&[create_with_limit_id, create_null_id])
.await
.expect("created keys should read by id");
assert_eq!(read_back.len(), 2);
assert!(read_back.iter().any(|key| {
key.name == "created key with concurrency" && key.concurrent_limit == Some(3)
}));
assert!(read_back.iter().any(|key| {
key.name == "created key with null concurrency" && key.concurrent_limit.is_none()
}));
let keys = provider_catalog_repository
.list_keys_by_provider_ids(&["provider-openai".to_string()])
.await
.expect("keys should read");
assert_eq!(keys.len(), 2);
assert!(keys
.iter()
.any(|key| key.name == "created key with concurrency" && key.concurrent_limit == Some(3)));
assert!(keys.iter().any(
|key| key.name == "created key with null concurrency" && key.concurrent_limit.is_none()
));
gateway_handle.abort();
}
#[tokio::test]
async fn provider_key_concurrent_limit_reads_existing_list_response() {
let mut key_a = sample_key(
"provider-key-a",
"test-provider-a",
"openai:chat",
"sk-provider-key-a",
);
key_a.concurrent_limit = Some(1);
let mut key_b = sample_key(
"provider-key-b",
"test-provider-a",
"openai:chat",
"sk-provider-key-b",
);
key_b.concurrent_limit = None;
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("test-provider-a", "openai", 10)],
vec![],
vec![key_a, key_b],
));
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(GatewayDataState::with_provider_catalog_reader_for_tests(
provider_catalog_repository,
)),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.get(format!(
"{gateway_url}/api/admin/endpoints/providers/test-provider-a/keys?skip=0&limit=50"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
let items = payload.as_array().expect("payload should be an array");
assert_eq!(items.len(), 2);
let provider_key_a = items
.iter()
.find(|item| item["id"].as_str() == Some("provider-key-a"))
.expect("provider-key-a should be listed");
let provider_key_b = items
.iter()
.find(|item| item["id"].as_str() == Some("provider-key-b"))
.expect("provider-key-b should be listed");
assert_eq!(provider_key_a["concurrent_limit"], 1);
assert_eq!(provider_key_b["concurrent_limit"], serde_json::Value::Null);
gateway_handle.abort();
}
#[test]
fn gateway_fetches_allowed_models_immediately_when_creating_key_with_auto_fetch() {
run_provider_keys_test(
"gateway_fetches_allowed_models_immediately_when_creating_key_with_auto_fetch",
gateway_fetches_allowed_models_immediately_when_creating_key_with_auto_fetch_impl,
);
}
async fn gateway_fetches_allowed_models_immediately_when_creating_key_with_auto_fetch_impl() {
let execution_runtime_hits = Arc::new(Mutex::new(0usize));
let execution_runtime_hits_clone = Arc::clone(&execution_runtime_hits);
let execution_runtime = Router::new().route(
"/v1/execute/sync",
any(move |Json(plan): Json<ExecutionPlan>| {
let execution_runtime_hits_inner = Arc::clone(&execution_runtime_hits_clone);
async move {
*execution_runtime_hits_inner
.lock()
.expect("mutex should lock") += 1;
assert_eq!(plan.url, "https://api.openai.example/v1/models");
assert_eq!(
plan.headers.get("authorization").map(String::as_str),
Some("Bearer sk-created-openai")
);
Json(json!({
"request_id": "req-create-key-auto-fetch",
"status_code": 200,
"headers": {
"content-type": "application/json"
},
"body": {
"json_body": {
"data": [
{"id": "gpt-5-mini"},
{"id": "gpt-4.1-mini"}
]
}
}
}))
}
}),
);
let (execution_runtime_url, execution_runtime_handle) = start_server(execution_runtime).await;
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-openai", "openai", 10)],
vec![sample_endpoint(
"endpoint-openai-chat",
"provider-openai",
"openai:chat",
"https://api.openai.example/v1",
)],
vec![],
));
let gateway = build_router_with_state(
build_state_with_execution_runtime_override(execution_runtime_url)
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_repository_for_tests(
provider_catalog_repository.clone(),
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!(
"{gateway_url}/api/admin/endpoints/providers/provider-openai/keys"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.json(&json!({
"api_formats": ["openai:chat"],
"api_key": "sk-created-openai",
"name": "created key with auto fetch",
"auto_fetch_models": true
}))
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["provider_id"], "provider-openai");
assert_eq!(payload["name"], "created key with auto fetch");
assert_eq!(payload["auto_fetch_models"], true);
assert_eq!(
payload["allowed_models"],
json!(["gpt-4.1-mini", "gpt-5-mini"])
);
assert_eq!(payload["last_models_fetch_error"], serde_json::Value::Null);
assert_eq!(
*execution_runtime_hits.lock().expect("mutex should lock"),
1
);
let keys = provider_catalog_repository
.list_keys_by_provider_ids(&["provider-openai".to_string()])
.await
.expect("keys should read");
assert_eq!(keys.len(), 1);
assert_eq!(keys[0].name, "created key with auto fetch");
assert!(keys[0].auto_fetch_models);
assert_eq!(
keys[0].allowed_models,
Some(json!(["gpt-4.1-mini", "gpt-5-mini"]))
);
gateway_handle.abort();
execution_runtime_handle.abort();
}
#[tokio::test]
async fn gateway_reveals_admin_provider_key_locally_with_trusted_admin_principal() {
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/endpoints/keys/key-openai-a/reveal",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-openai", "openai", 10)],
vec![],
vec![sample_key(
"key-openai-a",
"provider-openai",
"openai:chat",
"sk-test-a",
)],
));
let (upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_repository_for_tests(
provider_catalog_repository,
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.get(format!(
"{gateway_url}/api/admin/endpoints/keys/key-openai-a/reveal"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["auth_type"], "api_key");
assert_eq!(payload["api_key"], "sk-test-a");
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_exports_admin_provider_key_locally_with_trusted_admin_principal() {
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/endpoints/keys/key-kiro-a/export",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let mut key = sample_key(
"key-kiro-a",
"provider-kiro",
"claude:messages",
"oauth-access-token",
);
key.auth_type = "oauth".to_string();
key.encrypted_auth_config = Some(
encrypt_python_fernet_plaintext(
DEVELOPMENT_ENCRYPTION_KEY,
r#"{"provider_type":"kiro","auth_method":"idc","refresh_token":"rt-kiro-123"}"#,
)
.expect("auth config ciphertext should build"),
);
key.upstream_metadata = Some(json!({"kiro": {"email": "[email protected]"}}));
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-kiro", "kiro", 10)],
vec![],
vec![key],
));
let (upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_reader_for_tests(
provider_catalog_repository,
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.get(format!(
"{gateway_url}/api/admin/endpoints/keys/key-kiro-a/export"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["provider_type"], "kiro");
assert_eq!(payload["auth_method"], "idc");
assert_eq!(payload["refresh_token"], "rt-kiro-123");
assert_eq!(payload["email"], "[email protected]");
assert_eq!(payload["name"], "default");
assert!(payload.get("exported_at").is_some());
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_exports_admin_provider_key_access_token_when_refresh_token_is_missing() {
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/endpoints/keys/key-codex-a/export",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let mut key = sample_key(
"key-codex-a",
"provider-codex",
"openai:responses",
"codex-access-token",
);
key.auth_type = "oauth".to_string();
key.encrypted_auth_config = Some(
encrypt_python_fernet_plaintext(
DEVELOPMENT_ENCRYPTION_KEY,
r#"{"provider_type":"codex","email":"[email protected]","updated_at":1710000000}"#,
)
.expect("auth config ciphertext should build"),
);
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-codex", "codex", 10)],
vec![],
vec![key],
));
let (upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_reader_for_tests(
provider_catalog_repository,
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.get(format!(
"{gateway_url}/api/admin/endpoints/keys/key-codex-a/export"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["provider_type"], "codex");
assert_eq!(payload["email"], "[email protected]");
assert_eq!(payload["access_token"], "codex-access-token");
assert!(payload.get("refresh_token").is_none());
assert!(payload.get("updated_at").is_none());
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_export_does_not_emit_access_token_from_imported_authorization_header() {
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/endpoints/keys/key-codex-a/export",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let mut key = sample_key(
"key-codex-a",
"provider-codex",
"openai:responses",
"imported-session-token",
);
key.auth_type = "oauth".to_string();
key.encrypted_auth_config = Some(
encrypt_python_fernet_plaintext(
DEVELOPMENT_ENCRYPTION_KEY,
r#"{"provider_type":"codex","email":"[email protected]","headers":{"authorization":"Bearer imported-session-token"}}"#,
)
.expect("auth config ciphertext should build"),
);
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-codex", "codex", 10)],
vec![],
vec![key],
));
let (upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_reader_for_tests(
provider_catalog_repository,
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.get(format!(
"{gateway_url}/api/admin/endpoints/keys/key-codex-a/export"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["provider_type"], "codex");
assert_eq!(payload["email"], "[email protected]");
assert_eq!(
payload["headers"]["authorization"],
"Bearer imported-session-token"
);
assert!(payload.get("access_token").is_none());
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_export_preserves_distinct_imported_access_token_with_authorization_header() {
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/endpoints/keys/key-codex-a/export",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let mut key = sample_key(
"key-codex-a",
"provider-codex",
"openai:responses",
"jwt-access-token",
);
key.auth_type = "oauth".to_string();
key.encrypted_auth_config = Some(
encrypt_python_fernet_plaintext(
DEVELOPMENT_ENCRYPTION_KEY,
r#"{"provider_type":"codex","email":"[email protected]","access_token":"jwt-access-token","headers":{"authorization":"Bearer imported-session-token"}}"#,
)
.expect("auth config ciphertext should build"),
);
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-codex", "codex", 10)],
vec![],
vec![key],
));
let (_upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_reader_for_tests(
provider_catalog_repository,
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.get(format!(
"{gateway_url}/api/admin/endpoints/keys/key-codex-a/export"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["provider_type"], "codex");
assert_eq!(payload["access_token"], "jwt-access-token");
assert_eq!(
payload["headers"]["authorization"],
"Bearer imported-session-token"
);
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_generic_export_rejects_agent_identity_without_exposing_private_key() {
let private_key = "agent-private-key-must-not-leak";
let mut provider = sample_provider("provider-codex", "codex", 10);
provider.provider_type = "codex".to_string();
let mut key = sample_key(
"key-codex-agent",
"provider-codex",
"openai:responses",
"__placeholder__",
);
key.auth_type = "oauth".to_string();
key.encrypted_auth_config = Some(
encrypt_python_fernet_plaintext(
DEVELOPMENT_ENCRYPTION_KEY,
&json!({
"provider_type": "codex",
"auth_mode": "agentIdentity",
"agent_runtime_id": "runtime-must-not-leak",
"agent_private_key": private_key,
"task_id": "task-must-not-leak"
})
.to_string(),
)
.expect("Agent Identity auth config should encrypt"),
);
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![provider],
vec![],
vec![key],
));
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_reader_for_tests(
provider_catalog_repository,
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let client = reqwest::Client::new();
let reveal_response = client
.get(format!(
"{gateway_url}/api/admin/endpoints/keys/key-codex-agent/reveal"
))
.header(GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("reveal request should complete");
assert_eq!(reveal_response.status(), StatusCode::BAD_REQUEST);
let reveal_body = reveal_response
.text()
.await
.expect("reveal error body should read");
assert!(reveal_body.contains("专属 provider-oauth 管理面"));
assert!(!reveal_body.contains(private_key));
assert!(!reveal_body.contains("runtime-must-not-leak"));
assert!(!reveal_body.contains("task-must-not-leak"));
let export_response = client
.get(format!(
"{gateway_url}/api/admin/endpoints/keys/key-codex-agent/export"
))
.header(GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("export request should complete");
assert_eq!(export_response.status(), StatusCode::BAD_REQUEST);
let export_body = export_response
.text()
.await
.expect("export error body should read");
assert!(export_body.contains("专属 provider-oauth 管理面"));
assert!(!export_body.contains(private_key));
assert!(!export_body.contains("runtime-must-not-leak"));
assert!(!export_body.contains("task-must-not-leak"));
let list_response = client
.get(format!(
"{gateway_url}/api/admin/endpoints/providers/provider-codex/keys"
))
.header(GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("list request should complete");
assert_eq!(list_response.status(), StatusCode::OK);
let list_payload: serde_json::Value = list_response
.json()
.await
.expect("list body should be JSON");
let agent = list_payload
.as_array()
.and_then(|items| items.iter().find(|item| item["id"] == "key-codex-agent"))
.expect("Agent Identity key should be listed");
assert_eq!(agent["agent_identity"], true);
assert_eq!(agent["can_export_oauth"], false);
gateway_handle.abort();
}
#[tokio::test]
async fn gateway_clears_admin_provider_key_oauth_invalid_locally_with_trusted_admin_principal() {
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/endpoints/keys/key-openai-a/clear-oauth-invalid",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let mut key = sample_key(
"key-openai-a",
"provider-openai",
"openai:chat",
"sk-test-a",
);
key.auth_type = "oauth".to_string();
key.oauth_invalid_at_unix_secs = Some(1_710_000_000);
key.oauth_invalid_reason = Some("token expired".to_string());
key.expires_at_unix_secs = Some(1_900_000_000);
key.status_snapshot = Some(json!({
"oauth": {
"code": "invalid",
"label": "已失效",
"reason": "token expired",
"expires_at": 1_900_000_000u64,
"invalid_at": 1_710_000_000u64,
"source": "oauth_invalid",
"requires_reauth": true,
"expiring_soon": false
},
"account": {
"code": "ok",
"label": serde_json::Value::Null,
"reason": serde_json::Value::Null,
"blocked": false,
"source": serde_json::Value::Null,
"recoverable": false
},
"quota": {
"code": "unknown",
"label": serde_json::Value::Null,
"reason": serde_json::Value::Null,
"exhausted": false,
"usage_ratio": serde_json::Value::Null,
"updated_at": serde_json::Value::Null,
"reset_seconds": serde_json::Value::Null,
"plan_type": serde_json::Value::Null
}
}));
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-openai", "openai", 10)],
vec![],
vec![key],
));
let (upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_repository_for_tests(
provider_catalog_repository.clone(),
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!(
"{gateway_url}/api/admin/endpoints/keys/key-openai-a/clear-oauth-invalid"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["message"], "已清除 OAuth 失效标记");
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
let reloaded = provider_catalog_repository
.list_keys_by_ids(&["key-openai-a".to_string()])
.await
.expect("keys should read");
assert_eq!(reloaded.len(), 1);
assert_eq!(reloaded[0].oauth_invalid_at_unix_secs, None);
assert_eq!(reloaded[0].oauth_invalid_reason, None);
let oauth_snapshot = reloaded[0]
.status_snapshot
.as_ref()
.and_then(serde_json::Value::as_object)
.and_then(|snapshot| snapshot.get("oauth"))
.and_then(serde_json::Value::as_object)
.expect("oauth snapshot should exist");
assert_eq!(oauth_snapshot.get("code"), Some(&json!("valid")));
assert_eq!(oauth_snapshot.get("label"), Some(&json!("有效")));
assert_eq!(oauth_snapshot.get("reason"), Some(&serde_json::Value::Null));
assert_eq!(
oauth_snapshot.get("invalid_at"),
Some(&serde_json::Value::Null)
);
assert_eq!(oauth_snapshot.get("source"), Some(&json!("expires_at")));
assert_eq!(oauth_snapshot.get("requires_reauth"), Some(&json!(false)));
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_noops_admin_provider_key_oauth_invalid_clear_when_marker_absent() {
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/endpoints/keys/key-openai-a/clear-oauth-invalid",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-openai", "openai", 10)],
vec![],
vec![sample_key(
"key-openai-a",
"provider-openai",
"openai:chat",
"sk-test-a",
)],
));
let (upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_repository_for_tests(
provider_catalog_repository,
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!(
"{gateway_url}/api/admin/endpoints/keys/key-openai-a/clear-oauth-invalid"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["message"], "该 Key 当前无失效标记,无需清除");
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_updates_admin_provider_key_locally_with_trusted_admin_principal() {
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/endpoints/keys/key-openai-a",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let mut key = sample_key(
"key-openai-a",
"provider-openai",
"openai:chat",
"sk-test-a",
);
key.learned_rpm_limit = Some(88);
key.allowed_models = Some(json!(["gpt-4.1"]));
key.fingerprint = Some(json!({"user_agent": "old-ua"}));
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-openai", "openai", 10)],
vec![],
vec![key],
));
let (upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_repository_for_tests(
provider_catalog_repository.clone(),
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.put(format!(
"{gateway_url}/api/admin/endpoints/keys/key-openai-a"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.json(&json!({
"api_key": "sk-updated-openai",
"name": "updated key",
"internal_priority": 15,
"rpm_limit": null,
"allowed_models": [],
"note": "updated from rust",
"is_active": false,
"fingerprint": {"user_agent": "new-ua"}
}))
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["name"], "updated key");
assert_eq!(payload["internal_priority"], 15);
assert_eq!(payload["rpm_limit"], serde_json::Value::Null);
assert_eq!(payload["learned_rpm_limit"], serde_json::Value::Null);
assert_eq!(payload["allowed_models"], json!([]));
assert_eq!(payload["note"], "updated from rust");
assert_eq!(payload["is_active"], false);
assert_eq!(payload["fingerprint"]["user_agent"], "new-ua");
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
let reloaded = provider_catalog_repository
.list_keys_by_ids(&["key-openai-a".to_string()])
.await
.expect("keys should read");
assert_eq!(reloaded.len(), 1);
assert_eq!(reloaded[0].name, "updated key");
assert_eq!(reloaded[0].internal_priority, 15);
assert_eq!(reloaded[0].rpm_limit, None);
assert_eq!(reloaded[0].learned_rpm_limit, None);
assert_eq!(reloaded[0].allowed_models, None);
assert_eq!(reloaded[0].note.as_deref(), Some("updated from rust"));
assert!(!reloaded[0].is_active);
let decrypted = decrypt_python_fernet_ciphertext(
DEVELOPMENT_ENCRYPTION_KEY,
reloaded[0]
.encrypted_api_key
.as_deref()
.expect("api key should be present"),
)
.expect("ciphertext should decrypt");
assert_eq!(decrypted, "sk-updated-openai");
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn provider_key_concurrent_limit_update_presence_semantics() {
let mut key = sample_key(
"key-openai-a",
"provider-openai",
"openai:chat",
"sk-test-a",
);
key.concurrent_limit = Some(4);
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-openai", "openai", 10)],
vec![],
vec![key],
));
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_repository_for_tests(
provider_catalog_repository.clone(),
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let omitted_response = reqwest::Client::new()
.put(format!(
"{gateway_url}/api/admin/endpoints/keys/key-openai-a"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.json(&json!({
"name": "renamed without concurrency"
}))
.send()
.await
.expect("request should succeed");
assert_eq!(omitted_response.status(), StatusCode::OK);
let omitted_payload: serde_json::Value = omitted_response
.json()
.await
.expect("json body should parse");
assert_eq!(omitted_payload["name"], "renamed without concurrency");
assert_eq!(omitted_payload["concurrent_limit"], 4);
let set_response = reqwest::Client::new()
.put(format!(
"{gateway_url}/api/admin/endpoints/keys/key-openai-a"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.json(&json!({
"concurrent_limit": 7
}))
.send()
.await
.expect("request should succeed");
assert_eq!(set_response.status(), StatusCode::OK);
let set_payload: serde_json::Value = set_response.json().await.expect("json body should parse");
assert_eq!(set_payload["concurrent_limit"], 7);
let clear_response = reqwest::Client::new()
.put(format!(
"{gateway_url}/api/admin/endpoints/keys/key-openai-a"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.json(&json!({
"concurrent_limit": null
}))
.send()
.await
.expect("request should succeed");
assert_eq!(clear_response.status(), StatusCode::OK);
let clear_payload: serde_json::Value =
clear_response.json().await.expect("json body should parse");
assert_eq!(clear_payload["concurrent_limit"], serde_json::Value::Null);
let negative_response = reqwest::Client::new()
.put(format!(
"{gateway_url}/api/admin/endpoints/keys/key-openai-a"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.json(&json!({
"concurrent_limit": -1
}))
.send()
.await
.expect("request should succeed");
assert_eq!(negative_response.status(), StatusCode::BAD_REQUEST);
let negative_payload: serde_json::Value = negative_response
.json()
.await
.expect("json body should parse");
assert!(negative_payload["detail"]
.as_str()
.expect("detail should be string")
.contains("concurrent_limit"));
let reloaded = provider_catalog_repository
.list_keys_by_ids(&["key-openai-a".to_string()])
.await
.expect("keys should read");
assert_eq!(reloaded.len(), 1);
assert_eq!(reloaded[0].name, "renamed without concurrency");
assert_eq!(reloaded[0].concurrent_limit, None);
gateway_handle.abort();
}
#[test]
fn gateway_clears_allowed_models_when_disabling_auto_fetch_on_provider_key_update() {
run_provider_keys_test(
"gateway_clears_allowed_models_when_disabling_auto_fetch_on_provider_key_update",
gateway_clears_allowed_models_when_disabling_auto_fetch_on_provider_key_update_impl,
);
}
async fn gateway_clears_allowed_models_when_disabling_auto_fetch_on_provider_key_update_impl() {
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/endpoints/keys/key-openai-a",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let mut key = sample_key(
"key-openai-a",
"provider-openai",
"openai:chat",
"sk-test-a",
);
key.auto_fetch_models = true;
key.allowed_models = Some(json!(["gpt-5", "gpt-4.1-mini"]));
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-openai", "openai", 10)],
vec![],
vec![key],
));
let (_upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_repository_for_tests(
provider_catalog_repository.clone(),
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.put(format!(
"{gateway_url}/api/admin/endpoints/keys/key-openai-a"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.json(&json!({
"auto_fetch_models": false
}))
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["auto_fetch_models"], false);
assert_eq!(payload["allowed_models"], json!([]));
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
let reloaded = provider_catalog_repository
.list_keys_by_ids(&["key-openai-a".to_string()])
.await
.expect("keys should read");
assert_eq!(reloaded.len(), 1);
assert!(!reloaded[0].auto_fetch_models);
assert_eq!(reloaded[0].allowed_models, None);
gateway_handle.abort();
upstream_handle.abort();
}
#[test]
fn gateway_overwrites_allowed_models_immediately_when_enabling_auto_fetch() {
run_provider_keys_test(
"gateway_overwrites_allowed_models_immediately_when_enabling_auto_fetch",
gateway_overwrites_allowed_models_immediately_when_enabling_auto_fetch_impl,
);
}
async fn gateway_overwrites_allowed_models_immediately_when_enabling_auto_fetch_impl() {
let execution_runtime_hits = Arc::new(Mutex::new(0usize));
let execution_runtime_hits_clone = Arc::clone(&execution_runtime_hits);
let execution_runtime = Router::new().route(
"/v1/execute/sync",
any(move |Json(plan): Json<ExecutionPlan>| {
let execution_runtime_hits_inner = Arc::clone(&execution_runtime_hits_clone);
async move {
*execution_runtime_hits_inner
.lock()
.expect("mutex should lock") += 1;
assert_eq!(plan.url, "https://api.openai.example/v1/models");
assert_eq!(
plan.headers.get("authorization").map(String::as_str),
Some("Bearer sk-test-a")
);
Json(json!({
"request_id": "req-update-key-auto-fetch",
"status_code": 200,
"headers": {
"content-type": "application/json"
},
"body": {
"json_body": {
"data": [
{"id": "gpt-5"},
{"id": "gpt-4.1"},
{"id": "gpt-o1"}
]
}
}
}))
}
}),
);
let (execution_runtime_url, execution_runtime_handle) = start_server(execution_runtime).await;
let mut key = sample_key(
"key-openai-a",
"provider-openai",
"openai:chat",
"sk-test-a",
);
key.auto_fetch_models = false;
key.allowed_models = Some(json!(["manual-a", "manual-b"]));
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-openai", "openai", 10)],
vec![sample_endpoint(
"endpoint-openai-chat",
"provider-openai",
"openai:chat",
"https://api.openai.example/v1",
)],
vec![key],
));
let gateway = build_router_with_state(
build_state_with_execution_runtime_override(execution_runtime_url)
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_repository_for_tests(
provider_catalog_repository.clone(),
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.put(format!(
"{gateway_url}/api/admin/endpoints/keys/key-openai-a"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.json(&json!({
"auto_fetch_models": true
}))
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["auto_fetch_models"], true);
assert_eq!(
payload["allowed_models"],
json!(["gpt-4.1", "gpt-5", "gpt-o1"])
);
assert_eq!(payload["last_models_fetch_error"], serde_json::Value::Null);
assert_eq!(
*execution_runtime_hits.lock().expect("mutex should lock"),
1
);
let reloaded = provider_catalog_repository
.list_keys_by_ids(&["key-openai-a".to_string()])
.await
.expect("keys should read");
assert_eq!(reloaded.len(), 1);
assert!(reloaded[0].auto_fetch_models);
assert_eq!(
reloaded[0].allowed_models,
Some(json!(["gpt-4.1", "gpt-5", "gpt-o1"]))
);
assert_eq!(reloaded[0].locked_models, None);
gateway_handle.abort();
execution_runtime_handle.abort();
}
#[test]
fn gateway_fetches_allowed_models_immediately_when_enabling_auto_fetch_from_empty_state() {
run_provider_keys_test(
"gateway_fetches_allowed_models_immediately_when_enabling_auto_fetch_from_empty_state",
gateway_fetches_allowed_models_immediately_when_enabling_auto_fetch_from_empty_state_impl,
);
}
async fn gateway_fetches_allowed_models_immediately_when_enabling_auto_fetch_from_empty_state_impl()
{
let execution_runtime_hits = Arc::new(Mutex::new(0usize));
let execution_runtime_hits_clone = Arc::clone(&execution_runtime_hits);
let execution_runtime = Router::new().route(
"/v1/execute/sync",
any(move |Json(plan): Json<ExecutionPlan>| {
let execution_runtime_hits_inner = Arc::clone(&execution_runtime_hits_clone);
async move {
*execution_runtime_hits_inner
.lock()
.expect("mutex should lock") += 1;
assert_eq!(plan.url, "https://api.openai.example/v1/models");
Json(json!({
"request_id": "req-update-key-auto-fetch-empty",
"status_code": 200,
"headers": {
"content-type": "application/json"
},
"body": {
"json_body": {
"data": [
{"id": "gpt-5-mini"},
{"id": "gpt-4.1-nano"}
]
}
}
}))
}
}),
);
let (execution_runtime_url, execution_runtime_handle) = start_server(execution_runtime).await;
let mut key = sample_key(
"key-openai-a",
"provider-openai",
"openai:chat",
"sk-test-a",
);
key.auto_fetch_models = false;
key.allowed_models = None;
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-openai", "openai", 10)],
vec![sample_endpoint(
"endpoint-openai-chat",
"provider-openai",
"openai:chat",
"https://api.openai.example/v1",
)],
vec![key],
));
let gateway = build_router_with_state(
build_state_with_execution_runtime_override(execution_runtime_url)
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_repository_for_tests(
provider_catalog_repository.clone(),
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.put(format!(
"{gateway_url}/api/admin/endpoints/keys/key-openai-a"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.json(&json!({
"auto_fetch_models": true
}))
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["auto_fetch_models"], true);
assert_eq!(
payload["allowed_models"],
json!(["gpt-4.1-nano", "gpt-5-mini"])
);
assert_eq!(
*execution_runtime_hits.lock().expect("mutex should lock"),
1
);
let reloaded = provider_catalog_repository
.list_keys_by_ids(&["key-openai-a".to_string()])
.await
.expect("keys should read");
assert_eq!(reloaded.len(), 1);
assert!(reloaded[0].auto_fetch_models);
assert_eq!(
reloaded[0].allowed_models,
Some(json!(["gpt-4.1-nano", "gpt-5-mini"]))
);
gateway_handle.abort();
execution_runtime_handle.abort();
}
#[test]
fn gateway_refreshes_allowed_models_when_updating_include_patterns_with_auto_fetch_enabled() {
run_provider_keys_test(
"gateway_refreshes_allowed_models_when_updating_include_patterns_with_auto_fetch_enabled",
gateway_refreshes_allowed_models_when_updating_include_patterns_with_auto_fetch_enabled_impl,
);
}
async fn gateway_refreshes_allowed_models_when_updating_include_patterns_with_auto_fetch_enabled_impl(
) {
let execution_runtime_hits = Arc::new(Mutex::new(0usize));
let execution_runtime_hits_clone = Arc::clone(&execution_runtime_hits);
let execution_runtime = Router::new().route(
"/v1/execute/sync",
any(move |Json(plan): Json<ExecutionPlan>| {
let execution_runtime_hits_inner = Arc::clone(&execution_runtime_hits_clone);
async move {
*execution_runtime_hits_inner
.lock()
.expect("mutex should lock") += 1;
assert_eq!(plan.url, "https://api.openai.example/v1/models");
Json(json!({
"request_id": "req-update-key-include-patterns",
"status_code": 200,
"headers": {
"content-type": "application/json"
},
"body": {
"json_body": {
"data": [
{"id": "gpt-5"},
{"id": "gpt-4.1"},
{"id": "claude-3.7-sonnet"}
]
}
}
}))
}
}),
);
let (execution_runtime_url, execution_runtime_handle) = start_server(execution_runtime).await;
let mut key = sample_key(
"key-openai-a",
"provider-openai",
"openai:chat",
"sk-test-a",
);
key.auto_fetch_models = true;
key.allowed_models = Some(json!(["gpt-4.1", "gpt-5", "claude-3.7-sonnet"]));
key.model_include_patterns = Some(json!(["gpt-*", "claude-*"]));
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-openai", "openai", 10)],
vec![sample_endpoint(
"endpoint-openai-chat",
"provider-openai",
"openai:chat",
"https://api.openai.example/v1",
)],
vec![key],
));
let gateway = build_router_with_state(
build_state_with_execution_runtime_override(execution_runtime_url)
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_repository_for_tests(
provider_catalog_repository.clone(),
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.put(format!(
"{gateway_url}/api/admin/endpoints/keys/key-openai-a"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.json(&json!({
"model_include_patterns": ["gpt-5*"]
}))
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["auto_fetch_models"], true);
assert_eq!(payload["model_include_patterns"], json!(["gpt-5*"]));
assert_eq!(payload["allowed_models"], json!(["gpt-5"]));
assert_eq!(
*execution_runtime_hits.lock().expect("mutex should lock"),
1
);
let reloaded = provider_catalog_repository
.list_keys_by_ids(&["key-openai-a".to_string()])
.await
.expect("keys should read");
assert_eq!(reloaded.len(), 1);
assert!(reloaded[0].auto_fetch_models);
assert_eq!(reloaded[0].model_include_patterns, Some(json!(["gpt-5*"])));
assert_eq!(reloaded[0].allowed_models, Some(json!(["gpt-5"])));
gateway_handle.abort();
execution_runtime_handle.abort();
}
#[test]
fn gateway_refreshes_allowed_models_when_updating_exclude_patterns_with_auto_fetch_enabled() {
run_provider_keys_test(
"gateway_refreshes_allowed_models_when_updating_exclude_patterns_with_auto_fetch_enabled",
gateway_refreshes_allowed_models_when_updating_exclude_patterns_with_auto_fetch_enabled_impl,
);
}
async fn gateway_refreshes_allowed_models_when_updating_exclude_patterns_with_auto_fetch_enabled_impl(
) {
let execution_runtime_hits = Arc::new(Mutex::new(0usize));
let execution_runtime_hits_clone = Arc::clone(&execution_runtime_hits);
let execution_runtime = Router::new().route(
"/v1/execute/sync",
any(move |Json(plan): Json<ExecutionPlan>| {
let execution_runtime_hits_inner = Arc::clone(&execution_runtime_hits_clone);
async move {
*execution_runtime_hits_inner
.lock()
.expect("mutex should lock") += 1;
assert_eq!(plan.url, "https://api.openai.example/v1/models");
Json(json!({
"request_id": "req-update-key-exclude-patterns",
"status_code": 200,
"headers": {
"content-type": "application/json"
},
"body": {
"json_body": {
"data": [
{"id": "gpt-5"},
{"id": "gpt-4.1"},
{"id": "claude-3.7-sonnet"}
]
}
}
}))
}
}),
);
let (execution_runtime_url, execution_runtime_handle) = start_server(execution_runtime).await;
let mut key = sample_key(
"key-openai-a",
"provider-openai",
"openai:chat",
"sk-test-a",
);
key.auto_fetch_models = true;
key.allowed_models = Some(json!(["gpt-5", "gpt-4.1", "claude-3.7-sonnet"]));
key.model_exclude_patterns = Some(json!(["claude-*"]));
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-openai", "openai", 10)],
vec![sample_endpoint(
"endpoint-openai-chat",
"provider-openai",
"openai:chat",
"https://api.openai.example/v1",
)],
vec![key],
));
let gateway = build_router_with_state(
build_state_with_execution_runtime_override(execution_runtime_url)
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_repository_for_tests(
provider_catalog_repository.clone(),
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.put(format!(
"{gateway_url}/api/admin/endpoints/keys/key-openai-a"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.json(&json!({
"model_exclude_patterns": ["gpt-4*"]
}))
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["auto_fetch_models"], true);
assert_eq!(payload["model_exclude_patterns"], json!(["gpt-4*"]));
assert_eq!(
payload["allowed_models"],
json!(["claude-3.7-sonnet", "gpt-5"])
);
assert_eq!(
*execution_runtime_hits.lock().expect("mutex should lock"),
1
);
let reloaded = provider_catalog_repository
.list_keys_by_ids(&["key-openai-a".to_string()])
.await
.expect("keys should read");
assert_eq!(reloaded.len(), 1);
assert!(reloaded[0].auto_fetch_models);
assert_eq!(reloaded[0].model_exclude_patterns, Some(json!(["gpt-4*"])));
assert_eq!(
reloaded[0].allowed_models,
Some(json!(["claude-3.7-sonnet", "gpt-5"]))
);
gateway_handle.abort();
execution_runtime_handle.abort();
}
#[tokio::test]
async fn gateway_rejects_admin_provider_key_update_when_api_key_duplicates_existing_key() {
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/endpoints/keys/key-openai-a",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-openai", "openai", 10)],
vec![],
vec![
sample_key(
"key-openai-a",
"provider-openai",
"openai:chat",
"sk-test-a",
),
sample_key(
"key-openai-b",
"provider-openai",
"openai:chat",
"sk-test-b",
),
],
));
let (upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_repository_for_tests(
provider_catalog_repository,
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.put(format!(
"{gateway_url}/api/admin/endpoints/keys/key-openai-a"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.json(&json!({
"api_key": "sk-test-b"
}))
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert!(payload["detail"]
.as_str()
.expect("detail should be string")
.contains("该 API Key 已存在于当前 Provider 中"));
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_deletes_admin_provider_key_locally_with_trusted_admin_principal() {
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/endpoints/keys/key-openai-a",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-openai", "openai", 10)],
vec![],
vec![sample_key(
"key-openai-a",
"provider-openai",
"openai:chat",
"sk-test-a",
)],
));
let (upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_repository_for_tests(
provider_catalog_repository.clone(),
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.delete(format!(
"{gateway_url}/api/admin/endpoints/keys/key-openai-a"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["message"], "Key key-openai-a 已删除");
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
let reloaded = provider_catalog_repository
.list_keys_by_ids(&["key-openai-a".to_string()])
.await
.expect("keys should read");
assert!(reloaded.is_empty());
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_batch_deletes_admin_provider_keys_locally_with_trusted_admin_principal() {
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/endpoints/keys/batch-delete",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-openai", "openai", 10)],
vec![],
vec![
sample_key(
"key-openai-a",
"provider-openai",
"openai:chat",
"sk-test-a",
),
sample_key(
"key-openai-b",
"provider-openai",
"openai:chat",
"sk-test-b",
),
],
));
let (upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_repository_for_tests(
provider_catalog_repository.clone(),
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!(
"{gateway_url}/api/admin/endpoints/keys/batch-delete"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.json(&json!({
"ids": ["key-openai-a", "key-missing"]
}))
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["success_count"], 1);
assert_eq!(payload["failed_count"], 1);
assert_eq!(payload["failed"][0]["id"], "key-missing");
assert_eq!(payload["failed"][0]["error"], "not found");
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
let reloaded = provider_catalog_repository
.list_keys_by_provider_ids(&["provider-openai".to_string()])
.await
.expect("keys should read");
assert_eq!(reloaded.len(), 1);
assert_eq!(reloaded[0].id, "key-openai-b");
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_handles_admin_keys_grouped_by_format_locally_with_trusted_admin_principal() {
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/endpoints/keys/grouped-by-format",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let mut key_a = sample_key(
"key-openai-a",
"provider-openai",
"openai:chat",
"sk-test-a",
);
key_a.internal_priority = 10;
key_a.request_count = Some(12);
key_a.success_count = Some(9);
key_a.created_at_unix_ms = Some(1_711_000_000);
key_a.updated_at_unix_secs = Some(1_711_000_100);
key_a.capabilities = Some(json!({"cache_1h": true, "gemini_files": false}));
key_a.global_priority_by_format = Some(json!({"openai:chat": 3}));
key_a.health_by_format = Some(json!({"openai:chat": {"health_score": 0.8}}));
key_a.circuit_breaker_by_format = Some(json!({"openai:chat": {"open": false}}));
let mut key_b = sample_key(
"key-claude-a",
"provider-claude",
"claude:messages",
"sk-ant-a",
);
key_b.internal_priority = 20;
key_b.request_count = Some(2);
key_b.success_count = Some(1);
key_b.created_at_unix_ms = Some(1_711_100_000);
key_b.updated_at_unix_secs = Some(1_711_100_100);
let mut key_agent = sample_key(
"key-codex-agent",
"provider-codex",
"openai:responses",
"__placeholder__",
);
key_agent.auth_type = "oauth".to_string();
key_agent.encrypted_auth_config = Some(
encrypt_python_fernet_plaintext(
DEVELOPMENT_ENCRYPTION_KEY,
r#"{"provider_type":"codex","auth_mode":"agentIdentity","agent_runtime_id":"runtime-1","agent_private_key":"base64-private-key","task_id":"task-1"}"#,
)
.expect("Agent Identity auth config should encrypt"),
);
let mut codex_provider = sample_provider("provider-codex", "codex", 30);
codex_provider.provider_type = "codex".to_string();
let provider_catalog_repository = Arc::new(SummaryNullingProviderCatalogReadRepository::seed(
vec![
sample_provider("provider-openai", "openai", 10),
sample_provider("provider-claude", "claude", 20)
.with_transport_fields(false, false, true, None, None, None, None, None, None),
codex_provider,
],
vec![
sample_endpoint(
"endpoint-openai-chat",
"provider-openai",
"openai:chat",
"https://api.openai.example",
),
sample_endpoint(
"endpoint-claude-chat",
"provider-claude",
"claude:messages",
"https://api.claude.example",
),
sample_endpoint(
"endpoint-codex-responses",
"provider-codex",
"openai:responses",
"https://api.codex.example",
),
],
vec![key_a, key_b, key_agent],
));
let (upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_reader_for_tests(
provider_catalog_repository,
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.get(format!(
"{gateway_url}/api/admin/endpoints/keys/grouped-by-format"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["openai:chat"][0]["id"], "key-openai-a");
assert_eq!(payload["openai:chat"][0]["provider_name"], "openai");
assert_eq!(
payload["openai:chat"][0]["endpoint_base_url"],
"https://api.openai.example"
);
assert_eq!(payload["openai:chat"][0]["capabilities"], json!(["1h缓存"]));
assert_eq!(payload["openai:chat"][0]["format_priority"], 3);
assert_eq!(
payload["openai:chat"][0]["global_priority_by_format"]["openai:chat"],
3
);
assert_eq!(payload["openai:chat"][0]["internal_priority"], 10);
assert_eq!(payload["claude:messages"][0]["provider_active"], false);
let agent_item = payload["openai:responses"]
.as_array()
.and_then(|items| items.iter().find(|item| item["id"] == "key-codex-agent"))
.expect("Agent Identity key should be grouped");
assert_eq!(agent_item["api_key_masked"], "[Agent Identity]");
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
upstream_handle.abort();
}