diff --git a/Cargo.lock b/Cargo.lock index 1b4966d..e6365e1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1502,7 +1502,7 @@ dependencies = [ [[package]] name = "rawtree-cli" -version = "0.7.2" +version = "0.7.3" dependencies = [ "anyhow", "axoupdater", diff --git a/Cargo.toml b/Cargo.toml index 657885b..8600cdd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "rawtree-cli" -version = "0.7.2" +version = "0.7.3" edition = "2021" repository = "https://github.com/rawtreedb/rawtree-cli" diff --git a/README.md b/README.md index d7b94ff..a01fc8b 100644 --- a/README.md +++ b/README.md @@ -307,16 +307,16 @@ 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 \ +rtree workflow create --name errors --database analytics --schedule-cron "* * * * *" \ --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 - \ +rtree workflow create --name alerts --schedule-cron "* * * * *" --schedule-enabled false --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 --schedule-cron "*/5 * * * *" +rtree workflow update --schedule-cron null +rtree workflow update --schedule-enabled false rtree workflow update --clear-sinks rtree workflow delete @@ -333,17 +333,23 @@ 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. -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. +Creating a workflow without `--schedule-cron` makes it manual-only. The CLI explicitly +sends `schedule: null`, preserving its manual-only default. Use `--schedule-cron null` to +remove an existing schedule. The obsolete `--interval-seconds` flag is rejected. + +`--schedule-cron "*/5 * * * *"` runs every five minutes in UTC. Five-field expressions use +minute, hour, day-of-month, month, and day-of-week; six fields include leading +seconds. The API validates the expression. Requests nest cron and pause state in +`schedule`, for example `{"schedule":{"cron":"*/5 * * * *","enabled":false}}`. + +On `update`, omitting `--schedule-cron` preserves the schedule. `--schedule-enabled false` pauses it and +`--schedule-enabled true` resumes it without changing the cron expression. Adding cron to a +manual-only workflow enables scheduling by default; changing an existing cron +preserves its pause state unless you also pass `--schedule-enabled true|false`. +`--schedule-enabled` accepts an explicit boolean and requires a cron expression on `create`. `--schedule-cron null` cannot be +combined with pause/resume flags. Manual runs remain available while paused. +Text output shows `active`, `paused`, or `manual`, the UTC cron expression, and +the next scheduled run when available. JSON output preserves the API response. 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 2c659cc..047cf71 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -490,12 +490,12 @@ pub enum WorkflowCommand { /// SQL to run: a SELECT or INSERT INTO ... SELECT. Use "-" to read from stdin. #[arg(long)] sql: String, - /// 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, + /// UTC cron expression (5 fields, or 6 with seconds); null or omitted means manual-only + #[arg(long, value_name = "EXPRESSION|null")] + schedule_cron: Option, + /// Enable or pause the schedule (true or false); requires a cron expression + #[arg(long, action = clap::ArgAction::Set, value_name = "BOOL")] + schedule_enabled: Option, /// Sink as a JSON object; repeat for up to five sinks #[arg(long = "sink", value_name = "JSON")] sinks: Vec, @@ -514,15 +514,12 @@ pub enum WorkflowCommand { /// New SQL. Use "-" to read from stdin. #[arg(long)] sql: 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, - /// Pause scheduled runs - #[arg(long)] - disable: bool, + /// UTC cron expression (5 fields, or 6 with seconds); null removes the schedule + #[arg(long, value_name = "EXPRESSION|null")] + schedule_cron: Option, + /// Enable or pause scheduled runs (true or false); omit to preserve the state + #[arg(long, action = clap::ArgAction::Set, value_name = "BOOL")] + schedule_enabled: Option, /// Sink as a JSON object; replaces all sinks. Repeat for up to five. #[arg(long = "sink", value_name = "JSON", conflicts_with = "clear_sinks")] sinks: Vec, @@ -583,20 +580,21 @@ pub enum WorkflowCommand { }, } -#[derive(Clone, Copy, Debug)] -pub struct WorkflowIntervalArg(pub Option); +#[derive(Clone, Debug)] +pub struct WorkflowCronArg(pub Option); -impl FromStr for WorkflowIntervalArg { +impl FromStr for WorkflowCronArg { 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()), + let value = value.trim(); + if value.is_empty() { + return Err("expected a UTC cron expression or null".into()); } + Ok(Self(Some(value.to_string()))) } } diff --git a/src/cli/tests.rs b/src/cli/tests.rs index ce99527..ce27112 100644 --- a/src/cli/tests.rs +++ b/src/cli/tests.rs @@ -560,9 +560,10 @@ fn workflow_create_collects_repeated_sinks() { "alerts", "--sql", "SELECT 1", - "--interval-seconds", - "60", - "--disabled", + "--schedule-cron", + "* * * * *", + "--schedule-enabled", + "false", "--sink", r#"{"type":"table","settings":{"database":"a","table":"b"}}"#, "--sink", @@ -573,15 +574,18 @@ fn workflow_create_collects_repeated_sinks() { Command::Workflow { action: WorkflowCommand::Create { - interval_seconds, - disabled, + schedule_cron, + schedule_enabled, sinks, database, .. }, } => { - assert_eq!(interval_seconds.map(|interval| interval.0), Some(Some(60))); - assert!(disabled); + assert_eq!( + schedule_cron.map(|cron| cron.0), + Some(Some("* * * * *".into())) + ); + assert_eq!(schedule_enabled, Some(false)); assert_eq!(sinks.len(), 2); assert!(database.is_none()); } @@ -592,7 +596,7 @@ fn workflow_create_collects_repeated_sinks() { #[test] fn workflow_update_rejects_conflicting_flags() { for args in [ - vec!["--enable", "--disable"], + vec!["--schedule-enabled", "true", "--schedule-enabled", "false"], vec!["--clear-sinks", "--sink", "{}"], ] { let mut command = vec!["rtree", "workflow", "update", "wf-1"]; @@ -602,7 +606,7 @@ fn workflow_update_rejects_conflicting_flags() { } #[test] -fn workflow_interval_and_time_ranges_are_validated() { +fn workflow_cron_and_time_ranges_are_validated() { assert!(Cli::try_parse_from([ "rtree", "workflow", @@ -611,8 +615,8 @@ fn workflow_interval_and_time_ranges_are_validated() { "a", "--sql", "SELECT 1", - "--interval-seconds", - "86401", + "--schedule-cron", + "", ]) .is_err()); assert!(Cli::try_parse_from([ @@ -635,3 +639,52 @@ fn workflow_interval_and_time_ranges_are_validated() { } if range.since.as_deref() == Some("2h") )); } + +#[test] +fn workflow_schedule_enabled_requires_an_explicit_boolean() { + for base in [ + vec![ + "rtree", + "workflow", + "create", + "--name", + "alerts", + "--sql", + "SELECT 1", + "--schedule-cron", + "* * * * *", + ], + vec!["rtree", "workflow", "update", "wf-1"], + ] { + for (text, expected) in [("true", true), ("false", false)] { + let mut args = base.clone(); + args.extend(["--schedule-enabled", text]); + let parsed = Cli::try_parse_from(args).unwrap(); + match parsed.command { + Command::Workflow { + action: + WorkflowCommand::Create { + schedule_enabled, .. + } + | WorkflowCommand::Update { + schedule_enabled, .. + }, + } => assert_eq!(schedule_enabled, Some(expected)), + _ => panic!("expected workflow command"), + } + } + for flags in [ + vec!["--schedule-enabled"], + vec!["--schedule-enabled", "yes"], + vec!["--schedule-enabled", "1"], + vec!["--cron", "* * * * *"], + vec!["--enable"], + vec!["--disable"], + vec!["--disabled"], + ] { + let mut args = base.clone(); + args.extend(flags); + assert!(Cli::try_parse_from(args).is_err()); + } + } +} diff --git a/src/commands/workflow.rs b/src/commands/workflow.rs index 779738e..f3897fa 100644 --- a/src/commands/workflow.rs +++ b/src/commands/workflow.rs @@ -44,23 +44,27 @@ struct Workflow { id: String, name: String, query: WorkflowQuery, - enabled: bool, revision: i64, - /// `None` for manual-only workflows, which have no schedule. - interval_seconds: Option, - next_run_at: Option, + schedule: Option, created_at: String, updated_at: String, #[serde(default)] sinks: Vec, } +#[derive(Deserialize)] +struct WorkflowSchedule { + cron: String, + enabled: bool, + next_run_at: Option, +} + impl Workflow { fn status(&self) -> &'static str { - match (self.interval_seconds, self.enabled) { - (None, _) => "manual", - (Some(_), true) => "active", - (Some(_), false) => "paused", + match &self.schedule { + None => "manual", + Some(schedule) if schedule.enabled => "active", + Some(_) => "paused", } } } @@ -141,8 +145,8 @@ pub struct CreateWorkflow { pub name: String, pub database: String, pub sql: String, - pub interval_seconds: Option, - pub disabled: bool, + pub cron: Option, + pub enabled: Option, pub sinks: Vec, } @@ -150,7 +154,7 @@ pub struct UpdateWorkflow { pub name: Option, pub database: Option, pub sql: Option, - pub interval_seconds: Option>, + pub cron: Option>, pub enabled: Option, pub sinks: Option>, } @@ -183,11 +187,16 @@ fn parse_sinks(sinks: &[String]) -> Result> { } fn create_body(request: CreateWorkflow) -> Result { + if request.enabled.is_some() && request.cron.is_none() { + bail!("--schedule-enabled requires --schedule-cron with a cron expression; manual-only workflows have no schedule to pause"); + } + let schedule = request + .cron + .map(|cron| json!({"cron": cron, "enabled": request.enabled.unwrap_or(true)})); Ok(json!({ "name": request.name, "query": {"database": request.database, "sql": request.sql}, - "enabled": !request.disabled, - "interval_seconds": request.interval_seconds, + "schedule": schedule, "sinks": parse_sinks(&request.sinks)?, })) } @@ -207,17 +216,31 @@ fn update_body(request: UpdateWorkflow) -> Result { if !query.is_empty() { body.insert("query".into(), Value::Object(query)); } - if let Some(enabled) = request.enabled { - body.insert("enabled".into(), json!(enabled)); - } - if let Some(interval) = request.interval_seconds { - body.insert("interval_seconds".into(), json!(interval)); + match request.cron { + Some(None) => { + if request.enabled.is_some() { + bail!("--schedule-cron null cannot be combined with --schedule-enabled"); + } + body.insert("schedule".into(), Value::Null); + } + cron => { + let mut schedule = Map::new(); + if let Some(Some(cron)) = cron { + schedule.insert("cron".into(), json!(cron)); + } + if let Some(enabled) = request.enabled { + schedule.insert("enabled".into(), json!(enabled)); + } + if !schedule.is_empty() { + body.insert("schedule".into(), Value::Object(schedule)); + } + } } 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, --schedule-cron, --schedule-enabled, --sink, or --clear-sinks"); } Ok(Value::Object(body)) } @@ -298,15 +321,6 @@ fn format_ms(ms: i64) -> String { .unwrap_or_else(|| ms.to_string()) } -fn format_interval(seconds: Option) -> String { - match seconds { - 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"), - } -} - fn describe_sink(sink: &Value) -> String { let id = sink["id"].as_str().unwrap_or("—"); let settings = &sink["settings"]; @@ -343,10 +357,21 @@ fn print_workflow(workflow: &Workflow) { println!(" name: {}", workflow.name); println!(" database: {}", workflow.query.database); println!(" status: {}", workflow.status()); - println!(" interval: {}", format_interval(workflow.interval_seconds)); + println!( + " cron: {}", + workflow + .schedule + .as_ref() + .map(|s| s.cron.as_str()) + .unwrap_or("—") + ); println!( " next run: {}", - workflow.next_run_at.as_deref().unwrap_or("—") + workflow + .schedule + .as_ref() + .and_then(|s| s.next_run_at.as_deref()) + .unwrap_or("—") ); println!(" revision: {}", workflow.revision); println!(" created: {}", workflow.created_at); @@ -374,16 +399,34 @@ 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", "next run", "sinks", "updated", "id", + "name", + "database", + "status", + "cron (UTC)", + "next run", + "sinks", + "updated", + "id", ]); for workflow in &resp.workflows { table.add_row(vec![ Cell::new(&workflow.name), 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 + .schedule + .as_ref() + .map(|s| s.cron.as_str()) + .unwrap_or("—"), + ), + Cell::new( + workflow + .schedule + .as_ref() + .and_then(|s| s.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), @@ -651,13 +694,13 @@ mod tests { } #[test] - fn create_body_sends_defaults_explicitly_and_rejects_non_object_sinks() { + fn create_body_sends_manual_schedule_and_rejects_non_object_sinks() { let body = create_body(CreateWorkflow { name: "alerts".into(), database: "analytics".into(), sql: "SELECT 1".into(), - interval_seconds: None, - disabled: true, + cron: None, + enabled: None, sinks: vec![r#"{"type":"table","settings":{"database":"a","table":"b"}}"#.into()], }) .unwrap(); @@ -666,8 +709,7 @@ mod tests { json!({ "name": "alerts", "query": {"database": "analytics", "sql": "SELECT 1"}, - "enabled": false, - "interval_seconds": null, + "schedule": null, "sinks": [{"type": "table", "settings": {"database": "a", "table": "b"}}], }) ); @@ -676,8 +718,8 @@ mod tests { name: "alerts".into(), database: "analytics".into(), sql: "SELECT 1".into(), - interval_seconds: Some(60), - disabled: false, + cron: Some("* * * * *".into()), + enabled: None, sinks: vec!["[]".into()], }) .unwrap_err(); @@ -690,7 +732,7 @@ mod tests { name: None, database: None, sql: None, - interval_seconds: None, + cron: None, enabled: None, sinks: None, }; @@ -700,21 +742,21 @@ mod tests { name: None, database: None, sql: None, - interval_seconds: Some(Some(30)), + cron: Some(Some("*/30 * * * * *".into())), enabled: Some(false), sinks: Some(vec![]), }) .unwrap(); assert_eq!( body, - json!({"interval_seconds": 30, "enabled": false, "sinks": []}) + json!({"schedule": {"cron": "*/30 * * * * *", "enabled": false}, "sinks": []}) ); let body = update_body(UpdateWorkflow { name: None, database: None, sql: Some("SELECT 2".into()), - interval_seconds: None, + cron: None, enabled: None, sinks: None, }) @@ -722,14 +764,6 @@ mod tests { 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] fn window_uses_relative_or_absolute_bounds() { let now = DateTime::parse_from_rfc3339("2026-10-07T12:00:00Z") diff --git a/src/main.rs b/src/main.rs index 254d6c1..55e4c07 100644 --- a/src/main.rs +++ b/src/main.rs @@ -536,8 +536,8 @@ fn run(cli: Cli) -> Result { name, database, sql, - interval_seconds, - disabled, + schedule_cron, + schedule_enabled, sinks, } => commands::workflow::create( &client, @@ -546,8 +546,8 @@ fn run(cli: Cli) -> Result { name, database: resolve_database(database)?, sql: read_sql_arg(sql)?, - interval_seconds: interval_seconds.and_then(|interval| interval.0), - disabled, + cron: schedule_cron.and_then(|cron| cron.0), + enabled: schedule_enabled, sinks, }, json, @@ -557,9 +557,8 @@ fn run(cli: Cli) -> Result { name, database, sql, - interval_seconds, - enable, - disable, + schedule_cron, + schedule_enabled, sinks, clear_sinks, } => commands::workflow::update( @@ -570,8 +569,8 @@ fn run(cli: Cli) -> Result { name, database, sql: sql.map(read_sql_arg).transpose()?, - interval_seconds: interval_seconds.map(|interval| interval.0), - enabled: (enable || disable).then_some(enable), + cron: schedule_cron.map(|cron| cron.0), + enabled: schedule_enabled, sinks: (clear_sinks || !sinks.is_empty()).then_some(sinks), }, json, diff --git a/tests/api_contracts.rs b/tests/api_contracts.rs index 4a132e4..ab26154 100644 --- a/tests/api_contracts.rs +++ b/tests/api_contracts.rs @@ -182,8 +182,8 @@ 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", + "revision": 1, "schedule": {"cron": "* * * * *", "enabled": true, + "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"}}]}) } @@ -198,8 +198,8 @@ fn workflow_create_uses_saved_database_and_sends_sinks() { "alerts", "--sql", "SELECT 1", - "--interval-seconds", - "60", + "--schedule-cron", + "* * * * *", "--sink", r#"{"type":"table","settings":{"database":"analytics","table":"out"}}"#, ], @@ -211,8 +211,8 @@ fn workflow_create_uses_saved_database_and_sends_sinks() { assert_eq!(result, workflow_response()); assert_eq!( body, - json!({"name": "alerts", "query": {"database": "analytics", "sql": "SELECT 1"}, "enabled": true, - "interval_seconds": 60, + json!({"name": "alerts", "query": {"database": "analytics", "sql": "SELECT 1"}, + "schedule": {"cron": "* * * * *", "enabled": true}, "sinks": [{"type": "table", "settings": {"database": "analytics", "table": "out"}}]}) ); } @@ -247,8 +247,7 @@ fn workflow_list_renders_empty_and_populated_responses() { 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; + manual["schedule"] = Value::Null; let stdout = run_workflow_human( &["workflow", "list"], "GET", @@ -258,7 +257,7 @@ fn workflow_list_renders_empty_and_populated_responses() { for expected in [ "alerts", "analytics", - "1m", + "* * * * *", "wf-1", "backfill", "active", @@ -273,16 +272,23 @@ fn workflow_list_renders_empty_and_populated_responses() { #[test] fn workflow_human_output_reads_query_and_optional_metadata() { - for (interval, enabled, status, next_run) in [ + for (cron, 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")), + (json!("* * * * *"), false, "paused", Value::Null), + ( + json!("* * * * *"), + 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["schedule"] = if cron.is_null() { + Value::Null + } else { + json!({"cron": cron, "enabled": enabled, "next_run_at": next_run}) + }; response["sinks"] = json!([{"id": "sink-http", "type": "http", "settings": { "url": "https://example.com/hook", "url_configured": true, "header_names": ["Authorization"] }}]); @@ -295,8 +301,8 @@ fn workflow_human_output_reads_query_and_optional_metadata() { 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}"); + let expected_cron = if cron.is_null() { "—" } else { "* * * * *" }; + assert_eq!(cells[4], expected_cron, "{list}"); for (args, method, suffix) in [ (vec!["workflow", "get", "wf-1"], "GET", "/wf-1"), ( @@ -328,7 +334,7 @@ fn workflow_human_output_reads_query_and_optional_metadata() { assert!(text.contains("analytics"), "{text}"); assert!(text.contains(&format!(" status: {status}\n")), "{text}"); assert!( - text.contains(&format!(" interval: {expected_interval}\n")), + text.contains(&format!(" cron: {expected_cron}\n")), "{text}" ); assert!(text.contains(next_run.as_str().unwrap_or("—")), "{text}"); @@ -370,11 +376,14 @@ fn workflow_query_updates_preserve_unspecified_fields() { } #[test] -fn workflow_scheduling_distinguishes_omitted_null_and_numeric_intervals() { - for (flags, interval) in [ +fn workflow_scheduling_distinguishes_omitted_null_and_cron() { + for (flags, schedule) in [ (vec![], None), - (vec!["--interval-seconds", "null"], Some(Value::Null)), - (vec!["--interval-seconds", "300"], Some(json!(300))), + (vec!["--schedule-cron", "null"], Some(Value::Null)), + ( + vec!["--schedule-cron", "*/5 * * * *"], + Some(json!({"cron": "*/5 * * * *"})), + ), ] { let mut create = vec![ "workflow", "create", "--name", "alerts", "--sql", "SELECT 1", @@ -387,12 +396,13 @@ fn workflow_scheduling_distinguishes_omitted_null_and_numeric_intervals() { 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 expected = schedule.clone().unwrap_or(Value::Null); + if !expected.is_null() { + expected["enabled"] = json!(true); + } + assert_eq!(body["schedule"], expected); + assert!(body.get("enabled").is_none()); + assert!(body.get("interval_seconds").is_none()); let mut update = vec!["workflow", "update", "wf-1", "--name", "renamed"]; update.extend_from_slice(&flags); let (_, _, body) = run( @@ -402,7 +412,7 @@ fn workflow_scheduling_distinguishes_omitted_null_and_numeric_intervals() { workflow_response(), json!({}), ); - assert_eq!(body.get("interval_seconds"), interval.as_ref()); + assert_eq!(body.get("schedule"), schedule.as_ref()); assert!(body.get("enabled").is_none()); assert!(body.get("query").is_none()); } @@ -411,13 +421,20 @@ fn workflow_scheduling_distinguishes_omitted_null_and_numeric_intervals() { #[test] fn workflow_update_sends_only_changed_fields() { let (_, _, body) = run( - &["workflow", "update", "wf-1", "--disable", "--clear-sinks"], + &[ + "workflow", + "update", + "wf-1", + "--schedule-enabled", + "false", + "--clear-sinks", + ], "PATCH", &format!("/v1/workflows/wf-1?{WORKFLOW_SCOPE}"), workflow_response(), json!({}), ); - assert_eq!(body, json!({"enabled": false, "sinks": []})); + assert_eq!(body, json!({"schedule": {"enabled": false}, "sinks": []})); } #[test] @@ -491,7 +508,7 @@ fn workflow_commands_require_a_cluster() { } #[test] -fn workflow_invalid_intervals_are_rejected_before_sending_requests() { +fn workflow_obsolete_intervals_are_rejected_before_sending_requests() { for args in [ vec![ "workflow", "create", "--name", "alerts", "--sql", "SELECT 1", @@ -503,7 +520,8 @@ fn workflow_invalid_intervals_are_rejected_before_sending_requests() { 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")); + assert!(String::from_utf8_lossy(&output.stderr) + .contains("unexpected argument '--interval-seconds'")); } } } @@ -522,3 +540,140 @@ fn workflow_manual_flag_is_rejected() { assert!(String::from_utf8_lossy(&output.stderr).contains("unexpected argument '--manual'")); } } + +#[test] +fn workflow_schedule_pause_and_resume_fields_are_nested() { + let (_, _, body) = run( + &[ + "workflow", + "create", + "--name", + "alerts", + "--sql", + "SELECT 1", + "--schedule-cron", + "*/30 * * * * *", + "--schedule-enabled", + "false", + ], + "POST", + &format!("/v1/workflows?{WORKFLOW_SCOPE}"), + workflow_response(), + json!({"database":"analytics"}), + ); + assert_eq!( + body["schedule"], + json!({"cron":"*/30 * * * * *","enabled":false}) + ); + assert!(body.get("enabled").is_none()); + for (flags, schedule) in [ + (vec!["--schedule-enabled", "true"], json!({"enabled":true})), + ( + vec!["--schedule-cron", "0 * * * *", "--schedule-enabled", "true"], + json!({"cron":"0 * * * *","enabled":true}), + ), + ( + vec![ + "--schedule-cron", + "0 * * * *", + "--schedule-enabled", + "false", + ], + json!({"cron":"0 * * * *","enabled":false}), + ), + ] { + 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!({}), + ); + assert_eq!(body, json!({"schedule":schedule})); + } +} + +#[test] +fn workflow_manual_schedule_rejects_pause_flags_before_http() { + for args in [ + vec![ + "workflow", + "create", + "--name", + "alerts", + "--sql", + "SELECT 1", + "--schedule-enabled", + "false", + ], + vec![ + "workflow", + "create", + "--name", + "alerts", + "--sql", + "SELECT 1", + "--schedule-cron", + "null", + "--schedule-enabled", + "false", + ], + vec![ + "workflow", + "update", + "wf-1", + "--schedule-cron", + "null", + "--schedule-enabled", + "true", + ], + vec![ + "workflow", + "update", + "wf-1", + "--schedule-cron", + "null", + "--schedule-enabled", + "false", + ], + ] { + let mut command = vec!["--json", "--org", "team", "--cluster", "prod"]; + command.extend(args); + let (output, _, bodies) = common::run_cli(&[], &command, &json!({"database":"analytics"})); + assert!(!output.status.success()); + assert!(bodies.is_empty()); + let error = String::from_utf8_lossy(&output.stderr); + assert!( + error.contains("requires --schedule-cron") || error.contains("cannot be combined"), + "{error}" + ); + } +} + +#[test] +fn workflow_cron_validation_errors_are_returned_by_the_api() { + let (output, _, bodies) = common::run_cli( + &[( + "POST /v1/workflows?organization=team&cluster=prod".into(), + "400 Bad Request", + json!({"message":"Invalid workflow cron", "hint":"Cron must have 5 or 6 fields."}), + )], + &[ + "--json", + "workflow", + "create", + "--name", + "alerts", + "--sql", + "SELECT 1", + "--schedule-cron", + "invalid-cron", + ], + &json!({"default_organization":"team","cluster":"prod","database":"analytics"}), + ); + assert!(!output.status.success()); + assert_eq!(bodies[0]["schedule"]["cron"], "invalid-cron"); + assert!(String::from_utf8_lossy(&output.stderr).contains("Cron must have 5 or 6 fields")); +} diff --git a/tests/live_platform/workflows.rs b/tests/live_platform/workflows.rs index 7d95b68..95c6e2a 100644 --- a/tests/live_platform/workflows.rs +++ b/tests/live_platform/workflows.rs @@ -43,7 +43,7 @@ fn with_workflow(sql: &str, test: impl FnOnce(&LiveCli, &str, &str)) { ]); let id = created["id"].as_str().expect("workflow ID").to_string(); workflow_id = Some(id.clone()); - assert_eq!(created["interval_seconds"], Value::Null); + assert_eq!(created["schedule"], Value::Null); test(&cli, &database, &id); })); if let Some(id) = workflow_id { @@ -79,7 +79,7 @@ fn workflow_manual_runs_deliver_rows_and_expose_logs_and_metrics() { 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]); + let definition = cli.json(&["workflow", "update", id, "--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); @@ -139,9 +139,9 @@ fn workflow_manual_runs_deliver_rows_and_expose_logs_and_metrics() { #[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); + let scheduled = cli.json(&["workflow", "update", id, "--schedule-cron", "* * * * * *"]); + assert_eq!(scheduled["schedule"]["cron"], "* * * * * *"); + assert_eq!(scheduled["schedule"]["enabled"], true); wait_for( "scheduled execution", || cli.json(&["workflow", "runs", id]), @@ -153,13 +153,12 @@ fn workflow_scheduled_runs_can_be_paused_and_switched_to_manual() { .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 paused = cli.json(&["workflow", "update", id, "--schedule-enabled", "false"]); + assert_eq!(paused["schedule"]["enabled"], false); + assert_eq!(paused["schedule"]["cron"], "* * * * * *"); + assert_eq!(paused["schedule"]["next_run_at"], Value::Null); + let manual = cli.json(&["workflow", "update", id, "--schedule-cron", "null"]); + assert_eq!(manual["schedule"], 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"); @@ -167,13 +166,14 @@ fn workflow_scheduled_runs_can_be_paused_and_switched_to_manual() { "workflow", "update", id, - "--interval-seconds", - "3600", - "--enable", + "--schedule-cron", + "0 * * * *", + "--schedule-enabled", + "true", ]); - assert_eq!(resumed["enabled"], true); - assert_eq!(resumed["interval_seconds"], 3600); - assert!(resumed["next_run_at"].is_string()); + assert_eq!(resumed["schedule"]["enabled"], true); + assert_eq!(resumed["schedule"]["cron"], "0 * * * *"); + assert!(resumed["schedule"]["next_run_at"].is_string()); }); } @@ -213,7 +213,10 @@ fn workflow_query_and_schedule_round_trip_through_real_platform() { "default", "--sql", "SELECT 1", - "--disabled", + "--schedule-cron", + "0 * * * *", + "--schedule-enabled", + "false", ]); let id = created["id"].as_str().expect("workflow ID"); let result = std::panic::catch_unwind(|| { @@ -221,9 +224,9 @@ fn workflow_query_and_schedule_round_trip_through_real_platform() { 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); + assert_eq!(created["schedule"]["cron"], "0 * * * *"); + assert_eq!(created["schedule"]["next_run_at"], Value::Null); + assert_eq!(created["schedule"]["enabled"], false); let listed = cli.json(&["workflow", "list"]); assert!(listed["workflows"] .as_array() @@ -237,16 +240,16 @@ fn workflow_query_and_schedule_round_trip_through_real_platform() { updated["query"], json!({"database": "default", "sql": "SELECT 2"}) ); - assert_eq!(updated["interval_seconds"], Value::Null); + assert_eq!(updated["schedule"]["cron"], "0 * * * *"); 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 scheduled = cli.json(&["workflow", "update", id, "--schedule-cron", "*/5 * * * *"]); + assert_eq!(scheduled["schedule"]["cron"], "*/5 * * * *"); + assert_eq!(scheduled["schedule"]["enabled"], false); let renamed = cli.json(&[ "workflow", "update", @@ -254,11 +257,10 @@ fn workflow_query_and_schedule_round_trip_through_real_platform() { "--name", &format!("{name}-renamed"), ]); - assert_eq!(renamed["interval_seconds"], 300); + assert_eq!(renamed["schedule"]["cron"], "*/5 * * * *"); 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); + let manual = cli.json(&["workflow", "update", id, "--schedule-cron", "null"]); + assert_eq!(manual["schedule"], Value::Null); assert_eq!(manual["query"], updated["query"]); assert_eq!(cli.json(&["workflow", "get", id]), manual); });