diff --git a/Cargo.lock b/Cargo.lock index 9b0e14f..2d0ee03 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1502,7 +1502,7 @@ dependencies = [ [[package]] name = "rawtree-cli" -version = "0.6.13" +version = "0.6.14" dependencies = [ "anyhow", "axoupdater", diff --git a/Cargo.toml b/Cargo.toml index 4756a67..ba48682 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "rawtree-cli" -version = "0.6.13" +version = "0.6.14" edition = "2021" repository = "https://github.com/rawtreedb/rawtree-cli" diff --git a/README.md b/README.md index 9189b0f..8840753 100644 --- a/README.md +++ b/README.md @@ -249,17 +249,19 @@ creation; create a replacement key to change it. There is no `key update` comman ### Workflows -Workflows run saved SQL on a schedule and deliver results to sinks. +Workflows run saved SQL on demand or on a schedule and deliver results to sinks. ```sh rtree workflow list rtree workflow create --name errors --database analytics --interval-seconds 60 \ --sql "INSERT INTO error_counts SELECT count() FROM events WHERE level = 'error'" +rtree workflow create --name on-demand --database analytics --sql "SELECT 1" rtree workflow create --name alerts --disabled --sql - \ --sink '{"type":"http","settings":{"url":"https://example.com/hook","headers":{"Authorization":"Bearer ..."}}}' \ < alerts.sql rtree workflow get rtree workflow update --interval-seconds 300 +rtree workflow update --interval-seconds null rtree workflow update --disable rtree workflow update --clear-sinks rtree workflow delete @@ -274,6 +276,20 @@ rtree workflow metrics --since 7d Workflow commands need both an organization and a cluster (`--org`/`--cluster`, the environment, or the saved defaults). API keys must have admin permission. `create` uses the selected database when `--database` is omitted. +The CLI sends `--database` and `--sql` inside the API's `query` object. On `update`, +each flag changes only that query field; the other field stays unchanged. + +Creating a workflow without `--interval-seconds` makes it manual-only. The CLI +explicitly sends `interval_seconds: null`, so the API's default interval is not used. +Use `--interval-seconds null` to remove an existing schedule, matching the API's +`interval_seconds: null`. The same value is accepted on `create`. +An interval restores recurring execution, subject to the enabled state; use +`--enable` as well if the workflow is paused. Omitting `--interval-seconds` on `update` +preserves the current mode. +`--disable` pauses scheduled runs while retaining the interval. Manual runs remain +available in either mode, including while paused. Text output shows status as +`active`, `paused`, or `manual`, the next scheduled run when available, and HTTP +sink URLs. Sinks are JSON objects passed with `--sink`, up to five. Each has a `type` and a nested `settings` object: `{"database","table"}` for `table` sinks and @@ -367,6 +383,9 @@ database creation, insertion, querying, API key login, and deletion through the It also logs in with the local test user's email and password, verifies API access using the saved session, logs out, and checks that credentials are cleared and API access requires authentication again. +Workflow tests cover definition updates, manual and scheduled execution, +pause/resume, cancellation, idempotency, table sink delivery, logs, and metrics. +They run through the real CLI and clean up their workflows and databases. The Platform repository continues to test API endpoints directly. The Docker job checks out private `rawtreedb/rawtree-platform` at `main`. It diff --git a/src/cli.rs b/src/cli.rs index e6d23f9..d386dec 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -458,7 +458,7 @@ pub enum WorkflowCommand { /// Workflow ID id: String, }, - /// Create a workflow that runs SQL on a schedule + /// Create a workflow that runs SQL on demand or on a schedule #[command(after_help = SINK_HELP)] Create { /// Workflow name (letters, digits, '-' and '_') @@ -470,9 +470,9 @@ pub enum WorkflowCommand { /// SQL to run: a SELECT or INSERT INTO ... SELECT. Use "-" to read from stdin. #[arg(long)] sql: String, - /// Seconds between runs (server default: 1) - #[arg(long, value_parser = clap::value_parser!(u32).range(1..=86400))] - interval_seconds: Option, + /// Seconds between automatic runs (1–86400); null or omitted means manual-only + #[arg(long, value_name = "SECONDS|null")] + interval_seconds: Option, /// Create the workflow paused #[arg(long)] disabled: bool, @@ -494,9 +494,9 @@ pub enum WorkflowCommand { /// New SQL. Use "-" to read from stdin. #[arg(long)] sql: Option, - /// Seconds between runs - #[arg(long, value_parser = clap::value_parser!(u32).range(1..=86400))] - interval_seconds: Option, + /// Seconds between automatic runs (1–86400); null removes the schedule + #[arg(long, value_name = "SECONDS|null")] + interval_seconds: Option, /// Resume scheduled runs #[arg(long, conflicts_with = "disable")] enable: bool, @@ -563,6 +563,23 @@ pub enum WorkflowCommand { }, } +#[derive(Clone, Copy, Debug)] +pub struct WorkflowIntervalArg(pub Option); + +impl FromStr for WorkflowIntervalArg { + type Err = String; + + fn from_str(value: &str) -> Result { + if value == "null" { + return Ok(Self(None)); + } + match value.parse::() { + Ok(seconds) if (1..=86400).contains(&seconds) => Ok(Self(Some(seconds))), + _ => Err("expected a whole number from 1 to 86400, or null".into()), + } + } +} + /// Time window for workflow logs and metrics. Defaults to the last 24 hours; at most 7 days. #[derive(clap::Args, Debug, Default)] pub struct WorkflowTimeRange { @@ -1176,7 +1193,7 @@ mod tests { .. }, } => { - assert_eq!(interval_seconds, Some(60)); + assert_eq!(interval_seconds.map(|interval| interval.0), Some(Some(60))); assert!(disabled); assert_eq!(sinks.len(), 2); assert!(database.is_none()); diff --git a/src/commands/workflow.rs b/src/commands/workflow.rs index 091ff58..779738e 100644 --- a/src/commands/workflow.rs +++ b/src/commands/workflow.rs @@ -33,21 +33,38 @@ impl WorkflowScope<'_> { } } +#[derive(Deserialize)] +struct WorkflowQuery { + database: String, + sql: String, +} + #[derive(Deserialize)] struct Workflow { id: String, name: String, - database: String, - sql: String, + query: WorkflowQuery, enabled: bool, revision: i64, - interval_seconds: i64, + /// `None` for manual-only workflows, which have no schedule. + interval_seconds: Option, + next_run_at: Option, created_at: String, updated_at: String, #[serde(default)] sinks: Vec, } +impl Workflow { + fn status(&self) -> &'static str { + match (self.interval_seconds, self.enabled) { + (None, _) => "manual", + (Some(_), true) => "active", + (Some(_), false) => "paused", + } + } +} + #[derive(Deserialize)] struct ListWorkflowsResponse { workflows: Vec, @@ -133,7 +150,7 @@ pub struct UpdateWorkflow { pub name: Option, pub database: Option, pub sql: Option, - pub interval_seconds: Option, + pub interval_seconds: Option>, pub enabled: Option, pub sinks: Option>, } @@ -166,17 +183,13 @@ fn parse_sinks(sinks: &[String]) -> Result> { } fn create_body(request: CreateWorkflow) -> Result { - let mut body = json!({ + Ok(json!({ "name": request.name, - "database": request.database, - "sql": request.sql, + "query": {"database": request.database, "sql": request.sql}, "enabled": !request.disabled, + "interval_seconds": request.interval_seconds, "sinks": parse_sinks(&request.sinks)?, - }); - if let Some(interval) = request.interval_seconds { - body["interval_seconds"] = json!(interval); - } - Ok(body) + })) } fn update_body(request: UpdateWorkflow) -> Result { @@ -184,11 +197,15 @@ fn update_body(request: UpdateWorkflow) -> Result { if let Some(name) = request.name { body.insert("name".into(), json!(name)); } + let mut query = Map::new(); if let Some(database) = request.database { - body.insert("database".into(), json!(database)); + query.insert("database".into(), json!(database)); } if let Some(sql) = request.sql { - body.insert("sql".into(), json!(sql)); + query.insert("sql".into(), json!(sql)); + } + if !query.is_empty() { + body.insert("query".into(), Value::Object(query)); } if let Some(enabled) = request.enabled { body.insert("enabled".into(), json!(enabled)); @@ -281,11 +298,12 @@ fn format_ms(ms: i64) -> String { .unwrap_or_else(|| ms.to_string()) } -fn format_interval(seconds: i64) -> String { +fn format_interval(seconds: Option) -> String { match seconds { - s if s % 3600 == 0 => format!("{}h", s / 3600), - s if s % 60 == 0 => format!("{}m", s / 60), - s => format!("{s}s"), + None => "—".into(), + Some(s) if s % 3600 == 0 => format!("{}h", s / 3600), + Some(s) if s % 60 == 0 => format!("{}m", s / 60), + Some(s) => format!("{s}s"), } } @@ -299,6 +317,7 @@ fn describe_sink(sink: &Value) -> String { settings["table"].as_str().unwrap_or("?") ), Some("http") => { + let url = settings["url"].as_str().unwrap_or("URL unavailable"); let headers = settings["header_names"] .as_array() .map(|names| { @@ -310,9 +329,9 @@ fn describe_sink(sink: &Value) -> String { }) .unwrap_or_default(); if headers.is_empty() { - format!("http id={id}") + format!("http {url} id={id}") } else { - format!("http id={id} headers={headers}") + format!("http {url} id={id} headers={headers}") } } _ => sink.to_string(), @@ -322,16 +341,13 @@ fn describe_sink(sink: &Value) -> String { fn print_workflow(workflow: &Workflow) { println!(" id: {}", workflow.id); println!(" name: {}", workflow.name); - println!(" database: {}", workflow.database); + println!(" database: {}", workflow.query.database); + println!(" status: {}", workflow.status()); + println!(" interval: {}", format_interval(workflow.interval_seconds)); println!( - " status: {}", - if workflow.enabled { - "enabled" - } else { - "paused" - } + " next run: {}", + workflow.next_run_at.as_deref().unwrap_or("—") ); - println!(" interval: {}", format_interval(workflow.interval_seconds)); println!(" revision: {}", workflow.revision); println!(" created: {}", workflow.created_at); println!(" updated: {}", workflow.updated_at); @@ -344,7 +360,7 @@ fn print_workflow(workflow: &Workflow) { } } println!(" sql:"); - for line in workflow.sql.lines() { + for line in workflow.query.sql.lines() { println!(" {line}"); } } @@ -358,19 +374,16 @@ pub fn list(client: &ApiClient, scope: &WorkflowScope, json_mode: bool) -> Resul } let mut table = new_cli_table(); table.set_header(vec![ - "name", "database", "status", "interval", "sinks", "updated", "id", + "name", "database", "status", "interval", "next run", "sinks", "updated", "id", ]); for workflow in &resp.workflows { table.add_row(vec![ Cell::new(&workflow.name), - Cell::new(&workflow.database), - Cell::new(if workflow.enabled { - "enabled" - } else { - "paused" - }), + Cell::new(&workflow.query.database), + Cell::new(workflow.status()), Cell::new(format_interval(workflow.interval_seconds)) .set_alignment(CellAlignment::Right), + Cell::new(workflow.next_run_at.as_deref().unwrap_or("—")), Cell::new(workflow.sinks.len()).set_alignment(CellAlignment::Right), Cell::new(&workflow.updated_at), Cell::new(&workflow.id), @@ -652,9 +665,9 @@ mod tests { body, json!({ "name": "alerts", - "database": "analytics", - "sql": "SELECT 1", + "query": {"database": "analytics", "sql": "SELECT 1"}, "enabled": false, + "interval_seconds": null, "sinks": [{"type": "table", "settings": {"database": "a", "table": "b"}}], }) ); @@ -687,7 +700,7 @@ mod tests { name: None, database: None, sql: None, - interval_seconds: Some(30), + interval_seconds: Some(Some(30)), enabled: Some(false), sinks: Some(vec![]), }) @@ -696,6 +709,25 @@ mod tests { body, json!({"interval_seconds": 30, "enabled": false, "sinks": []}) ); + + let body = update_body(UpdateWorkflow { + name: None, + database: None, + sql: Some("SELECT 2".into()), + interval_seconds: None, + enabled: None, + sinks: None, + }) + .unwrap(); + assert_eq!(body, json!({"query": {"sql": "SELECT 2"}})); + } + + #[test] + fn format_interval_shows_only_scheduled_durations() { + assert_eq!(format_interval(None), "—"); + assert_eq!(format_interval(Some(7200)), "2h"); + assert_eq!(format_interval(Some(120)), "2m"); + assert_eq!(format_interval(Some(45)), "45s"); } #[test] @@ -765,7 +797,7 @@ mod tests { describe_sink( &json!({"type": "http", "id": "s2", "settings": {"url_configured": true, "header_names": ["Authorization"]}}) ), - "http id=s2 headers=Authorization" + "http URL unavailable id=s2 headers=Authorization" ); assert_eq!( describe_sink(&json!({"type": "queue"})), diff --git a/src/main.rs b/src/main.rs index 7794624..1f8ba85 100644 --- a/src/main.rs +++ b/src/main.rs @@ -531,7 +531,7 @@ fn run(cli: Cli) -> Result { name, database: resolve_database(database)?, sql: read_sql_arg(sql)?, - interval_seconds, + interval_seconds: interval_seconds.and_then(|interval| interval.0), disabled, sinks, }, @@ -555,7 +555,7 @@ fn run(cli: Cli) -> Result { name, database, sql: sql.map(read_sql_arg).transpose()?, - interval_seconds, + interval_seconds: interval_seconds.map(|interval| interval.0), enabled: (enable || disable).then_some(enable), sinks: (clear_sinks || !sinks.is_empty()).then_some(sinks), }, diff --git a/tests/api_contracts.rs b/tests/api_contracts.rs index 72f721f..6d49495 100644 --- a/tests/api_contracts.rs +++ b/tests/api_contracts.rs @@ -181,8 +181,9 @@ fn logs_use_cluster_scope_without_a_database_or_with_a_stale_saved_database() { const WORKFLOW_SCOPE: &str = "organization=team%20alpha&cluster=prod%2Feu"; fn workflow_response() -> Value { - json!({"id": "wf-1", "name": "alerts", "database": "analytics", "sql": "SELECT 1", + json!({"id": "wf-1", "name": "alerts", "query": {"database": "analytics", "sql": "SELECT 1"}, "enabled": true, "revision": 1, "interval_seconds": 60, + "next_run_at": "2026-10-07T10:01:00Z", "created_at": "2026-10-07 10:00:00+00", "updated_at": "2026-10-07 10:00:00+00", "sinks": [{"type": "table", "id": "sink-1", "settings": {"database": "analytics", "table": "out"}}]}) } @@ -210,12 +211,203 @@ fn workflow_create_uses_saved_database_and_sends_sinks() { assert_eq!(result, workflow_response()); assert_eq!( body, - json!({"name": "alerts", "database": "analytics", "sql": "SELECT 1", "enabled": true, + json!({"name": "alerts", "query": {"database": "analytics", "sql": "SELECT 1"}, "enabled": true, "interval_seconds": 60, "sinks": [{"type": "table", "settings": {"database": "analytics", "table": "out"}}]}) ); } +fn run_workflow_human(args: &[&str], method: &str, path: &str, response: Value) -> String { + let mut command = vec!["--org", "team alpha", "--cluster", "prod/eu"]; + command.extend_from_slice(args); + let (output, _, _) = common::run_cli( + &[(format!("{method} {path}"), "200 OK", response)], + &command, + &json!({}), + ); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + String::from_utf8(output.stdout).unwrap() +} + +#[test] +fn workflow_list_renders_empty_and_populated_responses() { + let path = format!("/v1/workflows?{WORKFLOW_SCOPE}"); + let stdout = run_workflow_human( + &["workflow", "list"], + "GET", + &path, + json!({"workflows": []}), + ); + assert!(stdout.contains("No workflows yet"), "{stdout}"); + + let mut manual = workflow_response(); + manual["id"] = json!("wf-2"); + manual["name"] = json!("backfill"); + manual["interval_seconds"] = Value::Null; + manual["next_run_at"] = Value::Null; + let stdout = run_workflow_human( + &["workflow", "list"], + "GET", + &path, + json!({"workflows": [workflow_response(), manual]}), + ); + for expected in [ + "alerts", + "analytics", + "1m", + "wf-1", + "backfill", + "active", + "manual", + "wf-2", + "2026-10-07T10:01:00Z", + "—", + ] { + assert!(stdout.contains(expected), "missing {expected}: {stdout}"); + } +} + +#[test] +fn workflow_human_output_reads_query_and_optional_metadata() { + for (interval, enabled, status, next_run) in [ + (Value::Null, true, "manual", Value::Null), + (Value::Null, false, "manual", Value::Null), + (json!(60), false, "paused", Value::Null), + (json!(60), true, "active", json!("2026-10-07T10:01:00Z")), + ] { + let mut response = workflow_response(); + response["interval_seconds"] = interval.clone(); + response["enabled"] = json!(enabled); + response["next_run_at"] = next_run.clone(); + response["sinks"] = json!([{"id": "sink-http", "type": "http", "settings": { + "url": "https://example.com/hook", "url_configured": true, "header_names": ["Authorization"] + }}]); + let list = run_workflow_human( + &["workflow", "list"], + "GET", + &format!("/v1/workflows?{WORKFLOW_SCOPE}"), + json!({"workflows": [response.clone()]}), + ); + let row = list.lines().find(|line| line.contains("alerts")).unwrap(); + let cells: Vec<_> = row.split(['│', '┆']).map(str::trim).collect(); + assert_eq!(cells[3], status, "{list}"); + let expected_interval = if interval.is_null() { "—" } else { "1m" }; + assert_eq!(cells[4], expected_interval, "{list}"); + for (args, method, suffix) in [ + (vec!["workflow", "get", "wf-1"], "GET", "/wf-1"), + ( + vec![ + "workflow", + "create", + "--name", + "alerts", + "--database", + "analytics", + "--sql", + "SELECT 1", + ], + "POST", + "", + ), + ( + vec!["workflow", "update", "wf-1", "--name", "alerts"], + "PATCH", + "/wf-1", + ), + ] { + let text = run_workflow_human( + &args, + method, + &format!("/v1/workflows{suffix}?{WORKFLOW_SCOPE}"), + response.clone(), + ); + assert!(text.contains("analytics"), "{text}"); + assert!(text.contains(&format!(" status: {status}\n")), "{text}"); + assert!( + text.contains(&format!(" interval: {expected_interval}\n")), + "{text}" + ); + assert!(text.contains(next_run.as_str().unwrap_or("—")), "{text}"); + assert!(text.contains("SELECT 1"), "{text}"); + assert!(text.contains("https://example.com/hook"), "{text}"); + assert!(text.contains("headers=Authorization"), "{text}"); + } + } +} + +#[test] +fn workflow_query_updates_preserve_unspecified_fields() { + for (flags, expected) in [ + ( + vec!["--sql", "SELECT 2"], + json!({"query": {"sql": "SELECT 2"}}), + ), + ( + vec!["--database", "other"], + json!({"query": {"database": "other"}}), + ), + ( + vec!["--database", "other", "--sql", "SELECT 2"], + json!({"query": {"database": "other", "sql": "SELECT 2"}}), + ), + (vec!["--name", "renamed"], json!({"name": "renamed"})), + ] { + let mut args = vec!["workflow", "update", "wf-1"]; + args.extend(flags); + let (_, _, body) = run( + &args, + "PATCH", + &format!("/v1/workflows/wf-1?{WORKFLOW_SCOPE}"), + workflow_response(), + json!({"database": "saved-db"}), + ); + assert_eq!(body, expected); + } +} + +#[test] +fn workflow_scheduling_distinguishes_omitted_null_and_numeric_intervals() { + for (flags, interval) in [ + (vec![], None), + (vec!["--interval-seconds", "null"], Some(Value::Null)), + (vec!["--interval-seconds", "300"], Some(json!(300))), + ] { + let mut create = vec![ + "workflow", "create", "--name", "alerts", "--sql", "SELECT 1", + ]; + create.extend_from_slice(&flags); + let (_, _, body) = run( + &create, + "POST", + &format!("/v1/workflows?{WORKFLOW_SCOPE}"), + workflow_response(), + json!({"database": "analytics"}), + ); + assert_eq!( + body.get("interval_seconds"), + Some(interval.as_ref().unwrap_or(&Value::Null)) + ); + assert_eq!(body["enabled"], true); + + let mut update = vec!["workflow", "update", "wf-1", "--name", "renamed"]; + update.extend_from_slice(&flags); + let (_, _, body) = run( + &update, + "PATCH", + &format!("/v1/workflows/wf-1?{WORKFLOW_SCOPE}"), + workflow_response(), + json!({}), + ); + assert_eq!(body.get("interval_seconds"), interval.as_ref()); + assert!(body.get("enabled").is_none()); + assert!(body.get("query").is_none()); + } +} + #[test] fn workflow_update_sends_only_changed_fields() { let (_, _, body) = run( @@ -297,3 +489,36 @@ fn workflow_commands_require_a_cluster() { assert!(!output.status.success()); assert!(String::from_utf8_lossy(&output.stderr).contains("No cluster specified")); } + +#[test] +fn workflow_invalid_intervals_are_rejected_before_sending_requests() { + for args in [ + vec![ + "workflow", "create", "--name", "alerts", "--sql", "SELECT 1", + ], + vec!["workflow", "update", "wf-1"], + ] { + for value in ["0", "86401", "1.5", "manual", ""] { + let mut command = args.clone(); + command.extend(["--interval-seconds", value]); + let (output, _, _) = common::run_cli(&[], &command, &json!({})); + assert_eq!(output.status.code(), Some(2)); + assert!(String::from_utf8_lossy(&output.stderr).contains("expected a whole number")); + } + } +} + +#[test] +fn workflow_manual_flag_is_rejected() { + for mut args in [ + vec![ + "workflow", "create", "--name", "alerts", "--sql", "SELECT 1", + ], + vec!["workflow", "update", "wf-1"], + ] { + args.push("--manual"); + let (output, _, _) = common::run_cli(&[], &args, &json!({})); + assert_eq!(output.status.code(), Some(2)); + assert!(String::from_utf8_lossy(&output.stderr).contains("unexpected argument '--manual'")); + } +} diff --git a/tests/live_platform.rs b/tests/live_platform.rs index 72f72e3..763bd5b 100644 --- a/tests/live_platform.rs +++ b/tests/live_platform.rs @@ -1,3 +1,6 @@ +#[path = "live_platform/workflows.rs"] +mod workflows; + use std::env; use std::process::{Command, Output}; use std::time::{SystemTime, UNIX_EPOCH}; diff --git a/tests/live_platform/workflows.rs b/tests/live_platform/workflows.rs new file mode 100644 index 0000000..7d95b68 --- /dev/null +++ b/tests/live_platform/workflows.rs @@ -0,0 +1,270 @@ +use super::LiveCli; +use serde_json::{json, Value}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +fn wait_for( + description: &str, + mut read: impl FnMut() -> Value, + ready: impl Fn(&Value) -> bool, +) -> Value { + let deadline = Instant::now() + Duration::from_secs(60); + loop { + let value = read(); + if ready(&value) { + return value; + } + assert!( + Instant::now() < deadline, + "Timed out waiting for {description}: {value}" + ); + std::thread::sleep(Duration::from_millis(250)); + } +} + +fn with_workflow(sql: &str, test: impl FnOnce(&LiveCli, &str, &str)) { + let cli = LiveCli::from_environment(); + let suffix = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let database = format!("cli_workflow_{suffix}"); + cli.json(&["database", "create", &database]); + let mut workflow_id = None; + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + let created = cli.json(&[ + "workflow", + "create", + "--name", + &database, + "--database", + &database, + "--sql", + sql, + ]); + let id = created["id"].as_str().expect("workflow ID").to_string(); + workflow_id = Some(id.clone()); + assert_eq!(created["interval_seconds"], Value::Null); + test(&cli, &database, &id); + })); + if let Some(id) = workflow_id { + assert_eq!(cli.json(&["workflow", "delete", &id])["deleted"], true); + } + assert_eq!( + cli.json(&["database", "delete", &database])["deleted"], + true + ); + if let Err(error) = result { + std::panic::resume_unwind(error); + } +} + +fn wait_for_run(cli: &LiveCli, id: &str, run_id: &str, status: &str) -> Value { + wait_for( + status, + || cli.json(&["workflow", "runs", id]), + |page| { + page["runs"] + .as_array() + .unwrap() + .iter() + .any(|run| run["id"] == run_id && run["status"] == status) + }, + ) +} + +#[test] +#[ignore = "requires a bootstrapped Platform Docker Compose stack"] +fn workflow_manual_runs_deliver_rows_and_expose_logs_and_metrics() { + with_workflow("SELECT 42 AS value", |cli, database, id| { + cli.json(&["table", "create", "results", "--database", database]); + let sink = json!({"type": "table", "settings": {"database": database, "table": "results"}}) + .to_string(); + let definition = cli.json(&["workflow", "update", id, "--disable", "--sink", &sink]); + assert_eq!(definition["sinks"].as_array().unwrap().len(), 1); + let run = cli.json(&["workflow", "run", id, "--idempotency-key", "cli-e2e-run"]); + assert_eq!(run["workflow_id"], id); + assert_eq!(run["source"], "manual"); + assert_eq!( + cli.json(&["workflow", "run", id, "--idempotency-key", "cli-e2e-run"]), + run + ); + let run_id = run["id"].as_str().unwrap(); + wait_for_run(cli, id, run_id, "completed"); + assert_eq!( + cli.json(&["workflow", "run", id, "--idempotency-key", "cli-e2e-run"]), + run + ); + wait_for( + "sink delivery", + || { + cli.json(&[ + "query", + "--database", + database, + "SELECT 1 AS delivered FROM results", + ]) + }, + |result| result["data"] == json!([{"delivered": 1}]), + ); + let rows = cli.json(&["query", "--database", database, "SELECT value FROM results"]); + assert_eq!(rows["data"], json!([{"value": 42}])); + let logs = wait_for( + "workflow logs", + || cli.json(&["workflow", "logs", id, "--since", "1h"]), + |result| { + result["logs"].as_array().unwrap().iter().any(|log| { + log["event_id"] == run_id + && log["event_type"] == "workflow.evaluated" + && log["outcome"] == "success" + }) + }, + ); + assert!(logs["from"].as_i64().unwrap() < logs["to"].as_i64().unwrap()); + wait_for( + "workflow metrics", + || cli.json(&["workflow", "metrics", id, "--since", "1h"]), + |result| { + result["totals"]["executions"] == 1 + && result["totals"]["matches"] == 1 + && result["totals"]["sink_events"] == 1 + }, + ); + assert_eq!(cli.json(&["workflow", "get", id]), definition); + let cleared = cli.json(&["workflow", "update", id, "--clear-sinks"]); + assert_eq!(cleared["sinks"], json!([])); + }); +} + +#[test] +#[ignore = "requires a bootstrapped Platform Docker Compose stack"] +fn workflow_scheduled_runs_can_be_paused_and_switched_to_manual() { + with_workflow("SELECT 1", |cli, _, id| { + let scheduled = cli.json(&["workflow", "update", id, "--interval-seconds", "1"]); + assert_eq!(scheduled["interval_seconds"], 1); + assert_eq!(scheduled["enabled"], true); + wait_for( + "scheduled execution", + || cli.json(&["workflow", "runs", id]), + |page| { + page["runs"] + .as_array() + .unwrap() + .iter() + .any(|run| run["source"] == "scheduled" && run["status"] == "completed") + }, + ); + let paused = cli.json(&["workflow", "update", id, "--disable"]); + assert_eq!(paused["enabled"], false); + assert_eq!(paused["interval_seconds"], 1); + assert_eq!(paused["next_run_at"], Value::Null); + let manual = cli.json(&["workflow", "update", id, "--interval-seconds", "null"]); + assert_eq!(manual["interval_seconds"], Value::Null); + assert_eq!(manual["next_run_at"], Value::Null); + let run = cli.json(&["workflow", "run", id]); + assert_eq!(run["source"], "manual"); + wait_for_run(cli, id, run["id"].as_str().unwrap(), "completed"); + let resumed = cli.json(&[ + "workflow", + "update", + id, + "--interval-seconds", + "3600", + "--enable", + ]); + assert_eq!(resumed["enabled"], true); + assert_eq!(resumed["interval_seconds"], 3600); + assert!(resumed["next_run_at"].is_string()); + }); +} + +#[test] +#[ignore = "requires a bootstrapped Platform Docker Compose stack"] +fn workflow_cancel_stops_a_run_and_preserves_the_definition() { + with_workflow("SELECT sleepEachRow(0.1) AS value FROM numbers(1000) SETTINGS max_execution_time = 0, max_block_size = 1, max_threads = 1", |cli, _, id| { + let definition = cli.json(&["workflow", "get", id]); + let run = cli.json(&["workflow", "run", id]); + let run_id = run["id"].as_str().unwrap(); + wait_for_run(cli, id, run_id, "running"); + assert_eq!(cli.json(&["workflow", "cancel", id, run_id]), json!({"workflow_id": id, "run_id": run_id, "cancel_requested": true})); + wait_for_run(cli, id, run_id, "canceled"); + assert_eq!(cli.json(&["workflow", "cancel", id, run_id])["cancel_requested"], true); + assert_eq!(cli.json(&["workflow", "get", id]), definition); + cli.json(&["workflow", "update", id, "--sql", "SELECT 1"]); + let next = cli.json(&["workflow", "run", id]); + wait_for_run(cli, id, next["id"].as_str().unwrap(), "completed"); + }); +} + +#[test] +#[ignore = "requires a bootstrapped Platform Docker Compose stack"] +fn workflow_query_and_schedule_round_trip_through_real_platform() { + let cli = LiveCli::from_environment(); + let suffix = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("clock after epoch") + .as_nanos(); + let name = format!("cli-workflow-{suffix}"); + let created = cli.json(&[ + "workflow", + "create", + "--name", + &name, + "--database", + "default", + "--sql", + "SELECT 1", + "--disabled", + ]); + let id = created["id"].as_str().expect("workflow ID"); + let result = std::panic::catch_unwind(|| { + assert_eq!( + created["query"], + json!({"database": "default", "sql": "SELECT 1"}) + ); + assert_eq!(created["interval_seconds"], Value::Null); + assert_eq!(created["next_run_at"], Value::Null); + assert_eq!(created["enabled"], false); + let listed = cli.json(&["workflow", "list"]); + assert!(listed["workflows"] + .as_array() + .unwrap() + .iter() + .any(|workflow| workflow == &created)); + assert_eq!(cli.json(&["workflow", "get", id]), created); + + let updated = cli.json(&["workflow", "update", id, "--sql", "SELECT 2"]); + assert_eq!( + updated["query"], + json!({"database": "default", "sql": "SELECT 2"}) + ); + assert_eq!(updated["interval_seconds"], Value::Null); + let updated = cli.json(&["workflow", "update", id, "--database", "default"]); + assert_eq!( + updated["query"], + json!({"database": "default", "sql": "SELECT 2"}) + ); + + let scheduled = cli.json(&["workflow", "update", id, "--interval-seconds", "300"]); + assert_eq!(scheduled["interval_seconds"], 300); + assert_eq!(scheduled["enabled"], false); + let renamed = cli.json(&[ + "workflow", + "update", + id, + "--name", + &format!("{name}-renamed"), + ]); + assert_eq!(renamed["interval_seconds"], 300); + assert_eq!(renamed["query"], updated["query"]); + let manual = cli.json(&["workflow", "update", id, "--interval-seconds", "null"]); + assert_eq!(manual["interval_seconds"], Value::Null); + assert_eq!(manual["next_run_at"], Value::Null); + assert_eq!(manual["query"], updated["query"]); + assert_eq!(cli.json(&["workflow", "get", id]), manual); + }); + let deleted = cli.json(&["workflow", "delete", id]); + assert_eq!(deleted["deleted"], true); + if let Err(error) = result { + std::panic::resume_unwind(error); + } +}