diff --git a/backend/tests/triggers.rs b/backend/tests/triggers.rs new file mode 100644 index 0000000000..47ef82ad19 --- /dev/null +++ b/backend/tests/triggers.rs @@ -0,0 +1,1274 @@ +/*! + * Integration tests for the trigger system (captures, HTTP triggers, trigger configs). + * + * These tests verify: + * 1. Capture config CRUD (create/ping/list/delete via API) + * 2. Capture payload insertion and retrieval + * 3. HTTP trigger CRUD and route matching + * 4. All trigger types DB schema validation + */ + +use serde::Deserialize; +use serde_json::json; +use sqlx::{Pool, Postgres}; + +mod common; +use common::*; + +// ============================================================================ +// Capture Config Tests (direct DB) +// ============================================================================ + +#[sqlx::test(fixtures("base"))] +async fn test_capture_config_insert_and_query(db: Pool) -> anyhow::Result<()> { + sqlx::query!( + r#" + INSERT INTO capture_config (workspace_id, path, is_flow, trigger_kind, owner, email) + VALUES ($1, $2, $3, $4::trigger_kind, $5, $6) + "#, + "test-workspace", + "f/test/script", + false, + "webhook" as _, + "test-user", + "test@windmill.dev", + ) + .execute(&db) + .await?; + + let config = sqlx::query!( + r#" + SELECT path, owner, email, trigger_kind AS "trigger_kind: String" + FROM capture_config + WHERE workspace_id = $1 AND path = $2 + "#, + "test-workspace", + "f/test/script", + ) + .fetch_one(&db) + .await?; + + assert_eq!(config.path, "f/test/script"); + assert_eq!(config.owner, "test-user"); + assert_eq!(config.email, "test@windmill.dev"); + assert_eq!(config.trigger_kind, "webhook"); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_capture_config_upsert(db: Pool) -> anyhow::Result<()> { + sqlx::query!( + r#" + INSERT INTO capture_config (workspace_id, path, is_flow, trigger_kind, owner, email) + VALUES ($1, $2, $3, $4::trigger_kind, $5, $6) + "#, + "test-workspace", + "f/test/script", + false, + "webhook" as _, + "test-user", + "test@windmill.dev", + ) + .execute(&db) + .await?; + + sqlx::query!( + r#" + INSERT INTO capture_config (workspace_id, path, is_flow, trigger_kind, owner, email) + VALUES ($1, $2, $3, $4::trigger_kind, $5, $6) + ON CONFLICT (workspace_id, path, is_flow, trigger_kind) + DO UPDATE SET owner = $5, email = $6, server_id = NULL, error = NULL + "#, + "test-workspace", + "f/test/script", + false, + "webhook" as _, + "new-owner", + "new@windmill.dev", + ) + .execute(&db) + .await?; + + let config = sqlx::query!( + "SELECT owner, email FROM capture_config WHERE workspace_id = $1 AND path = $2", + "test-workspace", + "f/test/script", + ) + .fetch_one(&db) + .await?; + + assert_eq!(config.owner, "new-owner"); + assert_eq!(config.email, "new@windmill.dev"); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_capture_config_ping_updates_timestamp(db: Pool) -> anyhow::Result<()> { + sqlx::query!( + r#" + INSERT INTO capture_config (workspace_id, path, is_flow, trigger_kind, owner, email) + VALUES ($1, $2, $3, $4::trigger_kind, $5, $6) + "#, + "test-workspace", + "f/test/script", + false, + "webhook" as _, + "test-user", + "test@windmill.dev", + ) + .execute(&db) + .await?; + + let before = sqlx::query!( + "SELECT last_client_ping FROM capture_config WHERE workspace_id = $1 AND path = $2", + "test-workspace", + "f/test/script", + ) + .fetch_one(&db) + .await?; + assert!(before.last_client_ping.is_none()); + + sqlx::query!( + r#" + UPDATE capture_config SET last_client_ping = NOW() + WHERE workspace_id = $1 AND path = $2 AND is_flow = $3 AND trigger_kind = $4::trigger_kind + "#, + "test-workspace", + "f/test/script", + false, + "webhook" as _, + ) + .execute(&db) + .await?; + + let after = sqlx::query!( + "SELECT last_client_ping FROM capture_config WHERE workspace_id = $1 AND path = $2", + "test-workspace", + "f/test/script", + ) + .fetch_one(&db) + .await?; + assert!(after.last_client_ping.is_some()); + + Ok(()) +} + +// ============================================================================ +// Capture Payload Tests (direct DB) +// ============================================================================ + +#[sqlx::test(fixtures("base"))] +async fn test_capture_insert_and_list(db: Pool) -> anyhow::Result<()> { + for i in 0..2 { + sqlx::query!( + r#" + INSERT INTO capture (workspace_id, path, is_flow, trigger_kind, main_args, preprocessor_args, created_by) + VALUES ($1, $2, $3, $4::trigger_kind, $5::jsonb, $6::jsonb, $7) + "#, + "test-workspace", + "f/test/script", + false, + "webhook" as _, + json!({"key": format!("value{}", i)}), + json!({"pre": format!("args{}", i)}), + "test-user", + ) + .execute(&db) + .await?; + } + + let count = sqlx::query_scalar!( + "SELECT COUNT(*) FROM capture WHERE workspace_id = $1 AND path = $2", + "test-workspace", + "f/test/script", + ) + .fetch_one(&db) + .await?; + + assert_eq!(count, Some(2)); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_capture_delete(db: Pool) -> anyhow::Result<()> { + let id = sqlx::query_scalar!( + r#" + INSERT INTO capture (workspace_id, path, is_flow, trigger_kind, main_args, preprocessor_args, created_by) + VALUES ($1, $2, $3, $4::trigger_kind, $5::jsonb, $6::jsonb, $7) + RETURNING id + "#, + "test-workspace", + "f/test/script", + false, + "webhook" as _, + json!({"key": "value"}), + json!({"pre": "args"}), + "test-user", + ) + .fetch_one(&db) + .await?; + + sqlx::query!("DELETE FROM capture WHERE id = $1", id) + .execute(&db) + .await?; + + let count = sqlx::query_scalar!( + "SELECT COUNT(*) FROM capture WHERE id = $1", + id, + ) + .fetch_one(&db) + .await?; + + assert_eq!(count, Some(0)); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_capture_filter_by_trigger_kind(db: Pool) -> anyhow::Result<()> { + sqlx::query!( + r#" + INSERT INTO capture (workspace_id, path, is_flow, trigger_kind, main_args, preprocessor_args, created_by) + VALUES ($1, $2, $3, $4::trigger_kind, $5::jsonb, $6::jsonb, $7) + "#, + "test-workspace", + "f/test/script", + false, + "webhook" as _, + json!({"source": "webhook"}), + json!({}), + "test-user", + ) + .execute(&db) + .await?; + + sqlx::query!( + r#" + INSERT INTO capture (workspace_id, path, is_flow, trigger_kind, main_args, preprocessor_args, created_by) + VALUES ($1, $2, $3, $4::trigger_kind, $5::jsonb, $6::jsonb, $7) + "#, + "test-workspace", + "f/test/script", + false, + "email" as _, + json!({"source": "email"}), + json!({}), + "test-user", + ) + .execute(&db) + .await?; + + let webhook_count = sqlx::query_scalar!( + r#" + SELECT COUNT(*) FROM capture + WHERE workspace_id = $1 AND path = $2 AND trigger_kind = $3::trigger_kind + "#, + "test-workspace", + "f/test/script", + "webhook" as _, + ) + .fetch_one(&db) + .await?; + + assert_eq!(webhook_count, Some(1)); + + let email_count = sqlx::query_scalar!( + r#" + SELECT COUNT(*) FROM capture + WHERE workspace_id = $1 AND path = $2 AND trigger_kind = $3::trigger_kind + "#, + "test-workspace", + "f/test/script", + "email" as _, + ) + .fetch_one(&db) + .await?; + + assert_eq!(email_count, Some(1)); + + Ok(()) +} + +// ============================================================================ +// Capture API Tests (via HTTP) +// ============================================================================ + +#[derive(Debug, Deserialize)] +struct CaptureResponse { + id: i64, + #[allow(dead_code)] + trigger_kind: String, + main_args: serde_json::Value, +} + +#[sqlx::test(fixtures("base"))] +async fn test_capture_api_set_config_and_list(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + let client = windmill_api_client::create_client( + &format!("http://localhost:{port}"), + "SECRET_TOKEN".to_string(), + ); + + let response = client + .client() + .post(format!( + "{}/w/test-workspace/capture/set_config", + client.baseurl() + )) + .json(&json!({ + "path": "f/test/my_script", + "is_flow": false, + "trigger_kind": "webhook", + })) + .send() + .await?; + + assert!( + response.status().is_success(), + "set_config should succeed, got: {}", + response.status() + ); + + let config = sqlx::query!( + "SELECT owner FROM capture_config WHERE workspace_id = $1 AND path = $2", + "test-workspace", + "f/test/my_script", + ) + .fetch_one(&db) + .await?; + + assert_eq!(config.owner, "test-user"); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_capture_api_list_captures(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + let client = windmill_api_client::create_client( + &format!("http://localhost:{port}"), + "SECRET_TOKEN".to_string(), + ); + + for i in 0..3 { + sqlx::query!( + r#" + INSERT INTO capture (workspace_id, path, is_flow, trigger_kind, main_args, preprocessor_args, created_by) + VALUES ($1, $2, $3, $4::trigger_kind, $5::jsonb, $6::jsonb, $7) + "#, + "test-workspace", + "f/test/script", + false, + "webhook" as _, + json!({"index": i}), + json!({}), + "test-user", + ) + .execute(&db) + .await?; + } + + let response = client + .client() + .get(format!( + "{}/w/test-workspace/capture/list/script/f/test/script", + client.baseurl() + )) + .send() + .await?; + + assert!(response.status().is_success(), "list captures should succeed"); + + let captures: Vec = response.json().await?; + assert_eq!(captures.len(), 3); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_capture_api_get_single(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + let client = windmill_api_client::create_client( + &format!("http://localhost:{port}"), + "SECRET_TOKEN".to_string(), + ); + + let id = sqlx::query_scalar!( + r#" + INSERT INTO capture (workspace_id, path, is_flow, trigger_kind, main_args, preprocessor_args, created_by) + VALUES ($1, $2, $3, $4::trigger_kind, $5::jsonb, $6::jsonb, $7) + RETURNING id + "#, + "test-workspace", + "f/test/script", + false, + "webhook" as _, + json!({"hello": "world"}), + json!({}), + "test-user", + ) + .fetch_one(&db) + .await?; + + let response = client + .client() + .get(format!( + "{}/w/test-workspace/capture/{}", + client.baseurl(), + id + )) + .send() + .await?; + + assert!(response.status().is_success(), "get capture should succeed"); + + let capture: CaptureResponse = response.json().await?; + assert_eq!(capture.id, id); + assert_eq!(capture.main_args["hello"], "world"); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_capture_api_delete(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + let client = windmill_api_client::create_client( + &format!("http://localhost:{port}"), + "SECRET_TOKEN".to_string(), + ); + + let id = sqlx::query_scalar!( + r#" + INSERT INTO capture (workspace_id, path, is_flow, trigger_kind, main_args, preprocessor_args, created_by) + VALUES ($1, $2, $3, $4::trigger_kind, $5::jsonb, $6::jsonb, $7) + RETURNING id + "#, + "test-workspace", + "f/test/script", + false, + "webhook" as _, + json!({"data": "to_delete"}), + json!({}), + "test-user", + ) + .fetch_one(&db) + .await?; + + let response = client + .client() + .delete(format!( + "{}/w/test-workspace/capture/{}", + client.baseurl(), + id + )) + .send() + .await?; + + assert!(response.status().is_success(), "delete should succeed"); + + let count = sqlx::query_scalar!( + "SELECT COUNT(*) FROM capture WHERE id = $1", + id, + ) + .fetch_one(&db) + .await?; + + assert_eq!(count, Some(0)); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_capture_api_pagination(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + let client = windmill_api_client::create_client( + &format!("http://localhost:{port}"), + "SECRET_TOKEN".to_string(), + ); + + for i in 0..5 { + sqlx::query!( + r#" + INSERT INTO capture (workspace_id, path, is_flow, trigger_kind, main_args, preprocessor_args, created_by) + VALUES ($1, $2, $3, $4::trigger_kind, $5::jsonb, $6::jsonb, $7) + "#, + "test-workspace", + "f/test/script", + false, + "webhook" as _, + json!({"index": i}), + json!({}), + "test-user", + ) + .execute(&db) + .await?; + } + + let response = client + .client() + .get(format!( + "{}/w/test-workspace/capture/list/script/f/test/script?per_page=2", + client.baseurl() + )) + .send() + .await?; + + assert!(response.status().is_success()); + let limited: Vec = response.json().await?; + assert_eq!(limited.len(), 2, "per_page=2 should limit to 2 results"); + + let response = client + .client() + .get(format!( + "{}/w/test-workspace/capture/list/script/f/test/script", + client.baseurl() + )) + .send() + .await?; + + assert!(response.status().is_success()); + let all: Vec = response.json().await?; + assert_eq!(all.len(), 5, "without limit should return all 5"); + + Ok(()) +} + +// ============================================================================ +// HTTP Trigger Tests (direct DB) +// ============================================================================ + +#[sqlx::test(fixtures("base"))] +async fn test_http_trigger_insert_and_query(db: Pool) -> anyhow::Result<()> { + sqlx::query!( + r#" + INSERT INTO http_trigger ( + path, route_path, route_path_key, script_path, is_flow, + workspace_id, edited_by, email, http_method, + authentication_method, is_static_website, workspaced_route, + wrap_body, raw_string + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::http_method, + $10::authentication_method, $11, $12, $13, $14) + "#, + "f/test/http_trigger", + "api/v1/users/:id", + "api/v1/users", + "f/test/handler_script", + false, + "test-workspace", + "test-user", + "test@windmill.dev", + "post" as _, + "none" as _, + false, + false, + false, + false, + ) + .execute(&db) + .await?; + + let trigger = sqlx::query!( + r#" + SELECT + path, route_path, script_path, + http_method AS "http_method: String", + authentication_method AS "authentication_method: String" + FROM http_trigger + WHERE workspace_id = $1 AND path = $2 + "#, + "test-workspace", + "f/test/http_trigger", + ) + .fetch_one(&db) + .await?; + + assert_eq!(trigger.path, "f/test/http_trigger"); + assert_eq!(trigger.route_path, "api/v1/users/:id"); + assert_eq!(trigger.script_path, "f/test/handler_script"); + assert_eq!(trigger.http_method, "post"); + assert_eq!(trigger.authentication_method, "none"); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_http_trigger_multiple_methods(db: Pool) -> anyhow::Result<()> { + let methods = ["get", "post", "put", "delete", "patch"]; + + for method in &methods { + sqlx::query!( + r#" + INSERT INTO http_trigger ( + path, route_path, route_path_key, script_path, is_flow, + workspace_id, edited_by, email, http_method, + authentication_method, is_static_website, workspaced_route, + wrap_body, raw_string + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::http_method, + $10::authentication_method, $11, $12, $13, $14) + "#, + format!("f/test/trigger_{}", method), + format!("api/{}", method), + format!("api/{}", method), + "f/test/handler", + false, + "test-workspace", + "test-user", + "test@windmill.dev", + *method as _, + "none" as _, + false, + false, + false, + false, + ) + .execute(&db) + .await?; + } + + let count = sqlx::query_scalar!( + "SELECT COUNT(*) FROM http_trigger WHERE workspace_id = $1", + "test-workspace", + ) + .fetch_one(&db) + .await?; + + assert_eq!(count, Some(methods.len() as i64)); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_http_trigger_authentication_methods(db: Pool) -> anyhow::Result<()> { + let auth_methods = ["none", "windmill", "api_key", "basic_http", "signature"]; + + for (i, auth) in auth_methods.iter().enumerate() { + sqlx::query!( + r#" + INSERT INTO http_trigger ( + path, route_path, route_path_key, script_path, is_flow, + workspace_id, edited_by, email, http_method, + authentication_method, is_static_website, workspaced_route, + wrap_body, raw_string + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::http_method, + $10::authentication_method, $11, $12, $13, $14) + "#, + format!("f/test/trigger_{}", i), + format!("api/{}", i), + format!("api/{}", i), + "f/test/handler", + false, + "test-workspace", + "test-user", + "test@windmill.dev", + "get" as _, + *auth as _, + false, + false, + false, + auth == &"signature", + ) + .execute(&db) + .await?; + } + + let sig_count = sqlx::query_scalar!( + r#" + SELECT COUNT(*) FROM http_trigger + WHERE workspace_id = $1 AND authentication_method = $2::authentication_method + "#, + "test-workspace", + "signature" as _, + ) + .fetch_one(&db) + .await?; + + assert_eq!(sig_count, Some(1)); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_http_trigger_update(db: Pool) -> anyhow::Result<()> { + sqlx::query!( + r#" + INSERT INTO http_trigger ( + path, route_path, route_path_key, script_path, is_flow, + workspace_id, edited_by, email, http_method, + authentication_method, is_static_website, workspaced_route, + wrap_body, raw_string + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::http_method, + $10::authentication_method, $11, $12, $13, $14) + "#, + "f/test/trigger", + "api/v1/old", + "api/v1/old", + "f/test/old_handler", + false, + "test-workspace", + "test-user", + "test@windmill.dev", + "get" as _, + "none" as _, + false, + false, + false, + false, + ) + .execute(&db) + .await?; + + sqlx::query!( + "UPDATE http_trigger SET script_path = $1 WHERE workspace_id = $2 AND path = $3", + "f/test/new_handler", + "test-workspace", + "f/test/trigger", + ) + .execute(&db) + .await?; + + let trigger = sqlx::query!( + "SELECT script_path FROM http_trigger WHERE workspace_id = $1 AND path = $2", + "test-workspace", + "f/test/trigger", + ) + .fetch_one(&db) + .await?; + + assert_eq!(trigger.script_path, "f/test/new_handler"); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_http_trigger_delete(db: Pool) -> anyhow::Result<()> { + sqlx::query!( + r#" + INSERT INTO http_trigger ( + path, route_path, route_path_key, script_path, is_flow, + workspace_id, edited_by, email, http_method, + authentication_method, is_static_website, workspaced_route, + wrap_body, raw_string + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::http_method, + $10::authentication_method, $11, $12, $13, $14) + "#, + "f/test/to_delete", + "api/delete_me", + "api/delete_me", + "f/test/handler", + false, + "test-workspace", + "test-user", + "test@windmill.dev", + "get" as _, + "none" as _, + false, + false, + false, + false, + ) + .execute(&db) + .await?; + + sqlx::query!( + "DELETE FROM http_trigger WHERE workspace_id = $1 AND path = $2", + "test-workspace", + "f/test/to_delete", + ) + .execute(&db) + .await?; + + let count = sqlx::query_scalar!( + "SELECT COUNT(*) FROM http_trigger WHERE workspace_id = $1 AND path = $2", + "test-workspace", + "f/test/to_delete", + ) + .fetch_one(&db) + .await?; + + assert_eq!(count, Some(0)); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_http_trigger_mode_filtering(db: Pool) -> anyhow::Result<()> { + let modes = ["enabled", "disabled", "suspended"]; + + for (i, mode) in modes.iter().enumerate() { + sqlx::query!( + r#" + INSERT INTO http_trigger ( + path, route_path, route_path_key, script_path, is_flow, + workspace_id, edited_by, email, http_method, + authentication_method, is_static_website, workspaced_route, + wrap_body, raw_string, mode + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::http_method, + $10::authentication_method, $11, $12, $13, $14, $15::trigger_mode) + "#, + format!("f/test/trigger_{}", i), + format!("api/{}", i), + format!("api/{}", i), + "f/test/handler", + false, + "test-workspace", + "test-user", + "test@windmill.dev", + "get" as _, + "none" as _, + false, + false, + false, + false, + *mode as _, + ) + .execute(&db) + .await?; + } + + // Query for active triggers (enabled or suspended, matching refresh_routers logic) + let active_count = sqlx::query_scalar!( + r#" + SELECT COUNT(*) FROM http_trigger + WHERE workspace_id = $1 + AND (mode = 'enabled'::trigger_mode OR mode = 'suspended'::trigger_mode) + "#, + "test-workspace", + ) + .fetch_one(&db) + .await?; + + assert_eq!(active_count, Some(2)); + + Ok(()) +} + +// ============================================================================ +// Other Trigger Types Tests (DB schema validation) +// ============================================================================ + +#[sqlx::test(fixtures("base"))] +async fn test_websocket_trigger_insert(db: Pool) -> anyhow::Result<()> { + sqlx::query!( + r#" + INSERT INTO websocket_trigger ( + path, url, script_path, is_flow, workspace_id, + edited_by, email + ) + VALUES ($1, $2, $3, $4, $5, $6, $7) + "#, + "f/test/ws_trigger", + "wss://example.com/feed", + "f/test/ws_handler", + false, + "test-workspace", + "test-user", + "test@windmill.dev", + ) + .execute(&db) + .await?; + + let trigger = sqlx::query!( + r#" + SELECT url, script_path, mode AS "mode: String" + FROM websocket_trigger WHERE workspace_id = $1 AND path = $2 + "#, + "test-workspace", + "f/test/ws_trigger", + ) + .fetch_one(&db) + .await?; + + assert_eq!(trigger.url, "wss://example.com/feed"); + assert_eq!(trigger.script_path, "f/test/ws_handler"); + assert_eq!(trigger.mode, "enabled"); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_kafka_trigger_insert(db: Pool) -> anyhow::Result<()> { + sqlx::query!( + r#" + INSERT INTO kafka_trigger ( + path, kafka_resource_path, topics, group_id, script_path, + is_flow, workspace_id, edited_by, email + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + "#, + "f/test/kafka_trigger", + "u/admin/kafka_resource", + &["topic-a", "topic-b"] as &[&str], + "my-consumer-group", + "f/test/kafka_handler", + false, + "test-workspace", + "test-user", + "test@windmill.dev", + ) + .execute(&db) + .await?; + + let trigger = sqlx::query!( + r#" + SELECT kafka_resource_path, topics, group_id, mode AS "mode: String" + FROM kafka_trigger + WHERE workspace_id = $1 AND path = $2 + "#, + "test-workspace", + "f/test/kafka_trigger", + ) + .fetch_one(&db) + .await?; + + assert_eq!(trigger.kafka_resource_path, "u/admin/kafka_resource"); + assert_eq!(trigger.topics, vec!["topic-a", "topic-b"]); + assert_eq!(trigger.group_id, "my-consumer-group"); + assert_eq!(trigger.mode, "enabled"); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_postgres_trigger_insert(db: Pool) -> anyhow::Result<()> { + sqlx::query!( + r#" + INSERT INTO postgres_trigger ( + path, script_path, is_flow, workspace_id, edited_by, email, + postgres_resource_path, replication_slot_name, publication_name + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + "#, + "f/test/pg_trigger", + "f/test/pg_handler", + false, + "test-workspace", + "test-user", + "test@windmill.dev", + "u/admin/pg_resource", + "test_slot", + "test_publication", + ) + .execute(&db) + .await?; + + let trigger = sqlx::query!( + r#" + SELECT postgres_resource_path, replication_slot_name, publication_name, mode AS "mode: String" + FROM postgres_trigger + WHERE workspace_id = $1 AND path = $2 + "#, + "test-workspace", + "f/test/pg_trigger", + ) + .fetch_one(&db) + .await?; + + assert_eq!(trigger.postgres_resource_path, "u/admin/pg_resource"); + assert_eq!(trigger.replication_slot_name, "test_slot"); + assert_eq!(trigger.publication_name, "test_publication"); + assert_eq!(trigger.mode, "enabled"); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_nats_trigger_insert(db: Pool) -> anyhow::Result<()> { + sqlx::query!( + r#" + INSERT INTO nats_trigger ( + path, nats_resource_path, subjects, script_path, + is_flow, workspace_id, edited_by, email, use_jetstream + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + "#, + "f/test/nats_trigger", + "u/admin/nats_resource", + &["orders.>", "payments.*"] as &[&str], + "f/test/nats_handler", + false, + "test-workspace", + "test-user", + "test@windmill.dev", + false, + ) + .execute(&db) + .await?; + + let trigger = sqlx::query!( + r#" + SELECT nats_resource_path, subjects, use_jetstream, mode AS "mode: String" + FROM nats_trigger + WHERE workspace_id = $1 AND path = $2 + "#, + "test-workspace", + "f/test/nats_trigger", + ) + .fetch_one(&db) + .await?; + + assert_eq!(trigger.nats_resource_path, "u/admin/nats_resource"); + assert_eq!(trigger.subjects, vec!["orders.>", "payments.*"]); + assert_eq!(trigger.use_jetstream, false); + assert_eq!(trigger.mode, "enabled"); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_sqs_trigger_insert(db: Pool) -> anyhow::Result<()> { + sqlx::query!( + r#" + INSERT INTO sqs_trigger ( + path, queue_url, aws_resource_path, script_path, + is_flow, workspace_id, edited_by, email + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8) + "#, + "f/test/sqs_trigger", + "https://sqs.us-east-1.amazonaws.com/123456789/my-queue", + "u/admin/aws_resource", + "f/test/sqs_handler", + false, + "test-workspace", + "test-user", + "test@windmill.dev", + ) + .execute(&db) + .await?; + + let trigger = sqlx::query!( + r#" + SELECT queue_url, aws_resource_path, + aws_auth_resource_type AS "aws_auth_resource_type: String", + mode AS "mode: String" + FROM sqs_trigger + WHERE workspace_id = $1 AND path = $2 + "#, + "test-workspace", + "f/test/sqs_trigger", + ) + .fetch_one(&db) + .await?; + + assert_eq!( + trigger.queue_url, + "https://sqs.us-east-1.amazonaws.com/123456789/my-queue" + ); + assert_eq!(trigger.aws_resource_path, "u/admin/aws_resource"); + assert_eq!(trigger.aws_auth_resource_type, "credentials"); + assert_eq!(trigger.mode, "enabled"); + + Ok(()) +} + +// ============================================================================ +// Cross-trigger tests +// ============================================================================ + +#[sqlx::test(fixtures("base"))] +async fn test_trigger_server_state_tracking(db: Pool) -> anyhow::Result<()> { + sqlx::query!( + r#" + INSERT INTO websocket_trigger ( + path, url, script_path, is_flow, workspace_id, + edited_by, email, server_id, error + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + "#, + "f/test/ws_with_error", + "wss://example.com/feed", + "f/test/handler", + false, + "test-workspace", + "test-user", + "test@windmill.dev", + "server-abc-123", + "connection refused", + ) + .execute(&db) + .await?; + + let trigger = sqlx::query!( + "SELECT server_id, error FROM websocket_trigger WHERE workspace_id = $1 AND path = $2", + "test-workspace", + "f/test/ws_with_error", + ) + .fetch_one(&db) + .await?; + + assert_eq!(trigger.server_id, Some("server-abc-123".to_string())); + assert_eq!(trigger.error, Some("connection refused".to_string())); + + sqlx::query!( + "UPDATE websocket_trigger SET error = NULL WHERE workspace_id = $1 AND path = $2", + "test-workspace", + "f/test/ws_with_error", + ) + .execute(&db) + .await?; + + let trigger = sqlx::query!( + "SELECT error FROM websocket_trigger WHERE workspace_id = $1 AND path = $2", + "test-workspace", + "f/test/ws_with_error", + ) + .fetch_one(&db) + .await?; + + assert!(trigger.error.is_none()); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_trigger_mode_filtering(db: Pool) -> anyhow::Result<()> { + let modes = ["enabled", "disabled", "enabled"]; + + for (i, mode) in modes.iter().enumerate() { + sqlx::query!( + r#" + INSERT INTO websocket_trigger ( + path, url, script_path, is_flow, workspace_id, + edited_by, email, mode + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8::trigger_mode) + "#, + format!("f/test/ws_trigger_{}", i), + "wss://example.com", + "f/test/handler", + false, + "test-workspace", + "test-user", + "test@windmill.dev", + *mode as _, + ) + .execute(&db) + .await?; + } + + let enabled_count = sqlx::query_scalar!( + "SELECT COUNT(*) FROM websocket_trigger WHERE workspace_id = $1 AND mode = 'enabled'::trigger_mode", + "test-workspace", + ) + .fetch_one(&db) + .await?; + + assert_eq!(enabled_count, Some(2)); + + let disabled_count = sqlx::query_scalar!( + "SELECT COUNT(*) FROM websocket_trigger WHERE workspace_id = $1 AND mode = 'disabled'::trigger_mode", + "test-workspace", + ) + .fetch_one(&db) + .await?; + + assert_eq!(disabled_count, Some(1)); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_multiple_capture_configs_per_path(db: Pool) -> anyhow::Result<()> { + let trigger_kinds = ["webhook", "email", "kafka"]; + + for kind in &trigger_kinds { + sqlx::query!( + r#" + INSERT INTO capture_config (workspace_id, path, is_flow, trigger_kind, owner, email) + VALUES ($1, $2, $3, $4::trigger_kind, $5, $6) + "#, + "test-workspace", + "f/test/multi_trigger_script", + false, + *kind as _, + "test-user", + "test@windmill.dev", + ) + .execute(&db) + .await?; + } + + let count = sqlx::query_scalar!( + "SELECT COUNT(*) FROM capture_config WHERE workspace_id = $1 AND path = $2", + "test-workspace", + "f/test/multi_trigger_script", + ) + .fetch_one(&db) + .await?; + + assert_eq!(count, Some(3)); + + Ok(()) +} + +// ============================================================================ +// Schedule Tests (DB-level) +// ============================================================================ + +#[sqlx::test(fixtures("base"))] +async fn test_schedule_insert_and_query(db: Pool) -> anyhow::Result<()> { + sqlx::query!( + r#" + INSERT INTO schedule ( + workspace_id, path, edited_by, schedule, enabled, + script_path, is_flow, email, timezone + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + "#, + "test-workspace", + "f/test/my_schedule", + "test-user", + "0 */5 * * *", + true, + "f/test/scheduled_script", + false, + "test@windmill.dev", + "UTC", + ) + .execute(&db) + .await?; + + let schedule = sqlx::query!( + r#" + SELECT path, schedule, enabled, script_path, timezone + FROM schedule + WHERE workspace_id = $1 AND path = $2 + "#, + "test-workspace", + "f/test/my_schedule", + ) + .fetch_one(&db) + .await?; + + assert_eq!(schedule.path, "f/test/my_schedule"); + assert_eq!(schedule.schedule, "0 */5 * * *"); + assert_eq!(schedule.enabled, true); + assert_eq!(schedule.script_path, "f/test/scheduled_script"); + assert_eq!(schedule.timezone, "UTC"); + + Ok(()) +} diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 381fe21db2..a08f546b24 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -1145,6 +1145,195 @@ public class Main { Ok(()) } +#[cfg(feature = "deno_core")] +#[sqlx::test(fixtures("base"))] +async fn test_nativets_job(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let content = r#" +export async function main(name: string): Promise { + return `hello ${name}`; +} + "# + .to_owned(); + + let result = RunJob::from(JobPayload::Code(RawCode { + hash: None, + content, + path: None, + lock: None, + language: ScriptLang::Nativets, + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: None, + concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default() + .into(), + debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(), + })) + .arg("name", json!("world")) + .run_until_complete(&db, false, port) + .await + .json_result() + .unwrap(); + + assert_eq!(result, serde_json::json!("hello world")); + Ok(()) +} + +#[cfg(feature = "deno_core")] +#[sqlx::test(fixtures("base"))] +async fn test_nativets_job_with_args(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let content = r#" +export async function main(a: number, b: number): Promise { + return a + b; +} + "# + .to_owned(); + + let result = RunJob::from(JobPayload::Code(RawCode { + hash: None, + content, + path: None, + lock: None, + language: ScriptLang::Nativets, + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: None, + concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default() + .into(), + debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(), + })) + .arg("a", json!(3)) + .arg("b", json!(7)) + .run_until_complete(&db, false, port) + .await + .json_result() + .unwrap(); + + assert_eq!(result, serde_json::json!(10)); + Ok(()) +} + +#[cfg(feature = "deno_core")] +#[sqlx::test(fixtures("base"))] +async fn test_nativets_job_object_return(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let content = r#" +export async function main(items: string[]): Promise<{ count: number; items: string[] }> { + return { count: items.length, items }; +} + "# + .to_owned(); + + let result = RunJob::from(JobPayload::Code(RawCode { + hash: None, + content, + path: None, + lock: None, + language: ScriptLang::Nativets, + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: None, + concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default() + .into(), + debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(), + })) + .arg("items", json!(["a", "b", "c"])) + .run_until_complete(&db, false, port) + .await + .json_result() + .unwrap(); + + assert_eq!(result, json!({"count": 3, "items": ["a", "b", "c"]})); + Ok(()) +} + +#[cfg(feature = "deno_core")] +#[sqlx::test(fixtures("base"))] +async fn test_nativets_job_datetime(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + // nativets passes Date-typed args as strings (no auto-conversion unlike Bun/Deno) + let content = r#" +export async function main(a: Date) { + return typeof a; +} + "# + .to_owned(); + + let result = RunJob::from(JobPayload::Code(RawCode { + hash: None, + content, + path: None, + lock: None, + language: ScriptLang::Nativets, + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: None, + concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default() + .into(), + debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(), + })) + .arg("a", json!("2024-09-24T10:00:00.000Z")) + .run_until_complete(&db, false, port) + .await + .json_result() + .unwrap(); + + assert_eq!(result, serde_json::json!("string")); + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_postgresql_job(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let content = r#" +-- $1 name +SELECT 'hello ' || $1::text AS result; +"# + .to_owned(); + + let result = RunJob::from(JobPayload::Code(RawCode { + hash: None, + content, + path: None, + lock: None, + language: ScriptLang::Postgresql, + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: None, + concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default() + .into(), + debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(), + })) + .arg("name", json!("world")) + .arg( + "database", + json!({"host": "localhost", "port": 5432, "dbname": "windmill", "user": "postgres", "password": "changeme"}), + ) + .run_until_complete(&db, false, port) + .await + .json_result() + .unwrap(); + + assert_eq!(result, json!([{"result": "hello world"}])); + Ok(()) +} + #[sqlx::test(fixtures("base"))] async fn test_bun_job_datetime(db: Pool) -> anyhow::Result<()> { initialize_tracing().await; diff --git a/backend/windmill-api-jobs/src/query.rs b/backend/windmill-api-jobs/src/query.rs index a863a8157a..ee7877438a 100644 --- a/backend/windmill-api-jobs/src/query.rs +++ b/backend/windmill-api-jobs/src/query.rs @@ -371,3 +371,441 @@ pub fn list_completed_jobs_query( filter_list_completed_query(sqlb, lq, w_id, join_outstanding_wait_times) } + +#[cfg(test)] +mod tests { + use super::*; + + fn empty_queue_query() -> ListQueueQuery { + ListQueueQuery { + script_path_start: None, + script_path_exact: None, + script_hash: None, + created_by: None, + started_before: None, + started_after: None, + created_before: None, + created_after: None, + created_or_started_before: None, + created_or_started_after: None, + running: None, + parent_job: None, + order_desc: None, + job_kinds: None, + suspended: None, + args: None, + tag: None, + schedule_path: None, + scheduled_for_before_now: None, + all_workspaces: None, + is_flow_step: None, + has_null_parent: None, + is_not_schedule: None, + concurrency_key: None, + worker: None, + allow_wildcards: None, + trigger_kind: None, + trigger_path: None, + include_args: None, + } + } + + fn empty_completed_query() -> ListCompletedQuery { + ListCompletedQuery { + script_path_start: None, + script_path_exact: None, + script_hash: None, + created_by: None, + started_before: None, + started_after: None, + created_before: None, + created_after: None, + created_or_started_before: None, + created_or_started_after: None, + created_or_started_after_completed_jobs: None, + created_before_queue: None, + created_after_queue: None, + completed_after: None, + completed_before: None, + success: None, + running: None, + parent_job: None, + order_desc: None, + job_kinds: None, + is_skipped: None, + is_flow_step: None, + suspended: None, + schedule_path: None, + args: None, + result: None, + tag: None, + scheduled_for_before_now: None, + all_workspaces: None, + has_null_parent: None, + label: None, + is_not_schedule: None, + concurrency_key: None, + worker: None, + allow_wildcards: None, + trigger_kind: None, + trigger_path: None, + include_args: None, + } + } + + fn build_sql(sqlb: SqlBuilder) -> String { + sqlb.sql().unwrap_or_default() + } + + // --- Queue query tests --- + + #[test] + fn test_queue_basic_query() { + let lq = empty_queue_query(); + let sqlb = list_queue_jobs_query( + "test_ws", + &lq, + &["v2_job_queue.id"], + Pagination { page: Some(1), per_page: Some(10) }, + false, + None, + ); + let sql = build_sql(sqlb); + assert!(sql.contains("v2_job_queue")); + assert!(sql.contains("v2_job_queue.workspace_id")); + } + + #[test] + fn test_queue_filter_script_path_start() { + let lq = ListQueueQuery { + script_path_start: Some("f/test".to_string()), + ..empty_queue_query() + }; + let sqlb = filter_list_queue_query( + SqlBuilder::select_from("v2_job_queue").clone(), + &lq, + "ws", + false, + ); + let sql = build_sql(sqlb); + assert!(sql.contains("runnable_path")); + assert!(sql.contains("LIKE")); + } + + #[test] + fn test_queue_filter_script_path_exact() { + let lq = ListQueueQuery { + script_path_exact: Some("f/test/script".to_string()), + ..empty_queue_query() + }; + let sqlb = filter_list_queue_query( + SqlBuilder::select_from("v2_job_queue").clone(), + &lq, + "ws", + false, + ); + let sql = build_sql(sqlb); + assert!(sql.contains("runnable_path")); + } + + #[test] + fn test_queue_filter_running() { + let lq = ListQueueQuery { + running: Some(true), + ..empty_queue_query() + }; + let sqlb = filter_list_queue_query( + SqlBuilder::select_from("v2_job_queue").clone(), + &lq, + "ws", + false, + ); + let sql = build_sql(sqlb); + assert!(sql.contains("running")); + } + + #[test] + fn test_queue_filter_job_kinds() { + let lq = ListQueueQuery { + job_kinds: Some("script,flow".to_string()), + ..empty_queue_query() + }; + let sqlb = filter_list_queue_query( + SqlBuilder::select_from("v2_job_queue").clone(), + &lq, + "ws", + false, + ); + let sql = build_sql(sqlb); + assert!(sql.contains("kind")); + assert!(sql.contains("IN")); + } + + #[test] + fn test_queue_filter_suspended() { + let lq = ListQueueQuery { + suspended: Some(true), + ..empty_queue_query() + }; + let sqlb = filter_list_queue_query( + SqlBuilder::select_from("v2_job_queue").clone(), + &lq, + "ws", + false, + ); + let sql = build_sql(sqlb); + assert!(sql.contains("suspend")); + } + + #[test] + fn test_queue_filter_is_not_schedule() { + let lq = ListQueueQuery { + is_not_schedule: Some(true), + ..empty_queue_query() + }; + let sqlb = filter_list_queue_query( + SqlBuilder::select_from("v2_job_queue").clone(), + &lq, + "ws", + false, + ); + let sql = build_sql(sqlb); + assert!(sql.contains("trigger_kind IS DISTINCT FROM")); + } + + #[test] + fn test_queue_filter_has_null_parent() { + let lq = ListQueueQuery { + has_null_parent: Some(true), + ..empty_queue_query() + }; + let sqlb = filter_list_queue_query( + SqlBuilder::select_from("v2_job_queue").clone(), + &lq, + "ws", + false, + ); + let sql = build_sql(sqlb); + assert!(sql.contains("parent_job IS NULL")); + } + + #[test] + fn test_queue_filter_is_flow_step_true() { + let lq = ListQueueQuery { + is_flow_step: Some(true), + ..empty_queue_query() + }; + let sqlb = filter_list_queue_query( + SqlBuilder::select_from("v2_job_queue").clone(), + &lq, + "ws", + false, + ); + let sql = build_sql(sqlb); + assert!(sql.contains("flow_step_id IS NOT NULL")); + } + + #[test] + fn test_queue_filter_is_flow_step_false() { + let lq = ListQueueQuery { + is_flow_step: Some(false), + ..empty_queue_query() + }; + let sqlb = filter_list_queue_query( + SqlBuilder::select_from("v2_job_queue").clone(), + &lq, + "ws", + false, + ); + let sql = build_sql(sqlb); + assert!(sql.contains("flow_step_id IS NULL")); + } + + #[test] + fn test_queue_admins_all_workspaces() { + let lq = ListQueueQuery { + all_workspaces: Some(true), + ..empty_queue_query() + }; + let sqlb = filter_list_queue_query( + SqlBuilder::select_from("v2_job_queue").clone(), + &lq, + "admins", + false, + ); + let sql = build_sql(sqlb); + assert!(!sql.contains("workspace_id")); + } + + #[test] + fn test_queue_non_admins_ignores_all_workspaces() { + let lq = ListQueueQuery { + all_workspaces: Some(true), + ..empty_queue_query() + }; + let sqlb = filter_list_queue_query( + SqlBuilder::select_from("v2_job_queue").clone(), + &lq, + "other_ws", + false, + ); + let sql = build_sql(sqlb); + assert!(sql.contains("workspace_id")); + } + + #[test] + fn test_queue_with_outstanding_wait_times() { + let lq = empty_queue_query(); + let sqlb = filter_list_queue_query( + SqlBuilder::select_from("v2_job_queue").clone(), + &lq, + "ws", + true, + ); + let sql = build_sql(sqlb); + assert!(sql.contains("outstanding_wait_time")); + } + + #[test] + fn test_queue_schedule_path_filter() { + let lq = ListQueueQuery { + schedule_path: Some("f/test/schedule".to_string()), + ..empty_queue_query() + }; + let sqlb = filter_list_queue_query( + SqlBuilder::select_from("v2_job_queue").clone(), + &lq, + "ws", + false, + ); + let sql = build_sql(sqlb); + assert!(sql.contains("trigger")); + assert!(sql.contains("'schedule'")); + } + + // --- Completed query tests --- + + #[test] + fn test_completed_basic_query() { + let lq = empty_completed_query(); + let sqlb = list_completed_jobs_query("test_ws", Some(10), 0, &lq, &["id"], false, None); + let sql = build_sql(sqlb); + assert!(sql.contains("v2_job_completed")); + } + + #[test] + fn test_completed_filter_success_true() { + let lq = ListCompletedQuery { + success: Some(true), + ..empty_completed_query() + }; + let sqlb = filter_list_completed_query( + SqlBuilder::select_from("v2_job_completed").clone(), + &lq, + "ws", + false, + ); + let sql = build_sql(sqlb); + assert!(sql.contains("'success'")); + } + + #[test] + fn test_completed_filter_success_false() { + let lq = ListCompletedQuery { + success: Some(false), + ..empty_completed_query() + }; + let sqlb = filter_list_completed_query( + SqlBuilder::select_from("v2_job_completed").clone(), + &lq, + "ws", + false, + ); + let sql = build_sql(sqlb); + assert!(sql.contains("'failure'")); + } + + #[test] + fn test_completed_order_by_completed_at() { + let lq = ListCompletedQuery { + completed_after: Some(chrono::Utc::now()), + ..empty_completed_query() + }; + let sqlb = list_completed_jobs_query("ws", Some(10), 0, &lq, &["id"], false, None); + let sql = build_sql(sqlb); + assert!(sql.contains("completed_at")); + } + + #[test] + fn test_completed_filter_label() { + let lq = ListCompletedQuery { + label: Some("deploy".to_string()), + ..empty_completed_query() + }; + let sqlb = filter_list_completed_query( + SqlBuilder::select_from("v2_job_completed").clone(), + &lq, + "ws", + false, + ); + let sql = build_sql(sqlb); + assert!(sql.contains("wm_labels")); + } + + #[test] + fn test_completed_filter_is_skipped() { + let lq = ListCompletedQuery { + is_skipped: Some(true), + ..empty_completed_query() + }; + let sqlb = filter_list_completed_query( + SqlBuilder::select_from("v2_job_completed").clone(), + &lq, + "ws", + false, + ); + let sql = build_sql(sqlb); + assert!(sql.contains("'skipped'")); + } + + #[test] + fn test_completed_with_tags() { + let lq = empty_completed_query(); + let sqlb = list_completed_jobs_query( + "ws", + Some(10), + 0, + &lq, + &["id"], + false, + Some(vec!["tag1", "tag2"]), + ); + let sql = build_sql(sqlb); + assert!(sql.contains("v2_job.tag")); + assert!(sql.contains("IN")); + } + + #[test] + fn test_completed_no_limit() { + let lq = empty_completed_query(); + let sqlb = list_completed_jobs_query("ws", None, 0, &lq, &["id"], false, None); + let sql = build_sql(sqlb); + assert!(!sql.contains("LIMIT")); + } + + #[test] + fn test_completed_result_filter() { + let lq = ListCompletedQuery { + result: Some(r#"{"status": "ok"}"#.to_string()), + ..empty_completed_query() + }; + let sqlb = filter_list_completed_query( + SqlBuilder::select_from("v2_job_completed").clone(), + &lq, + "ws", + false, + ); + let sql = build_sql(sqlb); + assert!(sql.contains("result @>")); + } +} diff --git a/backend/windmill-api-jobs/src/types.rs b/backend/windmill-api-jobs/src/types.rs index d9717b75a0..5ca13645e7 100644 --- a/backend/windmill-api-jobs/src/types.rs +++ b/backend/windmill-api-jobs/src/types.rs @@ -568,3 +568,279 @@ pub fn add_raw_string( use anyhow::Context; use base64::Engine; + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + // --- decode_payload --- + + #[test] + fn test_decode_payload_valid() { + let payload = base64::engine::general_purpose::STANDARD + .encode(r#"{"key": "value"}"#); + let result: HashMap = decode_payload(payload).unwrap(); + assert_eq!(result["key"], json!("value")); + } + + #[test] + fn test_decode_payload_invalid_base64() { + let result: anyhow::Result> = + decode_payload("not-valid-base64!!!".to_string()); + assert!(result.is_err()); + } + + #[test] + fn test_decode_payload_invalid_json() { + let payload = base64::engine::general_purpose::STANDARD.encode("not json"); + let result: anyhow::Result> = decode_payload(payload); + assert!(result.is_err()); + } + + #[test] + fn test_decode_payload_empty_object() { + let payload = base64::engine::general_purpose::STANDARD.encode("{}"); + let result: HashMap = decode_payload(payload).unwrap(); + assert!(result.is_empty()); + } + + // --- add_raw_string --- + + #[test] + fn test_add_raw_string_some() { + let args = serde_json::Map::new(); + let result = add_raw_string(Some("body content".to_string()), args); + assert_eq!( + result["raw_string"], + serde_json::Value::String("body content".to_string()) + ); + } + + #[test] + fn test_add_raw_string_none() { + let args = serde_json::Map::new(); + let result = add_raw_string(None, args); + assert!(!result.contains_key("raw_string")); + } + + #[test] + fn test_add_raw_string_preserves_existing() { + let mut args = serde_json::Map::new(); + args.insert("existing".to_string(), json!("value")); + let result = add_raw_string(Some("body".to_string()), args); + assert_eq!(result["existing"], json!("value")); + assert_eq!(result["raw_string"], json!("body")); + } + + // --- RunJobQuery --- + + #[test] + fn test_run_job_query_payload_as_args_none() { + let q = RunJobQuery::default(); + let result = q.payload_as_args().unwrap(); + assert!(result.is_empty()); + } + + #[test] + fn test_run_job_query_payload_as_args_valid() { + let encoded = base64::engine::general_purpose::STANDARD + .encode(r#"{"x": 42}"#); + let q = RunJobQuery { + payload: Some(encoded), + ..Default::default() + }; + let result = q.payload_as_args().unwrap(); + assert!(result.contains_key("x")); + } + + #[test] + fn test_run_job_query_payload_as_args_invalid() { + let q = RunJobQuery { + payload: Some("invalid!!!".to_string()), + ..Default::default() + }; + assert!(q.payload_as_args().is_err()); + } + + // --- ListCompletedQuery -> ListQueueQuery conversion --- + + #[test] + fn test_list_completed_to_queue_query_conversion() { + let lcq = ListCompletedQuery { + script_path_start: Some("f/test".to_string()), + script_path_exact: None, + script_hash: None, + created_by: Some("admin".to_string()), + started_before: None, + started_after: None, + created_before: Some(chrono::Utc::now()), + created_after: None, + created_or_started_before: None, + created_or_started_after: None, + created_or_started_after_completed_jobs: None, + created_before_queue: None, + created_after_queue: None, + completed_after: None, + completed_before: None, + success: None, + running: Some(true), + parent_job: None, + order_desc: Some(true), + job_kinds: Some("script,flow".to_string()), + is_skipped: None, + is_flow_step: None, + suspended: None, + schedule_path: None, + args: None, + result: None, + tag: Some("custom".to_string()), + scheduled_for_before_now: None, + all_workspaces: None, + has_null_parent: None, + label: None, + is_not_schedule: None, + concurrency_key: None, + worker: None, + allow_wildcards: None, + trigger_kind: None, + trigger_path: None, + include_args: None, + }; + + let lqq: ListQueueQuery = lcq.into(); + assert_eq!(lqq.script_path_start, Some("f/test".to_string())); + assert_eq!(lqq.created_by, Some("admin".to_string())); + assert_eq!(lqq.running, Some(true)); + assert_eq!(lqq.job_kinds, Some("script,flow".to_string())); + assert_eq!(lqq.tag, Some("custom".to_string())); + } + + #[test] + fn test_list_completed_to_queue_prefers_queue_created_fields() { + let specific_time = chrono::Utc::now(); + let other_time = specific_time - chrono::Duration::hours(1); + + let lcq = ListCompletedQuery { + script_path_start: None, + script_path_exact: None, + script_hash: None, + created_by: None, + started_before: None, + started_after: None, + created_before: Some(other_time), + created_after: Some(other_time), + created_or_started_before: None, + created_or_started_after: None, + created_or_started_after_completed_jobs: None, + created_before_queue: Some(specific_time), + created_after_queue: Some(specific_time), + completed_after: None, + completed_before: None, + success: None, + running: None, + parent_job: None, + order_desc: None, + job_kinds: None, + is_skipped: None, + is_flow_step: None, + suspended: None, + schedule_path: None, + args: None, + result: None, + tag: None, + scheduled_for_before_now: None, + all_workspaces: None, + has_null_parent: None, + label: None, + is_not_schedule: None, + concurrency_key: None, + worker: None, + allow_wildcards: None, + trigger_kind: None, + trigger_path: None, + include_args: None, + }; + + let lqq: ListQueueQuery = lcq.into(); + assert_eq!(lqq.created_before, Some(specific_time)); + assert_eq!(lqq.created_after, Some(specific_time)); + } + + // --- UnifiedJob field constants --- + + #[test] + fn test_completed_job_fields_not_empty() { + let fields = UnifiedJob::completed_job_fields(); + assert!(!fields.is_empty()); + assert!(fields.iter().any(|f| f.contains("typ"))); + assert!(fields.iter().any(|f| f.contains("workspace_id"))); + } + + #[test] + fn test_queued_job_fields_not_empty() { + let fields = UnifiedJob::queued_job_fields(); + assert!(!fields.is_empty()); + assert!(fields.iter().any(|f| f.contains("typ"))); + assert!(fields.iter().any(|f| f.contains("scheduled_for"))); + } + + #[test] + fn test_completed_and_queued_fields_same_count() { + assert_eq!( + UnifiedJob::completed_job_fields().len(), + UnifiedJob::queued_job_fields().len(), + "CJ and QJ field lists must have the same number of columns for UNION queries" + ); + } + + // --- ListableCompletedJob serialization --- + + #[test] + fn test_listable_completed_job_skip_none() { + let job = ListableCompletedJob { + r#type: "CompletedJob".to_string(), + workspace_id: "test".to_string(), + id: Uuid::nil(), + parent_job: None, + created_by: "admin".to_string(), + created_at: chrono::Utc::now(), + started_at: None, + duration_ms: 100, + success: true, + script_hash: None, + script_path: None, + deleted: false, + raw_code: None, + canceled: false, + canceled_by: None, + canceled_reason: None, + job_kind: JobKind::Script, + schedule_path: None, + permissioned_as: "u/admin".to_string(), + flow_status: None, + raw_flow: None, + is_flow_step: false, + language: None, + is_skipped: false, + email: "admin@test.com".to_string(), + visible_to_owner: true, + mem_peak: None, + tag: "default".to_string(), + priority: None, + labels: None, + args: None, + }; + + let json = serde_json::to_value(&job).unwrap(); + let obj = json.as_object().unwrap(); + assert!(!obj.contains_key("parent_job")); + assert!(!obj.contains_key("script_hash")); + assert!(!obj.contains_key("raw_code")); + assert!(!obj.contains_key("canceled_by")); + assert!(!obj.contains_key("mem_peak")); + assert!(!obj.contains_key("labels")); + assert!(obj.contains_key("type")); + assert!(obj.contains_key("workspace_id")); + } +} diff --git a/backend/windmill-common/src/webhook.rs b/backend/windmill-common/src/webhook.rs index 44097339ac..ffc50f932d 100644 --- a/backend/windmill-common/src/webhook.rs +++ b/backend/windmill-common/src/webhook.rs @@ -152,3 +152,125 @@ impl WebhookShared { let _ = self.channel.send(WebhookPayload::InstanceEvent(event)); } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_webhook_message_create_script() { + let msg = WebhookMessage::CreateScript { + workspace: "demo".to_string(), + path: "f/test/script".to_string(), + hash: "abc123".to_string(), + }; + let json = serde_json::to_value(&msg).unwrap(); + assert_eq!(json["type"], "CreateScript"); + assert_eq!(json["workspace"], "demo"); + assert_eq!(json["path"], "f/test/script"); + assert_eq!(json["hash"], "abc123"); + } + + #[test] + fn test_webhook_message_update_flow() { + let msg = WebhookMessage::UpdateFlow { + workspace: "staging".to_string(), + old_path: "f/old/flow".to_string(), + new_path: "f/new/flow".to_string(), + }; + let json = serde_json::to_value(&msg).unwrap(); + assert_eq!(json["type"], "UpdateFlow"); + assert_eq!(json["old_path"], "f/old/flow"); + assert_eq!(json["new_path"], "f/new/flow"); + } + + #[test] + fn test_webhook_message_delete_resource() { + let msg = WebhookMessage::DeleteResource { + workspace: "prod".to_string(), + path: "u/admin/db".to_string(), + }; + let json = serde_json::to_value(&msg).unwrap(); + assert_eq!(json["type"], "DeleteResource"); + assert_eq!(json["workspace"], "prod"); + assert_eq!(json["path"], "u/admin/db"); + } + + #[test] + fn test_webhook_message_create_folder() { + let msg = WebhookMessage::CreateFolder { + workspace: "demo".to_string(), + name: "shared".to_string(), + }; + let json = serde_json::to_value(&msg).unwrap(); + assert_eq!(json["type"], "CreateFolder"); + assert_eq!(json["name"], "shared"); + } + + #[test] + fn test_webhook_message_resource_type() { + let msg = WebhookMessage::CreateResourceType { + name: "postgresql".to_string(), + }; + let json = serde_json::to_value(&msg).unwrap(); + assert_eq!(json["type"], "CreateResourceType"); + assert_eq!(json["name"], "postgresql"); + // Should NOT have workspace field + assert!(json.get("workspace").is_none()); + } + + #[test] + fn test_webhook_message_all_variants_have_type_tag() { + let messages: Vec = vec![ + WebhookMessage::CreateApp { workspace: "w".into(), path: "p".into() }, + WebhookMessage::DeleteApp { workspace: "w".into(), path: "p".into() }, + WebhookMessage::UpdateApp { workspace: "w".into(), old_path: "o".into(), new_path: "n".into() }, + WebhookMessage::CreateFlow { workspace: "w".into(), path: "p".into() }, + WebhookMessage::UpdateFlow { workspace: "w".into(), old_path: "o".into(), new_path: "n".into() }, + WebhookMessage::ArchiveFlow { workspace: "w".into(), path: "p".into() }, + WebhookMessage::DeleteFlow { workspace: "w".into(), path: "p".into() }, + WebhookMessage::CreateFolder { workspace: "w".into(), name: "n".into() }, + WebhookMessage::UpdateFolder { workspace: "w".into(), name: "n".into() }, + WebhookMessage::DeleteFolder { workspace: "w".into(), name: "n".into() }, + WebhookMessage::DeleteResource { workspace: "w".into(), path: "p".into() }, + WebhookMessage::CreateResource { workspace: "w".into(), path: "p".into() }, + WebhookMessage::UpdateResource { workspace: "w".into(), old_path: "o".into(), new_path: "n".into() }, + WebhookMessage::CreateResourceType { name: "n".into() }, + WebhookMessage::DeleteResourceType { name: "n".into() }, + WebhookMessage::UpdateResourceType { name: "n".into() }, + WebhookMessage::CreateScript { workspace: "w".into(), path: "p".into(), hash: "h".into() }, + WebhookMessage::UpdateScript { workspace: "w".into(), path: "p".into(), hash: "h".into() }, + WebhookMessage::DeleteScript { workspace: "w".into(), hash: "h".into() }, + WebhookMessage::DeleteScriptPath { workspace: "w".into(), path: "p".into() }, + WebhookMessage::CreateVariable { workspace: "w".into(), path: "p".into() }, + WebhookMessage::UpdateVariable { workspace: "w".into(), old_path: "o".into(), new_path: "n".into() }, + WebhookMessage::DeleteVariable { workspace: "w".into(), path: "p".into() }, + ]; + + for msg in &messages { + let json = serde_json::to_value(msg).unwrap(); + assert!( + json.get("type").is_some(), + "Missing 'type' tag in: {}", + serde_json::to_string(msg).unwrap() + ); + } + } + + #[test] + fn test_webhook_message_type_tags_are_variant_names() { + let msg = WebhookMessage::CreateApp { + workspace: "w".into(), + path: "p".into(), + }; + let json = serde_json::to_value(&msg).unwrap(); + assert_eq!(json["type"], "CreateApp"); + + let msg = WebhookMessage::DeleteVariable { + workspace: "w".into(), + path: "p".into(), + }; + let json = serde_json::to_value(&msg).unwrap(); + assert_eq!(json["type"], "DeleteVariable"); + } +} diff --git a/backend/windmill-trigger-http/src/http_trigger_auth.rs b/backend/windmill-trigger-http/src/http_trigger_auth.rs index 85288d87b8..13462f9156 100644 --- a/backend/windmill-trigger-http/src/http_trigger_auth.rs +++ b/backend/windmill-trigger-http/src/http_trigger_auth.rs @@ -749,3 +749,1026 @@ impl IntoResponse for AuthenticationError { (status, headers, body.to_string()).into_response() } } + +#[cfg(test)] +mod tests { + use super::*; + + // --- calculate_hmac_signature --- + + #[test] + fn test_hmac_sha256_deterministic() { + let sig1 = calculate_hmac_signature(HmacAlgorithm::Sha256, "secret", "payload"); + let sig2 = calculate_hmac_signature(HmacAlgorithm::Sha256, "secret", "payload"); + assert_eq!(sig1, sig2); + } + + #[test] + fn test_hmac_sha256_different_keys() { + let sig1 = calculate_hmac_signature(HmacAlgorithm::Sha256, "key1", "payload"); + let sig2 = calculate_hmac_signature(HmacAlgorithm::Sha256, "key2", "payload"); + assert_ne!(sig1, sig2); + } + + #[test] + fn test_hmac_sha256_different_payloads() { + let sig1 = calculate_hmac_signature(HmacAlgorithm::Sha256, "secret", "payload1"); + let sig2 = calculate_hmac_signature(HmacAlgorithm::Sha256, "secret", "payload2"); + assert_ne!(sig1, sig2); + } + + #[test] + fn test_hmac_sha1_length() { + let sig = calculate_hmac_signature(HmacAlgorithm::Sha1, "secret", "payload"); + assert_eq!(sig.len(), 20); // SHA1 = 160 bits = 20 bytes + } + + #[test] + fn test_hmac_sha256_length() { + let sig = calculate_hmac_signature(HmacAlgorithm::Sha256, "secret", "payload"); + assert_eq!(sig.len(), 32); // SHA256 = 256 bits = 32 bytes + } + + #[test] + fn test_hmac_sha512_length() { + let sig = calculate_hmac_signature(HmacAlgorithm::Sha512, "secret", "payload"); + assert_eq!(sig.len(), 64); // SHA512 = 512 bits = 64 bytes + } + + #[test] + fn test_hmac_sha256_empty_payload() { + let sig = calculate_hmac_signature(HmacAlgorithm::Sha256, "secret", ""); + assert_eq!(sig.len(), 32); + } + + #[test] + fn test_hmac_sha256_empty_key() { + let sig = calculate_hmac_signature(HmacAlgorithm::Sha256, "", "payload"); + assert_eq!(sig.len(), 32); + } + + // --- encode_hmac_signature --- + + #[test] + fn test_encode_hex() { + let bytes = vec![0xde, 0xad, 0xbe, 0xef]; + assert_eq!(encode_hmac_signature(Encoding::Hex, &bytes), "deadbeef"); + } + + #[test] + fn test_encode_base64() { + let bytes = vec![0xde, 0xad, 0xbe, 0xef]; + let encoded = encode_hmac_signature(Encoding::Base64, &bytes); + assert_eq!(BASE64_STANDARD.decode(&encoded).unwrap(), bytes); + } + + #[test] + fn test_encode_base64uri() { + let bytes = vec![0xde, 0xad, 0xbe, 0xef]; + let encoded = encode_hmac_signature(Encoding::Base64Uri, &bytes); + assert_eq!(BASE64_URL_SAFE.decode(&encoded).unwrap(), bytes); + } + + #[test] + fn test_encode_hex_empty() { + assert_eq!(encode_hmac_signature(Encoding::Hex, &[]), ""); + } + + // --- verify_hmac_signature round-trip --- + + fn make_auth_data<'a>( + payload: &'a str, + header_value: &'a str, + prefix: Option<&'a str>, + algorithm: HmacAlgorithm, + encoding: Encoding, + ) -> SignatureAuthenticationData<'a, 'a, 'a> { + SignatureAuthenticationData::new( + Cow::Borrowed(payload), + header_value, + prefix, + SignatureAuthenticationDetails::new(algorithm, encoding), + ) + } + + #[test] + fn test_verify_hmac_sha256_hex_roundtrip() { + let secret = "my_webhook_secret"; + let payload = r#"{"event":"push"}"#; + let sig = calculate_hmac_signature(HmacAlgorithm::Sha256, secret, payload); + let encoded = encode_hmac_signature(Encoding::Hex, &sig); + + let data = make_auth_data(payload, &encoded, None, HmacAlgorithm::Sha256, Encoding::Hex); + assert!(verify_hmac_signature(data, secret).is_ok()); + } + + #[test] + fn test_verify_hmac_sha256_base64_roundtrip() { + let secret = "my_secret"; + let payload = "test body"; + let sig = calculate_hmac_signature(HmacAlgorithm::Sha256, secret, payload); + let encoded = encode_hmac_signature(Encoding::Base64, &sig); + + let data = make_auth_data( + payload, + &encoded, + None, + HmacAlgorithm::Sha256, + Encoding::Base64, + ); + assert!(verify_hmac_signature(data, secret).is_ok()); + } + + #[test] + fn test_verify_hmac_sha512_hex_roundtrip() { + let secret = "long_secret_key"; + let payload = "some data"; + let sig = calculate_hmac_signature(HmacAlgorithm::Sha512, secret, payload); + let encoded = encode_hmac_signature(Encoding::Hex, &sig); + + let data = make_auth_data(payload, &encoded, None, HmacAlgorithm::Sha512, Encoding::Hex); + assert!(verify_hmac_signature(data, secret).is_ok()); + } + + #[test] + fn test_verify_hmac_sha1_hex_roundtrip() { + let secret = "sha1_key"; + let payload = "data"; + let sig = calculate_hmac_signature(HmacAlgorithm::Sha1, secret, payload); + let encoded = encode_hmac_signature(Encoding::Hex, &sig); + + let data = make_auth_data(payload, &encoded, None, HmacAlgorithm::Sha1, Encoding::Hex); + assert!(verify_hmac_signature(data, secret).is_ok()); + } + + #[test] + fn test_verify_with_prefix() { + let secret = "key"; + let payload = "body"; + let sig = calculate_hmac_signature(HmacAlgorithm::Sha256, secret, payload); + let encoded = encode_hmac_signature(Encoding::Hex, &sig); + let header_value = format!("sha256={}", encoded); + + let data = make_auth_data( + payload, + &header_value, + Some("sha256="), + HmacAlgorithm::Sha256, + Encoding::Hex, + ); + assert!(verify_hmac_signature(data, secret).is_ok()); + } + + #[test] + fn test_verify_wrong_signature() { + let data = make_auth_data( + "payload", + "wrong_signature_value", + None, + HmacAlgorithm::Sha256, + Encoding::Hex, + ); + let result = verify_hmac_signature(data, "secret"); + assert!(matches!(result, Err(AuthenticationError::InvalidSignature))); + } + + #[test] + fn test_verify_wrong_key() { + let secret = "correct_key"; + let payload = "body"; + let sig = calculate_hmac_signature(HmacAlgorithm::Sha256, secret, payload); + let encoded = encode_hmac_signature(Encoding::Hex, &sig); + + let data = make_auth_data(payload, &encoded, None, HmacAlgorithm::Sha256, Encoding::Hex); + let result = verify_hmac_signature(data, "wrong_key"); + assert!(matches!(result, Err(AuthenticationError::InvalidSignature))); + } + + #[test] + fn test_verify_wrong_prefix() { + let secret = "key"; + let payload = "body"; + let sig = calculate_hmac_signature(HmacAlgorithm::Sha256, secret, payload); + let encoded = encode_hmac_signature(Encoding::Hex, &sig); + let header_value = format!("v0={}", encoded); + + let data = make_auth_data( + payload, + &header_value, + Some("sha256="), + HmacAlgorithm::Sha256, + Encoding::Hex, + ); + assert!(matches!( + verify_hmac_signature(data, secret), + Err(AuthenticationError::InvalidSignature) + )); + } + + // --- parse_signature --- + + #[test] + fn test_parse_signature_stripe_format() { + let header = "t=1234567890,v1=abc123def456"; + let parsed = parse_signature(header, (",", "=")); + assert_eq!(parsed.get("t"), Some(&"1234567890")); + assert_eq!(parsed.get("v1"), Some(&"abc123def456")); + } + + #[test] + fn test_parse_signature_tiktok_format() { + let header = "t=1234567890,s=signaturevalue"; + let parsed = parse_signature(header, (",", "=")); + assert_eq!(parsed.get("t"), Some(&"1234567890")); + assert_eq!(parsed.get("s"), Some(&"signaturevalue")); + } + + #[test] + fn test_parse_signature_single_entry() { + let header = "key=value"; + let parsed = parse_signature(header, (",", "=")); + assert_eq!(parsed.get("key"), Some(&"value")); + assert_eq!(parsed.len(), 1); + } + + #[test] + fn test_parse_signature_empty_string() { + let parsed = parse_signature("", (",", "=")); + assert!(parsed.is_empty() || parsed.len() == 1); + } + + #[test] + fn test_parse_signature_multiple_entries() { + let header = "a=1,b=2,c=3"; + let parsed = parse_signature(header, (",", "=")); + assert_eq!(parsed.len(), 3); + assert_eq!(parsed.get("a"), Some(&"1")); + assert_eq!(parsed.get("b"), Some(&"2")); + assert_eq!(parsed.get("c"), Some(&"3")); + } + + // --- WebhookType serde --- + + #[test] + fn test_webhook_type_serde_roundtrip() { + for wt in [ + WebhookType::Github, + WebhookType::Slack, + WebhookType::Stripe, + WebhookType::TikTok, + WebhookType::Twitch, + WebhookType::Zoom, + WebhookType::Custom, + ] { + let json = serde_json::to_value(wt).unwrap(); + let deserialized: WebhookType = serde_json::from_value(json).unwrap(); + assert_eq!(wt, deserialized); + } + } + + #[test] + fn test_webhook_type_handler_known_providers() { + assert!(WebhookType::Github.get_webhook_handler().is_some()); + assert!(WebhookType::Slack.get_webhook_handler().is_some()); + assert!(WebhookType::Stripe.get_webhook_handler().is_some()); + assert!(WebhookType::TikTok.get_webhook_handler().is_some()); + assert!(WebhookType::Twitch.get_webhook_handler().is_some()); + assert!(WebhookType::Zoom.get_webhook_handler().is_some()); + } + + #[test] + fn test_webhook_type_custom_has_no_handler() { + assert!(WebhookType::Custom.get_webhook_handler().is_none()); + } + + // --- HmacAlgorithm / Encoding serde --- + + #[test] + fn test_hmac_algorithm_serde() { + assert_eq!( + serde_json::to_value(HmacAlgorithm::Sha1).unwrap(), + "sha1" + ); + assert_eq!( + serde_json::to_value(HmacAlgorithm::Sha256).unwrap(), + "sha256" + ); + assert_eq!( + serde_json::to_value(HmacAlgorithm::Sha512).unwrap(), + "sha512" + ); + } + + #[test] + fn test_encoding_serde() { + assert_eq!(serde_json::to_value(Encoding::Hex).unwrap(), "hex"); + assert_eq!(serde_json::to_value(Encoding::Base64).unwrap(), "base64"); + assert_eq!( + serde_json::to_value(Encoding::Base64Uri).unwrap(), + "base64uri" + ); + } + + // --- TryGetWebhookHeader --- + + #[test] + fn test_try_get_header_present() { + let mut headers = HeaderMap::new(); + headers.insert("X-Custom-Header", HeaderValue::from_static("value123")); + assert_eq!( + headers.try_get_webhook_header("X-Custom-Header").unwrap(), + "value123" + ); + } + + #[test] + fn test_try_get_header_missing() { + let headers = HeaderMap::new(); + let result = headers.try_get_webhook_header("X-Missing"); + assert!(matches!(result, Err(AuthenticationError::MissingHeader(_)))); + } + + #[test] + fn test_try_get_header_case_insensitive() { + let mut headers = HeaderMap::new(); + headers.insert("x-hub-signature-256", HeaderValue::from_static("sig")); + assert_eq!( + headers + .try_get_webhook_header("X-Hub-Signature-256") + .unwrap(), + "sig" + ); + } + + // --- GitHub webhook end-to-end --- + + fn github_headers(secret: &str, payload: &str) -> HeaderMap { + let sig = calculate_hmac_signature(HmacAlgorithm::Sha256, secret, payload); + let encoded = encode_hmac_signature(Encoding::Hex, &sig); + let mut headers = HeaderMap::new(); + headers.insert( + "X-Hub-Signature-256", + HeaderValue::from_str(&format!("sha256={}", encoded)).unwrap(), + ); + headers + } + + #[test] + fn test_github_authenticate_valid() { + let secret = "github_webhook_secret"; + let payload = r#"{"action":"opened","number":1}"#.to_string(); + let headers = github_headers(secret, &payload); + + let method = AuthenticationMethod::Signature(SignatureAuthentication { + signature_provider: WebhookType::Github, + secret_key: secret.to_string(), + authentication_config: None, + }); + assert!(method + .authenticate_http_request(&headers, Some(&payload)) + .is_ok()); + } + + #[test] + fn test_github_authenticate_wrong_secret() { + let payload = r#"{"action":"opened"}"#.to_string(); + let headers = github_headers("correct_secret", &payload); + + let method = AuthenticationMethod::Signature(SignatureAuthentication { + signature_provider: WebhookType::Github, + secret_key: "wrong_secret".to_string(), + authentication_config: None, + }); + assert!(method + .authenticate_http_request(&headers, Some(&payload)) + .is_err()); + } + + #[test] + fn test_github_authenticate_missing_header() { + let payload = r#"{"action":"opened"}"#.to_string(); + let headers = HeaderMap::new(); + + let method = AuthenticationMethod::Signature(SignatureAuthentication { + signature_provider: WebhookType::Github, + secret_key: "secret".to_string(), + authentication_config: None, + }); + assert!(matches!( + method.authenticate_http_request(&headers, Some(&payload)), + Err(AuthenticationError::MissingHeader(_)) + )); + } + + #[test] + fn test_github_authenticate_no_payload() { + let headers = HeaderMap::new(); + let method = AuthenticationMethod::Signature(SignatureAuthentication { + signature_provider: WebhookType::Github, + secret_key: "secret".to_string(), + authentication_config: None, + }); + assert!(matches!( + method.authenticate_http_request(&headers, None), + Err(AuthenticationError::InvalidPayload) + )); + } + + // --- Slack webhook end-to-end --- + + fn slack_headers(secret: &str, payload: &str, timestamp: &str) -> HeaderMap { + let signed_payload = format!("v0:{}:{}", timestamp, payload); + let sig = calculate_hmac_signature(HmacAlgorithm::Sha256, secret, &signed_payload); + let encoded = encode_hmac_signature(Encoding::Hex, &sig); + let mut headers = HeaderMap::new(); + headers.insert( + "X-Slack-Signature", + HeaderValue::from_str(&format!("v0={}", encoded)).unwrap(), + ); + headers.insert( + "X-Slack-Request-Timestamp", + HeaderValue::from_str(timestamp).unwrap(), + ); + headers + } + + #[test] + fn test_slack_authenticate_valid() { + let secret = "slack_signing_secret"; + let payload = "token=xxx&command=%2Ftest".to_string(); + let timestamp = "1531420618"; + let headers = slack_headers(secret, &payload, timestamp); + + let method = AuthenticationMethod::Signature(SignatureAuthentication { + signature_provider: WebhookType::Slack, + secret_key: secret.to_string(), + authentication_config: None, + }); + assert!(method + .authenticate_http_request(&headers, Some(&payload)) + .is_ok()); + } + + #[test] + fn test_slack_authenticate_wrong_timestamp() { + let secret = "slack_secret"; + let payload = "data".to_string(); + let headers = slack_headers(secret, &payload, "1000000000"); + + let method = AuthenticationMethod::Signature(SignatureAuthentication { + signature_provider: WebhookType::Slack, + secret_key: secret.to_string(), + authentication_config: None, + }); + // Constructed with timestamp "1000000000" but that's valid - it just needs to match + assert!(method + .authenticate_http_request(&headers, Some(&payload)) + .is_ok()); + } + + // --- Stripe webhook end-to-end --- + + fn stripe_headers(secret: &str, payload: &str, timestamp: &str) -> HeaderMap { + let signed_payload = format!("{}.{}", timestamp, payload); + let sig = calculate_hmac_signature(HmacAlgorithm::Sha256, secret, &signed_payload); + let encoded = encode_hmac_signature(Encoding::Hex, &sig); + let mut headers = HeaderMap::new(); + headers.insert( + "STRIPE-SIGNATURE", + HeaderValue::from_str(&format!("t={},v1={}", timestamp, encoded)).unwrap(), + ); + headers + } + + #[test] + fn test_stripe_authenticate_valid() { + let secret = "whsec_stripe_secret"; + let payload = r#"{"id":"evt_123"}"#.to_string(); + let timestamp = "1614556800"; + let headers = stripe_headers(secret, &payload, timestamp); + + let method = AuthenticationMethod::Signature(SignatureAuthentication { + signature_provider: WebhookType::Stripe, + secret_key: secret.to_string(), + authentication_config: None, + }); + assert!(method + .authenticate_http_request(&headers, Some(&payload)) + .is_ok()); + } + + #[test] + fn test_stripe_authenticate_wrong_secret() { + let payload = r#"{"id":"evt_123"}"#.to_string(); + let headers = stripe_headers("correct", &payload, "12345"); + + let method = AuthenticationMethod::Signature(SignatureAuthentication { + signature_provider: WebhookType::Stripe, + secret_key: "wrong".to_string(), + authentication_config: None, + }); + assert!(method + .authenticate_http_request(&headers, Some(&payload)) + .is_err()); + } + + // --- TikTok webhook end-to-end --- + + fn tiktok_headers(secret: &str, payload: &str, timestamp: &str) -> HeaderMap { + let signed_payload = format!("{}.{}", timestamp, payload); + let sig = calculate_hmac_signature(HmacAlgorithm::Sha256, secret, &signed_payload); + let encoded = encode_hmac_signature(Encoding::Hex, &sig); + let mut headers = HeaderMap::new(); + headers.insert( + "TikTok-Signature", + HeaderValue::from_str(&format!("t={},s={}", timestamp, encoded)).unwrap(), + ); + headers + } + + #[test] + fn test_tiktok_authenticate_valid() { + let secret = "tiktok_secret"; + let payload = r#"{"event":"video.upload"}"#.to_string(); + let timestamp = "1700000000"; + let headers = tiktok_headers(secret, &payload, timestamp); + + let method = AuthenticationMethod::Signature(SignatureAuthentication { + signature_provider: WebhookType::TikTok, + secret_key: secret.to_string(), + authentication_config: None, + }); + assert!(method + .authenticate_http_request(&headers, Some(&payload)) + .is_ok()); + } + + // --- Twitch webhook end-to-end --- + + fn twitch_headers( + secret: &str, + payload: &str, + message_id: &str, + timestamp: &str, + message_type: &str, + ) -> HeaderMap { + let message = format!("{}{}{}", message_id, timestamp, payload); + let sig = calculate_hmac_signature(HmacAlgorithm::Sha256, secret, &message); + let encoded = encode_hmac_signature(Encoding::Hex, &sig); + let mut headers = HeaderMap::new(); + headers.insert( + "Twitch-Eventsub-Message-Signature", + HeaderValue::from_str(&format!("sha256={}", encoded)).unwrap(), + ); + headers.insert( + "Twitch-Eventsub-Message-Id", + HeaderValue::from_str(message_id).unwrap(), + ); + headers.insert( + "Twitch-Eventsub-Message-Timestamp", + HeaderValue::from_str(timestamp).unwrap(), + ); + headers.insert( + "Twitch-Eventsub-Message-Type", + HeaderValue::from_str(message_type).unwrap(), + ); + headers + } + + #[test] + fn test_twitch_authenticate_valid_notification() { + let secret = "twitch_secret"; + let payload = r#"{"subscription":{},"event":{"user_id":"123"}}"#.to_string(); + let headers = twitch_headers(secret, &payload, "msg-123", "2024-01-01T00:00:00Z", "notification"); + + let method = AuthenticationMethod::Signature(SignatureAuthentication { + signature_provider: WebhookType::Twitch, + secret_key: secret.to_string(), + authentication_config: None, + }); + assert!(method + .authenticate_http_request(&headers, Some(&payload)) + .is_ok()); + } + + #[test] + fn test_twitch_challenge_response() { + let secret = "twitch_secret"; + let payload = r#"{"challenge":"test_challenge_string","subscription":{"id":"sub-123"}}"#; + let headers = twitch_headers( + secret, + payload, + "msg-456", + "2024-01-01T00:00:00Z", + "webhook_callback_verification", + ); + + let handler = WebhookType::Twitch.get_webhook_handler().unwrap(); + let config_data = SignatureConfigData { secret_key: secret }; + let response = handler + .handle_challenge_request(&headers, &config_data, payload) + .unwrap(); + assert!(response.is_some()); + } + + #[test] + fn test_twitch_non_challenge_returns_none() { + let secret = "twitch_secret"; + let payload = r#"{"subscription":{},"event":{}}"#; + let headers = twitch_headers( + secret, + payload, + "msg-789", + "2024-01-01T00:00:00Z", + "notification", + ); + + let handler = WebhookType::Twitch.get_webhook_handler().unwrap(); + let config_data = SignatureConfigData { secret_key: secret }; + let response = handler + .handle_challenge_request(&headers, &config_data, payload) + .unwrap(); + assert!(response.is_none()); + } + + // --- Zoom webhook end-to-end --- + + fn zoom_headers(secret: &str, payload: &str, timestamp: &str) -> HeaderMap { + let message = format!("v0:{}:{}", timestamp, payload); + let sig = calculate_hmac_signature(HmacAlgorithm::Sha256, secret, &message); + let encoded = encode_hmac_signature(Encoding::Hex, &sig); + let mut headers = HeaderMap::new(); + headers.insert( + "x-zm-signature", + HeaderValue::from_str(&format!("v0={}", encoded)).unwrap(), + ); + headers.insert( + "x-zm-request-timestamp", + HeaderValue::from_str(timestamp).unwrap(), + ); + headers + } + + #[test] + fn test_zoom_authenticate_valid() { + let secret = "zoom_secret"; + let payload = r#"{"event":"meeting.started"}"#.to_string(); + let timestamp = "1700000000"; + let headers = zoom_headers(secret, &payload, timestamp); + + let method = AuthenticationMethod::Signature(SignatureAuthentication { + signature_provider: WebhookType::Zoom, + secret_key: secret.to_string(), + authentication_config: None, + }); + assert!(method + .authenticate_http_request(&headers, Some(&payload)) + .is_ok()); + } + + #[test] + fn test_zoom_challenge_response() { + let secret = "zoom_secret"; + let payload = r#"{"event":"endpoint.url_validation","event_ts":1234567890,"payload":{"plainToken":"abc123"}}"#; + + let handler = WebhookType::Zoom.get_webhook_handler().unwrap(); + let config_data = SignatureConfigData { secret_key: secret }; + let response = handler + .handle_challenge_request(&HeaderMap::new(), &config_data, payload) + .unwrap(); + assert!(response.is_some()); + } + + #[test] + fn test_zoom_non_challenge_returns_none() { + let payload = r#"{"event":"meeting.started","event_ts":1234567890,"payload":{"plainToken":"abc"}}"#; + + let handler = WebhookType::Zoom.get_webhook_handler().unwrap(); + let config_data = SignatureConfigData { secret_key: "secret" }; + let response = handler + .handle_challenge_request(&HeaderMap::new(), &config_data, payload) + .unwrap(); + assert!(response.is_none()); + } + + // --- Custom webhook end-to-end --- + + #[test] + fn test_custom_signature_authenticate_valid() { + let secret = "custom_key"; + let payload = "custom body".to_string(); + let sig = calculate_hmac_signature(HmacAlgorithm::Sha256, secret, &payload); + let encoded = encode_hmac_signature(Encoding::Hex, &sig); + + let mut headers = HeaderMap::new(); + headers.insert( + "X-My-Signature", + HeaderValue::from_str(&encoded).unwrap(), + ); + + let method = AuthenticationMethod::Signature(SignatureAuthentication { + signature_provider: WebhookType::Custom, + secret_key: secret.to_string(), + authentication_config: Some(SignatureAuthenticationMethod { + algorithm: HmacAlgorithm::Sha256, + encoding: Encoding::Hex, + signature_header_name: "X-My-Signature".to_string(), + signature_prefix: None, + }), + }); + assert!(method + .authenticate_http_request(&headers, Some(&payload)) + .is_ok()); + } + + #[test] + fn test_custom_signature_with_prefix() { + let secret = "key"; + let payload = "body".to_string(); + let sig = calculate_hmac_signature(HmacAlgorithm::Sha512, secret, &payload); + let encoded = encode_hmac_signature(Encoding::Base64, &sig); + + let mut headers = HeaderMap::new(); + headers.insert( + "X-Sig", + HeaderValue::from_str(&format!("hmac={}", encoded)).unwrap(), + ); + + let method = AuthenticationMethod::Signature(SignatureAuthentication { + signature_provider: WebhookType::Custom, + secret_key: secret.to_string(), + authentication_config: Some(SignatureAuthenticationMethod { + algorithm: HmacAlgorithm::Sha512, + encoding: Encoding::Base64, + signature_header_name: "X-Sig".to_string(), + signature_prefix: Some("hmac=".to_string()), + }), + }); + assert!(method + .authenticate_http_request(&headers, Some(&payload)) + .is_ok()); + } + + #[test] + fn test_custom_signature_missing_config() { + let payload = "body".to_string(); + let headers = HeaderMap::new(); + + let method = AuthenticationMethod::Signature(SignatureAuthentication { + signature_provider: WebhookType::Custom, + secret_key: "secret".to_string(), + authentication_config: None, + }); + assert!(matches!( + method.authenticate_http_request(&headers, Some(&payload)), + Err(AuthenticationError::InvalidCustomConfig) + )); + } + + // --- API key authentication --- + + #[test] + fn test_api_key_authenticate_valid() { + let mut headers = HeaderMap::new(); + headers.insert("X-API-Key", HeaderValue::from_static("my_secret_key_123")); + + let method = AuthenticationMethod::ApiKey(ApiKeyAuthentication { + api_key_header: "X-API-Key".to_string(), + api_key_secret: "my_secret_key_123".to_string(), + }); + assert!(method.authenticate_http_request(&headers, None).is_ok()); + } + + #[test] + fn test_api_key_authenticate_wrong_key() { + let mut headers = HeaderMap::new(); + headers.insert("X-API-Key", HeaderValue::from_static("wrong_key")); + + let method = AuthenticationMethod::ApiKey(ApiKeyAuthentication { + api_key_header: "X-API-Key".to_string(), + api_key_secret: "correct_key".to_string(), + }); + assert!(matches!( + method.authenticate_http_request(&headers, None), + Err(AuthenticationError::InvalidApiKey) + )); + } + + #[test] + fn test_api_key_authenticate_missing_header() { + let headers = HeaderMap::new(); + + let method = AuthenticationMethod::ApiKey(ApiKeyAuthentication { + api_key_header: "X-API-Key".to_string(), + api_key_secret: "secret".to_string(), + }); + assert!(matches!( + method.authenticate_http_request(&headers, None), + Err(AuthenticationError::InvalidApiKey) + )); + } + + // --- Basic auth --- + + fn basic_auth_header(username: &str, password: &str) -> HeaderMap { + let credentials = BASE64_STANDARD.encode(format!("{}:{}", username, password)); + let mut headers = HeaderMap::new(); + headers.insert( + "Authorization", + HeaderValue::from_str(&format!("Basic {}", credentials)).unwrap(), + ); + headers + } + + #[test] + fn test_basic_auth_valid() { + let headers = basic_auth_header("admin", "password123"); + + let method = AuthenticationMethod::BasicAuth(BasicAuthAuthentication { + username: "admin".to_string(), + password: "password123".to_string(), + }); + assert!(method.authenticate_http_request(&headers, None).is_ok()); + } + + #[test] + fn test_basic_auth_wrong_password() { + let headers = basic_auth_header("admin", "wrong"); + + let method = AuthenticationMethod::BasicAuth(BasicAuthAuthentication { + username: "admin".to_string(), + password: "correct".to_string(), + }); + assert!(matches!( + method.authenticate_http_request(&headers, None), + Err(AuthenticationError::UnauthorizedBasicHttpAuth) + )); + } + + #[test] + fn test_basic_auth_wrong_username() { + let headers = basic_auth_header("wrong_user", "password"); + + let method = AuthenticationMethod::BasicAuth(BasicAuthAuthentication { + username: "admin".to_string(), + password: "password".to_string(), + }); + assert!(matches!( + method.authenticate_http_request(&headers, None), + Err(AuthenticationError::UnauthorizedBasicHttpAuth) + )); + } + + #[test] + fn test_basic_auth_missing_header() { + let headers = HeaderMap::new(); + + let method = AuthenticationMethod::BasicAuth(BasicAuthAuthentication { + username: "admin".to_string(), + password: "password".to_string(), + }); + assert!(matches!( + method.authenticate_http_request(&headers, None), + Err(AuthenticationError::UnauthorizedBasicHttpAuth) + )); + } + + #[test] + fn test_basic_auth_bearer_instead_of_basic() { + let mut headers = HeaderMap::new(); + headers.insert( + "Authorization", + HeaderValue::from_static("Bearer sometoken"), + ); + + let method = AuthenticationMethod::BasicAuth(BasicAuthAuthentication { + username: "admin".to_string(), + password: "password".to_string(), + }); + assert!(matches!( + method.authenticate_http_request(&headers, None), + Err(AuthenticationError::UnauthorizedBasicHttpAuth) + )); + } + + #[test] + fn test_basic_auth_invalid_base64() { + let mut headers = HeaderMap::new(); + headers.insert( + "Authorization", + HeaderValue::from_static("Basic !!!invalid!!!"), + ); + + let method = AuthenticationMethod::BasicAuth(BasicAuthAuthentication { + username: "admin".to_string(), + password: "password".to_string(), + }); + assert!(matches!( + method.authenticate_http_request(&headers, None), + Err(AuthenticationError::UnauthorizedBasicHttpAuth) + )); + } + + // --- AuthenticationMethod serde (untagged enum) --- + + #[test] + fn test_authentication_method_signature_serde() { + let method = AuthenticationMethod::Signature(SignatureAuthentication { + signature_provider: WebhookType::Github, + secret_key: "secret".to_string(), + authentication_config: None, + }); + let json = serde_json::to_value(&method).unwrap(); + assert_eq!(json["signature_provider"], "Github"); + assert_eq!(json["secret_key"], "secret"); + + let deserialized: AuthenticationMethod = serde_json::from_value(json).unwrap(); + match deserialized { + AuthenticationMethod::Signature(sig) => { + assert_eq!(sig.signature_provider, WebhookType::Github); + assert_eq!(sig.secret_key, "secret"); + } + _ => panic!("expected Signature variant"), + } + } + + #[test] + fn test_authentication_method_api_key_serde() { + let method = AuthenticationMethod::ApiKey(ApiKeyAuthentication { + api_key_header: "X-Key".to_string(), + api_key_secret: "val".to_string(), + }); + let json = serde_json::to_value(&method).unwrap(); + let deserialized: AuthenticationMethod = serde_json::from_value(json).unwrap(); + match deserialized { + AuthenticationMethod::ApiKey(ak) => { + assert_eq!(ak.api_key_header, "X-Key"); + assert_eq!(ak.api_key_secret, "val"); + } + _ => panic!("expected ApiKey variant"), + } + } + + #[test] + fn test_authentication_method_basic_auth_serde() { + let method = AuthenticationMethod::BasicAuth(BasicAuthAuthentication { + username: "user".to_string(), + password: "pass".to_string(), + }); + let json = serde_json::to_value(&method).unwrap(); + let deserialized: AuthenticationMethod = serde_json::from_value(json).unwrap(); + match deserialized { + AuthenticationMethod::BasicAuth(ba) => { + assert_eq!(ba.username, "user"); + assert_eq!(ba.password, "pass"); + } + _ => panic!("expected BasicAuth variant"), + } + } + + // --- AuthenticationError into_response --- + + #[test] + fn test_error_invalid_signature_is_401() { + let response = AuthenticationError::InvalidSignature.into_response(); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + } + + #[test] + fn test_error_missing_header_is_400() { + let response = + AuthenticationError::MissingHeader("X-Sig".to_string()).into_response(); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + } + + #[test] + fn test_error_invalid_payload_is_400() { + let response = AuthenticationError::InvalidPayload.into_response(); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + } + + #[test] + fn test_error_invalid_api_key_is_401() { + let response = AuthenticationError::InvalidApiKey.into_response(); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + } + + #[test] + fn test_error_basic_auth_has_www_authenticate() { + let response = AuthenticationError::UnauthorizedBasicHttpAuth.into_response(); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + assert!(response.headers().contains_key("www-authenticate")); + } + + #[test] + fn test_error_invalid_custom_config_is_400() { + let response = AuthenticationError::InvalidCustomConfig.into_response(); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + } + + #[test] + fn test_error_invalid_timestamp_is_400() { + let response = AuthenticationError::InvalidTimestamp.into_response(); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + } +} diff --git a/backend/windmill-trigger-http/src/lib.rs b/backend/windmill-trigger-http/src/lib.rs index c0f46efc2f..2c109ff183 100644 --- a/backend/windmill-trigger-http/src/lib.rs +++ b/backend/windmill-trigger-http/src/lib.rs @@ -410,4 +410,186 @@ mod tests { let config: HttpConfigRequest = serde_json::from_str(json_both).unwrap(); assert_eq!(config.request_type, RequestType::SyncSse); } + + // --- HttpMethod --- + + #[test] + fn test_http_method_from_http_get() { + let method = HttpMethod::try_from(&http::Method::GET).unwrap(); + assert_eq!(method, HttpMethod::Get); + } + + #[test] + fn test_http_method_from_http_post() { + let method = HttpMethod::try_from(&http::Method::POST).unwrap(); + assert_eq!(method, HttpMethod::Post); + } + + #[test] + fn test_http_method_from_http_put() { + let method = HttpMethod::try_from(&http::Method::PUT).unwrap(); + assert_eq!(method, HttpMethod::Put); + } + + #[test] + fn test_http_method_from_http_delete() { + let method = HttpMethod::try_from(&http::Method::DELETE).unwrap(); + assert_eq!(method, HttpMethod::Delete); + } + + #[test] + fn test_http_method_from_http_patch() { + let method = HttpMethod::try_from(&http::Method::PATCH).unwrap(); + assert_eq!(method, HttpMethod::Patch); + } + + #[test] + fn test_http_method_unsupported() { + let result = HttpMethod::try_from(&http::Method::HEAD); + assert!(result.is_err()); + } + + #[test] + fn test_http_method_options_unsupported() { + let result = HttpMethod::try_from(&http::Method::OPTIONS); + assert!(result.is_err()); + } + + // --- HttpMethod serde --- + + #[test] + fn test_http_method_serde_roundtrip() { + for method in [HttpMethod::Get, HttpMethod::Post, HttpMethod::Put, HttpMethod::Delete, HttpMethod::Patch] { + let json = serde_json::to_value(method).unwrap(); + let deserialized: HttpMethod = serde_json::from_value(json).unwrap(); + assert_eq!(method, deserialized); + } + } + + #[test] + fn test_http_method_serialize_lowercase() { + assert_eq!(serde_json::to_value(HttpMethod::Get).unwrap(), "get"); + assert_eq!(serde_json::to_value(HttpMethod::Post).unwrap(), "post"); + } + + // --- RequestType serde --- + + #[test] + fn test_request_type_serde_roundtrip() { + for rt in [RequestType::Sync, RequestType::Async, RequestType::SyncSse] { + let json = serde_json::to_value(rt).unwrap(); + let deserialized: RequestType = serde_json::from_value(json).unwrap(); + assert_eq!(rt, deserialized); + } + } + + #[test] + fn test_request_type_serialize_values() { + assert_eq!(serde_json::to_value(RequestType::Sync).unwrap(), "sync"); + assert_eq!(serde_json::to_value(RequestType::Async).unwrap(), "async"); + assert_eq!(serde_json::to_value(RequestType::SyncSse).unwrap(), "sync_sse"); + } + + // --- AuthenticationMethod serde --- + + #[test] + fn test_authentication_method_serde_roundtrip() { + for method in [ + AuthenticationMethod::None, + AuthenticationMethod::Windmill, + AuthenticationMethod::ApiKey, + AuthenticationMethod::BasicHttp, + AuthenticationMethod::CustomScript, + AuthenticationMethod::Signature, + ] { + let json = serde_json::to_value(method).unwrap(); + let deserialized: AuthenticationMethod = serde_json::from_value(json).unwrap(); + assert_eq!(method, deserialized); + } + } + + // --- validate_authentication_method --- + + #[test] + fn test_validate_auth_none_ok() { + assert!(validate_authentication_method(AuthenticationMethod::None, None).is_ok()); + } + + #[test] + fn test_validate_auth_windmill_ok() { + assert!(validate_authentication_method(AuthenticationMethod::Windmill, None).is_ok()); + } + + #[test] + fn test_validate_auth_custom_script_requires_raw() { + assert!(validate_authentication_method(AuthenticationMethod::CustomScript, None).is_err()); + assert!(validate_authentication_method(AuthenticationMethod::CustomScript, Some(false)).is_err()); + assert!(validate_authentication_method(AuthenticationMethod::CustomScript, Some(true)).is_ok()); + } + + #[test] + fn test_validate_auth_signature_without_raw_ok() { + assert!(validate_authentication_method(AuthenticationMethod::Signature, None).is_ok()); + } + + // --- Route path regex --- + + #[test] + fn test_valid_route_path() { + assert!(VALID_ROUTE_PATH_RE.is_match("users")); + assert!(VALID_ROUTE_PATH_RE.is_match("users/:id")); + assert!(VALID_ROUTE_PATH_RE.is_match("api/v1/users")); + assert!(VALID_ROUTE_PATH_RE.is_match("api/v1/:id")); + assert!(VALID_ROUTE_PATH_RE.is_match("files/*path")); + } + + #[test] + fn test_invalid_route_path() { + assert!(!VALID_ROUTE_PATH_RE.is_match("")); + assert!(!VALID_ROUTE_PATH_RE.is_match("/leading-slash")); + } + + #[test] + fn test_route_path_key_regex() { + assert!(ROUTE_PATH_KEY_RE.is_match("/:id")); + assert!(ROUTE_PATH_KEY_RE.is_match("/*path")); + assert!(ROUTE_PATH_KEY_RE.is_match("/users/:userId/posts/:postId")); + } + + // --- HttpConfig deserialization --- + + #[test] + fn test_http_config_request_full() { + let json = r#"{ + "route_path": "api/v1/users", + "request_type": "async", + "authentication_method": "api_key", + "http_method": "post", + "is_static_website": false, + "workspaced_route": true, + "wrap_body": true, + "raw_string": false + }"#; + let config: HttpConfigRequest = serde_json::from_str(json).unwrap(); + assert_eq!(config.route_path, "api/v1/users"); + assert_eq!(config.request_type, RequestType::Async); + assert_eq!(config.authentication_method, AuthenticationMethod::ApiKey); + assert_eq!(config.http_method, HttpMethod::Post); + assert_eq!(config.workspaced_route, Some(true)); + assert_eq!(config.wrap_body, Some(true)); + } + + #[test] + fn test_http_config_request_minimal() { + let json = r#"{ + "authentication_method": "none", + "http_method": "get", + "is_static_website": false + }"#; + let config: HttpConfigRequest = serde_json::from_str(json).unwrap(); + assert_eq!(config.route_path, ""); + assert_eq!(config.request_type, RequestType::Sync); + assert!(config.workspaced_route.is_none()); + assert!(config.summary.is_none()); + } } diff --git a/backend/windmill-trigger-postgres/src/bool.rs b/backend/windmill-trigger-postgres/src/bool.rs index 13b0a8fa68..2acf42b460 100644 --- a/backend/windmill-trigger-postgres/src/bool.rs +++ b/backend/windmill-trigger-postgres/src/bool.rs @@ -22,3 +22,36 @@ pub fn parse_bool(s: &str) -> Result { _ => Err(ParseBoolError::InvalidInput(s.to_string())), } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_parse_true() { + assert_eq!(parse_bool("t").unwrap(), true); + } + + #[test] + fn test_parse_false() { + assert_eq!(parse_bool("f").unwrap(), false); + } + + #[test] + fn test_invalid_true_string() { + assert!(matches!( + parse_bool("true"), + Err(ParseBoolError::InvalidInput(s)) if s == "true" + )); + } + + #[test] + fn test_invalid_empty() { + assert!(parse_bool("").is_err()); + } + + #[test] + fn test_invalid_uppercase() { + assert!(parse_bool("T").is_err()); + } +} diff --git a/backend/windmill-trigger-postgres/src/converter.rs b/backend/windmill-trigger-postgres/src/converter.rs index 49162afd82..172b62df77 100644 --- a/backend/windmill-trigger-postgres/src/converter.rs +++ b/backend/windmill-trigger-postgres/src/converter.rs @@ -253,3 +253,244 @@ impl Converter { Ok(arr) } } + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + // --- Scalar type conversions --- + + #[test] + fn test_bool_true() { + let result = Converter::try_from_str(Some(Type::BOOL), "t").unwrap(); + assert_eq!(result, Value::Bool(true)); + } + + #[test] + fn test_bool_false() { + let result = Converter::try_from_str(Some(Type::BOOL), "f").unwrap(); + assert_eq!(result, Value::Bool(false)); + } + + #[test] + fn test_text() { + let result = Converter::try_from_str(Some(Type::TEXT), "hello world").unwrap(); + assert_eq!(result, Value::String("hello world".to_string())); + } + + #[test] + fn test_varchar() { + let result = Converter::try_from_str(Some(Type::VARCHAR), "test").unwrap(); + assert_eq!(result, Value::String("test".to_string())); + } + + #[test] + fn test_int2() { + let result = Converter::try_from_str(Some(Type::INT2), "42").unwrap(); + assert_eq!(result, json!(42)); + } + + #[test] + fn test_int4() { + let result = Converter::try_from_str(Some(Type::INT4), "-100").unwrap(); + assert_eq!(result, json!(-100)); + } + + #[test] + fn test_int8() { + let result = Converter::try_from_str(Some(Type::INT8), "9999999999").unwrap(); + assert_eq!(result, json!(9999999999i64)); + } + + #[test] + fn test_float4() { + let result = Converter::try_from_str(Some(Type::FLOAT4), "3.14").unwrap(); + assert!(result.as_f64().unwrap() - 3.14 < 0.001); + } + + #[test] + fn test_float8() { + let result = Converter::try_from_str(Some(Type::FLOAT8), "2.718281828").unwrap(); + assert!(result.as_f64().unwrap() - 2.718281828 < 0.0001); + } + + #[test] + fn test_numeric() { + let result = Converter::try_from_str(Some(Type::NUMERIC), "123.456").unwrap(); + assert_eq!(result.to_string(), "123.456"); + } + + #[test] + fn test_uuid() { + let result = + Converter::try_from_str(Some(Type::UUID), "550e8400-e29b-41d4-a716-446655440000") + .unwrap(); + assert_eq!( + result, + Value::String("550e8400-e29b-41d4-a716-446655440000".to_string()) + ); + } + + #[test] + fn test_json() { + let result = + Converter::try_from_str(Some(Type::JSON), r#"{"key": "value", "n": 1}"#).unwrap(); + assert_eq!(result, json!({"key": "value", "n": 1})); + } + + #[test] + fn test_jsonb() { + let result = Converter::try_from_str(Some(Type::JSONB), r#"[1,2,3]"#).unwrap(); + assert_eq!(result, json!([1, 2, 3])); + } + + #[test] + fn test_date() { + let result = Converter::try_from_str(Some(Type::DATE), "2024-01-15").unwrap(); + assert_eq!(result, Value::String("2024-01-15".to_string())); + } + + #[test] + fn test_time() { + let result = Converter::try_from_str(Some(Type::TIME), "14:30:00.0").unwrap(); + assert_eq!(result, Value::String("14:30:00".to_string())); + } + + #[test] + fn test_timestamp() { + let result = + Converter::try_from_str(Some(Type::TIMESTAMP), "2024-01-15 14:30:00.0").unwrap(); + assert_eq!( + result, + Value::String("2024-01-15 14:30:00".to_string()) + ); + } + + #[test] + fn test_timestamptz() { + let result = Converter::try_from_str( + Some(Type::TIMESTAMPTZ), + "2024-01-15 14:30:00.0+00", + ) + .unwrap(); + assert!(result.as_str().unwrap().contains("2024-01-15")); + } + + #[test] + fn test_bytea() { + let result = Converter::try_from_str(Some(Type::BYTEA), "\\x48656c6c6f").unwrap(); + assert_eq!(result, json!([72, 101, 108, 108, 111])); + } + + #[test] + fn test_oid() { + let result = Converter::try_from_str(Some(Type::OID), "12345").unwrap(); + assert_eq!(result, json!(12345u32)); + } + + #[test] + fn test_none_type_defaults_to_text() { + let result = Converter::try_from_str(None, "anything").unwrap(); + assert_eq!(result, Value::String("anything".to_string())); + } + + // --- Array type conversions --- + + #[test] + fn test_int4_array() { + let result = Converter::try_from_str(Some(Type::INT4_ARRAY), "{1,2,3}").unwrap(); + assert_eq!(result, json!([1, 2, 3])); + } + + #[test] + fn test_text_array() { + let result = Converter::try_from_str(Some(Type::TEXT_ARRAY), "{hello,world}").unwrap(); + assert_eq!(result, json!(["hello", "world"])); + } + + #[test] + fn test_bool_array() { + let result = Converter::try_from_str(Some(Type::BOOL_ARRAY), "{t,f,t}").unwrap(); + assert_eq!(result, json!([true, false, true])); + } + + #[test] + fn test_array_with_null() { + let result = Converter::try_from_str(Some(Type::INT4_ARRAY), "{1,NULL,3}").unwrap(); + assert_eq!(result, json!([1, null, 3])); + } + + #[test] + fn test_array_with_quoted_strings() { + let result = + Converter::try_from_str(Some(Type::TEXT_ARRAY), r#"{"hello, world","test"}"#).unwrap(); + assert_eq!(result, json!(["hello, world", "test"])); + } + + #[test] + fn test_empty_array() { + let result = Converter::try_from_str(Some(Type::INT4_ARRAY), "{}").unwrap(); + assert_eq!(result, json!([])); + } + + #[test] + fn test_uuid_array() { + let result = Converter::try_from_str( + Some(Type::UUID_ARRAY), + "{550e8400-e29b-41d4-a716-446655440000,6ba7b810-9dad-11d1-80b4-00c04fd430c8}", + ) + .unwrap(); + assert_eq!( + result, + json!([ + "550e8400-e29b-41d4-a716-446655440000", + "6ba7b810-9dad-11d1-80b4-00c04fd430c8" + ]) + ); + } + + // --- Error cases --- + + #[test] + fn test_invalid_int() { + assert!(Converter::try_from_str(Some(Type::INT4), "not_a_number").is_err()); + } + + #[test] + fn test_invalid_bool() { + assert!(Converter::try_from_str(Some(Type::BOOL), "yes").is_err()); + } + + #[test] + fn test_invalid_uuid() { + assert!(Converter::try_from_str(Some(Type::UUID), "not-a-uuid").is_err()); + } + + #[test] + fn test_invalid_json() { + assert!(Converter::try_from_str(Some(Type::JSON), "not json").is_err()); + } + + #[test] + fn test_array_missing_braces() { + assert!(Converter::try_from_str(Some(Type::INT4_ARRAY), "1,2,3").is_err()); + } + + #[test] + fn test_array_too_short() { + assert!(Converter::try_from_str(Some(Type::INT4_ARRAY), "{").is_err()); + } + + #[test] + fn test_array_with_escaped_backslash() { + let result = + Converter::try_from_str(Some(Type::TEXT_ARRAY), r#"{"a\\b","c"}"#).unwrap(); + assert_eq!(result, json!(["a\\b", "c"])); + } + + #[test] + fn test_float_nan_rejected() { + assert!(Converter::try_from_str(Some(Type::FLOAT4), "NaN").is_err()); + } +} diff --git a/backend/windmill-trigger-postgres/src/hex.rs b/backend/windmill-trigger-postgres/src/hex.rs index f0d155b554..9c8c914f2c 100644 --- a/backend/windmill-trigger-postgres/src/hex.rs +++ b/backend/windmill-trigger-postgres/src/hex.rs @@ -41,3 +41,86 @@ pub fn from_bytea_hex(s: &str) -> Result, ByteaHexParseError> { Ok(result) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_valid_bytea_hex() { + assert_eq!(from_bytea_hex("\\x48656c6c6f").unwrap(), b"Hello"); + } + + #[test] + fn test_empty_bytea_hex() { + assert_eq!(from_bytea_hex("\\x").unwrap(), Vec::::new()); + } + + #[test] + fn test_single_byte() { + assert_eq!(from_bytea_hex("\\xff").unwrap(), vec![0xff]); + } + + #[test] + fn test_all_zeros() { + assert_eq!(from_bytea_hex("\\x000000").unwrap(), vec![0, 0, 0]); + } + + #[test] + fn test_missing_prefix() { + assert!(matches!( + from_bytea_hex("48656c6c6f"), + Err(ByteaHexParseError::InvalidPrefix) + )); + } + + #[test] + fn test_wrong_prefix() { + assert!(matches!( + from_bytea_hex("0x48656c6c6f"), + Err(ByteaHexParseError::InvalidPrefix) + )); + } + + #[test] + fn test_too_short() { + assert!(matches!( + from_bytea_hex("\\"), + Err(ByteaHexParseError::InvalidPrefix) + )); + } + + #[test] + fn test_empty_string() { + assert!(matches!( + from_bytea_hex(""), + Err(ByteaHexParseError::InvalidPrefix) + )); + } + + #[test] + fn test_odd_digits() { + assert!(matches!( + from_bytea_hex("\\xabc"), + Err(ByteaHexParseError::OddNumerOfDigits) + )); + } + + #[test] + fn test_invalid_hex_chars() { + assert!(matches!( + from_bytea_hex("\\xzz"), + Err(ByteaHexParseError::ParseInt(_)) + )); + } + + #[test] + fn test_uppercase_hex() { + assert_eq!(from_bytea_hex("\\xABCD").unwrap(), vec![0xab, 0xcd]); + } + + #[test] + fn test_mixed_case_hex() { + assert_eq!(from_bytea_hex("\\xAbCd").unwrap(), vec![0xab, 0xcd]); + } +} diff --git a/backend/windmill-trigger-postgres/src/lib.rs b/backend/windmill-trigger-postgres/src/lib.rs index a1fef1cfa8..011453739c 100644 --- a/backend/windmill-trigger-postgres/src/lib.rs +++ b/backend/windmill-trigger-postgres/src/lib.rs @@ -550,3 +550,163 @@ pub fn generate_random_string() -> String { format!("{}_{}", timestamp, random_part) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_valid_publication_data() { + let json = r#"{ + "table_to_track": [ + { + "schema_name": "public", + "table_to_track": [ + {"table_name": "users"} + ] + } + ], + "transaction_to_track": ["insert", "update"] + }"#; + let result: std::result::Result = serde_json::from_str(json); + assert!(result.is_ok()); + } + + #[test] + fn test_publication_data_empty_schema_name() { + let json = r#"{ + "table_to_track": [ + { + "schema_name": "", + "table_to_track": [] + } + ], + "transaction_to_track": ["insert"] + }"#; + let result: std::result::Result = serde_json::from_str(json); + assert!(result.is_err()); + } + + #[test] + fn test_publication_data_empty_table_name() { + let json = r#"{ + "table_to_track": [ + { + "schema_name": "public", + "table_to_track": [ + {"table_name": " "} + ] + } + ], + "transaction_to_track": ["insert"] + }"#; + let result: std::result::Result = serde_json::from_str(json); + assert!(result.is_err()); + } + + #[test] + fn test_publication_data_invalid_transaction_type() { + let json = r#"{ + "transaction_to_track": ["insert", "truncate"] + }"#; + let result: std::result::Result = serde_json::from_str(json); + assert!(result.is_err()); + } + + #[test] + fn test_publication_data_too_many_transaction_types() { + let json = r#"{ + "transaction_to_track": ["insert", "update", "delete", "insert"] + }"#; + let result: std::result::Result = serde_json::from_str(json); + assert!(result.is_err()); + } + + #[test] + fn test_publication_data_duplicate_schema_names() { + let json = r#"{ + "table_to_track": [ + {"schema_name": "public", "table_to_track": [{"table_name": "a"}]}, + {"schema_name": "public", "table_to_track": [{"table_name": "b"}]} + ], + "transaction_to_track": ["insert"] + }"#; + let result: std::result::Result = serde_json::from_str(json); + assert!(result.is_err()); + } + + #[test] + fn test_publication_data_all_tables_in_schema() { + let json = r#"{ + "table_to_track": [ + {"schema_name": "public", "table_to_track": []} + ], + "transaction_to_track": ["insert"] + }"#; + let result: std::result::Result = serde_json::from_str(json); + assert!(result.is_ok()); + } + + #[test] + fn test_publication_data_incompatible_tracking() { + let json = r#"{ + "table_to_track": [ + {"schema_name": "schema1", "table_to_track": []}, + {"schema_name": "schema2", "table_to_track": [{"table_name": "t1", "columns_name": ["col1"]}]} + ], + "transaction_to_track": ["insert"] + }"#; + let result: std::result::Result = serde_json::from_str(json); + assert!(result.is_err()); + } + + #[test] + fn test_postgres_config_serialization() { + let config = PostgresConfig { + postgres_resource_path: "f/db/postgres".to_string(), + replication_slot_name: "slot_1".to_string(), + publication_name: "pub_1".to_string(), + basic_mode: Some(false), + }; + let json = serde_json::to_value(&config).unwrap(); + assert_eq!(json["postgres_resource_path"], "f/db/postgres"); + assert_eq!(json["replication_slot_name"], "slot_1"); + } + + #[test] + fn test_generate_random_string_format() { + let s = generate_random_string(); + assert!(s.contains('_')); + let parts: Vec<&str> = s.split('_').collect(); + assert_eq!(parts.len(), 2); + assert_eq!(parts[1].len(), 10); + } + + #[test] + fn test_relations_add_table() { + let mut rel = Relations::new("public".to_string(), vec![]); + rel.add_new_table(TableToTrack::new("users".to_string(), None, None)); + assert_eq!(rel.table_to_track.len(), 1); + assert_eq!(rel.table_to_track[0].table_name, "users"); + } + + #[test] + fn test_table_to_track_with_where_clause() { + let tt = TableToTrack::new( + "orders".to_string(), + Some("status = 'active'".to_string()), + None, + ); + assert_eq!(tt.where_clause, Some("status = 'active'".to_string())); + } + + #[test] + fn test_table_to_track_with_columns() { + let tt = TableToTrack::new( + "users".to_string(), + None, + Some(vec!["id".to_string(), "email".to_string()]), + ); + assert_eq!(tt.columns_name.as_ref().unwrap().len(), 2); + } +} diff --git a/backend/windmill-trigger-postgres/src/mapper.rs b/backend/windmill-trigger-postgres/src/mapper.rs index 727d6b8bb1..abd40a2838 100644 --- a/backend/windmill-trigger-postgres/src/mapper.rs +++ b/backend/windmill-trigger-postgres/src/mapper.rs @@ -133,3 +133,90 @@ export async function main( ) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_postgres_to_typescript_bool() { + assert_eq!(postgres_to_typescript_type(Some(Type::BOOL)), "boolean"); + } + + #[test] + fn test_postgres_to_typescript_text_types() { + assert_eq!(postgres_to_typescript_type(Some(Type::TEXT)), "string"); + assert_eq!(postgres_to_typescript_type(Some(Type::VARCHAR)), "string"); + assert_eq!(postgres_to_typescript_type(Some(Type::CHAR)), "string"); + } + + #[test] + fn test_postgres_to_typescript_number_types() { + assert_eq!(postgres_to_typescript_type(Some(Type::INT2)), "number"); + assert_eq!(postgres_to_typescript_type(Some(Type::INT4)), "number"); + assert_eq!(postgres_to_typescript_type(Some(Type::INT8)), "number"); + assert_eq!(postgres_to_typescript_type(Some(Type::FLOAT4)), "number"); + assert_eq!(postgres_to_typescript_type(Some(Type::FLOAT8)), "number"); + assert_eq!(postgres_to_typescript_type(Some(Type::NUMERIC)), "number"); + } + + #[test] + fn test_postgres_to_typescript_array_types() { + assert_eq!( + postgres_to_typescript_type(Some(Type::INT4_ARRAY)), + "Array" + ); + assert_eq!( + postgres_to_typescript_type(Some(Type::TEXT_ARRAY)), + "Array" + ); + assert_eq!( + postgres_to_typescript_type(Some(Type::BOOL_ARRAY)), + "Array" + ); + } + + #[test] + fn test_postgres_to_typescript_date_types() { + assert_eq!(postgres_to_typescript_type(Some(Type::DATE)), "string"); + assert_eq!(postgres_to_typescript_type(Some(Type::TIMESTAMP)), "string"); + assert_eq!(postgres_to_typescript_type(Some(Type::TIMESTAMPTZ)), "string"); + assert_eq!(postgres_to_typescript_type(Some(Type::UUID)), "string"); + } + + #[test] + fn test_postgres_to_typescript_json() { + assert_eq!(postgres_to_typescript_type(Some(Type::JSON)), "unknown"); + assert_eq!(postgres_to_typescript_type(Some(Type::JSONB)), "unknown"); + } + + #[test] + fn test_postgres_to_typescript_none() { + assert_eq!(postgres_to_typescript_type(None), "string"); + } + + #[test] + fn test_into_body_struct_typescript() { + let fields = vec![ + MappingInfo::new("id".to_string(), Some(Type::INT4), false), + MappingInfo::new("name".to_string(), Some(Type::TEXT), true), + ]; + let result = into_body_struct(Language::Typescript, fields); + assert!(result.contains("id: number,")); + assert!(result.contains("name?: string,")); + } + + #[test] + fn test_empty_template() { + let mapper = Mapper::new(HashMap::new(), Language::Typescript); + let template = mapper.get_template(); + assert!(template.contains("row: any")); + } + + #[test] + fn test_mapping_info_nullable_field() { + let info = MappingInfo::new("email".to_string(), Some(Type::VARCHAR), true); + assert!(info.is_nullable); + assert_eq!(info.column_name, "email"); + } +} diff --git a/backend/windmill-trigger-postgres/src/relation.rs b/backend/windmill-trigger-postgres/src/relation.rs index f313efb893..0907c134cf 100644 --- a/backend/windmill-trigger-postgres/src/relation.rs +++ b/backend/windmill-trigger-postgres/src/relation.rs @@ -72,3 +72,117 @@ impl RelationConverter { Ok(object) } } + +#[cfg(test)] +mod tests { + use super::*; + use bytes::Bytes; + use rust_postgres::types::Type; + use serde_json::json; + + use super::super::replication_message::{Column, ReplicaIdentity, RelationBody}; + + fn make_relation(o_id: Oid, columns: Vec) -> RelationBody { + RelationBody::new(None, o_id, "public".to_string(), "test".to_string(), ReplicaIdentity::Default, columns) + } + + fn text_col(name: &str, typ: Option) -> Column { + Column::new(0, name.to_string(), typ, -1) + } + + #[test] + fn test_row_to_json_text_columns() { + let mut converter = RelationConverter::new(); + converter.add_relation(make_relation(1, vec![ + text_col("id", Some(Type::INT4)), + text_col("name", Some(Type::TEXT)), + ])); + + let tuple = vec![ + TupleData::Text(Bytes::from("42")), + TupleData::Text(Bytes::from("Alice")), + ]; + + let result = converter.row_to_json((1, tuple)).unwrap(); + assert_eq!(result["id"], json!(42)); + assert_eq!(result["name"], json!("Alice")); + } + + #[test] + fn test_row_to_json_with_null() { + let mut converter = RelationConverter::new(); + converter.add_relation(make_relation(1, vec![ + text_col("id", Some(Type::INT4)), + text_col("email", Some(Type::TEXT)), + ])); + + let tuple = vec![ + TupleData::Text(Bytes::from("1")), + TupleData::Null, + ]; + + let result = converter.row_to_json((1, tuple)).unwrap(); + assert_eq!(result["id"], json!(1)); + assert_eq!(result["email"], Value::Null); + } + + #[test] + fn test_row_to_json_with_unchanged_toast() { + let mut converter = RelationConverter::new(); + converter.add_relation(make_relation(1, vec![ + text_col("data", Some(Type::TEXT)), + ])); + + let tuple = vec![TupleData::UnchangedToast]; + let result = converter.row_to_json((1, tuple)).unwrap(); + assert_eq!(result["data"], Value::Null); + } + + #[test] + fn test_row_to_json_binary_rejected() { + let mut converter = RelationConverter::new(); + converter.add_relation(make_relation(1, vec![ + text_col("data", Some(Type::BYTEA)), + ])); + + let tuple = vec![TupleData::Binary(Bytes::from("data"))]; + assert!(matches!( + converter.row_to_json((1, tuple)), + Err(RelationConversionError::BinaryFormatNotSupported) + )); + } + + #[test] + fn test_missing_relation() { + let converter = RelationConverter::new(); + let tuple = vec![TupleData::Null]; + assert!(matches!( + converter.row_to_json((999, tuple)), + Err(RelationConversionError::FailToFindMatchingTable) + )); + } + + #[test] + fn test_multiple_relations() { + let mut converter = RelationConverter::new(); + converter.add_relation(make_relation(1, vec![text_col("a", Some(Type::TEXT))])); + converter.add_relation(make_relation(2, vec![text_col("b", Some(Type::INT4))])); + + let result1 = converter.row_to_json((1, vec![TupleData::Text(Bytes::from("hello"))])).unwrap(); + let result2 = converter.row_to_json((2, vec![TupleData::Text(Bytes::from("42"))])).unwrap(); + + assert_eq!(result1["a"], json!("hello")); + assert_eq!(result2["b"], json!(42)); + } + + #[test] + fn test_row_to_json_bool_column() { + let mut converter = RelationConverter::new(); + converter.add_relation(make_relation(1, vec![ + text_col("active", Some(Type::BOOL)), + ])); + + let result = converter.row_to_json((1, vec![TupleData::Text(Bytes::from("t"))])).unwrap(); + assert_eq!(result["active"], json!(true)); + } +} diff --git a/backend/windmill-trigger-postgres/src/replication_message.rs b/backend/windmill-trigger-postgres/src/replication_message.rs index 5ae7f8cf2f..43dba75889 100644 --- a/backend/windmill-trigger-postgres/src/replication_message.rs +++ b/backend/windmill-trigger-postgres/src/replication_message.rs @@ -508,3 +508,309 @@ impl ReplicationMessage { Ok(replication_message) } } + +#[cfg(test)] +mod tests { + use super::*; + + fn build_keepalive(wal_end: u64, timestamp: i64, reply: bool) -> Bytes { + let mut buf = Vec::new(); + buf.push(PRIMARY_KEEPALIVE_BYTE); + buf.extend_from_slice(&wal_end.to_be_bytes()); + buf.extend_from_slice(×tamp.to_be_bytes()); + buf.push(if reply { 1 } else { 0 }); + Bytes::from(buf) + } + + fn build_xlog_data(wal_start: u64, wal_end: u64, timestamp: i64, data: &[u8]) -> Bytes { + let mut buf = Vec::new(); + buf.push(X_LOG_DATA_BYTE); + buf.extend_from_slice(&wal_start.to_be_bytes()); + buf.extend_from_slice(&wal_end.to_be_bytes()); + buf.extend_from_slice(×tamp.to_be_bytes()); + buf.extend_from_slice(data); + Bytes::from(buf) + } + + #[test] + fn test_parse_keepalive_with_reply() { + let buf = build_keepalive(100, 200, true); + match ReplicationMessage::parse(buf).unwrap() { + ReplicationMessage::PrimaryKeepAlive(body) => { + assert_eq!(body.wal_end, 100); + assert_eq!(body.timestamp, 200); + assert!(body.reply); + } + _ => panic!("expected PrimaryKeepAlive"), + } + } + + #[test] + fn test_parse_keepalive_without_reply() { + let buf = build_keepalive(500, 1000, false); + match ReplicationMessage::parse(buf).unwrap() { + ReplicationMessage::PrimaryKeepAlive(body) => { + assert_eq!(body.wal_end, 500); + assert_eq!(body.timestamp, 1000); + assert!(!body.reply); + } + _ => panic!("expected PrimaryKeepAlive"), + } + } + + #[test] + fn test_parse_xlog_data() { + let payload = b"test payload"; + let buf = build_xlog_data(10, 20, 30, payload); + match ReplicationMessage::parse(buf).unwrap() { + ReplicationMessage::XLogData(body) => { + assert_eq!(body.wal_start, 10); + assert_eq!(body.wal_end, 20); + assert_eq!(body.timestamp, 30); + assert_eq!(&body.data[..], payload); + } + _ => panic!("expected XLogData"), + } + } + + #[test] + fn test_parse_unknown_byte() { + let buf = Bytes::from(vec![0xFF, 0, 0, 0, 0, 0, 0, 0, 0]); + assert!(ReplicationMessage::parse(buf).is_err()); + } + + fn build_begin_message() -> Vec { + let mut buf = Vec::new(); + buf.push(BEGIN_BYTE); + buf.extend_from_slice(&0i64.to_be_bytes()); // lsn + buf.extend_from_slice(&0i64.to_be_bytes()); // timestamp + buf.extend_from_slice(&0i32.to_be_bytes()); // xid + buf + } + + fn build_commit_message() -> Vec { + let mut buf = Vec::new(); + buf.push(COMMIT_BYTE); + buf.push(0); // flags + buf.extend_from_slice(&0u64.to_be_bytes()); // lsn + buf.extend_from_slice(&0u64.to_be_bytes()); // end_lsn + buf.extend_from_slice(&0i64.to_be_bytes()); // timestamp + buf + } + + fn build_insert_message(o_id: u32, tuple_data: &[(u8, &[u8])]) -> Vec { + let mut buf = Vec::new(); + buf.push(INSERT_BYTE); + buf.extend_from_slice(&o_id.to_be_bytes()); + buf.push(TUPLE_NEW_BYTE); + buf.extend_from_slice(&(tuple_data.len() as i16).to_be_bytes()); + for (tag, data) in tuple_data { + buf.push(*tag); + match *tag { + TUPLE_DATA_TEXT_BYTE | TUPLE_DATA_BINARY_BYTE => { + buf.extend_from_slice(&(data.len() as i32).to_be_bytes()); + buf.extend_from_slice(data); + } + _ => {} + } + } + buf + } + + fn build_delete_message(o_id: u32, tuple_data: &[(u8, &[u8])]) -> Vec { + let mut buf = Vec::new(); + buf.push(DELETE_BYTE); + buf.extend_from_slice(&o_id.to_be_bytes()); + buf.push(TUPLE_OLD_BYTE); + buf.extend_from_slice(&(tuple_data.len() as i16).to_be_bytes()); + for (tag, data) in tuple_data { + buf.push(*tag); + match *tag { + TUPLE_DATA_TEXT_BYTE | TUPLE_DATA_BINARY_BYTE => { + buf.extend_from_slice(&(data.len() as i32).to_be_bytes()); + buf.extend_from_slice(data); + } + _ => {} + } + } + buf + } + + fn settings(streaming: bool) -> LogicalReplicationSettings { + LogicalReplicationSettings { streaming } + } + + #[test] + fn test_parse_begin() { + let data = Bytes::from(build_begin_message()); + let body = XLogDataBody::new(0, 0, 0, data); + match body.parse(&settings(false)).unwrap() { + LogicalReplicationMessage::Begin => {} + other => panic!("expected Begin, got {:?}", other), + } + } + + #[test] + fn test_parse_commit() { + let data = Bytes::from(build_commit_message()); + let body = XLogDataBody::new(0, 0, 0, data); + match body.parse(&settings(false)).unwrap() { + LogicalReplicationMessage::Commit => {} + other => panic!("expected Commit, got {:?}", other), + } + } + + #[test] + fn test_parse_insert_with_text_tuple() { + let data = Bytes::from(build_insert_message( + 42, + &[(TUPLE_DATA_TEXT_BYTE, b"hello")], + )); + let body = XLogDataBody::new(0, 0, 0, data); + match body.parse(&settings(false)).unwrap() { + LogicalReplicationMessage::Insert(insert) => { + assert_eq!(insert.o_id, 42); + assert_eq!(insert.tuple.len(), 1); + match &insert.tuple[0] { + TupleData::Text(b) => assert_eq!(&b[..], b"hello"), + other => panic!("expected Text, got {:?}", other), + } + } + other => panic!("expected Insert, got {:?}", other), + } + } + + #[test] + fn test_parse_insert_with_null_tuple() { + let data = Bytes::from(build_insert_message( + 10, + &[(TUPLE_DATA_NULL_BYTE, &[])], + )); + let body = XLogDataBody::new(0, 0, 0, data); + match body.parse(&settings(false)).unwrap() { + LogicalReplicationMessage::Insert(insert) => { + assert_eq!(insert.tuple.len(), 1); + assert!(matches!(insert.tuple[0], TupleData::Null)); + } + other => panic!("expected Insert, got {:?}", other), + } + } + + #[test] + fn test_parse_insert_with_toast_tuple() { + let data = Bytes::from(build_insert_message( + 10, + &[(TUPLE_DATA_TOAST_BYTE, &[])], + )); + let body = XLogDataBody::new(0, 0, 0, data); + match body.parse(&settings(false)).unwrap() { + LogicalReplicationMessage::Insert(insert) => { + assert!(matches!(insert.tuple[0], TupleData::UnchangedToast)); + } + other => panic!("expected Insert, got {:?}", other), + } + } + + #[test] + fn test_parse_insert_multiple_columns() { + let data = Bytes::from(build_insert_message( + 1, + &[ + (TUPLE_DATA_TEXT_BYTE, b"col1"), + (TUPLE_DATA_NULL_BYTE, &[]), + (TUPLE_DATA_TEXT_BYTE, b"col3"), + ], + )); + let body = XLogDataBody::new(0, 0, 0, data); + match body.parse(&settings(false)).unwrap() { + LogicalReplicationMessage::Insert(insert) => { + assert_eq!(insert.tuple.len(), 3); + assert!(matches!(&insert.tuple[0], TupleData::Text(_))); + assert!(matches!(insert.tuple[1], TupleData::Null)); + assert!(matches!(&insert.tuple[2], TupleData::Text(_))); + } + other => panic!("expected Insert, got {:?}", other), + } + } + + #[test] + fn test_parse_delete_with_old_tuple() { + let data = Bytes::from(build_delete_message( + 99, + &[(TUPLE_DATA_TEXT_BYTE, b"old_val")], + )); + let body = XLogDataBody::new(0, 0, 0, data); + match body.parse(&settings(false)).unwrap() { + LogicalReplicationMessage::Delete(delete) => { + assert_eq!(delete.o_id, 99); + assert!(delete.old_tuple.is_some()); + assert!(delete.key_tuple.is_none()); + } + other => panic!("expected Delete, got {:?}", other), + } + } + + #[test] + fn test_parse_relation() { + let mut buf = Vec::new(); + buf.push(RELATION_BYTE); + buf.extend_from_slice(&100u32.to_be_bytes()); // o_id + buf.extend_from_slice(b"public\0"); // namespace + buf.extend_from_slice(b"users\0"); // name + buf.push(REPLICA_IDENTITY_DEFAULT_BYTE as u8); // replica identity + buf.extend_from_slice(&1i16.to_be_bytes()); // num columns + // column: flags=0, name="id", type_oid=23 (INT4), type_modifier=-1 + buf.push(0); // flags + buf.extend_from_slice(b"id\0"); // name + buf.extend_from_slice(&23u32.to_be_bytes()); // type_oid (INT4) + buf.extend_from_slice(&(-1i32).to_be_bytes()); // type_modifier + + let data = Bytes::from(buf); + let body = XLogDataBody::new(0, 0, 0, data); + match body.parse(&settings(false)).unwrap() { + LogicalReplicationMessage::Relation(rel) => { + assert_eq!(rel.o_id, 100); + assert_eq!(rel.namespace, "public"); + assert_eq!(rel.name, "users"); + assert_eq!(rel.columns.len(), 1); + assert_eq!(rel.columns[0].name, "id"); + assert!(matches!(rel.columns[0].type_o_id, Some(Type::INT4))); + } + other => panic!("expected Relation, got {:?}", other), + } + } + + #[test] + fn test_parse_insert_with_streaming_transaction_id() { + let mut buf = Vec::new(); + buf.push(INSERT_BYTE); + buf.extend_from_slice(&42i32.to_be_bytes()); // transaction_id + buf.extend_from_slice(&10u32.to_be_bytes()); // o_id + buf.push(TUPLE_NEW_BYTE); + buf.extend_from_slice(&0i16.to_be_bytes()); // 0 columns + + let data = Bytes::from(buf); + let body = XLogDataBody::new(0, 0, 0, data); + match body.parse(&settings(true)).unwrap() { + LogicalReplicationMessage::Insert(insert) => { + assert_eq!(insert.transaction_id, Some(42)); + assert_eq!(insert.o_id, 10); + } + other => panic!("expected Insert, got {:?}", other), + } + } + + #[test] + fn test_unknown_tuple_data_byte() { + let mut buf = Vec::new(); + buf.push(INSERT_BYTE); + buf.extend_from_slice(&1u32.to_be_bytes()); // o_id + buf.push(TUPLE_NEW_BYTE); + buf.extend_from_slice(&1i16.to_be_bytes()); // 1 column + buf.push(0xFF); // invalid tuple data byte + + let data = Bytes::from(buf); + let body = XLogDataBody::new(0, 0, 0, data); + assert!(body.parse(&settings(false)).is_err()); + } +} diff --git a/backend/windmill-trigger/src/filter.rs b/backend/windmill-trigger/src/filter.rs index 57ef45839f..a1dd59f848 100644 --- a/backend/windmill-trigger/src/filter.rs +++ b/backend/windmill-trigger/src/filter.rs @@ -139,4 +139,82 @@ mod tests { let result = is_value_superset(&mut deserializer, key, &value).unwrap(); assert!(!result, "Should not match when key doesn't exist"); } + + // --- is_superset unit tests --- + + #[test] + fn test_superset_equal_scalars() { + assert!(is_superset(&json!(42), &json!(42))); + assert!(is_superset(&json!("hello"), &json!("hello"))); + assert!(is_superset(&json!(true), &json!(true))); + assert!(is_superset(&json!(null), &json!(null))); + } + + #[test] + fn test_superset_unequal_scalars() { + assert!(!is_superset(&json!(42), &json!(43))); + assert!(!is_superset(&json!("hello"), &json!("world"))); + assert!(!is_superset(&json!(true), &json!(false))); + } + + #[test] + fn test_superset_object_subset() { + let full = json!({"a": 1, "b": 2, "c": 3}); + let subset = json!({"a": 1, "b": 2}); + assert!(is_superset(&full, &subset)); + } + + #[test] + fn test_superset_object_not_subset() { + let full = json!({"a": 1, "b": 2}); + let check = json!({"a": 1, "b": 3}); + assert!(!is_superset(&full, &check)); + } + + #[test] + fn test_superset_object_missing_key() { + let full = json!({"a": 1}); + let check = json!({"a": 1, "b": 2}); + assert!(!is_superset(&full, &check)); + } + + #[test] + fn test_superset_nested_objects() { + let full = json!({"a": {"b": {"c": 1, "d": 2}, "e": 3}}); + let check = json!({"a": {"b": {"c": 1}}}); + assert!(is_superset(&full, &check)); + } + + #[test] + fn test_superset_array_subset() { + let full = json!([1, 2, 3, 4]); + let check = json!([2, 4]); + assert!(is_superset(&full, &check)); + } + + #[test] + fn test_superset_array_not_subset() { + let full = json!([1, 2, 3]); + let check = json!([4]); + assert!(!is_superset(&full, &check)); + } + + #[test] + fn test_superset_empty_check() { + assert!(is_superset(&json!({"a": 1}), &json!({}))); + assert!(is_superset(&json!([1, 2]), &json!([]))); + } + + #[test] + fn test_superset_array_of_objects() { + let full = json!([{"id": 1, "name": "a"}, {"id": 2, "name": "b"}]); + let check = json!([{"id": 1}]); + assert!(is_superset(&full, &check)); + } + + #[test] + fn test_superset_type_mismatch() { + assert!(!is_superset(&json!(42), &json!("42"))); + assert!(!is_superset(&json!([1]), &json!(1))); + } } diff --git a/backend/windmill-trigger/src/types.rs b/backend/windmill-trigger/src/types.rs index 07feae5831..1b471ab30c 100644 --- a/backend/windmill-trigger/src/types.rs +++ b/backend/windmill-trigger/src/types.rs @@ -156,3 +156,183 @@ pub enum TriggerMode { Disabled, Suspended, } + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + // --- TriggerMode serde --- + + #[test] + fn test_trigger_mode_serialize() { + assert_eq!(serde_json::to_value(TriggerMode::Enabled).unwrap(), json!("enabled")); + assert_eq!(serde_json::to_value(TriggerMode::Disabled).unwrap(), json!("disabled")); + assert_eq!(serde_json::to_value(TriggerMode::Suspended).unwrap(), json!("suspended")); + } + + #[test] + fn test_trigger_mode_deserialize() { + let enabled: TriggerMode = serde_json::from_value(json!("enabled")).unwrap(); + assert_eq!(enabled, TriggerMode::Enabled); + let disabled: TriggerMode = serde_json::from_value(json!("disabled")).unwrap(); + assert_eq!(disabled, TriggerMode::Disabled); + } + + #[test] + fn test_trigger_mode_invalid() { + let result: Result = serde_json::from_value(json!("paused")); + assert!(result.is_err()); + } + + // --- StandardTriggerQuery --- + + #[test] + fn test_query_default() { + let q = StandardTriggerQuery::default(); + assert_eq!(q.offset(), 0); + assert_eq!(q.limit(), 100); + } + + #[test] + fn test_query_offset_calculation() { + let q = StandardTriggerQuery { + page: Some(2), + per_page: Some(50), + path: None, + is_flow: None, + path_start: None, + }; + assert_eq!(q.offset(), 100); + assert_eq!(q.limit(), 50); + } + + #[test] + fn test_query_offset_defaults() { + let q = StandardTriggerQuery { + page: None, + per_page: None, + path: None, + is_flow: None, + path_start: None, + }; + assert_eq!(q.offset(), 0); + assert_eq!(q.limit(), 100); + } + + // --- BaseTriggerData backward compatibility --- + + #[test] + fn test_base_trigger_data_mode_field() { + let json = r#"{ + "path": "test", + "script_path": "f/test/script", + "is_flow": false, + "mode": "enabled" + }"#; + let data: BaseTriggerData = serde_json::from_str(json).unwrap(); + assert_eq!(data.mode(), &TriggerMode::Enabled); + } + + #[test] + fn test_base_trigger_data_legacy_enabled_true() { + let json = r#"{ + "path": "test", + "script_path": "f/test/script", + "is_flow": false, + "enabled": true + }"#; + let data: BaseTriggerData = serde_json::from_str(json).unwrap(); + assert_eq!(data.mode(), &TriggerMode::Enabled); + } + + #[test] + fn test_base_trigger_data_legacy_enabled_false() { + let json = r#"{ + "path": "test", + "script_path": "f/test/script", + "is_flow": false, + "enabled": false + }"#; + let data: BaseTriggerData = serde_json::from_str(json).unwrap(); + assert_eq!(data.mode(), &TriggerMode::Disabled); + } + + #[test] + fn test_base_trigger_data_mode_takes_precedence() { + let json = r#"{ + "path": "test", + "script_path": "f/test/script", + "is_flow": false, + "mode": "suspended", + "enabled": true + }"#; + let data: BaseTriggerData = serde_json::from_str(json).unwrap(); + assert_eq!(data.mode(), &TriggerMode::Suspended); + } + + #[test] + fn test_base_trigger_data_neither_field() { + let json = r#"{ + "path": "test", + "script_path": "f/test/script", + "is_flow": false + }"#; + let data: BaseTriggerData = serde_json::from_str(json).unwrap(); + assert_eq!(data.mode(), &TriggerMode::Enabled); + } + + // --- HandlerAction --- + + #[test] + fn test_handler_action_serialization() { + let action = HandlerAction::Trigger { + path: "f/test/trigger".to_string(), + trigger_kind: JobTriggerKind::Webhook, + }; + let json = serde_json::to_value(&action).unwrap(); + assert_eq!(json["type"], "trigger"); + assert_eq!(json["path"], "f/test/trigger"); + } + + #[test] + fn test_handler_action_deserialization() { + let json = r#"{"type": "trigger", "path": "f/test/trigger", "trigger_kind": "webhook"}"#; + let action: HandlerAction = serde_json::from_str(json).unwrap(); + match action { + HandlerAction::Trigger { path, trigger_kind } => { + assert_eq!(path, "f/test/trigger"); + assert_eq!( + serde_json::to_value(&trigger_kind).unwrap(), + serde_json::to_value(&JobTriggerKind::Webhook).unwrap() + ); + } + } + } + + // --- ServerState --- + + #[test] + fn test_server_state_skip_none_fields() { + let state = ServerState { + server_id: None, + last_server_ping: None, + error: None, + }; + let json = serde_json::to_value(&state).unwrap(); + assert!(!json.as_object().unwrap().contains_key("server_id")); + assert!(!json.as_object().unwrap().contains_key("error")); + } + + #[test] + fn test_server_state_with_error() { + let state = ServerState { + server_id: Some("srv-1".to_string()), + last_server_ping: None, + error: Some("connection timeout".to_string()), + }; + let json = serde_json::to_value(&state).unwrap(); + assert_eq!(json["server_id"], "srv-1"); + assert_eq!(json["error"], "connection timeout"); + } +}