From dc5e69481d1268edc73085c0e4710146b38ecfc9 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Mon, 9 Feb 2026 12:28:25 +0000 Subject: [PATCH] test: add end-to-end trigger integration tests and DB CRUD tests Add 7 #[ignore] e2e tests (one per trigger type) that fire real messages to external services and verify job creation in v2_job. Also add 9 DB-level CRUD tests for MQTT, GCP, and Email triggers. Includes helper shell scripts in tests/fixtures/ to start/stop each external service (MQTT, WebSocket, Postgres replication, Kafka, NATS, SQS via LocalStack, GCP Pub/Sub emulator). Co-Authored-By: Claude Opus 4.6 --- backend/Cargo.lock | 6 + backend/Cargo.toml | 6 + backend/tests/common/mod.rs | 6 + backend/tests/fixtures/start_all_triggers.sh | 58 ++ backend/tests/fixtures/start_gcp_pubsub.sh | 47 ++ backend/tests/fixtures/start_kafka.sh | 52 ++ backend/tests/fixtures/start_mqtt.sh | 31 + backend/tests/fixtures/start_nats.sh | 29 + .../fixtures/start_postgres_replication.sh | 53 ++ backend/tests/fixtures/start_sqs.sh | 45 ++ backend/tests/fixtures/start_websocket.sh | 31 + backend/tests/trigger_e2e.rs | 726 ++++++++++++++++++ backend/tests/triggers.rs | 428 +++++++++++ 13 files changed, 1518 insertions(+) create mode 100755 backend/tests/fixtures/start_all_triggers.sh create mode 100755 backend/tests/fixtures/start_gcp_pubsub.sh create mode 100755 backend/tests/fixtures/start_kafka.sh create mode 100755 backend/tests/fixtures/start_mqtt.sh create mode 100755 backend/tests/fixtures/start_nats.sh create mode 100755 backend/tests/fixtures/start_postgres_replication.sh create mode 100755 backend/tests/fixtures/start_sqs.sh create mode 100755 backend/tests/fixtures/start_websocket.sh create mode 100644 backend/tests/trigger_e2e.rs diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 0bba51de85..003c2589cc 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -15656,6 +15656,10 @@ name = "windmill" version = "1.628.3" dependencies = [ "anyhow", + "async-nats", + "aws-config", + "aws-credential-types", + "aws-sdk-sqs", "axum 0.7.9", "base64 0.22.1", "chrono", @@ -15669,7 +15673,9 @@ dependencies = [ "once_cell", "prometheus", "rand 0.9.0", + "rdkafka", "reqwest 0.13.1", + "rumqttc", "rustls 0.23.35", "serde", "serde_derive", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index a5a770fa0a..066050502c 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -229,6 +229,12 @@ axum.workspace = true serde.workspace = true windmill-api-client.workspace = true tempfile.workspace = true +rumqttc.workspace = true +rdkafka.workspace = true +async-nats.workspace = true +aws-sdk-sqs.workspace = true +aws-config.workspace = true +aws-credential-types.workspace = true [workspace.dependencies] diff --git a/backend/tests/common/mod.rs b/backend/tests/common/mod.rs index 2c569537fa..7f98d0937c 100644 --- a/backend/tests/common/mod.rs +++ b/backend/tests/common/mod.rs @@ -87,6 +87,12 @@ impl ApiServer { Self::start_inner(db, true).await } + /// Start the API server with server_mode=true so trigger listeners are active. + /// Alias for `start_agent_mode` with a clearer name for trigger e2e tests. + pub async fn start_with_listeners(db: Pool) -> anyhow::Result { + Self::start_inner(db, true).await + } + async fn start_inner(db: Pool, agent_mode: bool) -> anyhow::Result { let (tx, rx) = tokio::sync::broadcast::channel::<()>(1); diff --git a/backend/tests/fixtures/start_all_triggers.sh b/backend/tests/fixtures/start_all_triggers.sh new file mode 100755 index 0000000000..a7e5b911ca --- /dev/null +++ b/backend/tests/fixtures/start_all_triggers.sh @@ -0,0 +1,58 @@ +#!/usr/bin/env bash +# Starts (or stops) all external services needed for the trigger e2e tests. +# +# Usage: +# ./tests/fixtures/start_all_triggers.sh # start everything +# ./tests/fixtures/start_all_triggers.sh stop # stop everything +# ./tests/fixtures/start_all_triggers.sh oss # start only OSS services +set -euo pipefail + +DIR="$(cd "$(dirname "$0")" && pwd)" +ACTION="${1:-start}" + +SCRIPTS_OSS=( + "$DIR/start_mqtt.sh" + "$DIR/start_websocket.sh" + "$DIR/start_postgres_replication.sh" +) + +SCRIPTS_EE=( + "$DIR/start_kafka.sh" + "$DIR/start_nats.sh" + "$DIR/start_sqs.sh" + "$DIR/start_gcp_pubsub.sh" +) + +if [[ "$ACTION" == "stop" ]]; then + for s in "${SCRIPTS_OSS[@]}" "${SCRIPTS_EE[@]}"; do + echo "--- $(basename "$s" .sh) stop ---" + bash "$s" stop + done + exit 0 +fi + +if [[ "$ACTION" == "oss" ]]; then + SCRIPTS=("${SCRIPTS_OSS[@]}") +else + SCRIPTS=("${SCRIPTS_OSS[@]}" "${SCRIPTS_EE[@]}") +fi + +for s in "${SCRIPTS[@]}"; do + echo "--- $(basename "$s" .sh) ---" + bash "$s" + echo "" +done + +echo "============================================" +echo "All services ready. Run the e2e tests with:" +echo "" + +if [[ "$ACTION" == "oss" ]]; then + echo " cargo test --test trigger_e2e --features mqtt_trigger,websocket,postgres_trigger -- --ignored --nocapture" +else + echo " # OSS triggers" + echo " cargo test --test trigger_e2e --features mqtt_trigger,websocket,postgres_trigger -- --ignored --nocapture" + echo "" + echo " # Enterprise triggers" + echo " AWS_ENDPOINT_URL=http://localhost:4566 PUBSUB_EMULATOR_HOST=localhost:8085 cargo test --test trigger_e2e --features kafka,nats,sqs_trigger,gcp_trigger,enterprise,private -- --ignored --nocapture" +fi diff --git a/backend/tests/fixtures/start_gcp_pubsub.sh b/backend/tests/fixtures/start_gcp_pubsub.sh new file mode 100755 index 0000000000..88ef706982 --- /dev/null +++ b/backend/tests/fixtures/start_gcp_pubsub.sh @@ -0,0 +1,47 @@ +#!/usr/bin/env bash +# Starts the GCP Pub/Sub emulator for trigger_e2e::test_gcp_e2e (Enterprise) +# +# Usage: +# ./tests/fixtures/start_gcp_pubsub.sh # start +# ./tests/fixtures/start_gcp_pubsub.sh stop # stop & remove +set -euo pipefail + +NAME="windmill-test-pubsub" +PORT=8085 + +if [[ "${1:-}" == "stop" ]]; then + docker rm -f "$NAME" 2>/dev/null && echo "stopped $NAME" || echo "$NAME not running" + exit 0 +fi + +if docker ps --format '{{.Names}}' | grep -q "^${NAME}$"; then + echo "$NAME is already running" + exit 0 +fi + +docker rm -f "$NAME" 2>/dev/null || true + +docker run -d --name "$NAME" -p "${PORT}:8085" \ + gcr.io/google.com/cloudsdktool/google-cloud-cli:emulators \ + gcloud beta emulators pubsub start --host-port="0.0.0.0:${PORT}" + +echo "Waiting for Pub/Sub emulator to become ready..." +for i in $(seq 1 30); do + if curl -sf "http://localhost:${PORT}" &>/dev/null; then + break + fi + sleep 1 +done + +# Create the test topic and subscription +curl -sX PUT "http://localhost:${PORT}/v1/projects/local-project/topics/windmill-e2e-test" >/dev/null +curl -sX PUT "http://localhost:${PORT}/v1/projects/local-project/subscriptions/windmill-e2e-sub" \ + -H "Content-Type: application/json" \ + -d '{"topic": "projects/local-project/topics/windmill-e2e-test"}' >/dev/null + +echo "GCP Pub/Sub emulator listening on localhost:${PORT}" +echo " topic: windmill-e2e-test" +echo " subscription: windmill-e2e-sub" +echo "" +echo "Run the test:" +echo " PUBSUB_EMULATOR_HOST=localhost:${PORT} cargo test --test trigger_e2e test_gcp_e2e --features gcp_trigger,enterprise,private -- --ignored --nocapture" diff --git a/backend/tests/fixtures/start_kafka.sh b/backend/tests/fixtures/start_kafka.sh new file mode 100755 index 0000000000..567c6e1893 --- /dev/null +++ b/backend/tests/fixtures/start_kafka.sh @@ -0,0 +1,52 @@ +#!/usr/bin/env bash +# Starts a Kafka broker for trigger_e2e::test_kafka_e2e (Enterprise) +# +# Usage: +# ./tests/fixtures/start_kafka.sh # start +# ./tests/fixtures/start_kafka.sh stop # stop & remove +set -euo pipefail + +NAME="windmill-test-kafka" +PORT=9092 + +if [[ "${1:-}" == "stop" ]]; then + docker rm -f "$NAME" 2>/dev/null && echo "stopped $NAME" || echo "$NAME not running" + exit 0 +fi + +if docker ps --format '{{.Names}}' | grep -q "^${NAME}$"; then + echo "$NAME is already running" + exit 0 +fi + +docker rm -f "$NAME" 2>/dev/null || true + +docker run -d --name "$NAME" -p "${PORT}:9092" \ + -e KAFKA_NODE_ID=1 \ + -e KAFKA_PROCESS_ROLES=broker,controller \ + -e KAFKA_LISTENERS="PLAINTEXT://0.0.0.0:${PORT},CONTROLLER://0.0.0.0:9093" \ + -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \ + -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093 \ + -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \ + -e KAFKA_ADVERTISED_LISTENERS="PLAINTEXT://localhost:${PORT}" \ + -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \ + -e KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0 \ + apache/kafka:latest + +echo "Waiting for Kafka to become ready..." +for i in $(seq 1 60); do + if docker exec "$NAME" /opt/kafka/bin/kafka-topics.sh --list --bootstrap-server "localhost:${PORT}" &>/dev/null; then + break + fi + sleep 1 +done + +docker exec "$NAME" /opt/kafka/bin/kafka-topics.sh --create \ + --topic windmill-e2e-test \ + --bootstrap-server "localhost:${PORT}" \ + --partitions 1 --replication-factor 1 2>/dev/null || true + +echo "Kafka broker listening on localhost:${PORT} with topic 'windmill-e2e-test'" +echo "" +echo "Run the test:" +echo " cargo test --test trigger_e2e test_kafka_e2e --features kafka,enterprise,private -- --ignored --nocapture" diff --git a/backend/tests/fixtures/start_mqtt.sh b/backend/tests/fixtures/start_mqtt.sh new file mode 100755 index 0000000000..e48e5043cf --- /dev/null +++ b/backend/tests/fixtures/start_mqtt.sh @@ -0,0 +1,31 @@ +#!/usr/bin/env bash +# Starts a Mosquitto MQTT broker for trigger_e2e::test_mqtt_e2e +# +# Usage: +# ./tests/fixtures/start_mqtt.sh # start +# ./tests/fixtures/start_mqtt.sh stop # stop & remove +set -euo pipefail + +NAME="windmill-test-mqtt" +PORT=1883 + +if [[ "${1:-}" == "stop" ]]; then + docker rm -f "$NAME" 2>/dev/null && echo "stopped $NAME" || echo "$NAME not running" + exit 0 +fi + +if docker ps --format '{{.Names}}' | grep -q "^${NAME}$"; then + echo "$NAME is already running" + exit 0 +fi + +docker rm -f "$NAME" 2>/dev/null || true + +docker run -d --name "$NAME" -p "${PORT}:1883" \ + eclipse-mosquitto:latest \ + mosquitto -c /mosquitto-no-auth.conf + +echo "MQTT broker listening on localhost:${PORT}" +echo "" +echo "Run the test:" +echo " cargo test --test trigger_e2e test_mqtt_e2e --features mqtt_trigger -- --ignored --nocapture" diff --git a/backend/tests/fixtures/start_nats.sh b/backend/tests/fixtures/start_nats.sh new file mode 100755 index 0000000000..fe6cf7be68 --- /dev/null +++ b/backend/tests/fixtures/start_nats.sh @@ -0,0 +1,29 @@ +#!/usr/bin/env bash +# Starts a NATS server for trigger_e2e::test_nats_e2e (Enterprise) +# +# Usage: +# ./tests/fixtures/start_nats.sh # start +# ./tests/fixtures/start_nats.sh stop # stop & remove +set -euo pipefail + +NAME="windmill-test-nats" +PORT=4222 + +if [[ "${1:-}" == "stop" ]]; then + docker rm -f "$NAME" 2>/dev/null && echo "stopped $NAME" || echo "$NAME not running" + exit 0 +fi + +if docker ps --format '{{.Names}}' | grep -q "^${NAME}$"; then + echo "$NAME is already running" + exit 0 +fi + +docker rm -f "$NAME" 2>/dev/null || true + +docker run -d --name "$NAME" -p "${PORT}:4222" nats:latest + +echo "NATS server listening on localhost:${PORT}" +echo "" +echo "Run the test:" +echo " cargo test --test trigger_e2e test_nats_e2e --features nats,enterprise,private -- --ignored --nocapture" diff --git a/backend/tests/fixtures/start_postgres_replication.sh b/backend/tests/fixtures/start_postgres_replication.sh new file mode 100755 index 0000000000..5bde571e23 --- /dev/null +++ b/backend/tests/fixtures/start_postgres_replication.sh @@ -0,0 +1,53 @@ +#!/usr/bin/env bash +# Configures the local PostgreSQL for logical replication trigger tests. +# +# Prerequisites: wal_level=logical must be set (requires PG restart). +# Check with: SHOW wal_level; +# +# Usage: +# ./tests/fixtures/start_postgres_replication.sh # setup +# ./tests/fixtures/start_postgres_replication.sh stop # teardown +set -euo pipefail + +PGURL="${DATABASE_URL:-postgres://postgres:changeme@localhost:5432/windmill}" + +if [[ "${1:-}" == "stop" ]]; then + psql "$PGURL" <<'SQL' +SELECT pg_drop_replication_slot('test_e2e_slot') + WHERE EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = 'test_e2e_slot'); +DROP PUBLICATION IF EXISTS test_e2e_pub; +DROP TABLE IF EXISTS test_trigger_table; +SQL + echo "Postgres replication teardown complete" + exit 0 +fi + +# Check wal_level +WAL_LEVEL=$(psql "$PGURL" -tAc "SHOW wal_level;") +if [[ "$WAL_LEVEL" != "logical" ]]; then + echo "ERROR: wal_level is '$WAL_LEVEL', must be 'logical'" + echo "" + echo "Fix with:" + echo " psql \"$PGURL\" -c \"ALTER SYSTEM SET wal_level = logical;\"" + echo " # then restart PostgreSQL" + exit 1 +fi + +psql "$PGURL" <<'SQL' +CREATE TABLE IF NOT EXISTS test_trigger_table (id serial PRIMARY KEY, data text); + +DO $$ +BEGIN + IF NOT EXISTS (SELECT 1 FROM pg_publication WHERE pubname = 'test_e2e_pub') THEN + CREATE PUBLICATION test_e2e_pub FOR TABLE test_trigger_table; + END IF; +END $$; + +SELECT pg_create_logical_replication_slot('test_e2e_slot', 'pgoutput') + WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = 'test_e2e_slot'); +SQL + +echo "Postgres logical replication ready (publication=test_e2e_pub, slot=test_e2e_slot)" +echo "" +echo "Run the test:" +echo " cargo test --test trigger_e2e test_postgres_e2e --features postgres_trigger -- --ignored --nocapture" diff --git a/backend/tests/fixtures/start_sqs.sh b/backend/tests/fixtures/start_sqs.sh new file mode 100755 index 0000000000..79563505e2 --- /dev/null +++ b/backend/tests/fixtures/start_sqs.sh @@ -0,0 +1,45 @@ +#!/usr/bin/env bash +# Starts LocalStack for SQS trigger_e2e::test_sqs_e2e (Enterprise) +# +# Usage: +# ./tests/fixtures/start_sqs.sh # start +# ./tests/fixtures/start_sqs.sh stop # stop & remove +set -euo pipefail + +NAME="windmill-test-localstack" +PORT=4566 + +if [[ "${1:-}" == "stop" ]]; then + docker rm -f "$NAME" 2>/dev/null && echo "stopped $NAME" || echo "$NAME not running" + exit 0 +fi + +if docker ps --format '{{.Names}}' | grep -q "^${NAME}$"; then + echo "$NAME is already running" + exit 0 +fi + +docker rm -f "$NAME" 2>/dev/null || true + +docker run -d --name "$NAME" -p "${PORT}:4566" \ + -e SERVICES=sqs \ + localstack/localstack + +echo "Waiting for LocalStack to become ready..." +for i in $(seq 1 30); do + if curl -sf "http://localhost:${PORT}/_localstack/health" &>/dev/null; then + break + fi + sleep 1 +done + +# Create the test queue +aws --endpoint-url="http://localhost:${PORT}" \ + --region us-east-1 \ + --no-sign-request \ + sqs create-queue --queue-name windmill-e2e-test 2>/dev/null || true + +echo "LocalStack SQS listening on localhost:${PORT} with queue 'windmill-e2e-test'" +echo "" +echo "Run the test:" +echo " AWS_ENDPOINT_URL=http://localhost:${PORT} cargo test --test trigger_e2e test_sqs_e2e --features sqs_trigger,enterprise,private -- --ignored --nocapture" diff --git a/backend/tests/fixtures/start_websocket.sh b/backend/tests/fixtures/start_websocket.sh new file mode 100755 index 0000000000..9bab96debe --- /dev/null +++ b/backend/tests/fixtures/start_websocket.sh @@ -0,0 +1,31 @@ +#!/usr/bin/env bash +# Starts a WebSocket echo server for trigger_e2e::test_websocket_e2e +# +# Usage: +# ./tests/fixtures/start_websocket.sh # start +# ./tests/fixtures/start_websocket.sh stop # stop & remove +set -euo pipefail + +NAME="windmill-test-ws-echo" +PORT=8765 + +if [[ "${1:-}" == "stop" ]]; then + docker rm -f "$NAME" 2>/dev/null && echo "stopped $NAME" || echo "$NAME not running" + exit 0 +fi + +if docker ps --format '{{.Names}}' | grep -q "^${NAME}$"; then + echo "$NAME is already running" + exit 0 +fi + +docker rm -f "$NAME" 2>/dev/null || true + +docker run -d --name "$NAME" -p "${PORT}:8080" \ + -e PORT=8080 \ + jmalloc/echo-server + +echo "WebSocket echo server listening on localhost:${PORT}" +echo "" +echo "Run the test:" +echo " cargo test --test trigger_e2e test_websocket_e2e --features websocket -- --ignored --nocapture" diff --git a/backend/tests/trigger_e2e.rs b/backend/tests/trigger_e2e.rs new file mode 100644 index 0000000000..7bfb735a5c --- /dev/null +++ b/backend/tests/trigger_e2e.rs @@ -0,0 +1,726 @@ +/*! + * End-to-end integration tests for Windmill trigger listeners. + * + * Each test is `#[ignore]` because it requires a running external service + * (MQTT broker, NATS server, Kafka broker, etc.). See individual test doc + * comments for setup instructions. + * + * Quick start — use the helper scripts in `tests/fixtures/`: + * ```bash + * ./tests/fixtures/start_all_triggers.sh # start all services + * ./tests/fixtures/start_all_triggers.sh oss # start OSS services only + * ./tests/fixtures/start_all_triggers.sh stop # tear down everything + * ``` + * + * The general pattern: + * 1. Insert a test script + trigger row + resource into the DB + * 2. Start the API server with listeners enabled (server_mode=true) + * 3. Connect to the external service and send a test message + * 4. Poll `v2_job` for a job matching the trigger path + trigger_kind + * 5. Verify the args shape/content + */ + +use serde_json::json; +use sqlx::{Pool, Postgres}; +use std::time::Duration; + +mod common; +use common::*; + +// ============================================================================ +// Helpers +// ============================================================================ + +/// Row shape for polling v2_job. +#[derive(Debug)] +#[allow(dead_code)] +struct TriggerJobRow { + id: uuid::Uuid, + runnable_path: Option, + trigger_kind: Option, + args: Option>, +} + +/// Poll `v2_job` every 500ms for up to `timeout` for a job whose +/// `runnable_path` and `trigger_kind` match the expected values. +async fn poll_for_trigger_job( + db: &Pool, + script_path: &str, + trigger_kind: &str, + timeout: Duration, +) -> anyhow::Result { + let deadline = tokio::time::Instant::now() + timeout; + loop { + let row = sqlx::query_as!( + TriggerJobRow, + r#" + SELECT id, runnable_path, trigger_kind AS "trigger_kind: String", + args AS "args: sqlx::types::Json" + FROM v2_job + WHERE runnable_path = $1 + AND trigger_kind = $2::job_trigger_kind + ORDER BY created_at DESC + LIMIT 1 + "#, + script_path, + trigger_kind as _, + ) + .fetch_optional(db) + .await?; + + if let Some(job) = row { + return Ok(job); + } + + if tokio::time::Instant::now() >= deadline { + anyhow::bail!( + "timed out waiting for trigger job (script_path={}, trigger_kind={})", + script_path, + trigger_kind + ); + } + + tokio::time::sleep(Duration::from_millis(500)).await; + } +} + +/// Insert a minimal test script row that trigger listeners can reference. +async fn insert_test_script(db: &Pool, path: &str) -> anyhow::Result { + let hash: i64 = rand::random::().unsigned_abs() as i64; + sqlx::query( + "INSERT INTO script (workspace_id, hash, path, summary, description, content, + created_by, language, kind, lock) + VALUES ('test-workspace', $1, $2, '', '', 'def main(): pass', + 'test-user', 'python3', 'script', '')", + ) + .bind(hash) + .bind(path) + .execute(db) + .await?; + Ok(hash) +} + +/// Insert a resource row for triggers that resolve connection details from the +/// `resource` table. +async fn insert_resource( + db: &Pool, + path: &str, + resource_type: &str, + value: serde_json::Value, +) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO resource (workspace_id, path, value, resource_type, extra_perms, created_by) + VALUES ('test-workspace', $1, $2::jsonb, $3, '{}'::jsonb, 'test-user')", + ) + .bind(path) + .bind(value) + .bind(resource_type) + .execute(db) + .await?; + Ok(()) +} + +// ============================================================================ +// MQTT Trigger E2E +// ============================================================================ + +/// End-to-end test for MQTT trigger. +/// +/// Requires a running MQTT broker. Setup: +/// ```bash +/// ./tests/fixtures/start_mqtt.sh +/// ``` +/// +/// Run: +/// ```bash +/// cargo test --test trigger_e2e test_mqtt_e2e --features mqtt_trigger \ +/// -- --ignored --nocapture +/// ``` +#[ignore = "requires running MQTT broker on localhost:1883"] +#[sqlx::test(fixtures("base"))] +async fn test_mqtt_e2e(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + + let script_path = "f/test/mqtt_e2e_handler"; + insert_test_script(&db, script_path).await?; + + insert_resource( + &db, + "u/test-user/mqtt_res", + "mqtt", + json!({ + "broker": "localhost", + "port": 1883 + }), + ) + .await?; + + sqlx::query( + r#" + INSERT INTO mqtt_trigger ( + path, mqtt_resource_path, subscribe_topics, client_version, + script_path, is_flow, workspace_id, edited_by, email + ) + VALUES ($1, $2, ARRAY[$3::jsonb], $4::mqtt_client_version, $5, $6, $7, $8, $9) + "#, + ) + .bind("f/test/mqtt_e2e_trigger") + .bind("u/test-user/mqtt_res") + .bind(json!({"topic": "windmill/test/e2e", "qos": "qos0"})) + .bind("v5") + .bind(script_path) + .bind(false) + .bind("test-workspace") + .bind("test-user") + .bind("test@windmill.dev") + .execute(&db) + .await?; + + let _server = ApiServer::start_with_listeners(db.clone()).await?; + tokio::time::sleep(Duration::from_secs(3)).await; + + // Publish a message using rumqttc + let mut mqtt_opts = rumqttc::MqttOptions::new("windmill-e2e-test", "localhost", 1883); + mqtt_opts.set_keep_alive(Duration::from_secs(5)); + let (client, mut eventloop) = rumqttc::AsyncClient::new(mqtt_opts, 10); + + // Drive the event loop in the background + let el_handle = tokio::spawn(async move { + loop { + match eventloop.poll().await { + Ok(_) => {} + Err(_) => break, + } + } + }); + + tokio::time::sleep(Duration::from_millis(500)).await; + client + .publish( + "windmill/test/e2e", + rumqttc::QoS::AtLeastOnce, + false, + b"hello from e2e test".to_vec(), + ) + .await?; + + let job = poll_for_trigger_job(&db, script_path, "mqtt", Duration::from_secs(30)).await?; + assert!(job.args.is_some(), "job should have args"); + + client.disconnect().await.ok(); + el_handle.abort(); + + Ok(()) +} + +// ============================================================================ +// WebSocket Trigger E2E +// ============================================================================ + +/// End-to-end test for WebSocket trigger. +/// +/// Requires a WebSocket echo server. Setup: +/// ```bash +/// ./tests/fixtures/start_websocket.sh +/// ``` +/// +/// Run: +/// ```bash +/// cargo test --test trigger_e2e test_websocket_e2e --features websocket \ +/// -- --ignored --nocapture +/// ``` +#[ignore = "requires running WebSocket echo server on localhost:8765"] +#[sqlx::test(fixtures("base"))] +async fn test_websocket_e2e(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + + let script_path = "f/test/ws_e2e_handler"; + insert_test_script(&db, script_path).await?; + + sqlx::query!( + r#" + INSERT INTO websocket_trigger ( + path, url, script_path, is_flow, workspace_id, + edited_by, email, initial_messages + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8) + "#, + "f/test/ws_e2e_trigger", + "ws://localhost:8765", + script_path, + false, + "test-workspace", + "test-user", + "test@windmill.dev", + &[json!({"type": "RawMessage", "content": "hello from e2e test"})] as &[serde_json::Value], + ) + .execute(&db) + .await?; + + let _server = ApiServer::start_with_listeners(db.clone()).await?; + + // The WebSocket trigger connects to the server and sends initial_messages, + // and each received message triggers a job. + let job = poll_for_trigger_job(&db, script_path, "websocket", Duration::from_secs(30)).await?; + assert!(job.args.is_some(), "job should have args"); + + Ok(()) +} + +// ============================================================================ +// Postgres Trigger E2E +// ============================================================================ + +/// End-to-end test for Postgres trigger (logical replication). +/// +/// Requires PostgreSQL with `wal_level=logical`. Setup: +/// ```bash +/// ./tests/fixtures/start_postgres_replication.sh +/// ``` +/// (The script checks wal_level and creates the table/publication/slot in the +/// main DB. This test re-creates them in its isolated sqlx::test database.) +/// +/// Run: +/// ```bash +/// cargo test --test trigger_e2e test_postgres_e2e --features postgres_trigger \ +/// -- --ignored --nocapture +/// ``` +#[ignore = "requires PostgreSQL with wal_level=logical"] +#[sqlx::test(fixtures("base"))] +async fn test_postgres_e2e(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + + let script_path = "f/test/pg_e2e_handler"; + insert_test_script(&db, script_path).await?; + + // Create the tracked table + publication + replication slot inside the + // isolated test database (sqlx::test gives us a fresh DB each run). + // Replication slots are server-wide so we use a random suffix. + let suffix: u32 = rand::random(); + let slot_name = format!("test_e2e_slot_{suffix}"); + let pub_name = format!("test_e2e_pub_{suffix}"); + + sqlx::query("CREATE TABLE test_trigger_table (id serial PRIMARY KEY, data text)") + .execute(&db) + .await?; + sqlx::query(&format!("CREATE PUBLICATION {pub_name} FOR TABLE test_trigger_table")) + .execute(&db) + .await?; + sqlx::query(&format!( + "SELECT pg_create_logical_replication_slot('{slot_name}', 'pgoutput')" + )) + .execute(&db) + .await?; + + // Extract the test DB name from the pool so the resource points here, + // not at the main windmill database. + let test_db_name: String = + sqlx::query_scalar("SELECT current_database()") + .fetch_one(&db) + .await?; + + insert_resource( + &db, + "u/test-user/pg_res", + "postgresql", + json!({ + "user": "postgres", + "password": "changeme", + "host": "localhost", + "port": 5432, + "dbname": test_db_name, + "sslmode": "disable" + }), + ) + .await?; + + 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) + "#, + ) + .bind("f/test/pg_e2e_trigger") + .bind(script_path) + .bind(false) + .bind("test-workspace") + .bind("test-user") + .bind("test@windmill.dev") + .bind("u/test-user/pg_res") + .bind(&slot_name) + .bind(&pub_name) + .execute(&db) + .await?; + + let _server = ApiServer::start_with_listeners(db.clone()).await?; + tokio::time::sleep(Duration::from_secs(3)).await; + + // Insert a row into the tracked table to trigger a change event + sqlx::query("INSERT INTO test_trigger_table (data) VALUES ('e2e test data')") + .execute(&db) + .await?; + + let job = poll_for_trigger_job(&db, script_path, "postgres", Duration::from_secs(30)).await?; + assert!(job.args.is_some(), "job should have args"); + + Ok(()) +} + +// ============================================================================ +// Kafka Trigger E2E (Enterprise) +// ============================================================================ + +/// End-to-end test for Kafka trigger (Enterprise only). +/// +/// Requires a running Kafka broker with the test topic. Setup: +/// ```bash +/// ./tests/fixtures/start_kafka.sh +/// ``` +/// +/// Run: +/// ```bash +/// cargo test --test trigger_e2e test_kafka_e2e \ +/// --features kafka,enterprise,private -- --ignored --nocapture +/// ``` +#[cfg(all(feature = "enterprise", feature = "private"))] +#[ignore = "requires running Kafka broker on localhost:9092"] +#[sqlx::test(fixtures("base"))] +async fn test_kafka_e2e(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + + let script_path = "f/test/kafka_e2e_handler"; + insert_test_script(&db, script_path).await?; + + insert_resource( + &db, + "u/test-user/kafka_res", + "kafka", + json!({ + "brokers": ["localhost:9092"], + "security": { "label": "PLAINTEXT" } + }), + ) + .await?; + + 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_e2e_trigger", + "u/test-user/kafka_res", + &["windmill-e2e-test"] as &[&str], + "windmill-e2e-test-group", + script_path, + false, + "test-workspace", + "test-user", + "test@windmill.dev", + ) + .execute(&db) + .await?; + + let _server = ApiServer::start_with_listeners(db.clone()).await?; + tokio::time::sleep(Duration::from_secs(5)).await; + + // Produce messages using rdkafka. The consumer starts with auto.offset.reset=latest + // and needs time for group rebalance, so we send repeatedly until a job appears. + use rdkafka::config::ClientConfig; + use rdkafka::producer::{FutureProducer, FutureRecord}; + + let producer: FutureProducer = ClientConfig::new() + .set("bootstrap.servers", "localhost:9092") + .create()?; + + let db2 = db.clone(); + let produce_handle = tokio::spawn(async move { + for _ in 0..30 { + let _ = producer + .send( + FutureRecord::to("windmill-e2e-test") + .payload("hello from kafka e2e test") + .key("test-key"), + Duration::from_secs(5), + ) + .await; + tokio::time::sleep(Duration::from_secs(1)).await; + } + }); + + let job = poll_for_trigger_job(&db2, script_path, "kafka", Duration::from_secs(30)).await?; + produce_handle.abort(); + assert!(job.args.is_some(), "job should have args"); + + Ok(()) +} + +// ============================================================================ +// NATS Trigger E2E (Enterprise) +// ============================================================================ + +/// End-to-end test for NATS trigger (Enterprise only). +/// +/// Requires a running NATS server. Setup: +/// ```bash +/// ./tests/fixtures/start_nats.sh +/// ``` +/// +/// Run: +/// ```bash +/// cargo test --test trigger_e2e test_nats_e2e \ +/// --features nats,enterprise,private -- --ignored --nocapture +/// ``` +#[cfg(all(feature = "enterprise", feature = "private"))] +#[ignore = "requires running NATS server on localhost:4222"] +#[sqlx::test(fixtures("base"))] +async fn test_nats_e2e(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + + let script_path = "f/test/nats_e2e_handler"; + insert_test_script(&db, script_path).await?; + + insert_resource( + &db, + "u/test-user/nats_res", + "nats", + json!({ + "servers": ["nats://localhost:4222"], + "auth": { "label": "NO_AUTH" }, + "require_tls": false + }), + ) + .await?; + + 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_e2e_trigger", + "u/test-user/nats_res", + &["windmill.e2e.test"] as &[&str], + script_path, + false, + "test-workspace", + "test-user", + "test@windmill.dev", + false, + ) + .execute(&db) + .await?; + + let _server = ApiServer::start_with_listeners(db.clone()).await?; + tokio::time::sleep(Duration::from_secs(3)).await; + + // Publish a message using async-nats + let nats_client = async_nats::connect("localhost:4222").await?; + nats_client + .publish("windmill.e2e.test", "hello from nats e2e test".into()) + .await?; + nats_client.flush().await?; + + let job = poll_for_trigger_job(&db, script_path, "nats", Duration::from_secs(30)).await?; + assert!(job.args.is_some(), "job should have args"); + + Ok(()) +} + +// ============================================================================ +// SQS Trigger E2E (Enterprise) +// ============================================================================ + +/// End-to-end test for SQS trigger (Enterprise only). +/// +/// Requires LocalStack with the test queue. Setup: +/// ```bash +/// ./tests/fixtures/start_sqs.sh +/// ``` +/// +/// Run: +/// ```bash +/// AWS_ENDPOINT_URL=http://localhost:4566 \ +/// cargo test --test trigger_e2e test_sqs_e2e \ +/// --features sqs_trigger,enterprise,private -- --ignored --nocapture +/// ``` +#[cfg(all(feature = "enterprise", feature = "private"))] +#[ignore = "requires LocalStack SQS on localhost:4566"] +#[sqlx::test(fixtures("base"))] +async fn test_sqs_e2e(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + + // The SQS listener uses aws_config which respects AWS_ENDPOINT_URL for LocalStack. + std::env::set_var("AWS_ENDPOINT_URL", "http://localhost:4566"); + + let script_path = "f/test/sqs_e2e_handler"; + insert_test_script(&db, script_path).await?; + + insert_resource( + &db, + "u/test-user/aws_res", + "aws", + json!({ + "awsAccessKeyId": "test", + "awsSecretAccessKey": "test", + "region": "us-east-1" + }), + ) + .await?; + + 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_e2e_trigger", + "http://localhost:4566/000000000000/windmill-e2e-test", + "u/test-user/aws_res", + script_path, + false, + "test-workspace", + "test-user", + "test@windmill.dev", + ) + .execute(&db) + .await?; + + let _server = ApiServer::start_with_listeners(db.clone()).await?; + tokio::time::sleep(Duration::from_secs(3)).await; + + // Send a message using aws-sdk-sqs + let config = aws_config::defaults(aws_config::BehaviorVersion::latest()) + .endpoint_url("http://localhost:4566") + .region(aws_config::Region::new("us-east-1")) + .credentials_provider(aws_credential_types::Credentials::new( + "test", "test", None, None, "test", + )) + .load() + .await; + let sqs_client = aws_sdk_sqs::Client::new(&config); + + sqs_client + .send_message() + .queue_url("http://localhost:4566/000000000000/windmill-e2e-test") + .message_body("hello from sqs e2e test") + .send() + .await?; + + let job = poll_for_trigger_job(&db, script_path, "sqs", Duration::from_secs(30)).await?; + assert!(job.args.is_some(), "job should have args"); + + Ok(()) +} + +// ============================================================================ +// GCP Pub/Sub Trigger E2E (Enterprise) +// ============================================================================ + +/// End-to-end test for GCP Pub/Sub trigger (Enterprise only). +/// +/// Requires the GCP Pub/Sub emulator with test topic/subscription. Setup: +/// ```bash +/// ./tests/fixtures/start_gcp_pubsub.sh +/// ``` +/// +/// Run: +/// ```bash +/// PUBSUB_EMULATOR_HOST=localhost:8085 \ +/// cargo test --test trigger_e2e test_gcp_e2e \ +/// --features gcp_trigger,enterprise,private -- --ignored --nocapture +/// ``` +#[cfg(all(feature = "enterprise", feature = "private"))] +#[ignore = "requires GCP Pub/Sub emulator on localhost:8085"] +#[sqlx::test(fixtures("base"))] +async fn test_gcp_e2e(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + + let script_path = "f/test/gcp_e2e_handler"; + insert_test_script(&db, script_path).await?; + + // The GCP emulator doesn't require real credentials, but the resource + // row must still exist for the listener to resolve it. + // The private_key must use literal \n (backslash-n) as in real GCP service + // account JSON files. The trigger code re-parses it through serde_json to + // convert those escape sequences to actual newlines. + insert_resource( + &db, + "u/test-user/gcp_res", + "google", + json!({ + "project_id": "test-project", + "private_key_id": "test", + "private_key": "-----BEGIN RSA PRIVATE KEY-----\\nMIIBogIBAAJBALRiMLAH\\n-----END RSA PRIVATE KEY-----\\n", + "client_email": "test@test-project.iam.gserviceaccount.com", + "auth_uri": "https://accounts.google.com/o/oauth2/auth", + "token_uri": "https://oauth2.googleapis.com/token", + "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs" + }), + ) + .await?; + + sqlx::query( + r#" + INSERT INTO gcp_trigger ( + path, gcp_resource_path, topic_id, subscription_id, + delivery_type, subscription_mode, script_path, is_flow, + workspace_id, edited_by, email + ) + VALUES ($1, $2, $3, $4, $5::delivery_mode, $6::gcp_subscription_mode, $7, $8, $9, $10, $11) + "#, + ) + .bind("f/test/gcp_e2e_trigger") + .bind("u/test-user/gcp_res") + .bind("windmill-e2e-test") + .bind("windmill-e2e-sub") + .bind("pull") + .bind("existing") + .bind(script_path) + .bind(false) + .bind("test-workspace") + .bind("test-user") + .bind("test@windmill.dev") + .execute(&db) + .await?; + + let _server = ApiServer::start_with_listeners(db.clone()).await?; + tokio::time::sleep(Duration::from_secs(3)).await; + + // Publish a message to the emulator via HTTP + let client = reqwest::Client::new(); + let emulator_host = + std::env::var("PUBSUB_EMULATOR_HOST").unwrap_or_else(|_| "localhost:8085".to_string()); + // The google-cloud-pubsub crate uses "local-project" as the default project ID + // when PUBSUB_EMULATOR_HOST is set, so we must publish to that project's topic. + let publish_url = format!( + "http://{}/v1/projects/local-project/topics/windmill-e2e-test:publish", + emulator_host + ); + + let message_data = base64::Engine::encode( + &base64::engine::general_purpose::STANDARD, + "hello from gcp e2e test", + ); + client + .post(&publish_url) + .json(&json!({ + "messages": [{ "data": message_data }] + })) + .send() + .await?; + + let job = poll_for_trigger_job(&db, script_path, "gcp", Duration::from_secs(30)).await?; + assert!(job.args.is_some(), "job should have args"); + + Ok(()) +} diff --git a/backend/tests/triggers.rs b/backend/tests/triggers.rs index 47ef82ad19..ff7d34decd 100644 --- a/backend/tests/triggers.rs +++ b/backend/tests/triggers.rs @@ -1272,3 +1272,431 @@ async fn test_schedule_insert_and_query(db: Pool) -> anyhow::Result<() Ok(()) } + +// ============================================================================ +// MQTT Trigger Tests (DB-level) +// ============================================================================ + +#[sqlx::test(fixtures("base"))] +async fn test_mqtt_trigger_insert(db: Pool) -> anyhow::Result<()> { + sqlx::query( + r#" + INSERT INTO mqtt_trigger ( + path, mqtt_resource_path, subscribe_topics, client_version, + script_path, is_flow, workspace_id, edited_by, email + ) + VALUES ($1, $2, ARRAY[$3::jsonb], $4::mqtt_client_version, $5, $6, $7, $8, $9) + "#, + ) + .bind("f/test/mqtt_trigger") + .bind("u/admin/mqtt_resource") + .bind(json!({"topic": "test/+", "qos": "qos1"})) + .bind("v5") + .bind("f/test/mqtt_handler") + .bind(false) + .bind("test-workspace") + .bind("test-user") + .bind("test@windmill.dev") + .execute(&db) + .await?; + + let trigger = sqlx::query!( + r#" + SELECT mqtt_resource_path, client_version AS "client_version: String", + script_path, mode AS "mode: String" + FROM mqtt_trigger + WHERE workspace_id = $1 AND path = $2 + "#, + "test-workspace", + "f/test/mqtt_trigger", + ) + .fetch_one(&db) + .await?; + + assert_eq!(trigger.mqtt_resource_path, "u/admin/mqtt_resource"); + assert_eq!(trigger.client_version, "v5"); + assert_eq!(trigger.script_path, "f/test/mqtt_handler"); + assert_eq!(trigger.mode, "enabled"); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_mqtt_trigger_update(db: Pool) -> anyhow::Result<()> { + sqlx::query( + r#" + INSERT INTO mqtt_trigger ( + path, mqtt_resource_path, subscribe_topics, client_version, + script_path, is_flow, workspace_id, edited_by, email + ) + VALUES ($1, $2, ARRAY[$3::jsonb], $4::mqtt_client_version, $5, $6, $7, $8, $9) + "#, + ) + .bind("f/test/mqtt_trigger") + .bind("u/admin/mqtt_resource") + .bind(json!({"topic": "test/+", "qos": "qos1"})) + .bind("v5") + .bind("f/test/old_handler") + .bind(false) + .bind("test-workspace") + .bind("test-user") + .bind("test@windmill.dev") + .execute(&db) + .await?; + + sqlx::query!( + "UPDATE mqtt_trigger SET script_path = $1 WHERE workspace_id = $2 AND path = $3", + "f/test/new_handler", + "test-workspace", + "f/test/mqtt_trigger", + ) + .execute(&db) + .await?; + + let trigger = sqlx::query!( + "SELECT script_path FROM mqtt_trigger WHERE workspace_id = $1 AND path = $2", + "test-workspace", + "f/test/mqtt_trigger", + ) + .fetch_one(&db) + .await?; + + assert_eq!(trigger.script_path, "f/test/new_handler"); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_mqtt_trigger_delete(db: Pool) -> anyhow::Result<()> { + sqlx::query( + r#" + INSERT INTO mqtt_trigger ( + path, mqtt_resource_path, subscribe_topics, client_version, + script_path, is_flow, workspace_id, edited_by, email + ) + VALUES ($1, $2, ARRAY[$3::jsonb], $4::mqtt_client_version, $5, $6, $7, $8, $9) + "#, + ) + .bind("f/test/mqtt_trigger") + .bind("u/admin/mqtt_resource") + .bind(json!({"topic": "test/+", "qos": "qos1"})) + .bind("v5") + .bind("f/test/mqtt_handler") + .bind(false) + .bind("test-workspace") + .bind("test-user") + .bind("test@windmill.dev") + .execute(&db) + .await?; + + sqlx::query!( + "DELETE FROM mqtt_trigger WHERE workspace_id = $1 AND path = $2", + "test-workspace", + "f/test/mqtt_trigger", + ) + .execute(&db) + .await?; + + let count = sqlx::query_scalar!( + "SELECT COUNT(*) FROM mqtt_trigger WHERE workspace_id = $1 AND path = $2", + "test-workspace", + "f/test/mqtt_trigger", + ) + .fetch_one(&db) + .await?; + + assert_eq!(count, Some(0)); + + Ok(()) +} + +// ============================================================================ +// GCP Trigger Tests (DB-level) +// ============================================================================ + +#[sqlx::test(fixtures("base"))] +async fn test_gcp_trigger_insert_pull(db: Pool) -> anyhow::Result<()> { + sqlx::query( + r#" + INSERT INTO gcp_trigger ( + path, gcp_resource_path, topic_id, subscription_id, + delivery_type, subscription_mode, script_path, is_flow, + workspace_id, edited_by, email + ) + VALUES ($1, $2, $3, $4, $5::delivery_mode, $6::gcp_subscription_mode, $7, $8, $9, $10, $11) + "#, + ) + .bind("f/test/gcp_trigger_pull") + .bind("u/admin/gcp_resource") + .bind("my-topic") + .bind("my-subscription") + .bind("pull") + .bind("create_update") + .bind("f/test/gcp_handler") + .bind(false) + .bind("test-workspace") + .bind("test-user") + .bind("test@windmill.dev") + .execute(&db) + .await?; + + let trigger = sqlx::query!( + r#" + SELECT gcp_resource_path, topic_id, subscription_id, + delivery_type AS "delivery_type: String", + subscription_mode AS "subscription_mode: String", + mode AS "mode: String" + FROM gcp_trigger + WHERE workspace_id = $1 AND path = $2 + "#, + "test-workspace", + "f/test/gcp_trigger_pull", + ) + .fetch_one(&db) + .await?; + + assert_eq!(trigger.gcp_resource_path, "u/admin/gcp_resource"); + assert_eq!(trigger.topic_id, "my-topic"); + assert_eq!(trigger.subscription_id, "my-subscription"); + assert_eq!(trigger.delivery_type, "pull"); + assert_eq!(trigger.subscription_mode, "create_update"); + assert_eq!(trigger.mode, "enabled"); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_gcp_trigger_insert_push(db: Pool) -> anyhow::Result<()> { + sqlx::query( + r#" + INSERT INTO gcp_trigger ( + path, gcp_resource_path, topic_id, subscription_id, + delivery_type, delivery_config, subscription_mode, + script_path, is_flow, workspace_id, edited_by, email + ) + VALUES ($1, $2, $3, $4, $5::delivery_mode, $6::jsonb, $7::gcp_subscription_mode, $8, $9, $10, $11, $12) + "#, + ) + .bind("f/test/gcp_trigger_push") + .bind("u/admin/gcp_resource") + .bind("my-topic") + .bind("my-push-subscription") + .bind("push") + .bind(json!({"endpoint": "https://example.com/push"})) + .bind("create_update") + .bind("f/test/gcp_handler") + .bind(false) + .bind("test-workspace") + .bind("test-user") + .bind("test@windmill.dev") + .execute(&db) + .await?; + + let trigger = sqlx::query!( + r#" + SELECT delivery_type AS "delivery_type: String", + delivery_config + FROM gcp_trigger + WHERE workspace_id = $1 AND path = $2 + "#, + "test-workspace", + "f/test/gcp_trigger_push", + ) + .fetch_one(&db) + .await?; + + assert_eq!(trigger.delivery_type, "push"); + assert!(trigger.delivery_config.is_some()); + assert_eq!( + trigger.delivery_config.unwrap()["endpoint"], + "https://example.com/push" + ); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_gcp_trigger_unique_constraint(db: Pool) -> anyhow::Result<()> { + let insert_query = r#" + INSERT INTO gcp_trigger ( + path, gcp_resource_path, topic_id, subscription_id, + delivery_type, subscription_mode, script_path, is_flow, + workspace_id, edited_by, email + ) + VALUES ($1, $2, $3, $4, $5::delivery_mode, $6::gcp_subscription_mode, $7, $8, $9, $10, $11) + "#; + + sqlx::query(insert_query) + .bind("f/test/gcp_trigger_1") + .bind("u/admin/gcp_resource") + .bind("my-topic") + .bind("shared-subscription") + .bind("pull") + .bind("create_update") + .bind("f/test/gcp_handler") + .bind(false) + .bind("test-workspace") + .bind("test-user") + .bind("test@windmill.dev") + .execute(&db) + .await?; + + // Inserting a second trigger with same (subscription_id, gcp_resource_path, workspace_id) should fail + let result = sqlx::query(insert_query) + .bind("f/test/gcp_trigger_2") + .bind("u/admin/gcp_resource") + .bind("my-topic") + .bind("shared-subscription") + .bind("pull") + .bind("create_update") + .bind("f/test/gcp_handler_2") + .bind(false) + .bind("test-workspace") + .bind("test-user") + .bind("test@windmill.dev") + .execute(&db) + .await; + + assert!( + result.is_err(), + "should fail due to unique constraint on (subscription_id, gcp_resource_path, workspace_id)" + ); + + Ok(()) +} + +// ============================================================================ +// Email Trigger Tests (DB-level) +// ============================================================================ + +#[sqlx::test(fixtures("base"))] +async fn test_email_trigger_insert(db: Pool) -> anyhow::Result<()> { + sqlx::query!( + r#" + INSERT INTO email_trigger ( + path, local_part, workspaced_local_part, script_path, + is_flow, workspace_id, edited_by, email + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8) + "#, + "f/test/email_trigger", + "support", + true, + "f/test/email_handler", + false, + "test-workspace", + "test-user", + "test@windmill.dev", + ) + .execute(&db) + .await?; + + let trigger = sqlx::query!( + r#" + SELECT local_part, workspaced_local_part, script_path, + mode AS "mode: String" + FROM email_trigger + WHERE workspace_id = $1 AND path = $2 + "#, + "test-workspace", + "f/test/email_trigger", + ) + .fetch_one(&db) + .await?; + + assert_eq!(trigger.local_part, "support"); + assert_eq!(trigger.workspaced_local_part, true); + assert_eq!(trigger.script_path, "f/test/email_handler"); + assert_eq!(trigger.mode, "enabled"); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_email_trigger_update(db: Pool) -> anyhow::Result<()> { + sqlx::query!( + r#" + INSERT INTO email_trigger ( + path, local_part, workspaced_local_part, script_path, + is_flow, workspace_id, edited_by, email + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8) + "#, + "f/test/email_trigger", + "support", + true, + "f/test/old_handler", + false, + "test-workspace", + "test-user", + "test@windmill.dev", + ) + .execute(&db) + .await?; + + sqlx::query!( + "UPDATE email_trigger SET script_path = $1, local_part = $2 WHERE workspace_id = $3 AND path = $4", + "f/test/new_handler", + "billing", + "test-workspace", + "f/test/email_trigger", + ) + .execute(&db) + .await?; + + let trigger = sqlx::query!( + "SELECT script_path, local_part FROM email_trigger WHERE workspace_id = $1 AND path = $2", + "test-workspace", + "f/test/email_trigger", + ) + .fetch_one(&db) + .await?; + + assert_eq!(trigger.script_path, "f/test/new_handler"); + assert_eq!(trigger.local_part, "billing"); + + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +async fn test_email_trigger_delete(db: Pool) -> anyhow::Result<()> { + sqlx::query!( + r#" + INSERT INTO email_trigger ( + path, local_part, workspaced_local_part, script_path, + is_flow, workspace_id, edited_by, email + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8) + "#, + "f/test/email_trigger", + "support", + true, + "f/test/email_handler", + false, + "test-workspace", + "test-user", + "test@windmill.dev", + ) + .execute(&db) + .await?; + + sqlx::query!( + "DELETE FROM email_trigger WHERE workspace_id = $1 AND path = $2", + "test-workspace", + "f/test/email_trigger", + ) + .execute(&db) + .await?; + + let count = sqlx::query_scalar!( + "SELECT COUNT(*) FROM email_trigger WHERE workspace_id = $1 AND path = $2", + "test-workspace", + "f/test/email_trigger", + ) + .fetch_one(&db) + .await?; + + assert_eq!(count, Some(0)); + + Ok(()) +}