From 62890704e5e66df29494cef1f146f873b6a9e887 Mon Sep 17 00:00:00 2001 From: jordiori Date: Thu, 8 Oct 2026 13:09:47 +0200 Subject: [PATCH 1/6] Read and send workflow SQL and database under query The platform API now nests database and sql in a query object (rawtreedb/rawtree-platform#1368), which broke workflow list, get, create and update. Also accept a null interval for manual-only workflows. --- src/commands/workflow.rs | 60 ++++++++++++++++++++++--------- tests/api_contracts.rs | 78 ++++++++++++++++++++++++++++++++++++++-- 2 files changed, 120 insertions(+), 18 deletions(-) diff --git a/src/commands/workflow.rs b/src/commands/workflow.rs index 091ff58..2a301dc 100644 --- a/src/commands/workflow.rs +++ b/src/commands/workflow.rs @@ -33,15 +33,21 @@ 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, created_at: String, updated_at: String, #[serde(default)] @@ -168,8 +174,7 @@ fn parse_sinks(sinks: &[String]) -> Result> { fn create_body(request: CreateWorkflow) -> Result { let mut body = json!({ "name": request.name, - "database": request.database, - "sql": request.sql, + "query": {"database": request.database, "sql": request.sql}, "enabled": !request.disabled, "sinks": parse_sinks(&request.sinks)?, }); @@ -184,11 +189,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 +290,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 => "manual".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"), } } @@ -322,7 +332,7 @@ 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: {}", if workflow.enabled { @@ -344,7 +354,7 @@ fn print_workflow(workflow: &Workflow) { } } println!(" sql:"); - for line in workflow.sql.lines() { + for line in workflow.query.sql.lines() { println!(" {line}"); } } @@ -363,7 +373,7 @@ pub fn list(client: &ApiClient, scope: &WorkflowScope, json_mode: bool) -> Resul for workflow in &resp.workflows { table.add_row(vec![ Cell::new(&workflow.name), - Cell::new(&workflow.database), + Cell::new(&workflow.query.database), Cell::new(if workflow.enabled { "enabled" } else { @@ -652,8 +662,7 @@ mod tests { body, json!({ "name": "alerts", - "database": "analytics", - "sql": "SELECT 1", + "query": {"database": "analytics", "sql": "SELECT 1"}, "enabled": false, "sinks": [{"type": "table", "settings": {"database": "a", "table": "b"}}], }) @@ -696,6 +705,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_marks_manual_workflows() { + assert_eq!(format_interval(None), "manual"); + assert_eq!(format_interval(Some(7200)), "2h"); + assert_eq!(format_interval(Some(120)), "2m"); + assert_eq!(format_interval(Some(45)), "45s"); } #[test] diff --git a/tests/api_contracts.rs b/tests/api_contracts.rs index 72f721f..de4ebb3 100644 --- a/tests/api_contracts.rs +++ b/tests/api_contracts.rs @@ -181,7 +181,7 @@ 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, "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 +210,86 @@ 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; + let stdout = run_workflow_human( + &["workflow", "list"], + "GET", + &path, + json!({"workflows": [workflow_response(), manual]}), + ); + for expected in [ + "alerts", + "analytics", + "1m", + "wf-1", + "backfill", + "manual", + "wf-2", + ] { + assert!(stdout.contains(expected), "missing {expected}: {stdout}"); + } +} + +#[test] +fn workflow_get_renders_query() { + let stdout = run_workflow_human( + &["workflow", "get", "wf-1"], + "GET", + &format!("/v1/workflows/wf-1?{WORKFLOW_SCOPE}"), + workflow_response(), + ); + assert!(stdout.contains("database: analytics"), "{stdout}"); + assert!(stdout.contains("SELECT 1"), "{stdout}"); +} + +#[test] +fn workflow_update_nests_query_fields() { + let (_, _, body) = run( + &["workflow", "update", "wf-1", "--sql", "SELECT 2"], + "PATCH", + &format!("/v1/workflows/wf-1?{WORKFLOW_SCOPE}"), + workflow_response(), + json!({}), + ); + assert_eq!(body, json!({"query": {"sql": "SELECT 2"}})); +} + #[test] fn workflow_update_sends_only_changed_fields() { let (_, _, body) = run( From 420d576222a485291fe86d59a5dddd92230a0108 Mon Sep 17 00:00:00 2001 From: jordiori Date: Thu, 8 Oct 2026 14:02:07 +0200 Subject: [PATCH 2/6] Add --manual to workflow create and update Sends a null interval so the workflow has no schedule and only runs via rtree workflow run. --- src/cli.rs | 22 +++++++++++++++++++++- src/commands/workflow.rs | 17 ++++++++++++++--- src/main.rs | 4 ++++ tests/api_contracts.rs | 24 ++++++++++++++++++++++++ 4 files changed, 63 insertions(+), 4 deletions(-) diff --git a/src/cli.rs b/src/cli.rs index e6d23f9..f57ca8f 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -473,6 +473,9 @@ pub enum WorkflowCommand { /// Seconds between runs (server default: 1) #[arg(long, value_parser = clap::value_parser!(u32).range(1..=86400))] interval_seconds: Option, + /// Run only on demand with `rtree workflow run`, with no schedule + #[arg(long, conflicts_with = "interval_seconds")] + manual: bool, /// Create the workflow paused #[arg(long)] disabled: bool, @@ -494,9 +497,12 @@ pub enum WorkflowCommand { /// New SQL. Use "-" to read from stdin. #[arg(long)] sql: Option, - /// Seconds between runs + /// Seconds between runs; also restores the schedule of a manual workflow #[arg(long, value_parser = clap::value_parser!(u32).range(1..=86400))] interval_seconds: Option, + /// Remove the schedule so the workflow only runs with `rtree workflow run` + #[arg(long, conflicts_with = "interval_seconds")] + manual: bool, /// Resume scheduled runs #[arg(long, conflicts_with = "disable")] enable: bool, @@ -1190,6 +1196,7 @@ mod tests { for args in [ vec!["--enable", "--disable"], vec!["--clear-sinks", "--sink", "{}"], + vec!["--manual", "--interval-seconds", "60"], ] { let mut command = vec!["rtree", "workflow", "update", "wf-1"]; command.extend(args); @@ -1211,6 +1218,19 @@ mod tests { "86401", ]) .is_err()); + assert!(Cli::try_parse_from([ + "rtree", + "workflow", + "create", + "--name", + "a", + "--sql", + "SELECT 1", + "--manual", + "--interval-seconds", + "60", + ]) + .is_err()); assert!(Cli::try_parse_from([ "rtree", "workflow", diff --git a/src/commands/workflow.rs b/src/commands/workflow.rs index 2a301dc..995df67 100644 --- a/src/commands/workflow.rs +++ b/src/commands/workflow.rs @@ -131,6 +131,7 @@ pub struct CreateWorkflow { pub database: String, pub sql: String, pub interval_seconds: Option, + pub manual: bool, pub disabled: bool, pub sinks: Vec, } @@ -140,6 +141,7 @@ pub struct UpdateWorkflow { pub database: Option, pub sql: Option, pub interval_seconds: Option, + pub manual: bool, pub enabled: Option, pub sinks: Option>, } @@ -178,7 +180,9 @@ fn create_body(request: CreateWorkflow) -> Result { "enabled": !request.disabled, "sinks": parse_sinks(&request.sinks)?, }); - if let Some(interval) = request.interval_seconds { + if request.manual { + body["interval_seconds"] = Value::Null; + } else if let Some(interval) = request.interval_seconds { body["interval_seconds"] = json!(interval); } Ok(body) @@ -202,14 +206,16 @@ fn update_body(request: UpdateWorkflow) -> Result { if let Some(enabled) = request.enabled { body.insert("enabled".into(), json!(enabled)); } - if let Some(interval) = request.interval_seconds { + if request.manual { + body.insert("interval_seconds".into(), Value::Null); + } else if let Some(interval) = request.interval_seconds { body.insert("interval_seconds".into(), json!(interval)); } if let Some(sinks) = request.sinks { body.insert("sinks".into(), Value::Array(parse_sinks(&sinks)?)); } if body.is_empty() { - bail!("nothing to update: pass at least one of --name, --database, --sql, --interval-seconds, --enable, --disable, --sink, or --clear-sinks"); + bail!("nothing to update: pass at least one of --name, --database, --sql, --interval-seconds, --manual, --enable, --disable, --sink, or --clear-sinks"); } Ok(Value::Object(body)) } @@ -654,6 +660,7 @@ mod tests { database: "analytics".into(), sql: "SELECT 1".into(), interval_seconds: None, + manual: false, disabled: true, sinks: vec![r#"{"type":"table","settings":{"database":"a","table":"b"}}"#.into()], }) @@ -673,6 +680,7 @@ mod tests { database: "analytics".into(), sql: "SELECT 1".into(), interval_seconds: Some(60), + manual: false, disabled: false, sinks: vec!["[]".into()], }) @@ -687,6 +695,7 @@ mod tests { database: None, sql: None, interval_seconds: None, + manual: false, enabled: None, sinks: None, }; @@ -697,6 +706,7 @@ mod tests { database: None, sql: None, interval_seconds: Some(30), + manual: false, enabled: Some(false), sinks: Some(vec![]), }) @@ -711,6 +721,7 @@ mod tests { database: None, sql: Some("SELECT 2".into()), interval_seconds: None, + manual: false, enabled: None, sinks: None, }) diff --git a/src/main.rs b/src/main.rs index 7794624..7501fa5 100644 --- a/src/main.rs +++ b/src/main.rs @@ -522,6 +522,7 @@ fn run(cli: Cli) -> Result { database, sql, interval_seconds, + manual, disabled, sinks, } => commands::workflow::create( @@ -532,6 +533,7 @@ fn run(cli: Cli) -> Result { database: resolve_database(database)?, sql: read_sql_arg(sql)?, interval_seconds, + manual, disabled, sinks, }, @@ -543,6 +545,7 @@ fn run(cli: Cli) -> Result { database, sql, interval_seconds, + manual, enable, disable, sinks, @@ -556,6 +559,7 @@ fn run(cli: Cli) -> Result { database, sql: sql.map(read_sql_arg).transpose()?, interval_seconds, + manual, 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 de4ebb3..f4740d3 100644 --- a/tests/api_contracts.rs +++ b/tests/api_contracts.rs @@ -290,6 +290,30 @@ fn workflow_update_nests_query_fields() { assert_eq!(body, json!({"query": {"sql": "SELECT 2"}})); } +#[test] +fn workflow_manual_sends_null_interval() { + let (_, _, body) = run( + &[ + "workflow", "create", "--name", "alerts", "--sql", "SELECT 1", "--manual", + ], + "POST", + &format!("/v1/workflows?{WORKFLOW_SCOPE}"), + workflow_response(), + json!({"database": "analytics"}), + ); + assert_eq!(body["interval_seconds"], Value::Null); + assert!(body.as_object().unwrap().contains_key("interval_seconds")); + + let (_, _, body) = run( + &["workflow", "update", "wf-1", "--manual"], + "PATCH", + &format!("/v1/workflows/wf-1?{WORKFLOW_SCOPE}"), + workflow_response(), + json!({}), + ); + assert_eq!(body, json!({"interval_seconds": null})); +} + #[test] fn workflow_update_sends_only_changed_fields() { let (_, _, body) = run( From 5acbd5a77c32a074f0963b9d38d9e65a54edd47f Mon Sep 17 00:00:00 2001 From: jordiori Date: Thu, 8 Oct 2026 14:02:49 +0200 Subject: [PATCH 3/6] Bump version to 0.6.14 --- Cargo.lock | 2 +- Cargo.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) 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" From 700a85182c50227c97a2707a0d819f973b661908 Mon Sep 17 00:00:00 2001 From: Rafa Moreno Date: Thu, 8 Oct 2026 15:03:07 +0200 Subject: [PATCH 4/6] Complete workflow display metadata and Compose CLI coverage --- README.md | 18 +- src/commands/workflow.rs | 15 +- tests/api_contracts.rs | 151 ++++++++++++----- tests/live_platform.rs | 3 + tests/live_platform/workflows.rs | 271 +++++++++++++++++++++++++++++++ 5 files changed, 415 insertions(+), 43 deletions(-) create mode 100644 tests/live_platform/workflows.rs diff --git a/README.md b/README.md index 9189b0f..2575328 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 --manual --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 --manual rtree workflow update --disable rtree workflow update --clear-sinks rtree workflow delete @@ -274,6 +276,17 @@ 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. + +Use `--manual` on `create` or `update` for manual-only execution +(`interval_seconds: null`). It cannot be combined with `--interval-seconds`. +An interval restores recurring execution, subject to the enabled state; use +`--enable` as well if the workflow is paused. Omitting both flags on `create` +uses the server default of one second; on `update` it 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 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 +380,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/commands/workflow.rs b/src/commands/workflow.rs index 995df67..fe42abf 100644 --- a/src/commands/workflow.rs +++ b/src/commands/workflow.rs @@ -48,6 +48,7 @@ struct Workflow { revision: 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)] @@ -315,6 +316,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| { @@ -326,9 +328,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(), @@ -348,6 +350,10 @@ fn print_workflow(workflow: &Workflow) { } ); println!(" interval: {}", format_interval(workflow.interval_seconds)); + println!( + " next run: {}", + workflow.next_run_at.as_deref().unwrap_or("—") + ); println!(" revision: {}", workflow.revision); println!(" created: {}", workflow.created_at); println!(" updated: {}", workflow.updated_at); @@ -374,7 +380,7 @@ 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![ @@ -387,6 +393,7 @@ pub fn list(client: &ApiClient, scope: &WorkflowScope, json_mode: bool) -> Resul }), 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), @@ -804,7 +811,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/tests/api_contracts.rs b/tests/api_contracts.rs index f4740d3..37b376f 100644 --- a/tests/api_contracts.rs +++ b/tests/api_contracts.rs @@ -183,6 +183,7 @@ const WORKFLOW_SCOPE: &str = "organization=team%20alpha&cluster=prod%2Feu"; fn workflow_response() -> Value { 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"}}]}) } @@ -247,6 +248,7 @@ fn workflow_list_renders_empty_and_populated_responses() { 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", @@ -261,57 +263,130 @@ fn workflow_list_renders_empty_and_populated_responses() { "backfill", "manual", "wf-2", + "2026-10-07T10:01:00Z", + "—", ] { assert!(stdout.contains(expected), "missing {expected}: {stdout}"); } } #[test] -fn workflow_get_renders_query() { - let stdout = run_workflow_human( - &["workflow", "get", "wf-1"], - "GET", - &format!("/v1/workflows/wf-1?{WORKFLOW_SCOPE}"), - workflow_response(), - ); - assert!(stdout.contains("database: analytics"), "{stdout}"); - assert!(stdout.contains("SELECT 1"), "{stdout}"); +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, "enabled", json!("2026-10-07T10:01:00Z")), + ] { + let mut response = workflow_response(); + response["interval_seconds"] = interval; + 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"] + }}]); + 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(status), "{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_update_nests_query_fields() { - let (_, _, body) = run( - &["workflow", "update", "wf-1", "--sql", "SELECT 2"], - "PATCH", - &format!("/v1/workflows/wf-1?{WORKFLOW_SCOPE}"), - workflow_response(), - json!({}), - ); - assert_eq!(body, json!({"query": {"sql": "SELECT 2"}})); +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_manual_sends_null_interval() { - let (_, _, body) = run( - &[ - "workflow", "create", "--name", "alerts", "--sql", "SELECT 1", "--manual", - ], - "POST", - &format!("/v1/workflows?{WORKFLOW_SCOPE}"), - workflow_response(), - json!({"database": "analytics"}), - ); - assert_eq!(body["interval_seconds"], Value::Null); - assert!(body.as_object().unwrap().contains_key("interval_seconds")); +fn workflow_scheduling_distinguishes_omitted_null_and_numeric_intervals() { + for (flags, interval) in [ + (vec![], None), + (vec!["--manual"], 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"), interval.as_ref()); + assert_eq!(body["enabled"], true); - let (_, _, body) = run( - &["workflow", "update", "wf-1", "--manual"], - "PATCH", - &format!("/v1/workflows/wf-1?{WORKFLOW_SCOPE}"), - workflow_response(), - json!({}), - ); - assert_eq!(body, json!({"interval_seconds": null})); + 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] 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..a4c8240 --- /dev/null +++ b/tests/live_platform/workflows.rs @@ -0,0 +1,271 @@ +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, + "--manual", + ]); + 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, "--manual"]); + 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"], 1); + 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"], 1); + 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, "--manual"]); + 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); + } +} From dbceb3841ffb6440174e18cb776f3c285757752c Mon Sep 17 00:00:00 2001 From: Rafa Moreno Date: Thu, 8 Oct 2026 15:11:28 +0200 Subject: [PATCH 5/6] Use nullable workflow intervals instead of a manual flag --- README.md | 14 +++++---- src/cli.rs | 51 +++++++++++++++----------------- src/commands/workflow.rs | 29 +++++------------- src/main.rs | 8 ++--- tests/api_contracts.rs | 40 +++++++++++++++++++++++-- tests/live_platform/workflows.rs | 9 +++--- 6 files changed, 84 insertions(+), 67 deletions(-) diff --git a/README.md b/README.md index 2575328..9191964 100644 --- a/README.md +++ b/README.md @@ -255,13 +255,13 @@ Workflows run saved SQL on demand or on a schedule and deliver results to sinks. 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 --manual --sql "SELECT 1" +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 --manual +rtree workflow update --interval-seconds null rtree workflow update --disable rtree workflow update --clear-sinks rtree workflow delete @@ -279,11 +279,13 @@ the environment, or the saved defaults). API keys must have admin permission. 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. -Use `--manual` on `create` or `update` for manual-only execution -(`interval_seconds: null`). It cannot be combined with `--interval-seconds`. +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 both flags on `create` -uses the server default of one second; on `update` it preserves the current mode. +`--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 the next scheduled run when available and HTTP sink URLs. diff --git a/src/cli.rs b/src/cli.rs index f57ca8f..a125a4c 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -470,12 +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, - /// Run only on demand with `rtree workflow run`, with no schedule - #[arg(long, conflicts_with = "interval_seconds")] - manual: bool, + /// 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, @@ -497,12 +494,9 @@ pub enum WorkflowCommand { /// New SQL. Use "-" to read from stdin. #[arg(long)] sql: Option, - /// Seconds between runs; also restores the schedule of a manual workflow - #[arg(long, value_parser = clap::value_parser!(u32).range(1..=86400))] - interval_seconds: Option, - /// Remove the schedule so the workflow only runs with `rtree workflow run` - #[arg(long, conflicts_with = "interval_seconds")] - manual: bool, + /// 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, @@ -569,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 { @@ -1182,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()); @@ -1196,7 +1207,6 @@ mod tests { for args in [ vec!["--enable", "--disable"], vec!["--clear-sinks", "--sink", "{}"], - vec!["--manual", "--interval-seconds", "60"], ] { let mut command = vec!["rtree", "workflow", "update", "wf-1"]; command.extend(args); @@ -1218,19 +1228,6 @@ mod tests { "86401", ]) .is_err()); - assert!(Cli::try_parse_from([ - "rtree", - "workflow", - "create", - "--name", - "a", - "--sql", - "SELECT 1", - "--manual", - "--interval-seconds", - "60", - ]) - .is_err()); assert!(Cli::try_parse_from([ "rtree", "workflow", diff --git a/src/commands/workflow.rs b/src/commands/workflow.rs index fe42abf..aa1efc4 100644 --- a/src/commands/workflow.rs +++ b/src/commands/workflow.rs @@ -132,7 +132,6 @@ pub struct CreateWorkflow { pub database: String, pub sql: String, pub interval_seconds: Option, - pub manual: bool, pub disabled: bool, pub sinks: Vec, } @@ -141,8 +140,7 @@ pub struct UpdateWorkflow { pub name: Option, pub database: Option, pub sql: Option, - pub interval_seconds: Option, - pub manual: bool, + pub interval_seconds: Option>, pub enabled: Option, pub sinks: Option>, } @@ -175,18 +173,13 @@ fn parse_sinks(sinks: &[String]) -> Result> { } fn create_body(request: CreateWorkflow) -> Result { - let mut body = json!({ + Ok(json!({ "name": request.name, "query": {"database": request.database, "sql": request.sql}, "enabled": !request.disabled, + "interval_seconds": request.interval_seconds, "sinks": parse_sinks(&request.sinks)?, - }); - if request.manual { - body["interval_seconds"] = Value::Null; - } else if let Some(interval) = request.interval_seconds { - body["interval_seconds"] = json!(interval); - } - Ok(body) + })) } fn update_body(request: UpdateWorkflow) -> Result { @@ -207,16 +200,14 @@ fn update_body(request: UpdateWorkflow) -> Result { if let Some(enabled) = request.enabled { body.insert("enabled".into(), json!(enabled)); } - if request.manual { - body.insert("interval_seconds".into(), Value::Null); - } else if let Some(interval) = request.interval_seconds { + if let Some(interval) = request.interval_seconds { body.insert("interval_seconds".into(), json!(interval)); } if let Some(sinks) = request.sinks { body.insert("sinks".into(), Value::Array(parse_sinks(&sinks)?)); } if body.is_empty() { - bail!("nothing to update: pass at least one of --name, --database, --sql, --interval-seconds, --manual, --enable, --disable, --sink, or --clear-sinks"); + bail!("nothing to update: pass at least one of --name, --database, --sql, --interval-seconds, --enable, --disable, --sink, or --clear-sinks"); } Ok(Value::Object(body)) } @@ -667,7 +658,6 @@ mod tests { database: "analytics".into(), sql: "SELECT 1".into(), interval_seconds: None, - manual: false, disabled: true, sinks: vec![r#"{"type":"table","settings":{"database":"a","table":"b"}}"#.into()], }) @@ -678,6 +668,7 @@ mod tests { "name": "alerts", "query": {"database": "analytics", "sql": "SELECT 1"}, "enabled": false, + "interval_seconds": null, "sinks": [{"type": "table", "settings": {"database": "a", "table": "b"}}], }) ); @@ -687,7 +678,6 @@ mod tests { database: "analytics".into(), sql: "SELECT 1".into(), interval_seconds: Some(60), - manual: false, disabled: false, sinks: vec!["[]".into()], }) @@ -702,7 +692,6 @@ mod tests { database: None, sql: None, interval_seconds: None, - manual: false, enabled: None, sinks: None, }; @@ -712,8 +701,7 @@ mod tests { name: None, database: None, sql: None, - interval_seconds: Some(30), - manual: false, + interval_seconds: Some(Some(30)), enabled: Some(false), sinks: Some(vec![]), }) @@ -728,7 +716,6 @@ mod tests { database: None, sql: Some("SELECT 2".into()), interval_seconds: None, - manual: false, enabled: None, sinks: None, }) diff --git a/src/main.rs b/src/main.rs index 7501fa5..1f8ba85 100644 --- a/src/main.rs +++ b/src/main.rs @@ -522,7 +522,6 @@ fn run(cli: Cli) -> Result { database, sql, interval_seconds, - manual, disabled, sinks, } => commands::workflow::create( @@ -532,8 +531,7 @@ fn run(cli: Cli) -> Result { name, database: resolve_database(database)?, sql: read_sql_arg(sql)?, - interval_seconds, - manual, + interval_seconds: interval_seconds.and_then(|interval| interval.0), disabled, sinks, }, @@ -545,7 +543,6 @@ fn run(cli: Cli) -> Result { database, sql, interval_seconds, - manual, enable, disable, sinks, @@ -558,8 +555,7 @@ fn run(cli: Cli) -> Result { name, database, sql: sql.map(read_sql_arg).transpose()?, - interval_seconds, - manual, + 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 37b376f..78f5c4f 100644 --- a/tests/api_contracts.rs +++ b/tests/api_contracts.rs @@ -357,7 +357,7 @@ fn workflow_query_updates_preserve_unspecified_fields() { fn workflow_scheduling_distinguishes_omitted_null_and_numeric_intervals() { for (flags, interval) in [ (vec![], None), - (vec!["--manual"], Some(Value::Null)), + (vec!["--interval-seconds", "null"], Some(Value::Null)), (vec!["--interval-seconds", "300"], Some(json!(300))), ] { let mut create = vec![ @@ -371,7 +371,10 @@ fn workflow_scheduling_distinguishes_omitted_null_and_numeric_intervals() { workflow_response(), json!({"database": "analytics"}), ); - assert_eq!(body.get("interval_seconds"), interval.as_ref()); + 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"]; @@ -470,3 +473,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/workflows.rs b/tests/live_platform/workflows.rs index a4c8240..7d95b68 100644 --- a/tests/live_platform/workflows.rs +++ b/tests/live_platform/workflows.rs @@ -40,7 +40,6 @@ fn with_workflow(sql: &str, test: impl FnOnce(&LiveCli, &str, &str)) { &database, "--sql", sql, - "--manual", ]); let id = created["id"].as_str().expect("workflow ID").to_string(); workflow_id = Some(id.clone()); @@ -158,7 +157,7 @@ fn workflow_scheduled_runs_can_be_paused_and_switched_to_manual() { 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, "--manual"]); + 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]); @@ -222,7 +221,7 @@ fn workflow_query_and_schedule_round_trip_through_real_platform() { created["query"], json!({"database": "default", "sql": "SELECT 1"}) ); - assert_eq!(created["interval_seconds"], 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"]); @@ -238,7 +237,7 @@ fn workflow_query_and_schedule_round_trip_through_real_platform() { updated["query"], json!({"database": "default", "sql": "SELECT 2"}) ); - assert_eq!(updated["interval_seconds"], 1); + assert_eq!(updated["interval_seconds"], Value::Null); let updated = cli.json(&["workflow", "update", id, "--database", "default"]); assert_eq!( updated["query"], @@ -257,7 +256,7 @@ fn workflow_query_and_schedule_round_trip_through_real_platform() { ]); assert_eq!(renamed["interval_seconds"], 300); assert_eq!(renamed["query"], updated["query"]); - let manual = cli.json(&["workflow", "update", id, "--manual"]); + 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"]); From a0c0eb97dd8c222d25802c5c3a7832c9774d3509 Mon Sep 17 00:00:00 2001 From: Rafa Moreno Date: Thu, 8 Oct 2026 15:44:09 +0200 Subject: [PATCH 6/6] Unify workflow status labels and clarify create help --- README.md | 5 +++-- src/cli.rs | 2 +- src/commands/workflow.rs | 31 +++++++++++++++---------------- tests/api_contracts.rs | 22 +++++++++++++++++++--- 4 files changed, 38 insertions(+), 22 deletions(-) diff --git a/README.md b/README.md index 9191964..8840753 100644 --- a/README.md +++ b/README.md @@ -287,8 +287,9 @@ 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 the next -scheduled run when available and HTTP sink URLs. +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 diff --git a/src/cli.rs b/src/cli.rs index a125a4c..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 '_') diff --git a/src/commands/workflow.rs b/src/commands/workflow.rs index aa1efc4..779738e 100644 --- a/src/commands/workflow.rs +++ b/src/commands/workflow.rs @@ -55,6 +55,16 @@ struct Workflow { 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, @@ -290,7 +300,7 @@ fn format_ms(ms: i64) -> String { fn format_interval(seconds: Option) -> String { match seconds { - None => "manual".into(), + 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"), @@ -332,14 +342,7 @@ fn print_workflow(workflow: &Workflow) { println!(" id: {}", workflow.id); println!(" name: {}", workflow.name); println!(" database: {}", workflow.query.database); - println!( - " status: {}", - if workflow.enabled { - "enabled" - } else { - "paused" - } - ); + println!(" status: {}", workflow.status()); println!(" interval: {}", format_interval(workflow.interval_seconds)); println!( " next run: {}", @@ -377,11 +380,7 @@ pub fn list(client: &ApiClient, scope: &WorkflowScope, json_mode: bool) -> Resul table.add_row(vec![ Cell::new(&workflow.name), Cell::new(&workflow.query.database), - Cell::new(if workflow.enabled { - "enabled" - } else { - "paused" - }), + 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("—")), @@ -724,8 +723,8 @@ mod tests { } #[test] - fn format_interval_marks_manual_workflows() { - assert_eq!(format_interval(None), "manual"); + 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"); diff --git a/tests/api_contracts.rs b/tests/api_contracts.rs index 78f5c4f..6d49495 100644 --- a/tests/api_contracts.rs +++ b/tests/api_contracts.rs @@ -261,6 +261,7 @@ fn workflow_list_renders_empty_and_populated_responses() { "1m", "wf-1", "backfill", + "active", "manual", "wf-2", "2026-10-07T10:01:00Z", @@ -276,15 +277,26 @@ fn workflow_human_output_reads_query_and_optional_metadata() { (Value::Null, true, "manual", Value::Null), (Value::Null, false, "manual", Value::Null), (json!(60), false, "paused", Value::Null), - (json!(60), true, "enabled", json!("2026-10-07T10:01:00Z")), + (json!(60), true, "active", json!("2026-10-07T10:01:00Z")), ] { let mut response = workflow_response(); - response["interval_seconds"] = interval; + 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"), ( @@ -314,7 +326,11 @@ fn workflow_human_output_reads_query_and_optional_metadata() { response.clone(), ); assert!(text.contains("analytics"), "{text}"); - assert!(text.contains(status), "{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}");