/*! * Integration tests for windmill-operator db_sync module. * * Tests verify full declarative sync of global_settings and worker configs: * - Upsert desired settings into DB * - Delete settings present in DB but absent from desired state * - Protect certain internal settings from deletion */ #[cfg(feature = "operator")] mod tests { use std::collections::BTreeMap; use sqlx::{Pool, Postgres}; // ======================================================================== // Helpers // ======================================================================== async fn get_global_setting(db: &Pool, name: &str) -> Option { sqlx::query_as::<_, (serde_json::Value,)>( "SELECT value FROM global_settings WHERE name = $1", ) .bind(name) .fetch_optional(db) .await .expect("query should succeed") .map(|(v,)| v) } async fn get_config(db: &Pool, name: &str) -> Option { sqlx::query_as::<_, (serde_json::Value,)>("SELECT config FROM config WHERE name = $1") .bind(name) .fetch_optional(db) .await .expect("query should succeed") .map(|(v,)| v) } async fn insert_global_setting(db: &Pool, name: &str, value: serde_json::Value) { sqlx::query( "INSERT INTO global_settings (name, value) VALUES ($1, $2) \ ON CONFLICT (name) DO UPDATE SET value = EXCLUDED.value", ) .bind(name) .bind(&value) .execute(db) .await .expect("insert should succeed"); } async fn insert_config(db: &Pool, name: &str, config: serde_json::Value) { sqlx::query( "INSERT INTO config (name, config) VALUES ($1, $2) \ ON CONFLICT (name) DO UPDATE SET config = EXCLUDED.config", ) .bind(name) .bind(&config) .execute(db) .await .expect("insert should succeed"); } // ======================================================================== // sync_global_settings tests // ======================================================================== #[sqlx::test(fixtures("base"))] async fn test_sync_global_settings_upserts(db: Pool) { let mut desired = BTreeMap::new(); desired.insert( "test_op_setting_a".to_string(), serde_json::json!("value_a"), ); desired.insert("test_op_setting_b".to_string(), serde_json::json!(42)); windmill_operator::db_sync::sync_global_settings(&db, &desired) .await .expect("sync should succeed"); assert_eq!( get_global_setting(&db, "test_op_setting_a").await, Some(serde_json::json!("value_a")) ); assert_eq!( get_global_setting(&db, "test_op_setting_b").await, Some(serde_json::json!(42)) ); } #[sqlx::test(fixtures("base"))] async fn test_sync_global_settings_updates_existing(db: Pool) { // Pre-populate a setting insert_global_setting(&db, "test_op_existing", serde_json::json!("old")).await; // Sync with new value let mut desired = BTreeMap::new(); desired.insert("test_op_existing".to_string(), serde_json::json!("new")); windmill_operator::db_sync::sync_global_settings(&db, &desired) .await .expect("sync should succeed"); assert_eq!( get_global_setting(&db, "test_op_existing").await, Some(serde_json::json!("new")) ); } #[sqlx::test(fixtures("base"))] async fn test_sync_global_settings_deletes_absent(db: Pool) { // Pre-populate settings insert_global_setting(&db, "test_op_keep", serde_json::json!("keep")).await; insert_global_setting(&db, "test_op_remove", serde_json::json!("remove")).await; // Sync with only one of them let mut desired = BTreeMap::new(); desired.insert("test_op_keep".to_string(), serde_json::json!("keep")); windmill_operator::db_sync::sync_global_settings(&db, &desired) .await .expect("sync should succeed"); assert!(get_global_setting(&db, "test_op_keep").await.is_some()); assert!( get_global_setting(&db, "test_op_remove").await.is_none(), "Setting absent from desired should be deleted" ); } #[sqlx::test(fixtures("base"))] async fn test_sync_global_settings_protects_ducklake(db: Pool) { // Pre-populate a protected setting insert_global_setting( &db, "ducklake_settings", serde_json::json!({"protected": true}), ) .await; // Sync with empty desired — protected key should survive let desired = BTreeMap::new(); windmill_operator::db_sync::sync_global_settings(&db, &desired) .await .expect("sync should succeed"); assert!( get_global_setting(&db, "ducklake_settings").await.is_some(), "Protected setting ducklake_settings should not be deleted" ); } #[sqlx::test(fixtures("base"))] async fn test_sync_global_settings_protects_all_protected_keys(db: Pool) { let protected_keys = [ "ducklake_user_pg_pwd", "ducklake_settings", "custom_instance_pg_databases", ]; for key in &protected_keys { insert_global_setting(&db, key, serde_json::json!("protected_value")).await; } // Sync with empty desired let desired = BTreeMap::new(); windmill_operator::db_sync::sync_global_settings(&db, &desired) .await .expect("sync should succeed"); for key in &protected_keys { assert!( get_global_setting(&db, key).await.is_some(), "Protected key {key} should not be deleted" ); } } #[sqlx::test(fixtures("base"))] async fn test_sync_global_settings_empty_desired(db: Pool) { // Pre-populate a non-protected setting insert_global_setting(&db, "test_op_ephemeral", serde_json::json!("gone")).await; let desired = BTreeMap::new(); windmill_operator::db_sync::sync_global_settings(&db, &desired) .await .expect("sync should succeed"); assert!( get_global_setting(&db, "test_op_ephemeral").await.is_none(), "Non-protected settings should be deleted when desired is empty" ); } #[sqlx::test(fixtures("base"))] async fn test_sync_global_settings_complex_json_values(db: Pool) { let mut desired = BTreeMap::new(); desired.insert( "test_op_complex".to_string(), serde_json::json!({ "host": "smtp.example.com", "port": 587, "tls": true, "nested": {"array": [1, 2, 3]} }), ); windmill_operator::db_sync::sync_global_settings(&db, &desired) .await .expect("sync should succeed"); let stored = get_global_setting(&db, "test_op_complex") .await .expect("Setting should exist"); assert_eq!(stored["host"], "smtp.example.com"); assert_eq!(stored["port"], 587); assert_eq!(stored["nested"]["array"][1], 2); } // ======================================================================== // sync_worker_configs tests // ======================================================================== #[sqlx::test(fixtures("base"))] async fn test_sync_worker_configs_upserts_with_prefix(db: Pool) { let mut desired = BTreeMap::new(); desired.insert( "test_group".to_string(), serde_json::json!({"dedicated_worker": false}), ); windmill_operator::db_sync::sync_worker_configs(&db, &desired) .await .expect("sync should succeed"); let config = get_config(&db, "worker__test_group") .await .expect("Config should exist with worker__ prefix"); assert_eq!(config["dedicated_worker"], false); } #[sqlx::test(fixtures("base"))] async fn test_sync_worker_configs_updates_existing(db: Pool) { // Pre-populate insert_config( &db, "worker__test_wc_existing", serde_json::json!({"old": true}), ) .await; let mut desired = BTreeMap::new(); desired.insert( "test_wc_existing".to_string(), serde_json::json!({"new": true}), ); windmill_operator::db_sync::sync_worker_configs(&db, &desired) .await .expect("sync should succeed"); let config = get_config(&db, "worker__test_wc_existing") .await .expect("Config should exist"); assert_eq!(config["new"], true); assert!(config.get("old").is_none()); } #[sqlx::test(fixtures("base"))] async fn test_sync_worker_configs_deletes_absent(db: Pool) { // Pre-populate two worker configs insert_config( &db, "worker__test_wc_keep", serde_json::json!({"keep": true}), ) .await; insert_config( &db, "worker__test_wc_remove", serde_json::json!({"remove": true}), ) .await; // Sync with only one let mut desired = BTreeMap::new(); desired.insert( "test_wc_keep".to_string(), serde_json::json!({"keep": true}), ); windmill_operator::db_sync::sync_worker_configs(&db, &desired) .await .expect("sync should succeed"); assert!(get_config(&db, "worker__test_wc_keep").await.is_some()); assert!( get_config(&db, "worker__test_wc_remove").await.is_none(), "Worker config absent from desired should be deleted" ); } #[sqlx::test(fixtures("base"))] async fn test_sync_worker_configs_does_not_touch_non_worker_configs(db: Pool) { // Insert a non-worker config (no worker__ prefix) insert_config(&db, "server_config", serde_json::json!({"important": true})).await; // Sync with empty worker configs let desired = BTreeMap::new(); windmill_operator::db_sync::sync_worker_configs(&db, &desired) .await .expect("sync should succeed"); assert!( get_config(&db, "server_config").await.is_some(), "Non-worker configs should not be touched" ); } #[sqlx::test(fixtures("base"))] async fn test_sync_worker_configs_multiple_groups(db: Pool) { let mut desired = BTreeMap::new(); desired.insert( "default".to_string(), serde_json::json!({"init_bash": "echo default"}), ); desired.insert( "gpu".to_string(), serde_json::json!({"dedicated_worker": true}), ); desired.insert( "native".to_string(), serde_json::json!({"init_bash": "echo native"}), ); windmill_operator::db_sync::sync_worker_configs(&db, &desired) .await .expect("sync should succeed"); assert!(get_config(&db, "worker__default").await.is_some()); assert!(get_config(&db, "worker__gpu").await.is_some()); assert!(get_config(&db, "worker__native").await.is_some()); } #[sqlx::test(fixtures("base"))] async fn test_sync_worker_configs_empty_desired(db: Pool) { insert_config( &db, "worker__test_wc_gone", serde_json::json!({"ephemeral": true}), ) .await; let desired = BTreeMap::new(); windmill_operator::db_sync::sync_worker_configs(&db, &desired) .await .expect("sync should succeed"); assert!( get_config(&db, "worker__test_wc_gone").await.is_none(), "Worker config should be deleted when desired is empty" ); } // ======================================================================== // End-to-end: both syncs together // ======================================================================== #[sqlx::test(fixtures("base"))] async fn test_full_declarative_sync(db: Pool) { // Pre-populate some existing state insert_global_setting(&db, "test_op_stale_setting", serde_json::json!("stale")).await; insert_config( &db, "worker__test_stale_group", serde_json::json!({"stale": true}), ) .await; // Define desired state let mut global_settings = BTreeMap::new(); global_settings.insert( "test_op_base_url".to_string(), serde_json::json!("https://windmill.example.com"), ); global_settings.insert( "test_op_license_key".to_string(), serde_json::json!("my-license"), ); let mut worker_configs = BTreeMap::new(); worker_configs.insert( "default".to_string(), serde_json::json!({"init_bash": "echo hello"}), ); // Sync both windmill_operator::db_sync::sync_global_settings(&db, &global_settings) .await .expect("global sync should succeed"); windmill_operator::db_sync::sync_worker_configs(&db, &worker_configs) .await .expect("worker sync should succeed"); // Verify desired state is present assert_eq!( get_global_setting(&db, "test_op_base_url").await, Some(serde_json::json!("https://windmill.example.com")) ); assert_eq!( get_global_setting(&db, "test_op_license_key").await, Some(serde_json::json!("my-license")) ); assert!(get_config(&db, "worker__default").await.is_some()); // Verify stale state is removed assert!( get_global_setting(&db, "test_op_stale_setting") .await .is_none(), "Stale global setting should be removed" ); assert!( get_config(&db, "worker__test_stale_group").await.is_none(), "Stale worker config should be removed" ); } #[sqlx::test(fixtures("base"))] async fn test_idempotent_sync(db: Pool) { let mut desired = BTreeMap::new(); desired.insert("test_op_idempotent".to_string(), serde_json::json!("value")); // Run sync twice — should be idempotent windmill_operator::db_sync::sync_global_settings(&db, &desired) .await .expect("first sync should succeed"); windmill_operator::db_sync::sync_global_settings(&db, &desired) .await .expect("second sync should succeed"); assert_eq!( get_global_setting(&db, "test_op_idempotent").await, Some(serde_json::json!("value")) ); } // ======================================================================== // jwt_secret and rsa_keys via sync_global_settings // ======================================================================== #[sqlx::test(fixtures("base"))] async fn test_sync_jwt_secret_via_settings_map(db: Pool) { // Simulate the full operator pipeline: parse GlobalSettings with a // resolved jwt_secret (StringOrSecretRef::Literal), convert to // settings map, and sync to DB. let settings = windmill_common::instance_config::GlobalSettings { jwt_secret: Some( windmill_common::instance_config::StringOrSecretRef::Literal( "resolved-jwt-value".to_string(), ), ), ..Default::default() }; let map = settings.to_settings_map(); windmill_operator::db_sync::sync_global_settings(&db, &map) .await .expect("sync should succeed"); let stored = get_global_setting(&db, "jwt_secret") .await .expect("jwt_secret should exist"); assert_eq!(stored, serde_json::json!("resolved-jwt-value")); } #[sqlx::test(fixtures("base"))] async fn test_sync_rsa_keys_via_settings_map(db: Pool) { // rsa_keys flows through GlobalSettings.extra as opaque JSON. // After operator resolves secretKeyRef, the value is a plain JSON // object that gets synced to DB. let json_str = r#"{ "rsa_keys": { "private_key": "-----BEGIN RSA PRIVATE KEY-----\ntest\n-----END RSA PRIVATE KEY-----" } }"#; let settings: windmill_common::instance_config::GlobalSettings = serde_json::from_str(json_str).unwrap(); let map = settings.to_settings_map(); windmill_operator::db_sync::sync_global_settings(&db, &map) .await .expect("sync should succeed"); let stored = get_global_setting(&db, "rsa_keys") .await .expect("rsa_keys should exist"); assert_eq!( stored["private_key"], "-----BEGIN RSA PRIVATE KEY-----\ntest\n-----END RSA PRIVATE KEY-----" ); } #[sqlx::test(fixtures("base"))] async fn test_sync_protects_jwt_secret_and_rsa_keys(db: Pool) { // Pre-populate both protected settings insert_global_setting(&db, "jwt_secret", serde_json::json!("original-jwt")).await; insert_global_setting( &db, "rsa_keys", serde_json::json!({"private_key": "original-rsa"}), ) .await; // Sync with empty desired — protected keys should survive let desired = BTreeMap::new(); windmill_operator::db_sync::sync_global_settings(&db, &desired) .await .expect("sync should succeed"); assert_eq!( get_global_setting(&db, "jwt_secret").await, Some(serde_json::json!("original-jwt")), "jwt_secret is protected and should survive empty sync" ); assert_eq!( get_global_setting(&db, "rsa_keys").await, Some(serde_json::json!({"private_key": "original-rsa"})), "rsa_keys is protected and should survive empty sync" ); } }