From 9081d0fcc54f1a6d96e7051df1fc05bf9efa6e22 Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Fri, 30 Oct 2026 10:53:24 -0400 Subject: [PATCH] test: tolerate valid startup interleavings in event snapshots --- lib/apps/fabro-cli/tests/it/cmd/events.rs | 215 ++++++++++++------ lib/apps/fabro-cli/tests/it/scenario/petri.rs | 24 +- lib/apps/fabro-cli/tests/it/support/mod.rs | 24 ++ 3 files changed, 177 insertions(+), 86 deletions(-) diff --git a/lib/apps/fabro-cli/tests/it/cmd/events.rs b/lib/apps/fabro-cli/tests/it/cmd/events.rs index ada175646..f67205912 100644 --- a/lib/apps/fabro-cli/tests/it/cmd/events.rs +++ b/lib/apps/fabro-cli/tests/it/cmd/events.rs @@ -1,7 +1,12 @@ -use fabro_test::{fabro_snapshot, json_elapsed_ms_snapshot_filters, test_context}; +use fabro_test::{apply_filters, fabro_snapshot, test_context}; +use httpmock::MockServer; use serde_json::Value; -use super::support::{setup_detached_dry_run, setup_seeded_completed_dry_run}; +use super::support::{ + remote_run_summary_json, run_stream_items, setup_detached_dry_run, + setup_seeded_completed_dry_run, +}; +use crate::support::{pretty_stream_snapshot_filters, run_projection_json, unique_run_id}; fn parse_ndjson(stdout: &[u8]) -> Vec { String::from_utf8(stdout.to_vec()) @@ -99,6 +104,12 @@ fn events_completed_run_outputs_raw_ndjson() { let events = parse_ndjson(&output.stdout); assert_events_belong_to_run(&events, &run.run_id); + let recorded: Vec = run_stream_items(&run.run_dir) + .into_iter() + .map(|item| serde_json::to_value(item).expect("stream item serializes")) + .collect(); + assert_eq!(events, recorded, "events must preserve the recorded stream"); + assert!(output.stderr.is_empty(), "{output:?}"); assert_event_sequence_contains(&events, &[ "run.created", "run.lifecycle", @@ -111,64 +122,29 @@ fn events_completed_run_outputs_raw_ndjson() { } #[test] -fn events_completed_run_reads_store_without_progress_jsonl() { +fn events_tail_matches_recorded_stream_suffix() { let context = test_context!(); let run = setup_seeded_completed_dry_run(&context); + let expected: Vec = run_stream_items(&run.run_dir) + .into_iter() + .map(|item| serde_json::to_value(item).expect("stream item serializes")) + .collect(); + assert!(expected.len() > 2, "the fixture must exercise truncation"); - let mut filters = json_elapsed_ms_snapshot_filters(context.filters()); - filters.push(( - r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z".to_string(), - "[TIMESTAMP]".to_string(), - )); - filters.push(( - r#""id":"[0-9a-f-]+""#.to_string(), - r#""id":"[EVENT_ID]""#.to_string(), - )); - filters.push(( - r#""recorded_at":\d{13}"#.to_string(), - r#""recorded_at":[EPOCH_MS]"#.to_string(), - )); - let mut cmd = context.command(); - cmd.args(["events", "--tail", "2", &run.run_id]); - - fabro_snapshot!(filters, cmd, @r#" - success: true - exit_code: 0 - ----- stdout ----- - {"run_id":"[ULID]","stream_seq":67,"kind":"petri","id":"coordinator/7/0","recorded_at":[EPOCH_MS],"item":{"id":{"log":"coordinator","seq":7,"index":0},"origin":"external","context":{},"recorded_at":[EPOCH_MS],"record":{"seq":7,"origin":"external","recorded_at":[EPOCH_MS],"body":{"event":"run.finished","status":"success"}}}} - {"run_id":"[ULID]","stream_seq":68,"kind":"platform","id":"[EVENT_ID]","recorded_at":[EPOCH_MS],"item":{"seq":14,"recorded_at":[EPOCH_MS],"record":{"kind":"run.lifecycle","transition":"succeeded","status":{"kind":"succeeded","reason":"completed"}}}} - ----- stderr ----- - "#); -} - -#[test] -fn events_tail_limits_output() { - let context = test_context!(); - let run = setup_seeded_completed_dry_run(&context); - let mut filters = json_elapsed_ms_snapshot_filters(context.filters()); - filters.push(( - r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z".to_string(), - "[TIMESTAMP]".to_string(), - )); - filters.push(( - r#""id":"[0-9a-f-]+""#.to_string(), - r#""id":"[EVENT_ID]""#.to_string(), - )); - filters.push(( - r#""recorded_at":\d{13}"#.to_string(), - r#""recorded_at":[EPOCH_MS]"#.to_string(), - )); - let mut cmd = context.command(); - cmd.args(["events", "--tail", "2", &run.run_id]); - - fabro_snapshot!(filters, cmd, @r#" - success: true - exit_code: 0 - ----- stdout ----- - {"run_id":"[ULID]","stream_seq":67,"kind":"petri","id":"coordinator/7/0","recorded_at":[EPOCH_MS],"item":{"id":{"log":"coordinator","seq":7,"index":0},"origin":"external","context":{},"recorded_at":[EPOCH_MS],"record":{"seq":7,"origin":"external","recorded_at":[EPOCH_MS],"body":{"event":"run.finished","status":"success"}}}} - {"run_id":"[ULID]","stream_seq":68,"kind":"platform","id":"[EVENT_ID]","recorded_at":[EPOCH_MS],"item":{"seq":14,"recorded_at":[EPOCH_MS],"record":{"kind":"run.lifecycle","transition":"succeeded","status":{"kind":"succeeded","reason":"completed"}}}} - ----- stderr ----- - "#); + for tail in [0, 1, 2, expected.len() + 1] { + let output = context + .command() + .args(["events", "--tail", &tail.to_string(), &run.run_id]) + .output() + .expect("events should execute"); + assert!(output.status.success(), "{output:?}"); + assert!(output.stderr.is_empty(), "{output:?}"); + assert_eq!( + parse_ndjson(&output.stdout), + expected[expected.len().saturating_sub(tail)..], + "--tail {tail} must preserve the recorded suffix, including cursors and payloads" + ); + } } #[test] @@ -190,16 +166,7 @@ fn events_since_filters_output() { fn events_pretty_formats_small_run() { let context = test_context!(); let run = setup_seeded_completed_dry_run(&context); - let mut filters = context.filters(); - filters.push((r"\b\d{2}:\d{2}:\d{2}\b".to_string(), "[CLOCK]".to_string())); - filters.push(( - r"\b\d+(\.\d+)?(ms|s)\b".to_string(), - "[DURATION]".to_string(), - )); - filters.push(( - r"Checkpoint [0-9a-f]{7}\b".to_string(), - "Checkpoint [SHA]".to_string(), - )); + let filters = pretty_stream_snapshot_filters(&context); let mut cmd = context.command(); cmd.args(["events", "--pretty", &run.run_id]); @@ -238,6 +205,120 @@ fn events_pretty_formats_small_run() { "); } +#[test] +fn events_preserve_each_valid_startup_interleaving() { + let context = test_context!(); + let run_id = unique_run_id(); + let status = serde_json::json!({"kind": "running"}); + let records = [ + ( + "platform", + "6", + serde_json::json!({ + "seq": 6, "recorded_at": 0, + "record": {"kind": "run.lifecycle", "transition": "running", "status": status} + }), + ), + ( + "petri", + "coordinator/0/0", + serde_json::json!({ + "record": {"body": {"event": "run.started"}} + }), + ), + ( + "petri", + "execution 0/0/1", + serde_json::json!({ + "subject": {"node": {"name": "Start", "meta": {"kind": "start"}}}, + "derived": {"event": "visit.started"} + }), + ), + ]; + let rendered = [ + "00:00:00 · running\n", + "00:00:00 Engine: petri run started\n", + "00:00:00 ▶ Start\n", + ]; + for order in [[0, 1, 2], [1, 0, 2], [1, 2, 0]] { + let items: Vec = order + .iter() + .enumerate() + .map(|(index, source)| { + let (kind, id, item) = &records[*source]; + serde_json::json!({ + "run_id": run_id, "stream_seq": index + 6, "kind": kind, + "id": id, "recorded_at": 0, "item": item + }) + }) + .collect(); + let server = MockServer::start(); + server.mock(|when, then| { + when.method("GET").path("/api/v1/runs/resolve"); + then.status(200).json_body(remote_run_summary_json( + &run_id, + "Startup", + "startup", + "Startup", + &status, + "2026-10-30T12:00:00Z", + )); + }); + server.mock(|when, then| { + when.method("GET") + .path(format!("/api/v1/runs/{run_id}/state")); + then.status(200) + .json_body(run_projection_json(&run_id, &status)); + }); + server.mock(|when, then| { + when.method("GET") + .path(format!("/api/v1/runs/{run_id}/events")); + then.status(200).json_body(serde_json::json!({ + "data": items, "meta": {"has_more": false}, "event_contract_version": 3 + })); + }); + let raw = context + .command() + .args(["events", "--server", &server.base_url(), &run_id]) + .output() + .expect("events should execute"); + assert!(raw.status.success(), "{raw:?}"); + assert!(raw.stderr.is_empty(), "{raw:?}"); + assert_eq!(parse_ndjson(&raw.stdout), items, "order {order:?}"); + + let pretty = context + .command() + .args([ + "events", + "--pretty", + "--server", + &server.base_url(), + &run_id, + ]) + .output() + .expect("pretty events should execute"); + assert!(pretty.status.success(), "{pretty:?}"); + assert!(pretty.stderr.is_empty(), "{pretty:?}"); + let stdout = String::from_utf8(pretty.stdout).expect("pretty output is UTF-8"); + let expected: String = order.iter().map(|index| rendered[*index]).collect(); + assert_eq!( + stdout, expected, + "pretty output must retain delivery order {order:?}" + ); + // Canonicalize only the snapshot, after checking both real output modes. + insta::allow_duplicates! { + insta::assert_snapshot!( + apply_filters(&stdout, &pretty_stream_snapshot_filters(&context)), + @" + [CLOCK] · running + [CLOCK] Engine: petri run started + [CLOCK] ▶ Start + " + ); + } + } +} + #[test] #[ignore = "pre-existing flake: events --follow hangs against detached dry-run on this branch and on origin/main; tracked separately"] fn events_follow_detached_run_streams_until_completion() { diff --git a/lib/apps/fabro-cli/tests/it/scenario/petri.rs b/lib/apps/fabro-cli/tests/it/scenario/petri.rs index 8e5fc4337..27c93b8cb 100644 --- a/lib/apps/fabro-cli/tests/it/scenario/petri.rs +++ b/lib/apps/fabro-cli/tests/it/scenario/petri.rs @@ -42,7 +42,9 @@ use fabro_types::RunId; use fabro_vault::{SecretType, Vault}; use crate::cmd::support::created_run_id; -use crate::support::{TEST_DEV_TOKEN, TEST_SESSION_SECRET, seed_dev_token_auth}; +use crate::support::{ + TEST_DEV_TOKEN, TEST_SESSION_SECRET, pretty_stream_snapshot_filters, seed_dev_token_auth, +}; pub(super) const RUN_TIMEOUT: Duration = Duration::from_mins(1); pub(super) const POLL: Duration = Duration::from_millis(50); @@ -1198,22 +1200,6 @@ fn stream_line_name(line: &serde_json::Value) -> String { .to_string() } -/// The snapshot filters for `events --pretty` over a Petri run: clocks, -/// durations and the run id vary per run. -fn pretty_filters(context: &fabro_test::TestContext) -> Vec<(String, String)> { - let mut filters = context.filters(); - filters.push((r"\b\d{2}:\d{2}:\d{2}\b".to_string(), "[CLOCK]".to_string())); - filters.push(( - r"\b\d+(\.\d+)?(ms|s)\b".to_string(), - "[DURATION]".to_string(), - )); - filters.push(( - r"Checkpoint [0-9a-f]{7}".to_string(), - "Checkpoint [SHA]".to_string(), - )); - filters -} - /// A finished Petri run reads back through the CLI: `events` prints the /// stream envelope raw, dense in `stream_seq`; `events --pretty` renders /// the stages by `.` with their labels and the platform @@ -1273,12 +1259,12 @@ async fn a_finished_petri_run_reads_back_through_the_cli() { // Tail: the last two items only. let tail = cli(&context, &server, &["events", "--tail", "2", &run_id]); - assert_eq!(ndjson(&tail).len(), 2); + assert_eq!(ndjson(&tail), lines[lines.len() - 2..]); // Pretty: stages and platform records. let mut cmd = context.command(); cmd.args(["events", "--pretty", "--server", &target, &run_id]); - fabro_snapshot!(pretty_filters(&context), cmd, @" + fabro_snapshot!(pretty_stream_snapshot_filters(&context), cmd, @" success: true exit_code: 0 ----- stdout ----- diff --git a/lib/apps/fabro-cli/tests/it/support/mod.rs b/lib/apps/fabro-cli/tests/it/support/mod.rs index 25364a67c..1545bd762 100644 --- a/lib/apps/fabro-cli/tests/it/support/mod.rs +++ b/lib/apps/fabro-cli/tests/it/support/mod.rs @@ -13,6 +13,30 @@ use fabro_test::{EnvVars, TestContext, preserve_coverage_env}; use fabro_types::{RunGraph, RunId, RunSpec, RunStreamItem, WorkflowSettings}; pub(crate) use mcp_client::McpStdioTestClient; +/// Snapshot a real run's pretty stream without depending on wall-clock values +/// or the relative delivery of the platform and Petri startup records. +pub(crate) fn pretty_stream_snapshot_filters(context: &TestContext) -> Vec<(String, String)> { + let mut filters = context.filters(); + filters.push((r"\b\d{2}:\d{2}:\d{2}\b".to_string(), "[CLOCK]".to_string())); + filters.push(( + r"\b\d+(\.\d+)?(ms|s)\b".to_string(), + "[DURATION]".to_string(), + )); + filters.push(( + r"Checkpoint [0-9a-f]{7}\b".to_string(), + "Checkpoint [SHA]".to_string(), + )); + // Only move the independent `running` line ahead of Petri's start. Keep + // every line (including any intervening output) and all other ordering. + // Missing or repeated records must still fail the full transcript snapshot. + filters.push(( + r"(?m)^(\[CLOCK\] Engine: petri run started\n)((?:[^\n]*\n)*?)(\[CLOCK\] · running\n)" + .to_string(), + "${3}${1}${2}".to_string(), + )); + filters +} + pub(crate) fn run_output_filters(context: &TestContext) -> Vec<(String, String)> { let mut filters = context.filters(); filters.push((r"\b\d+(\.\d+)?(ms|s)\b".to_string(), "[TIME]".to_string()));