starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-18 16:02:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

test: add end-to-end trigger integration tests and DB CRUD tests

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-02-09 12:28:25 +00:00
parent edb0d4a05d
commit dc5e69481d
13 changed files with 1518 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/Cargo.lock
Generated
+6
View File
@@ -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",
backend/Cargo.toml
+6
View File
@@ -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]
backend/tests/common/mod.rs
+6
View File
@@ -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<Postgres>) -> anyhow::Result<Self> {
Self::start_inner(db, true).await
}
async fn start_inner(db: Pool<Postgres>, agent_mode: bool) -> anyhow::Result<Self> {
let (tx, rx) = tokio::sync::broadcast::channel::<()>(1);
backend/tests/fixtures/start_all_triggers.sh
Vendored Executable
+58
View File
@@ -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
backend/tests/fixtures/start_gcp_pubsub.sh
Vendored Executable
+47
View File
@@ -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"
backend/tests/fixtures/start_kafka.sh
Vendored Executable
+52
View File
@@ -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"
backend/tests/fixtures/start_mqtt.sh
Vendored Executable
+31
View File
@@ -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"
backend/tests/fixtures/start_nats.sh
Vendored Executable
+29
View File
@@ -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"
backend/tests/fixtures/start_postgres_replication.sh
Vendored Executable
+53
View File
@@ -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"
backend/tests/fixtures/start_sqs.sh
Vendored Executable
+45
View File
@@ -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"
backend/tests/fixtures/start_websocket.sh
Vendored Executable
+31
View File
@@ -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"
backend/tests/trigger_e2e.rs
+726
View File
@@ -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<String>,
trigger_kind: Option<String>,
args: Option<sqlx::types::Json<serde_json::Value>>,
}
/// 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<Postgres>,
script_path: &str,
trigger_kind: &str,
timeout: Duration,
) -> anyhow::Result<TriggerJobRow> {
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<serde_json::Value>"
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<Postgres>, path: &str) -> anyhow::Result<i64> {
let hash: i64 = rand::random::<i64>().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<Postgres>,
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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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(())
}
backend/tests/triggers.rs
+428
View File
@@ -1272,3 +1272,431 @@ async fn test_schedule_insert_and_query(db: Pool<Postgres>) -> anyhow::Result<()
Ok(())
}
// ============================================================================
// MQTT Trigger Tests (DB-level)
// ============================================================================
#[sqlx::test(fixtures("base"))]
async fn test_mqtt_trigger_insert(db: Pool<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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(())
}