litellm-rs 0.9.0

use super::*;
use crate::server::HttpServer;

fn sync(key: u8) -> Arc<ConfigSync> {
    Arc::new(ConfigSync {
        key: vec![key; 32],
        status: Default::default(),
    })
}

#[tokio::test(flavor = "current_thread")]
async fn repeated_candidate_commit_failures_release_cache_tasks_and_keep_live_revision() {
    let state = state().await;
    let mut config = (*state.config()).clone();
    config.gateway.cache.enabled = true;
    state.apply_runtime_at(config.clone(), 0).await.unwrap();
    let active = state.pin_runtime();
    let cache = Arc::clone(active.response_cache.as_ref().unwrap());
    // A different node advances the authoritative row after this node pins
    // its generation. Every local build succeeds, then its real SQL CAS fails.
    let encrypted = state
        .config_sync
        .as_ref()
        .unwrap()
        .encrypt(&config, 2)
        .unwrap();
    state
        .storage
        .database
        .commit_runtime_config(1, encrypted)
        .await
        .unwrap();
    tokio::task::yield_now().await;
    let runtime = tokio::runtime::Handle::current();
    let baseline = runtime.metrics().num_alive_tasks();
    for _ in 0..100 {
        assert!(matches!(
            state.apply_runtime_at(config.clone(), 1).await,
            Err(GatewayError::Conflict(_))
        ));
        let live = state.pin_runtime();
        assert_eq!(live.generation, 1);
        assert!(Arc::ptr_eq(&active, &live));
        assert!(Arc::ptr_eq(&cache, live.response_cache.as_ref().unwrap()));
    }
    tokio::time::timeout(Duration::from_secs(2), async {
        while runtime.metrics().num_alive_tasks() > baseline {
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("failed candidates must not retain cache cleanup tasks");
    assert_eq!(
        state
            .storage
            .database
            .runtime_config()
            .await
            .unwrap()
            .revision,
        2
    );
    assert!(state.config_sync_status().last_sync_error.is_some());
}

async fn state() -> AppState {
    let mut config = crate::server::valid_test_config();
    config.gateway.storage.database.enabled = false;
    config.gateway.storage.redis.enabled = false;
    config.gateway.pricing.source = None;
    // These requests exercise config sync, not the process-wide metrics counters.
    config.gateway.monitoring.metrics.enabled = false;
    config.gateway.auth.enable_jwt = true;
    config.gateway.auth.jwt_secret = format!("Aa1!{}", uuid::Uuid::new_v4());
    config.gateway.auth.enable_api_key = false;
    config.gateway.auth.allow_anonymous = true;
    let server = HttpServer::new(&config).await.unwrap();
    let mut state = server.state().clone();
    state.config_sync = Some(sync(7));
    state
}

#[test]
fn snapshot_is_authenticated_and_excludes_node_local_settings() {
    let mut config = crate::server::valid_test_config();
    config.gateway.providers[0].api_key = "secret-provider-sentinel".into();
    let crypt = sync(7);
    let ciphertext = crypt.encrypt(&config, 4).unwrap();
    assert!(!ciphertext.contains("secret-provider-sentinel"));
    let json = decrypt_data(&crypt.key, &ciphertext).unwrap();
    assert!(json.contains("secret-provider-sentinel"));
    assert!(!json.contains("jwt_secret"));
    assert!(!json.contains("database"));
    let mut other = config.clone();
    other.gateway.server.port = 8123;
    let restored = crypt.candidate(&other, 4, &ciphertext).unwrap();
    assert_eq!(restored.gateway.server.port, 8123);
    assert_eq!(
        restored.gateway.providers[0].api_key,
        "secret-provider-sentinel"
    );
    assert!(crypt.candidate(&other, 5, &ciphertext).is_err());
    assert!(sync(9).candidate(&other, 4, &ciphertext).is_err());
}

#[tokio::test]
async fn two_nodes_converge_and_duplicates_do_not_rebuild() {
    let a = state().await;
    let mut b = state().await;
    b.storage = a.storage.clone();
    let mut candidate = (*a.config()).clone();
    candidate.gateway.providers[0].weight = 7.0;
    // Disabled Redis makes notification fail after the durable commit.
    assert!(a.apply_runtime(candidate).await.is_err());
    assert_eq!(a.pin_runtime().generation, 1);
    assert!(a.config_sync_status().last_sync_error.is_some());
    b.reconcile_runtime_config().await.unwrap();
    assert_eq!(b.pin_runtime().generation, 1);
    assert_eq!(b.config().gateway.providers[0].weight, 7.0);
    let pin = b.pin_runtime();
    b.reconcile_runtime_config().await.unwrap();
    assert!(Arc::ptr_eq(&pin, &b.pin_runtime()));
    assert!(matches!(
        b.apply_runtime_at((*b.config()).clone(), 0).await,
        Err(GatewayError::Conflict(_))
    ));
}

#[tokio::test]
async fn concurrent_database_writers_cannot_overwrite_a_new_revision() {
    let a = state().await;
    let mut b = state().await;
    b.storage = a.storage.clone();
    let mut first = (*a.config()).clone();
    first.gateway.providers[0].weight = 3.0;
    let mut second = (*b.config()).clone();
    second.gateway.providers[0].weight = 8.0;
    let (left, right) = tokio::join!(a.apply_runtime(first), b.apply_runtime(second));
    assert_eq!(
        [left, right]
            .iter()
            .filter(|r| matches!(r, Err(GatewayError::Conflict(_))))
            .count(),
        1
    );
    a.reconcile_runtime_config().await.unwrap();
    b.reconcile_runtime_config().await.unwrap();
    assert_eq!(a.pin_runtime().generation, 1);
    assert_eq!(b.pin_runtime().generation, 1);
    assert_eq!(
        a.config().gateway.providers[0].weight,
        b.config().gateway.providers[0].weight
    );
}

#[tokio::test]
async fn one_nodes_decryption_failure_keeps_healthy_nodes_active_and_is_visible() {
    let a = state().await;
    let mut b = state().await;
    b.storage = a.storage.clone();
    b.config_sync = Some(sync(9));
    let before = b.pin_runtime();
    assert!(a.apply_runtime((*a.config()).clone()).await.is_err());
    assert!(b.reconcile_runtime_config().await.is_err());
    assert!(Arc::ptr_eq(&before, &b.pin_runtime()));
    assert_eq!(a.pin_runtime().generation, 1);
    let status = b.config_sync_status();
    assert_eq!(status.observed_revision, 1);
    assert!(
        status
            .last_apply_error
            .unwrap()
            .contains("decryption failed")
    );
    b.config_sync = Some(sync(7));
    b.reconcile_runtime_config().await.unwrap();
    assert_eq!(b.pin_runtime().generation, 1);
    assert!(b.config_sync_status().last_apply_error.is_none());
}

#[tokio::test]
async fn invalid_candidate_never_changes_authoritative_revision() {
    let state = state().await;
    let mut candidate = (*state.config()).clone();
    candidate.gateway.providers[0].weight = -1.0;
    assert!(state.apply_runtime_at(candidate, 0).await.is_err());
    assert_eq!(
        state
            .storage
            .database
            .runtime_config()
            .await
            .unwrap()
            .revision,
        0
    );
    assert_eq!(state.pin_runtime().generation, 0);
    assert!(state.config_sync_status().last_apply_error.is_some());
}

#[tokio::test]
async fn revision_diagnostics_require_authentication_and_never_return_ciphertext() {
    use actix_web::{http::StatusCode, test, web};
    let state = state().await;
    let token = crate::server::test_admin_token(&state).await;
    let app = test::init_service(HttpServer::create_app(web::Data::new(state.clone()))).await;
    let response = test::call_service(
        &app,
        test::TestRequest::get()
            .uri("/admin/routing/revision")
            .insert_header(("Authorization", format!("Bearer {token}")))
            .to_request(),
    )
    .await;
    assert_eq!(response.status(), StatusCode::OK);
    let body: serde_json::Value = test::read_body_json(response).await;
    assert_eq!(body["active_revision"], 0);
    assert!(body.get("encrypted_payload").is_none());
    let mut config = (*state.config()).clone();
    config.gateway.auth.enable_api_key = true;
    state.config.store(config);
    let response = test::call_service(
        &app,
        test::TestRequest::get()
            .uri("/admin/routing/revision")
            .to_request(),
    )
    .await;
    assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
}

#[tokio::test]
async fn redis_broadcasts_only_ids_and_reconnect_reads_latest() {
    let Ok(url) = std::env::var("REDIS_URL") else {
        assert!(std::env::var("CI").is_err(), "REDIS_URL is required in CI");
        return;
    };
    let mut a = state().await;
    let mut storage = (*a.storage).clone();
    storage.redis = Arc::new(
        crate::storage::redis::RedisPool::new(&crate::config::models::storage::RedisConfig {
            url,
            enabled: true,
            allow_degraded: false,
            ..Default::default()
        })
        .await
        .unwrap(),
    );
    a.storage = Arc::new(storage);
    let mut b = state().await;
    b.storage = a.storage.clone();
    let mut capture = a
        .storage
        .redis
        .subscribe(&[REVISION_CHANNEL.into()])
        .await
        .unwrap();
    a.apply_runtime((*a.config()).clone()).await.unwrap();
    let message = tokio::time::timeout(Duration::from_secs(3), capture.next_message())
        .await
        .unwrap()
        .unwrap();
    assert_eq!(message.get_payload::<String>().unwrap(), "1");
    let task = start(b.clone()).unwrap();
    async fn reached(state: &AppState, revision: u64) {
        tokio::time::timeout(Duration::from_secs(5), async {
            while state.pin_runtime().generation != revision {
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
        })
        .await
        .expect("node must converge");
    }
    reached(&b, 1).await;
    drop(task); // Lose notifications while disconnected.
    a.apply_runtime((*a.config()).clone()).await.unwrap();
    a.apply_runtime((*a.config()).clone()).await.unwrap();
    let _task = start(b.clone()).unwrap();
    reached(&b, 3).await;
    let pin = b.pin_runtime();
    a.storage
        .redis
        .publish(REVISION_CHANNEL, "3")
        .await
        .unwrap();
    a.storage
        .redis
        .publish(REVISION_CHANNEL, "1")
        .await
        .unwrap();
    tokio::time::sleep(Duration::from_millis(100)).await;
    assert!(
        Arc::ptr_eq(&pin, &b.pin_runtime()),
        "duplicate/out-of-order events must not rebuild"
    );
    capture.unsubscribe_all().await.unwrap();
}

#[tokio::test]
async fn notification_failure_preserves_committed_admin_revision_records() {
    use actix_web::{http::StatusCode, test, web};
    let state = state().await;
    let token = crate::server::test_admin_token(&state).await;
    let provider = state.config().gateway.providers[0].name.clone();
    let app = test::init_service(HttpServer::create_app(web::Data::new(state.clone()))).await;
    for (path, body, method) in [
        (
            format!("/admin/providers/{provider}"),
            serde_json::json!({"weight": 2}),
            "PATCH",
        ),
        (
            "/admin/routing/policy".into(),
            serde_json::json!({"model_aliases": {}}),
            "PUT",
        ),
    ] {
        let response = test::call_service(
            &app,
            test::TestRequest::default()
                .method(method.parse().unwrap())
                .uri(&path)
                .insert_header(("Authorization", format!("Bearer {token}")))
                .set_json(body)
                .to_request(),
        )
        .await;
        assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
        let body: serde_json::Value = test::read_body_json(response).await;
        assert!(
            body["error"]
                .as_str()
                .unwrap()
                .contains("committed and applied")
        );
    }
    assert_eq!(state.pin_runtime().generation, 2);
    assert_eq!(
        state
            .storage
            .database
            .latest_provider_config_revision()
            .await
            .unwrap()
            .unwrap()
            .generation,
        1
    );
    assert_eq!(
        state
            .storage
            .database
            .latest_routing_policy_revision()
            .await
            .unwrap()
            .unwrap()
            .generation,
        2
    );
}