i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-10 05:00:19 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
45c840b8d30612b51f6b375a9a10e23ff3097abd
Aether/apps/aether-gateway/src/tests/ai_execute/sync/gemini/cli.rs
T

2374 lines
98 KiB
Rust
Raw Blame History

use super::{
any, build_router_with_state, build_state_with_execution_runtime_override,
encrypt_python_fernet_plaintext, json, start_server, to_bytes, Arc, Body, Digest,
InMemoryAuthApiKeySnapshotRepository, InMemoryMinimalCandidateSelectionReadRepository,
InMemoryProviderCatalogReadRepository, InMemoryRequestCandidateRepository, Json, Mutex,
Request, RequestCandidateReadRepository, RequestCandidateStatus, Router, Sha256, StatusCode,
StoredAuthApiKeySnapshot, StoredMinimalCandidateSelectionRow, StoredProviderCatalogEndpoint,
StoredProviderCatalogKey, StoredProviderCatalogProvider, StoredProviderModelMapping,
DEVELOPMENT_ENCRYPTION_KEY, EXECUTION_PATH_EXECUTION_RUNTIME_SYNC, EXECUTION_PATH_HEADER,
TRACE_ID_HEADER,
};
const GEMINI_CLI_SYNC_TEST_STACK_BYTES: usize = 16 * 1024 * 1024;
fn run_gemini_cli_sync_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(GEMINI_CLI_SYNC_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("gemini cli sync test thread should spawn");
if let Err(payload) = handle.join() {
std::panic::resume_unwind(payload);
}
}
#[test]
fn gateway_executes_gemini_cli_sync_via_local_decision_gate_with_local_sync_decision() {
run_gemini_cli_sync_test(
"gateway_executes_gemini_cli_sync_via_local_decision_gate_with_local_sync_decision",
gateway_executes_gemini_cli_sync_via_local_decision_gate_with_local_sync_decision_impl,
);
}
async fn gateway_executes_gemini_cli_sync_via_local_decision_gate_with_local_sync_decision_impl() {
#[derive(Debug, Clone)]
struct SeenExecutionRuntimeSyncRequest {
trace_id: String,
url: String,
has_model_field: bool,
authorization: String,
exact_temperature: f64,
endpoint_tag: String,
metadata_mode: String,
metadata_source: String,
tool_config_present: bool,
proxy_node_id: String,
transport_profile_id: String,
}
fn hash_api_key(value: &str) -> String {
let mut hasher = Sha256::new();
hasher.update(value.as_bytes());
format!("{:x}", hasher.finalize())
}
fn sample_auth_snapshot(api_key_id: &str, user_id: &str) -> StoredAuthApiKeySnapshot {
StoredAuthApiKeySnapshot::new(
user_id.to_string(),
"alice".to_string(),
Some("alice@example.com".to_string()),
"user".to_string(),
"local".to_string(),
true,
false,
Some(serde_json::json!(["gemini"])),
Some(serde_json::json!(["gemini:generate_content"])),
Some(serde_json::json!(["gemini-cli"])),
api_key_id.to_string(),
Some("default".to_string()),
true,
false,
false,
Some(60),
Some(5),
Some(4_102_444_800),
Some(serde_json::json!(["gemini"])),
Some(serde_json::json!(["gemini:generate_content"])),
Some(serde_json::json!(["gemini-cli"])),
)
.expect("auth snapshot should build")
}
fn sample_candidate_row() -> StoredMinimalCandidateSelectionRow {
StoredMinimalCandidateSelectionRow {
provider_id: "provider-gemini-cli-local-1".to_string(),
provider_name: "gemini".to_string(),
provider_type: "custom".to_string(),
provider_priority: 10,
provider_is_active: true,
endpoint_id: "endpoint-gemini-cli-local-1".to_string(),
endpoint_api_format: "gemini:generate_content".to_string(),
endpoint_api_family: Some("gemini".to_string()),
endpoint_kind: Some("cli".to_string()),
endpoint_is_active: true,
key_id: "key-gemini-cli-local-1".to_string(),
key_name: "prod".to_string(),
key_auth_type: "bearer".to_string(),
key_is_active: true,
key_api_formats: Some(vec!["gemini:generate_content".to_string()]),
key_allowed_models: None,
key_capabilities: None,
key_internal_priority: 5,
key_global_priority_by_format: Some(serde_json::json!({"gemini:generate_content": 1})),
model_id: "model-gemini-cli-local-1".to_string(),
global_model_id: "global-model-gemini-cli-local-1".to_string(),
global_model_name: "gemini-cli".to_string(),
global_model_mappings: None,
global_model_supports_streaming: Some(true),
model_provider_model_name: "gemini-cli-upstream".to_string(),
model_provider_model_mappings: Some(vec![StoredProviderModelMapping {
name: "gemini-cli-upstream".to_string(),
priority: 1,
api_formats: Some(vec!["gemini:generate_content".to_string()]),
endpoint_ids: None,
operations: None,
}]),
model_supports_streaming: Some(true),
model_is_active: true,
model_is_available: true,
}
}
fn sample_provider_catalog_provider() -> StoredProviderCatalogProvider {
StoredProviderCatalogProvider::new(
"provider-gemini-cli-local-1".to_string(),
"gemini".to_string(),
Some("https://example.com".to_string()),
"custom".to_string(),
)
.expect("provider should build")
.with_transport_fields(
true,
false,
false,
None,
Some(2),
Some(serde_json::json!({"url":"http://provider-proxy.internal:8080"})),
Some(20.0),
None,
None,
)
}
fn sample_provider_catalog_endpoint() -> StoredProviderCatalogEndpoint {
StoredProviderCatalogEndpoint::new(
"endpoint-gemini-cli-local-1".to_string(),
"provider-gemini-cli-local-1".to_string(),
"gemini:generate_content".to_string(),
Some("gemini".to_string()),
Some("cli".to_string()),
true,
)
.expect("endpoint should build")
.with_transport_fields(
"https://generativelanguage.googleapis.com".to_string(),
Some(serde_json::json!([
{"action":"set","key":"x-endpoint-tag","value":"gemini-cli-local"}
])),
Some(serde_json::json!([
{"action":"set","path":"metadata.mode","value":"safe"},
{"action":"rename","from":"metadata.client","to":"metadata.source"},
{"action":"drop","path":"toolConfig"}
])),
Some(2),
Some("/custom/v1beta/models/gemini-cli-upstream:generateContent".to_string()),
None,
None,
None,
)
.expect("endpoint transport should build")
}
fn sample_provider_catalog_key() -> StoredProviderCatalogKey {
StoredProviderCatalogKey::new(
"key-gemini-cli-local-1".to_string(),
"provider-gemini-cli-local-1".to_string(),
"prod".to_string(),
"bearer".to_string(),
None,
true,
)
.expect("key should build")
.with_transport_fields(
Some(serde_json::json!(["gemini:generate_content"])),
encrypt_python_fernet_plaintext(DEVELOPMENT_ENCRYPTION_KEY, "sk-upstream-gemini-cli")
.expect("api key should encrypt"),
None,
None,
Some(serde_json::json!({"gemini:generate_content": 1})),
None,
None,
Some(serde_json::json!({"enabled": true, "node_id":"proxy-node-gemini-cli-local"})),
Some(serde_json::json!({"transport_profile":"chrome_136"})),
)
.expect("key transport should build")
}
let seen_execution_runtime = Arc::new(Mutex::new(None::<SeenExecutionRuntimeSyncRequest>));
let seen_execution_runtime_clone = Arc::clone(&seen_execution_runtime);
let seen_report = Arc::new(Mutex::new(false));
let seen_report_clone = Arc::clone(&seen_report);
let decision_hits = Arc::new(Mutex::new(0usize));
let decision_hits_clone = Arc::clone(&decision_hits);
let plan_hits = Arc::new(Mutex::new(0usize));
let plan_hits_clone = Arc::clone(&plan_hits);
let public_hits = Arc::new(Mutex::new(0usize));
let public_hits_clone = Arc::clone(&public_hits);
let upstream = Router::new()
.route(
"/api/internal/gateway/decision-sync",
any(move |_request: Request| {
let decision_hits_inner = Arc::clone(&decision_hits_clone);
async move {
*decision_hits_inner.lock().expect("mutex should lock") += 1;
Json(json!({"action": "proxy_public"}))
}
}),
)
.route(
"/api/internal/gateway/plan-sync",
any(move |_request: Request| {
let plan_hits_inner = Arc::clone(&plan_hits_clone);
async move {
*plan_hits_inner.lock().expect("mutex should lock") += 1;
Json(json!({"action": "proxy_public"}))
}
}),
)
.route(
"/api/internal/gateway/report-sync",
any(move |request: Request| {
let seen_report_inner = Arc::clone(&seen_report_clone);
async move {
let (_parts, body) = request.into_parts();
let _raw_body = to_bytes(body, usize::MAX).await.expect("body should read");
*seen_report_inner.lock().expect("mutex should lock") = true;
Json(json!({"ok": true}))
}
}),
)
.route(
"/v1beta/models/gemini-cli:generateContent",
any(move |_request: Request| {
let public_hits_inner = Arc::clone(&public_hits_clone);
async move {
*public_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::IM_A_TEAPOT, Body::from("public-route-hit"))
}
}),
);
let execution_runtime = Router::new().route(
"/v1/execute/sync",
any(move |request: Request| {
let seen_execution_runtime_inner = Arc::clone(&seen_execution_runtime_clone);
async move {
let (parts, body) = request.into_parts();
let raw_body = to_bytes(body, usize::MAX).await.expect("body should read");
let payload: serde_json::Value = serde_json::from_slice(&raw_body)
.expect("execution runtime payload should parse");
*seen_execution_runtime_inner
.lock()
.expect("mutex should lock") = Some(SeenExecutionRuntimeSyncRequest {
trace_id: parts
.headers
.get(TRACE_ID_HEADER)
.and_then(|value| value.to_str().ok())
.unwrap_or_default()
.to_string(),
url: payload
.get("url")
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
has_model_field: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("model"))
.is_some(),
authorization: payload
.get("headers")
.and_then(|value| value.get("authorization"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
exact_temperature: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("generationConfig"))
.and_then(|value| value.get("temperature"))
.and_then(|value| value.as_f64())
.unwrap_or_default(),
endpoint_tag: payload
.get("headers")
.and_then(|value| value.get("x-endpoint-tag"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
metadata_mode: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("metadata"))
.and_then(|value| value.get("mode"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
metadata_source: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("metadata"))
.and_then(|value| value.get("source"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
tool_config_present: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("toolConfig"))
.is_some(),
proxy_node_id: payload
.get("proxy")
.and_then(|value| value.get("node_id"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
transport_profile_id: payload
.get("transport_profile")
.and_then(|value| value.get("profile_id"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
});
Json(json!({
"request_id": "trace-gemini-cli-local-sync-123",
"status_code": 200,
"headers": {
"content-type": "application/json"
},
"body": {
"json_body": {
"candidates": [{
"content": {
"role": "model",
"parts": [{"text": "Hello from Gemini CLI"}]
},
"finishReason": "STOP"
}],
"usageMetadata": {
"promptTokenCount": 1,
"candidatesTokenCount": 2,
"totalTokenCount": 3
}
}
},
"telemetry": {
"elapsed_ms": 27
}
}))
}
}),
);
let auth_repository = Arc::new(InMemoryAuthApiKeySnapshotRepository::seed(vec![(
Some(hash_api_key("client-gemini-cli-local")),
sample_auth_snapshot("api-key-gemini-cli-local-1", "user-gemini-cli-local-1"),
)]));
let candidate_selection_repository =
Arc::new(InMemoryMinimalCandidateSelectionReadRepository::seed(vec![
sample_candidate_row(),
]));
let request_candidate_repository = Arc::new(InMemoryRequestCandidateRepository::default());
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider_catalog_provider()],
vec![sample_provider_catalog_endpoint()],
vec![sample_provider_catalog_key()],
));
let (upstream_url, upstream_handle) = start_server(upstream).await;
let (execution_runtime_url, execution_runtime_handle) = start_server(execution_runtime).await;
let gateway_state = build_state_with_execution_runtime_override(execution_runtime_url.clone())
.with_data_state_for_tests(
crate::data::GatewayDataState::with_auth_candidate_selection_provider_catalog_and_request_candidate_repository_for_tests(
auth_repository,
candidate_selection_repository,
provider_catalog_repository,
Arc::clone(&request_candidate_repository),
DEVELOPMENT_ENCRYPTION_KEY,
),
);
let gateway = build_router_with_state(gateway_state);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!(
"{gateway_url}/v1beta/models/gemini-cli:generateContent"
))
.header(http::header::CONTENT_TYPE, "application/json")
.header("user-agent", "GeminiCLI/1.0")
.header("x-goog-api-key", "client-gemini-cli-local")
.header(TRACE_ID_HEADER, "trace-gemini-cli-local-sync-123")
.body(
"{\"contents\":[],\"generationConfig\":{\"temperature\":0.2},\"metadata\":{\"client\":\"desktop-gemini-cli\"},\"toolConfig\":{\"functionCallingConfig\":{\"mode\":\"AUTO\"}}}",
)
.send()
.await
.expect("request should succeed");
let response_status = response.status();
let response_body = response.text().await.expect("body should read");
assert_eq!(
response_status,
StatusCode::OK,
"decision_hits={} plan_hits={} public_hits={} body={response_body}",
*decision_hits.lock().expect("mutex should lock"),
*plan_hits.lock().expect("mutex should lock"),
*public_hits.lock().expect("mutex should lock"),
);
let seen_execution_runtime_request = seen_execution_runtime
.lock()
.expect("mutex should lock")
.clone()
.expect("execution runtime sync should be captured");
assert_eq!(
seen_execution_runtime_request.trace_id,
"trace-gemini-cli-local-sync-123"
);
assert_eq!(
seen_execution_runtime_request.url,
"https://generativelanguage.googleapis.com/custom/v1beta/models/gemini-cli-upstream:generateContent"
);
assert!(!seen_execution_runtime_request.has_model_field);
assert_eq!(
seen_execution_runtime_request.authorization,
"Bearer sk-upstream-gemini-cli"
);
assert!((seen_execution_runtime_request.exact_temperature - 0.2).abs() < f64::EPSILON);
assert_eq!(
seen_execution_runtime_request.endpoint_tag,
"gemini-cli-local"
);
assert_eq!(seen_execution_runtime_request.metadata_mode, "safe");
assert_eq!(
seen_execution_runtime_request.metadata_source,
"desktop-gemini-cli"
);
assert!(!seen_execution_runtime_request.tool_config_present);
assert_eq!(
seen_execution_runtime_request.proxy_node_id,
"proxy-node-gemini-cli-local"
);
assert_eq!(
seen_execution_runtime_request.transport_profile_id,
"chrome_136"
);
let stored_candidates = request_candidate_repository
.list_by_request_id("trace-gemini-cli-local-sync-123")
.await
.expect("request candidate trace should read");
assert_eq!(stored_candidates.len(), 1);
assert_eq!(stored_candidates[0].status, RequestCandidateStatus::Success);
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
assert!(
!*seen_report.lock().expect("mutex should lock"),
"report-sync should stay local when request candidate persistence is available"
);
assert_eq!(*decision_hits.lock().expect("mutex should lock"), 0);
assert_eq!(*plan_hits.lock().expect("mutex should lock"), 0);
assert_eq!(*public_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
execution_runtime_handle.abort();
upstream_handle.abort();
}
#[test]
fn gateway_returns_gemini_cli_error_for_local_sync_failure() {
run_gemini_cli_sync_test(
"gateway_returns_gemini_cli_error_for_local_sync_failure",
gateway_returns_gemini_cli_error_for_local_sync_failure_impl,
);
}
async fn gateway_returns_gemini_cli_error_for_local_sync_failure_impl() {
fn hash_api_key(value: &str) -> String {
let mut hasher = Sha256::new();
hasher.update(value.as_bytes());
format!("{:x}", hasher.finalize())
}
fn sample_auth_snapshot(api_key_id: &str, user_id: &str) -> StoredAuthApiKeySnapshot {
StoredAuthApiKeySnapshot::new(
user_id.to_string(),
"alice".to_string(),
Some("alice@example.com".to_string()),
"user".to_string(),
"local".to_string(),
true,
false,
Some(serde_json::json!(["gemini"])),
Some(serde_json::json!(["gemini:generate_content"])),
Some(serde_json::json!(["gemini-cli"])),
api_key_id.to_string(),
Some("default".to_string()),
true,
false,
false,
Some(60),
Some(5),
Some(4_102_444_800),
Some(serde_json::json!(["gemini"])),
Some(serde_json::json!(["gemini:generate_content"])),
Some(serde_json::json!(["gemini-cli"])),
)
.expect("auth snapshot should build")
}
fn sample_candidate_row() -> StoredMinimalCandidateSelectionRow {
StoredMinimalCandidateSelectionRow {
provider_id: "provider-gemini-cli-local-1".to_string(),
provider_name: "gemini".to_string(),
provider_type: "custom".to_string(),
provider_priority: 10,
provider_is_active: true,
endpoint_id: "endpoint-gemini-cli-local-1".to_string(),
endpoint_api_format: "gemini:generate_content".to_string(),
endpoint_api_family: Some("gemini".to_string()),
endpoint_kind: Some("cli".to_string()),
endpoint_is_active: true,
key_id: "key-gemini-cli-local-1".to_string(),
key_name: "prod".to_string(),
key_auth_type: "bearer".to_string(),
key_is_active: true,
key_api_formats: Some(vec!["gemini:generate_content".to_string()]),
key_allowed_models: None,
key_capabilities: None,
key_internal_priority: 5,
key_global_priority_by_format: Some(serde_json::json!({"gemini:generate_content": 1})),
model_id: "model-gemini-cli-local-1".to_string(),
global_model_id: "global-model-gemini-cli-local-1".to_string(),
global_model_name: "gemini-cli".to_string(),
global_model_mappings: None,
global_model_supports_streaming: Some(true),
model_provider_model_name: "gemini-cli-upstream".to_string(),
model_provider_model_mappings: Some(vec![StoredProviderModelMapping {
name: "gemini-cli-upstream".to_string(),
priority: 1,
api_formats: Some(vec!["gemini:generate_content".to_string()]),
endpoint_ids: None,
operations: None,
}]),
model_supports_streaming: Some(true),
model_is_active: true,
model_is_available: true,
}
}
fn sample_provider_catalog_provider() -> StoredProviderCatalogProvider {
StoredProviderCatalogProvider::new(
"provider-gemini-cli-local-1".to_string(),
"gemini".to_string(),
Some("https://example.com".to_string()),
"custom".to_string(),
)
.expect("provider should build")
.with_transport_fields(
true,
false,
false,
None,
Some(2),
Some(serde_json::json!({"url":"http://provider-proxy.internal:8080"})),
Some(20.0),
None,
None,
)
}
fn sample_provider_catalog_endpoint() -> StoredProviderCatalogEndpoint {
StoredProviderCatalogEndpoint::new(
"endpoint-gemini-cli-local-1".to_string(),
"provider-gemini-cli-local-1".to_string(),
"gemini:generate_content".to_string(),
Some("gemini".to_string()),
Some("cli".to_string()),
true,
)
.expect("endpoint should build")
.with_transport_fields(
"https://generativelanguage.googleapis.com".to_string(),
Some(serde_json::json!([
{"action":"set","key":"x-endpoint-tag","value":"gemini-cli-local"}
])),
Some(serde_json::json!([
{"action":"set","path":"metadata.mode","value":"safe"},
{"action":"rename","from":"metadata.client","to":"metadata.source"},
{"action":"drop","path":"toolConfig"}
])),
Some(2),
Some("/custom/v1beta/models/gemini-cli-upstream:generateContent".to_string()),
None,
None,
None,
)
.expect("endpoint transport should build")
}
fn sample_provider_catalog_key() -> StoredProviderCatalogKey {
StoredProviderCatalogKey::new(
"key-gemini-cli-local-1".to_string(),
"provider-gemini-cli-local-1".to_string(),
"prod".to_string(),
"bearer".to_string(),
None,
true,
)
.expect("key should build")
.with_transport_fields(
Some(serde_json::json!(["gemini:generate_content"])),
encrypt_python_fernet_plaintext(DEVELOPMENT_ENCRYPTION_KEY, "sk-upstream-gemini-cli")
.expect("api key should encrypt"),
None,
None,
Some(serde_json::json!({"gemini:generate_content": 1})),
None,
None,
Some(serde_json::json!({"enabled": true, "node_id":"proxy-node-gemini-cli-local"})),
Some(serde_json::json!({"transport_profile":"chrome_136"})),
)
.expect("key transport should build")
}
let seen_report = Arc::new(Mutex::new(false));
let seen_report_clone = Arc::clone(&seen_report);
let upstream = Router::new().route(
"/api/internal/gateway/report-sync",
any(move |request: Request| {
let seen_report_inner = Arc::clone(&seen_report_clone);
async move {
let (_parts, body) = request.into_parts();
let _raw_body = to_bytes(body, usize::MAX).await.expect("body should read");
*seen_report_inner.lock().expect("mutex should lock") = true;
Json(json!({"ok": true}))
}
}),
);
let execution_runtime = Router::new().route(
"/v1/execute/sync",
any(move |_request: Request| async move {
Json(json!({
"request_id": "trace-gemini-cli-local-error-123",
"status_code": 200,
"headers": {
"content-type": "application/json"
},
"body": {
"json_body": {
"error": {
"message": "quota reached",
"status": "RESOURCE_EXHAUSTED"
}
}
},
"telemetry": {
"elapsed_ms": 27
}
}))
}),
);
let auth_repository = Arc::new(InMemoryAuthApiKeySnapshotRepository::seed(vec![(
Some(hash_api_key("client-gemini-cli-local-error")),
sample_auth_snapshot(
"api-key-gemini-cli-local-error-1",
"user-gemini-cli-local-error-1",
),
)]));
let candidate_selection_repository =
Arc::new(InMemoryMinimalCandidateSelectionReadRepository::seed(vec![
sample_candidate_row(),
]));
let request_candidate_repository = Arc::new(InMemoryRequestCandidateRepository::default());
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider_catalog_provider()],
vec![sample_provider_catalog_endpoint()],
vec![sample_provider_catalog_key()],
));
let (_upstream_url, upstream_handle) = start_server(upstream).await;
let (execution_runtime_url, execution_runtime_handle) = start_server(execution_runtime).await;
let gateway_state = build_state_with_execution_runtime_override(execution_runtime_url.clone())
.with_data_state_for_tests(
crate::data::GatewayDataState::with_auth_candidate_selection_provider_catalog_and_request_candidate_repository_for_tests(
auth_repository,
candidate_selection_repository,
provider_catalog_repository,
Arc::clone(&request_candidate_repository),
DEVELOPMENT_ENCRYPTION_KEY,
),
);
let gateway = build_router_with_state(gateway_state);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!("{gateway_url}/v1beta/models/gemini-cli:generateContent"))
.header(http::header::CONTENT_TYPE, "application/json")
.header("user-agent", "GeminiCLI/1.0")
.header("x-goog-api-key", "client-gemini-cli-local-error")
.header(TRACE_ID_HEADER, "trace-gemini-cli-local-error-123")
.body(
"{\"contents\":[],\"generationConfig\":{\"temperature\":0.2},\"metadata\":{\"client\":\"desktop-gemini-cli\"},\"toolConfig\":{\"functionCallingConfig\":{\"mode\":\"AUTO\"}}}",
)
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::TOO_MANY_REQUESTS);
assert_eq!(
response
.headers()
.get(EXECUTION_PATH_HEADER)
.and_then(|value| value.to_str().ok()),
Some(EXECUTION_PATH_EXECUTION_RUNTIME_SYNC)
);
let response_json: serde_json::Value = response.json().await.expect("body should parse");
assert_eq!(
response_json,
json!({
"error": {
"message": "quota reached",
"status": "RESOURCE_EXHAUSTED"
}
})
);
let stored_candidates = request_candidate_repository
.list_by_request_id("trace-gemini-cli-local-error-123")
.await
.expect("request candidate trace should read");
assert_eq!(stored_candidates.len(), 1);
assert_eq!(stored_candidates[0].status, RequestCandidateStatus::Failed);
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
assert!(
!*seen_report.lock().expect("mutex should lock"),
"report-sync should stay local when request candidate persistence is available"
);
gateway_handle.abort();
execution_runtime_handle.abort();
upstream_handle.abort();
}
#[test]
fn gateway_executes_gemini_cli_sync_via_local_decision_gate_after_oauth_refresh() {
run_gemini_cli_sync_test(
"gateway_executes_gemini_cli_sync_via_local_decision_gate_after_oauth_refresh",
gateway_executes_gemini_cli_sync_via_local_decision_gate_after_oauth_refresh_impl,
);
}
async fn gateway_executes_gemini_cli_sync_via_local_decision_gate_after_oauth_refresh_impl() {
#[derive(Debug, Clone)]
struct SeenExecutionRuntimeSyncRequest {
trace_id: String,
url: String,
project: String,
outer_model: String,
user_prompt_id: String,
inner_model_present: bool,
authorization: String,
exact_temperature: f64,
endpoint_tag: String,
metadata_mode: String,
metadata_source: String,
tool_config_present: bool,
proxy_node_id: String,
transport_profile_id: String,
}
#[derive(Debug, Clone)]
struct SeenRefreshRequest {
content_type: String,
body: String,
}
fn hash_api_key(value: &str) -> String {
let mut hasher = Sha256::new();
hasher.update(value.as_bytes());
format!("{:x}", hasher.finalize())
}
fn sample_auth_snapshot(api_key_id: &str, user_id: &str) -> StoredAuthApiKeySnapshot {
StoredAuthApiKeySnapshot::new(
user_id.to_string(),
"alice".to_string(),
Some("alice@example.com".to_string()),
"user".to_string(),
"local".to_string(),
true,
false,
Some(serde_json::json!(["gemini", "gemini_cli"])),
Some(serde_json::json!(["gemini:generate_content"])),
Some(serde_json::json!(["gemini-cli"])),
api_key_id.to_string(),
Some("default".to_string()),
true,
false,
false,
Some(60),
Some(5),
Some(4_102_444_800),
Some(serde_json::json!(["gemini", "gemini_cli"])),
Some(serde_json::json!(["gemini:generate_content"])),
Some(serde_json::json!(["gemini-cli"])),
)
.expect("auth snapshot should build")
}
fn sample_candidate_row() -> StoredMinimalCandidateSelectionRow {
StoredMinimalCandidateSelectionRow {
provider_id: "provider-gemini-cli-oauth-local-1".to_string(),
provider_name: "gemini_cli".to_string(),
provider_type: "gemini_cli".to_string(),
provider_priority: 10,
provider_is_active: true,
endpoint_id: "endpoint-gemini-cli-oauth-local-1".to_string(),
endpoint_api_format: "gemini:generate_content".to_string(),
endpoint_api_family: Some("gemini".to_string()),
endpoint_kind: Some("cli".to_string()),
endpoint_is_active: true,
key_id: "key-gemini-cli-oauth-local-1".to_string(),
key_name: "oauth".to_string(),
key_auth_type: "oauth".to_string(),
key_is_active: true,
key_api_formats: Some(vec!["gemini:generate_content".to_string()]),
key_allowed_models: None,
key_capabilities: None,
key_internal_priority: 5,
key_global_priority_by_format: Some(serde_json::json!({"gemini:generate_content": 1})),
model_id: "model-gemini-cli-oauth-local-1".to_string(),
global_model_id: "global-model-gemini-cli-oauth-local-1".to_string(),
global_model_name: "gemini-cli".to_string(),
global_model_mappings: None,
global_model_supports_streaming: Some(true),
model_provider_model_name: "gemini-cli-upstream".to_string(),
model_provider_model_mappings: Some(vec![StoredProviderModelMapping {
name: "gemini-cli-upstream".to_string(),
priority: 1,
api_formats: Some(vec!["gemini:generate_content".to_string()]),
endpoint_ids: None,
operations: None,
}]),
model_supports_streaming: Some(true),
model_is_active: true,
model_is_available: true,
}
}
fn sample_provider_catalog_provider() -> StoredProviderCatalogProvider {
StoredProviderCatalogProvider::new(
"provider-gemini-cli-oauth-local-1".to_string(),
"gemini_cli".to_string(),
Some("https://example.com".to_string()),
"gemini_cli".to_string(),
)
.expect("provider should build")
.with_transport_fields(
true,
false,
false,
None,
Some(2),
Some(serde_json::json!({"url":"http://provider-proxy.internal:8080"})),
Some(20.0),
None,
None,
)
}
fn sample_provider_catalog_endpoint() -> StoredProviderCatalogEndpoint {
StoredProviderCatalogEndpoint::new(
"endpoint-gemini-cli-oauth-local-1".to_string(),
"provider-gemini-cli-oauth-local-1".to_string(),
"gemini:generate_content".to_string(),
Some("gemini".to_string()),
Some("cli".to_string()),
true,
)
.expect("endpoint should build")
.with_transport_fields(
"https://cloudcode-pa.googleapis.com".to_string(),
Some(serde_json::json!([
{"action":"set","key":"x-endpoint-tag","value":"gemini-cli-oauth-local"}
])),
Some(serde_json::json!([
{"action":"set","path":"metadata.mode","value":"safe"},
{"action":"rename","from":"metadata.client","to":"metadata.source"},
{"action":"drop","path":"toolConfig"}
])),
Some(2),
None,
None,
None,
None,
)
.expect("endpoint transport should build")
}
fn sample_provider_catalog_key() -> StoredProviderCatalogKey {
let encrypted_auth_config = encrypt_python_fernet_plaintext(
DEVELOPMENT_ENCRYPTION_KEY,
r#"{"provider_type":"gemini_cli","refresh_token":"rt-gemini-cli-local-123","project_id":"gemini-cli-project-1"}"#,
)
.expect("auth config should encrypt");
StoredProviderCatalogKey::new(
"key-gemini-cli-oauth-local-1".to_string(),
"provider-gemini-cli-oauth-local-1".to_string(),
"oauth".to_string(),
"oauth".to_string(),
None,
true,
)
.expect("key should build")
.with_transport_fields(
Some(serde_json::json!(["gemini:generate_content"])),
encrypt_python_fernet_plaintext(DEVELOPMENT_ENCRYPTION_KEY, "__placeholder__")
.expect("placeholder api key should encrypt"),
Some(encrypted_auth_config),
None,
Some(serde_json::json!({"gemini:generate_content": 1})),
None,
None,
Some(
serde_json::json!({"enabled": true, "node_id":"proxy-node-gemini-cli-oauth-local"}),
),
Some(serde_json::json!({"transport_profile":"chrome_136"})),
)
.expect("key transport should build")
}
let seen_execution_runtime = Arc::new(Mutex::new(None::<SeenExecutionRuntimeSyncRequest>));
let seen_execution_runtime_clone = Arc::clone(&seen_execution_runtime);
let seen_report = Arc::new(Mutex::new(false));
let seen_report_clone = Arc::clone(&seen_report);
let seen_refresh = Arc::new(Mutex::new(None::<SeenRefreshRequest>));
let seen_refresh_clone = Arc::clone(&seen_refresh);
let decision_hits = Arc::new(Mutex::new(0usize));
let decision_hits_clone = Arc::clone(&decision_hits);
let plan_hits = Arc::new(Mutex::new(0usize));
let plan_hits_clone = Arc::clone(&plan_hits);
let public_hits = Arc::new(Mutex::new(0usize));
let public_hits_clone = Arc::clone(&public_hits);
let upstream = Router::new()
.route(
"/api/internal/gateway/decision-sync",
any(move |_request: Request| {
let decision_hits_inner = Arc::clone(&decision_hits_clone);
async move {
*decision_hits_inner.lock().expect("mutex should lock") += 1;
Json(json!({"action": "proxy_public"}))
}
}),
)
.route(
"/api/internal/gateway/plan-sync",
any(move |_request: Request| {
let plan_hits_inner = Arc::clone(&plan_hits_clone);
async move {
*plan_hits_inner.lock().expect("mutex should lock") += 1;
Json(json!({"action": "proxy_public"}))
}
}),
)
.route(
"/api/internal/gateway/report-sync",
any(move |request: Request| {
let seen_report_inner = Arc::clone(&seen_report_clone);
async move {
let (_parts, body) = request.into_parts();
let _raw_body = to_bytes(body, usize::MAX).await.expect("body should read");
*seen_report_inner.lock().expect("mutex should lock") = true;
Json(json!({"ok": true}))
}
}),
)
.route(
"/v1beta/models/gemini-cli:generateContent",
any(move |_request: Request| {
let public_hits_inner = Arc::clone(&public_hits_clone);
async move {
*public_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::IM_A_TEAPOT, Body::from("public-route-hit"))
}
}),
);
let refresh = Router::new().route(
"/oauth/token",
any(move |request: Request| {
let seen_refresh_inner = Arc::clone(&seen_refresh_clone);
async move {
let (parts, body) = request.into_parts();
let raw_body = to_bytes(body, usize::MAX).await.expect("body should read");
*seen_refresh_inner.lock().expect("mutex should lock") = Some(SeenRefreshRequest {
content_type: parts
.headers
.get(http::header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.unwrap_or_default()
.to_string(),
body: String::from_utf8(raw_body.to_vec())
.expect("refresh body should be utf8"),
});
Json(json!({
"access_token": "refreshed-gemini-cli-access-token",
"refresh_token": "rt-gemini-cli-local-456",
"token_type": "Bearer",
"expires_in": 3600
}))
}
}),
);
let execution_runtime = Router::new().route(
"/v1/execute/sync",
any(move |request: Request| {
let seen_execution_runtime_inner = Arc::clone(&seen_execution_runtime_clone);
async move {
let (parts, body) = request.into_parts();
let raw_body = to_bytes(body, usize::MAX).await.expect("body should read");
let payload: serde_json::Value = serde_json::from_slice(&raw_body)
.expect("execution runtime payload should parse");
*seen_execution_runtime_inner
.lock()
.expect("mutex should lock") = Some(SeenExecutionRuntimeSyncRequest {
trace_id: parts
.headers
.get(TRACE_ID_HEADER)
.and_then(|value| value.to_str().ok())
.unwrap_or_default()
.to_string(),
url: payload
.get("url")
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
outer_model: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("model"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
user_prompt_id: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("user_prompt_id"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
inner_model_present: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("request"))
.and_then(|value| value.get("model"))
.is_some(),
project: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("project"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
authorization: payload
.get("headers")
.and_then(|value| value.get("authorization"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
exact_temperature: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("request"))
.and_then(|value| value.get("generationConfig"))
.and_then(|value| value.get("temperature"))
.and_then(|value| value.as_f64())
.unwrap_or_default(),
endpoint_tag: payload
.get("headers")
.and_then(|value| value.get("x-endpoint-tag"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
metadata_mode: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("request"))
.and_then(|value| value.get("metadata"))
.and_then(|value| value.get("mode"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
metadata_source: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("request"))
.and_then(|value| value.get("metadata"))
.and_then(|value| value.get("source"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
tool_config_present: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("request"))
.and_then(|value| value.get("toolConfig"))
.is_some(),
proxy_node_id: payload
.get("proxy")
.and_then(|value| value.get("node_id"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
transport_profile_id: payload
.get("transport_profile")
.and_then(|value| value.get("profile_id"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
});
Json(json!({
"request_id": "trace-gemini-cli-oauth-local-sync-123",
"status_code": 200,
"headers": {
"content-type": "application/json"
},
"body": {
"json_body": {
"response": {
"candidates": [{
"content": {
"role": "model",
"parts": [{"text": "Hello from Gemini CLI"}]
},
"finishReason": "STOP"
}],
"usageMetadata": {
"promptTokenCount": 1,
"candidatesTokenCount": 2,
"totalTokenCount": 3
}
}
,"remainingCredits": 41,
"consumedCredits": 1,
"traceId": "trace-upstream-sync-1"
}
},
"telemetry": {
"elapsed_ms": 27
}
}))
}
}),
);
let client_api_key = "client-gemini-cli-oauth-local";
let auth_repository = Arc::new(InMemoryAuthApiKeySnapshotRepository::seed(vec![(
Some(hash_api_key(client_api_key)),
sample_auth_snapshot(
"api-key-gemini-cli-oauth-local-1",
"user-gemini-cli-oauth-local-1",
),
)]));
let candidate_selection_repository =
Arc::new(InMemoryMinimalCandidateSelectionReadRepository::seed(vec![
sample_candidate_row(),
]));
let request_candidate_repository = Arc::new(InMemoryRequestCandidateRepository::default());
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider_catalog_provider()],
vec![sample_provider_catalog_endpoint()],
vec![sample_provider_catalog_key()],
));
let (upstream_url, upstream_handle) = start_server(upstream).await;
let (refresh_url, refresh_handle) = start_server(refresh).await;
let (execution_runtime_url, execution_runtime_handle) = start_server(execution_runtime).await;
let oauth_refresh =
crate::provider_transport::LocalOAuthRefreshCoordinator::with_adapters_for_tests(vec![
Arc::new(
crate::provider_transport::oauth_refresh::GenericOAuthRefreshAdapter::default()
.with_token_url_for_tests("gemini_cli", format!("{refresh_url}/oauth/token")),
),
]);
let gateway_state = build_state_with_execution_runtime_override(execution_runtime_url.clone())
.with_data_state_for_tests(
crate::data::GatewayDataState::with_auth_candidate_selection_provider_catalog_and_request_candidate_repository_for_tests(
auth_repository,
candidate_selection_repository,
provider_catalog_repository,
Arc::clone(&request_candidate_repository),
DEVELOPMENT_ENCRYPTION_KEY,
),
)
.with_oauth_refresh_coordinator_for_tests(oauth_refresh);
let gateway = build_router_with_state(gateway_state);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!(
"{gateway_url}/v1beta/models/gemini-cli:generateContent"
))
.header(http::header::CONTENT_TYPE, "application/json")
.header("user-agent", "GeminiCLI/1.0")
.header("x-goog-api-key", client_api_key)
.header(TRACE_ID_HEADER, "trace-gemini-cli-oauth-local-sync-123")
.body(
"{\"contents\":[],\"generationConfig\":{\"temperature\":0.2},\"metadata\":{\"client\":\"desktop-gemini-cli\"},\"toolConfig\":{\"functionCallingConfig\":{\"mode\":\"AUTO\"}}}",
)
.send()
.await
.expect("request should succeed");
let response_status = response.status();
let response_body = response.text().await.expect("response body should read");
assert_eq!(response_status, StatusCode::OK, "body={response_body}");
let seen_refresh_request = seen_refresh
.lock()
.expect("mutex should lock")
.clone()
.expect("refresh request should be captured");
assert_eq!(
seen_refresh_request.content_type,
"application/x-www-form-urlencoded"
);
assert!(seen_refresh_request
.body
.contains("grant_type=refresh_token"));
assert!(seen_refresh_request.body.contains(
"client_id=681255809395-oo8ft2oprdrnp9e3aqf6av3hmdib135j.apps.googleusercontent.com"
));
assert!(seen_refresh_request
.body
.contains("client_secret=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"));
assert!(seen_refresh_request
.body
.contains("refresh_token=rt-gemini-cli-local-123"));
let seen_execution_runtime_request = seen_execution_runtime
.lock()
.expect("mutex should lock")
.clone()
.expect("execution runtime sync should be captured");
assert_eq!(
seen_execution_runtime_request.trace_id,
"trace-gemini-cli-oauth-local-sync-123"
);
assert_eq!(
seen_execution_runtime_request.url,
"https://cloudcode-pa.googleapis.com/v1internal:generateContent"
);
assert_eq!(
seen_execution_runtime_request.project,
"gemini-cli-project-1"
);
assert_eq!(
seen_execution_runtime_request.outer_model,
"gemini-cli-upstream"
);
assert_eq!(
seen_execution_runtime_request.user_prompt_id,
"trace-gemini-cli-oauth-local-sync-123"
);
assert!(!seen_execution_runtime_request.inner_model_present);
assert_eq!(
seen_execution_runtime_request.authorization,
"Bearer refreshed-gemini-cli-access-token"
);
assert!((seen_execution_runtime_request.exact_temperature - 0.2).abs() < f64::EPSILON);
assert_eq!(
seen_execution_runtime_request.endpoint_tag,
"gemini-cli-oauth-local"
);
assert_eq!(seen_execution_runtime_request.metadata_mode, "safe");
assert_eq!(
seen_execution_runtime_request.metadata_source,
"desktop-gemini-cli"
);
assert!(!seen_execution_runtime_request.tool_config_present);
assert_eq!(
seen_execution_runtime_request.proxy_node_id,
"proxy-node-gemini-cli-oauth-local"
);
assert_eq!(
seen_execution_runtime_request.transport_profile_id,
"chrome_136"
);
let stored_candidates = request_candidate_repository
.list_by_request_id("trace-gemini-cli-oauth-local-sync-123")
.await
.expect("request candidate trace should read");
assert_eq!(stored_candidates.len(), 1);
assert_eq!(stored_candidates[0].status, RequestCandidateStatus::Success);
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
assert!(
!*seen_report.lock().expect("mutex should lock"),
"report-sync should stay local when request candidate persistence is available"
);
assert_eq!(*decision_hits.lock().expect("mutex should lock"), 0);
assert_eq!(*plan_hits.lock().expect("mutex should lock"), 0);
assert_eq!(*public_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
execution_runtime_handle.abort();
refresh_handle.abort();
upstream_handle.abort();
}
#[test]
fn gateway_executes_vertex_ai_gemini_cli_sync_via_local_decision_gate_with_local_sync_decision() {
run_gemini_cli_sync_test(
"gateway_executes_vertex_ai_gemini_cli_sync_via_local_decision_gate_with_local_sync_decision",
gateway_executes_vertex_ai_gemini_cli_sync_via_local_decision_gate_with_local_sync_decision_impl,
);
}
async fn gateway_executes_vertex_ai_gemini_cli_sync_via_local_decision_gate_with_local_sync_decision_impl(
) {
#[derive(Debug, Clone)]
struct SeenExecutionRuntimeSyncRequest {
trace_id: String,
url: String,
authorization: String,
x_goog_api_key: String,
user_agent: String,
has_model_field: bool,
exact_temperature: f64,
endpoint_tag: String,
metadata_mode: String,
metadata_source: String,
tool_config_present: bool,
}
fn hash_api_key(value: &str) -> String {
let mut hasher = Sha256::new();
hasher.update(value.as_bytes());
format!("{:x}", hasher.finalize())
}
fn sample_auth_snapshot(api_key_id: &str, user_id: &str) -> StoredAuthApiKeySnapshot {
StoredAuthApiKeySnapshot::new(
user_id.to_string(),
"alice".to_string(),
Some("alice@example.com".to_string()),
"user".to_string(),
"local".to_string(),
true,
false,
Some(serde_json::json!(["gemini", "vertex"])),
Some(serde_json::json!(["gemini:generate_content"])),
Some(serde_json::json!(["gemini-cli"])),
api_key_id.to_string(),
Some("default".to_string()),
true,
false,
false,
Some(60),
Some(5),
Some(4_102_444_800),
Some(serde_json::json!(["gemini", "vertex"])),
Some(serde_json::json!(["gemini:generate_content"])),
Some(serde_json::json!(["gemini-cli"])),
)
.expect("auth snapshot should build")
}
fn sample_candidate_row() -> StoredMinimalCandidateSelectionRow {
StoredMinimalCandidateSelectionRow {
provider_id: "provider-vertex-cli-local-1".to_string(),
provider_name: "vertex".to_string(),
provider_type: "vertex_ai".to_string(),
provider_priority: 10,
provider_is_active: true,
endpoint_id: "endpoint-vertex-cli-local-1".to_string(),
endpoint_api_format: "gemini:generate_content".to_string(),
endpoint_api_family: Some("gemini".to_string()),
endpoint_kind: Some("cli".to_string()),
endpoint_is_active: true,
key_id: "key-vertex-cli-local-1".to_string(),
key_name: "prod".to_string(),
key_auth_type: "api_key".to_string(),
key_is_active: true,
key_api_formats: Some(vec!["gemini:generate_content".to_string()]),
key_allowed_models: None,
key_capabilities: None,
key_internal_priority: 5,
key_global_priority_by_format: Some(serde_json::json!({"gemini:generate_content": 1})),
model_id: "model-vertex-cli-local-1".to_string(),
global_model_id: "global-model-vertex-cli-local-1".to_string(),
global_model_name: "gemini-cli".to_string(),
global_model_mappings: None,
global_model_supports_streaming: Some(true),
model_provider_model_name: "gemini-cli-upstream".to_string(),
model_provider_model_mappings: Some(vec![StoredProviderModelMapping {
name: "gemini-cli-upstream".to_string(),
priority: 1,
api_formats: Some(vec!["gemini:generate_content".to_string()]),
endpoint_ids: None,
operations: None,
}]),
model_supports_streaming: Some(true),
model_is_active: true,
model_is_available: true,
}
}
fn sample_provider_catalog_provider() -> StoredProviderCatalogProvider {
StoredProviderCatalogProvider::new(
"provider-vertex-cli-local-1".to_string(),
"vertex".to_string(),
Some("https://example.com".to_string()),
"vertex_ai".to_string(),
)
.expect("provider should build")
.with_transport_fields(
true,
false,
false,
None,
Some(2),
None,
Some(20.0),
None,
None,
)
}
fn sample_provider_catalog_endpoint() -> StoredProviderCatalogEndpoint {
StoredProviderCatalogEndpoint::new(
"endpoint-vertex-cli-local-1".to_string(),
"provider-vertex-cli-local-1".to_string(),
"gemini:generate_content".to_string(),
Some("gemini".to_string()),
Some("cli".to_string()),
true,
)
.expect("endpoint should build")
.with_transport_fields(
"https://aiplatform.googleapis.com".to_string(),
Some(serde_json::json!([
{"action":"set","key":"x-endpoint-tag","value":"vertex-cli-local"}
])),
Some(serde_json::json!([
{"action":"set","path":"metadata.mode","value":"safe"},
{"action":"rename","from":"metadata.client","to":"metadata.source"},
{"action":"drop","path":"toolConfig"}
])),
Some(2),
None,
None,
None,
None,
)
.expect("endpoint transport should build")
}
fn sample_provider_catalog_key() -> StoredProviderCatalogKey {
let mut key = StoredProviderCatalogKey::new(
"key-vertex-cli-local-1".to_string(),
"provider-vertex-cli-local-1".to_string(),
"prod".to_string(),
"api_key".to_string(),
None,
true,
)
.expect("key should build")
.with_transport_fields(
Some(serde_json::json!(["gemini:generate_content"])),
encrypt_python_fernet_plaintext(DEVELOPMENT_ENCRYPTION_KEY, "vertex-upstream-secret")
.expect("api key should encrypt"),
None,
None,
Some(serde_json::json!({"gemini:generate_content": 1})),
None,
None,
None,
None,
)
.expect("key transport should build");
key.allow_auth_channel_mismatch_formats =
Some(serde_json::json!(["gemini:generate_content"]));
key
}
let seen_execution_runtime = Arc::new(Mutex::new(None::<SeenExecutionRuntimeSyncRequest>));
let seen_execution_runtime_clone = Arc::clone(&seen_execution_runtime);
let seen_report = Arc::new(Mutex::new(false));
let seen_report_clone = Arc::clone(&seen_report);
let decision_hits = Arc::new(Mutex::new(0usize));
let decision_hits_clone = Arc::clone(&decision_hits);
let plan_hits = Arc::new(Mutex::new(0usize));
let plan_hits_clone = Arc::clone(&plan_hits);
let public_hits = Arc::new(Mutex::new(0usize));
let public_hits_clone = Arc::clone(&public_hits);
let upstream = Router::new()
.route(
"/api/internal/gateway/decision-sync",
any(move |_request: Request| {
let decision_hits_inner = Arc::clone(&decision_hits_clone);
async move {
*decision_hits_inner.lock().expect("mutex should lock") += 1;
Json(json!({"action": "proxy_public"}))
}
}),
)
.route(
"/api/internal/gateway/plan-sync",
any(move |_request: Request| {
let plan_hits_inner = Arc::clone(&plan_hits_clone);
async move {
*plan_hits_inner.lock().expect("mutex should lock") += 1;
Json(json!({"action": "proxy_public"}))
}
}),
)
.route(
"/api/internal/gateway/report-sync",
any(move |request: Request| {
let seen_report_inner = Arc::clone(&seen_report_clone);
async move {
let (_parts, body) = request.into_parts();
let _raw_body = to_bytes(body, usize::MAX).await.expect("body should read");
*seen_report_inner.lock().expect("mutex should lock") = true;
Json(json!({"ok": true}))
}
}),
)
.route(
"/v1beta/models/gemini-cli:generateContent",
any(move |_request: Request| {
let public_hits_inner = Arc::clone(&public_hits_clone);
async move {
*public_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::IM_A_TEAPOT, Body::from("public-route-hit"))
}
}),
);
let execution_runtime = Router::new().route(
"/v1/execute/sync",
any(move |request: Request| {
let seen_execution_runtime_inner = Arc::clone(&seen_execution_runtime_clone);
async move {
let (parts, body) = request.into_parts();
let raw_body = to_bytes(body, usize::MAX).await.expect("body should read");
let payload: serde_json::Value = serde_json::from_slice(&raw_body)
.expect("execution runtime payload should parse");
*seen_execution_runtime_inner
.lock()
.expect("mutex should lock") = Some(SeenExecutionRuntimeSyncRequest {
trace_id: parts
.headers
.get(TRACE_ID_HEADER)
.and_then(|value| value.to_str().ok())
.unwrap_or_default()
.to_string(),
url: payload
.get("url")
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
authorization: payload
.get("headers")
.and_then(|value| value.get("authorization"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
x_goog_api_key: payload
.get("headers")
.and_then(|value| value.get("x-goog-api-key"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
user_agent: payload
.get("headers")
.and_then(|value| value.get("user-agent"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
has_model_field: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("model"))
.is_some(),
exact_temperature: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("generationConfig"))
.and_then(|value| value.get("temperature"))
.and_then(|value| value.as_f64())
.unwrap_or_default(),
endpoint_tag: payload
.get("headers")
.and_then(|value| value.get("x-endpoint-tag"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
metadata_mode: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("metadata"))
.and_then(|value| value.get("mode"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
metadata_source: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("metadata"))
.and_then(|value| value.get("source"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
tool_config_present: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("toolConfig"))
.is_some(),
});
Json(json!({
"request_id": "trace-vertex-cli-local-sync-123",
"status_code": 200,
"headers": {
"content-type": "application/json"
},
"body": {
"json_body": {
"candidates": [{
"content": {
"role": "model",
"parts": [{"text": "Hello from Vertex Gemini CLI"}]
},
"finishReason": "STOP"
}],
"usageMetadata": {
"promptTokenCount": 1,
"candidatesTokenCount": 2,
"totalTokenCount": 3
}
}
},
"telemetry": {
"elapsed_ms": 27
}
}))
}
}),
);
let auth_repository = Arc::new(InMemoryAuthApiKeySnapshotRepository::seed(vec![(
Some(hash_api_key("client-vertex-cli-local")),
sample_auth_snapshot("api-key-vertex-cli-local-1", "user-vertex-cli-local-1"),
)]));
let candidate_selection_repository =
Arc::new(InMemoryMinimalCandidateSelectionReadRepository::seed(vec![
sample_candidate_row(),
]));
let request_candidate_repository = Arc::new(InMemoryRequestCandidateRepository::default());
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider_catalog_provider()],
vec![sample_provider_catalog_endpoint()],
vec![sample_provider_catalog_key()],
));
let (upstream_url, upstream_handle) = start_server(upstream).await;
let (execution_runtime_url, execution_runtime_handle) = start_server(execution_runtime).await;
let gateway_state = build_state_with_execution_runtime_override(execution_runtime_url.clone())
.with_data_state_for_tests(
crate::data::GatewayDataState::with_auth_candidate_selection_provider_catalog_and_request_candidate_repository_for_tests(
auth_repository,
candidate_selection_repository,
provider_catalog_repository,
Arc::clone(&request_candidate_repository),
DEVELOPMENT_ENCRYPTION_KEY,
),
);
let gateway = build_router_with_state(gateway_state);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!(
"{gateway_url}/v1beta/models/gemini-cli:generateContent"
))
.header(http::header::CONTENT_TYPE, "application/json")
.header("user-agent", "GeminiCLI/1.0")
.header("x-goog-api-key", "client-vertex-cli-local")
.header(TRACE_ID_HEADER, "trace-vertex-cli-local-sync-123")
.body(
"{\"contents\":[],\"generationConfig\":{\"temperature\":0.2},\"metadata\":{\"client\":\"desktop-vertex-cli\"},\"toolConfig\":{\"functionCallingConfig\":{\"mode\":\"AUTO\"}}}",
)
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let seen_execution_runtime_request = seen_execution_runtime
.lock()
.expect("mutex should lock")
.clone()
.expect("execution runtime sync should be captured");
assert_eq!(
seen_execution_runtime_request.trace_id,
"trace-vertex-cli-local-sync-123"
);
assert_eq!(
seen_execution_runtime_request.url,
"https://aiplatform.googleapis.com/v1/publishers/google/models/gemini-cli-upstream:generateContent?key=vertex-upstream-secret"
);
assert!(seen_execution_runtime_request.authorization.is_empty());
assert!(seen_execution_runtime_request.x_goog_api_key.is_empty());
assert_eq!(seen_execution_runtime_request.user_agent, "GeminiCLI/1.0");
assert!(!seen_execution_runtime_request.has_model_field);
assert!((seen_execution_runtime_request.exact_temperature - 0.2).abs() < f64::EPSILON);
assert_eq!(
seen_execution_runtime_request.endpoint_tag,
"vertex-cli-local"
);
assert_eq!(seen_execution_runtime_request.metadata_mode, "safe");
assert_eq!(
seen_execution_runtime_request.metadata_source,
"desktop-vertex-cli"
);
assert!(!seen_execution_runtime_request.tool_config_present);
let stored_candidates = request_candidate_repository
.list_by_request_id("trace-vertex-cli-local-sync-123")
.await
.expect("request candidate trace should read");
assert_eq!(stored_candidates.len(), 1);
assert_eq!(stored_candidates[0].status, RequestCandidateStatus::Success);
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
assert!(
!*seen_report.lock().expect("mutex should lock"),
"report-sync should stay local when request candidate persistence is available"
);
assert_eq!(*decision_hits.lock().expect("mutex should lock"), 0);
assert_eq!(*plan_hits.lock().expect("mutex should lock"), 0);
assert_eq!(*public_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
execution_runtime_handle.abort();
upstream_handle.abort();
}
#[test]
fn gateway_executes_antigravity_gemini_cli_sync_via_local_decision_gate_after_oauth_refresh() {
run_gemini_cli_sync_test(
"gateway_executes_antigravity_gemini_cli_sync_via_local_decision_gate_after_oauth_refresh",
gateway_executes_antigravity_gemini_cli_sync_via_local_decision_gate_after_oauth_refresh_impl,
);
}
async fn gateway_executes_antigravity_gemini_cli_sync_via_local_decision_gate_after_oauth_refresh_impl(
) {
use base64::Engine as _;
#[derive(Debug, Clone)]
struct SeenExecutionRuntimeSyncRequest {
trace_id: String,
url: String,
accept: String,
authorization: String,
x_client_name: String,
x_client_version: String,
x_vscode_sessionid: String,
x_goog_api_client: String,
project: String,
request_id: String,
model: String,
user_agent: String,
request_type: String,
contents_len: usize,
exact_temperature: f64,
request_has_model: bool,
}
#[derive(Debug, Clone)]
struct SeenRefreshRequest {
content_type: String,
body: String,
}
fn hash_api_key(value: &str) -> String {
let mut hasher = Sha256::new();
hasher.update(value.as_bytes());
format!("{:x}", hasher.finalize())
}
fn sample_auth_snapshot(api_key_id: &str, user_id: &str) -> StoredAuthApiKeySnapshot {
StoredAuthApiKeySnapshot::new(
user_id.to_string(),
"alice".to_string(),
Some("alice@example.com".to_string()),
"user".to_string(),
"local".to_string(),
true,
false,
Some(serde_json::json!(["gemini", "antigravity"])),
Some(serde_json::json!(["gemini:generate_content"])),
Some(serde_json::json!(["gemini-cli"])),
api_key_id.to_string(),
Some("default".to_string()),
true,
false,
false,
Some(60),
Some(5),
Some(4_102_444_800),
Some(serde_json::json!(["gemini", "antigravity"])),
Some(serde_json::json!(["gemini:generate_content"])),
Some(serde_json::json!(["gemini-cli"])),
)
.expect("auth snapshot should build")
}
fn sample_candidate_row() -> StoredMinimalCandidateSelectionRow {
StoredMinimalCandidateSelectionRow {
provider_id: "provider-antigravity-cli-oauth-local-1".to_string(),
provider_name: "antigravity".to_string(),
provider_type: "antigravity".to_string(),
provider_priority: 10,
provider_is_active: true,
endpoint_id: "endpoint-antigravity-cli-oauth-local-1".to_string(),
endpoint_api_format: "gemini:generate_content".to_string(),
endpoint_api_family: Some("gemini".to_string()),
endpoint_kind: Some("cli".to_string()),
endpoint_is_active: true,
key_id: "key-antigravity-cli-oauth-local-1".to_string(),
key_name: "oauth".to_string(),
key_auth_type: "oauth".to_string(),
key_is_active: true,
key_api_formats: Some(vec!["gemini:generate_content".to_string()]),
key_allowed_models: None,
key_capabilities: None,
key_internal_priority: 5,
key_global_priority_by_format: Some(serde_json::json!({"gemini:generate_content": 1})),
model_id: "model-antigravity-cli-oauth-local-1".to_string(),
global_model_id: "global-model-antigravity-cli-oauth-local-1".to_string(),
global_model_name: "gemini-cli".to_string(),
global_model_mappings: None,
global_model_supports_streaming: Some(true),
model_provider_model_name: "claude-sonnet-4-5".to_string(),
model_provider_model_mappings: Some(vec![StoredProviderModelMapping {
name: "claude-sonnet-4-5".to_string(),
priority: 1,
api_formats: Some(vec!["gemini:generate_content".to_string()]),
endpoint_ids: None,
operations: None,
}]),
model_supports_streaming: Some(true),
model_is_active: true,
model_is_available: true,
}
}
fn sample_provider_catalog_provider() -> StoredProviderCatalogProvider {
StoredProviderCatalogProvider::new(
"provider-antigravity-cli-oauth-local-1".to_string(),
"antigravity".to_string(),
Some("https://example.com".to_string()),
"antigravity".to_string(),
)
.expect("provider should build")
.with_transport_fields(
true,
false,
false,
None,
Some(2),
None,
Some(20.0),
None,
None,
)
}
fn sample_provider_catalog_endpoint() -> StoredProviderCatalogEndpoint {
StoredProviderCatalogEndpoint::new(
"endpoint-antigravity-cli-oauth-local-1".to_string(),
"provider-antigravity-cli-oauth-local-1".to_string(),
"gemini:generate_content".to_string(),
Some("gemini".to_string()),
Some("cli".to_string()),
true,
)
.expect("endpoint should build")
.with_transport_fields(
"https://antigravity.googleapis.com".to_string(),
None,
None,
Some(2),
None,
None,
None,
None,
)
.expect("endpoint transport should build")
}
fn sample_provider_catalog_key() -> StoredProviderCatalogKey {
let encrypted_auth_config = encrypt_python_fernet_plaintext(
DEVELOPMENT_ENCRYPTION_KEY,
r#"{"provider_type":"antigravity","project_id":"project-antigravity-local-1","client_version":"1.2.3","session_id":"sess-antigravity-local-123","refresh_token":"rt-antigravity-cli-local-123"}"#,
)
.expect("auth config should encrypt");
StoredProviderCatalogKey::new(
"key-antigravity-cli-oauth-local-1".to_string(),
"provider-antigravity-cli-oauth-local-1".to_string(),
"oauth".to_string(),
"oauth".to_string(),
None,
true,
)
.expect("key should build")
.with_transport_fields(
Some(serde_json::json!(["gemini:generate_content"])),
encrypt_python_fernet_plaintext(DEVELOPMENT_ENCRYPTION_KEY, "__placeholder__")
.expect("placeholder api key should encrypt"),
Some(encrypted_auth_config),
None,
Some(serde_json::json!({"gemini:generate_content": 1})),
None,
None,
None,
None,
)
.expect("key transport should build")
}
let seen_execution_runtime = Arc::new(Mutex::new(None::<SeenExecutionRuntimeSyncRequest>));
let seen_execution_runtime_clone = Arc::clone(&seen_execution_runtime);
let seen_refresh = Arc::new(Mutex::new(None::<SeenRefreshRequest>));
let seen_refresh_clone = Arc::clone(&seen_refresh);
let seen_report = Arc::new(Mutex::new(false));
let seen_report_clone = Arc::clone(&seen_report);
let decision_hits = Arc::new(Mutex::new(0usize));
let decision_hits_clone = Arc::clone(&decision_hits);
let plan_hits = Arc::new(Mutex::new(0usize));
let plan_hits_clone = Arc::clone(&plan_hits);
let public_hits = Arc::new(Mutex::new(0usize));
let public_hits_clone = Arc::clone(&public_hits);
let finalize_hits = Arc::new(Mutex::new(0usize));
let finalize_hits_clone = Arc::clone(&finalize_hits);
let upstream = Router::new()
.route(
"/api/internal/gateway/decision-sync",
any(move |_request: Request| {
let decision_hits_inner = Arc::clone(&decision_hits_clone);
async move {
*decision_hits_inner.lock().expect("mutex should lock") += 1;
Json(json!({"action": "proxy_public"}))
}
}),
)
.route(
"/api/internal/gateway/plan-sync",
any(move |_request: Request| {
let plan_hits_inner = Arc::clone(&plan_hits_clone);
async move {
*plan_hits_inner.lock().expect("mutex should lock") += 1;
Json(json!({"action": "proxy_public"}))
}
}),
)
.route(
"/api/internal/gateway/report-sync",
any(move |request: Request| {
let seen_report_inner = Arc::clone(&seen_report_clone);
async move {
let (_parts, body) = request.into_parts();
let _raw_body = to_bytes(body, usize::MAX).await.expect("body should read");
*seen_report_inner.lock().expect("mutex should lock") = true;
Json(json!({"ok": true}))
}
}),
)
.route(
"/api/internal/gateway/finalize-sync",
any(move |_request: Request| {
let finalize_hits_inner = Arc::clone(&finalize_hits_clone);
async move {
*finalize_hits_inner.lock().expect("mutex should lock") += 1;
Json(json!({"unexpected": true}))
}
}),
)
.route(
"/v1beta/models/gemini-cli:generateContent",
any(move |_request: Request| {
let public_hits_inner = Arc::clone(&public_hits_clone);
async move {
*public_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::IM_A_TEAPOT, Body::from("public-route-hit"))
}
}),
);
let refresh = Router::new().route(
"/oauth/token",
any(move |request: Request| {
let seen_refresh_inner = Arc::clone(&seen_refresh_clone);
async move {
let (parts, body) = request.into_parts();
let raw_body = to_bytes(body, usize::MAX).await.expect("body should read");
*seen_refresh_inner.lock().expect("mutex should lock") = Some(SeenRefreshRequest {
content_type: parts
.headers
.get(http::header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.unwrap_or_default()
.to_string(),
body: String::from_utf8(raw_body.to_vec())
.expect("refresh body should be utf8"),
});
Json(json!({
"access_token": "refreshed-antigravity-cli-access-token",
"refresh_token": "rt-antigravity-cli-local-456",
"token_type": "Bearer",
"expires_in": 3600
}))
}
}),
);
let execution_runtime = Router::new().route(
"/v1/execute/sync",
any(move |request: Request| {
let seen_execution_runtime_inner = Arc::clone(&seen_execution_runtime_clone);
async move {
let (parts, body) = request.into_parts();
let raw_body = to_bytes(body, usize::MAX).await.expect("body should read");
let payload: serde_json::Value =
serde_json::from_slice(&raw_body).expect("execution runtime payload should parse");
*seen_execution_runtime_inner.lock().expect("mutex should lock") =
Some(SeenExecutionRuntimeSyncRequest {
trace_id: parts
.headers
.get(TRACE_ID_HEADER)
.and_then(|value| value.to_str().ok())
.unwrap_or_default()
.to_string(),
url: payload
.get("url")
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
accept: payload
.get("headers")
.and_then(|value| value.get("accept"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
authorization: payload
.get("headers")
.and_then(|value| value.get("authorization"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
x_client_name: payload
.get("headers")
.and_then(|value| value.get("x-client-name"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
x_client_version: payload
.get("headers")
.and_then(|value| value.get("x-client-version"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
x_vscode_sessionid: payload
.get("headers")
.and_then(|value| value.get("x-vscode-sessionid"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
x_goog_api_client: payload
.get("headers")
.and_then(|value| value.get("x-goog-api-client"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
project: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("project"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
request_id: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("requestId"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
model: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("model"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
user_agent: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("userAgent"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
request_type: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("requestType"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
contents_len: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("request"))
.and_then(|value| value.get("contents"))
.and_then(|value| value.as_array())
.map(Vec::len)
.unwrap_or_default(),
exact_temperature: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("request"))
.and_then(|value| value.get("generationConfig"))
.and_then(|value| value.get("temperature"))
.and_then(|value| value.as_f64())
.unwrap_or_default(),
request_has_model: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("request"))
.and_then(|value| value.get("model"))
.is_some(),
});
Json(json!({
"request_id": "trace-antigravity-cli-oauth-local-sync-123",
"status_code": 200,
"headers": {
"content-type": "text/event-stream"
},
"body": {
"body_bytes_b64": base64::engine::general_purpose::STANDARD.encode(
concat!(
"data: {\"response\":{\"candidates\":[{\"content\":{\"parts\":[{\"text\":\"Hello Antigravity CLI\"}],\"role\":\"model\"},\"finishReason\":\"STOP\",\"index\":0}],\"modelVersion\":\"claude-sonnet-4-5\",\"usageMetadata\":{\"promptTokenCount\":2,\"candidatesTokenCount\":3,\"totalTokenCount\":5}},\"responseId\":\"resp_antigravity_cli_local_sync_123\"}\n\n"
)
)
},
"telemetry": {
"elapsed_ms": 27
}
}))
}
}),
);
let client_api_key = "client-antigravity-cli-oauth-local";
let auth_repository = Arc::new(InMemoryAuthApiKeySnapshotRepository::seed(vec![(
Some(hash_api_key(client_api_key)),
sample_auth_snapshot(
"api-key-antigravity-cli-oauth-local-1",
"user-antigravity-cli-oauth-local-1",
),
)]));
let candidate_selection_repository =
Arc::new(InMemoryMinimalCandidateSelectionReadRepository::seed(vec![
sample_candidate_row(),
]));
let request_candidate_repository = Arc::new(InMemoryRequestCandidateRepository::default());
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider_catalog_provider()],
vec![sample_provider_catalog_endpoint()],
vec![sample_provider_catalog_key()],
));
let (upstream_url, upstream_handle) = start_server(upstream).await;
let (refresh_url, refresh_handle) = start_server(refresh).await;
let (execution_runtime_url, execution_runtime_handle) = start_server(execution_runtime).await;
let oauth_refresh =
crate::provider_transport::LocalOAuthRefreshCoordinator::with_adapters_for_tests(vec![
Arc::new(
crate::provider_transport::oauth_refresh::GenericOAuthRefreshAdapter::default()
.with_token_url_for_tests("antigravity", format!("{refresh_url}/oauth/token")),
),
]);
let gateway_state = build_state_with_execution_runtime_override(execution_runtime_url.clone())
.with_data_state_for_tests(
crate::data::GatewayDataState::with_auth_candidate_selection_provider_catalog_and_request_candidate_repository_for_tests(
auth_repository,
candidate_selection_repository,
provider_catalog_repository,
Arc::clone(&request_candidate_repository),
DEVELOPMENT_ENCRYPTION_KEY,
),
)
.with_oauth_refresh_coordinator_for_tests(oauth_refresh);
let gateway = build_router_with_state(gateway_state);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!(
"{gateway_url}/v1beta/models/gemini-cli:generateContent"
))
.header(http::header::CONTENT_TYPE, "application/json")
.header("user-agent", "GeminiCLI/1.0")
.header("x-goog-api-key", client_api_key)
.header(
TRACE_ID_HEADER,
"trace-antigravity-cli-oauth-local-sync-123",
)
.body("{\"contents\":[],\"generationConfig\":{\"temperature\":0.2}}")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let response_json: serde_json::Value = response.json().await.expect("body should parse");
assert_eq!(
response_json,
json!({
"responseId": "resp-local-stream",
"_v1internal_response_id": "resp_antigravity_cli_local_sync_123",
"candidates": [{
"content": {
"parts": [{"text": "Hello Antigravity CLI"}],
"role": "model"
},
"finishReason": "STOP",
"index": 0
}],
"modelVersion": "claude-sonnet-4-5",
"usageMetadata": {
"promptTokenCount": 2,
"candidatesTokenCount": 3,
"totalTokenCount": 5
}
})
);
let seen_refresh_request = seen_refresh
.lock()
.expect("mutex should lock")
.clone()
.expect("refresh request should be captured");
assert_eq!(
seen_refresh_request.content_type,
"application/x-www-form-urlencoded"
);
assert!(seen_refresh_request
.body
.contains("grant_type=refresh_token"));
assert!(seen_refresh_request.body.contains(
"client_id=1071006060591-tmhssin2h21lcre235vtolojh4g403ep.apps.googleusercontent.com"
));
assert!(seen_refresh_request
.body
.contains("client_secret=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"));
assert!(seen_refresh_request
.body
.contains("refresh_token=rt-antigravity-cli-local-123"));
let seen_execution_runtime_request = seen_execution_runtime
.lock()
.expect("mutex should lock")
.clone()
.expect("execution runtime sync should be captured");
assert_eq!(
seen_execution_runtime_request.trace_id,
"trace-antigravity-cli-oauth-local-sync-123"
);
assert_eq!(
seen_execution_runtime_request.url,
"https://antigravity.googleapis.com/v1internal:streamGenerateContent?alt=sse"
);
assert_eq!(seen_execution_runtime_request.accept, "text/event-stream");
assert_eq!(
seen_execution_runtime_request.authorization,
"Bearer refreshed-antigravity-cli-access-token"
);
assert_eq!(seen_execution_runtime_request.x_client_name, "antigravity");
assert_eq!(seen_execution_runtime_request.x_client_version, "4.3.0");
assert_eq!(
seen_execution_runtime_request.x_vscode_sessionid,
"sess-antigravity-local-123"
);
assert_eq!(
seen_execution_runtime_request.x_goog_api_client,
"gl-node/18.18.2 fire/0.8.6 grpc/1.10.x"
);
assert_eq!(
seen_execution_runtime_request.project,
"project-antigravity-local-1"
);
assert_eq!(
seen_execution_runtime_request.request_id,
"trace-antigravity-cli-oauth-local-sync-123"
);
assert_eq!(seen_execution_runtime_request.model, "claude-sonnet-4-5");
assert_eq!(
seen_execution_runtime_request.user_agent,
aether_provider_transport::antigravity::ANTIGRAVITY_REQUEST_USER_AGENT
);
assert_eq!(seen_execution_runtime_request.request_type, "agent");
assert_eq!(seen_execution_runtime_request.contents_len, 0);
assert!((seen_execution_runtime_request.exact_temperature - 0.2).abs() < f64::EPSILON);
assert!(!seen_execution_runtime_request.request_has_model);
let stored_candidates = request_candidate_repository
.list_by_request_id("trace-antigravity-cli-oauth-local-sync-123")
.await
.expect("request candidate trace should read");
assert_eq!(stored_candidates.len(), 1);
assert_eq!(stored_candidates[0].status, RequestCandidateStatus::Success);
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
assert!(
!*seen_report.lock().expect("mutex should lock"),
"report-sync should stay local when request candidate persistence is available"
);
assert_eq!(*decision_hits.lock().expect("mutex should lock"), 0);
assert_eq!(*plan_hits.lock().expect("mutex should lock"), 0);
assert_eq!(*public_hits.lock().expect("mutex should lock"), 0);
assert_eq!(*finalize_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
execution_runtime_handle.abort();
refresh_handle.abort();
upstream_handle.abort();
}
Reference in New Issue View Git Blame Copy Permalink