bynk_driver/
test_stream.rs

1//! `bynk test --format ndjson` (#2267, Slice D stage 1): the supported result
2//! stream.
3//!
4//! The runner already writes one JSON event per line (`BYNK_TEST_FORMAT=ndjson`),
5//! an internal protocol between the generated entry and the driver. `--format
6//! json` collects all of it and prints one document at the end; `--format ndjson`
7//! forwards it as it arrives, so a consumer sees each suite when that suite
8//! finishes. The events the driver prints are its own, not the runner's lines
9//! passed through, so the contract does not depend on the runner's wording:
10//!
11//! - `run-begin {suites}`
12//! - `suite-begin {name, kind, tests}`
13//! - `case {suite, <the document's case fields>}`: the same object the JSON
14//!   document holds for the case, in the same field order, behind `type` and
15//!   `suite`, so it carries `tier` and the generative fields too
16//! - `suite-end {name}`
17//! - `run-end {passed, failed, skipped?, transport?}`
18//! - `error {kind, message?, diagnostics?, stderr?}`: the document's `error`
19//!   object, for a compile failure, a run the driver could not start or finish,
20//!   and a runner that died before `run-end`
21//!
22//! The stream ends with exactly one of `run-end` or `error`. A case a remote run
23//! skipped (`--skip-unsupported`) is a `case` event with outcome `skipped`
24//! between the last `suite-end` and `run-end`. A line that is not a recognised
25//! runner event is dropped. Field order is the contract; a consumer ignores
26//! event types and fields it does not know.
27
28use std::io::{BufRead, BufReader, Write};
29use std::process::{Child, Command, ExitStatus, Stdio};
30
31use serde::Serialize;
32
33use crate::test_json::{
34    Case, TestError, TestRun, Unmatched, case_from_event, unmatched_from_event,
35};
36
37#[derive(Serialize)]
38struct RunBegin {
39    r#type: &'static str,
40    suites: usize,
41}
42
43#[derive(Serialize)]
44struct SuiteBegin<'a> {
45    r#type: &'static str,
46    name: &'a str,
47    kind: &'a str,
48    tests: usize,
49}
50
51#[derive(Serialize)]
52struct CaseEvent<'a> {
53    r#type: &'static str,
54    suite: &'a str,
55    #[serde(flatten)]
56    case: &'a Case,
57    /// After the document's case fields, so a later optional `Case` field still
58    /// lands before it.
59    #[serde(rename = "durationMs", skip_serializing_if = "Option::is_none")]
60    duration_ms: Option<f64>,
61}
62
63#[derive(Serialize)]
64struct SuiteEnd<'a> {
65    r#type: &'static str,
66    name: &'a str,
67}
68
69#[derive(Serialize)]
70struct RunEnd<'a> {
71    r#type: &'static str,
72    passed: u32,
73    failed: u32,
74    #[serde(skip_serializing_if = "is_zero")]
75    skipped: u32,
76    #[serde(skip_serializing_if = "Option::is_none")]
77    transport: Option<&'a str>,
78}
79
80#[derive(Serialize)]
81struct ErrorEvent<'a> {
82    r#type: &'static str,
83    #[serde(flatten)]
84    error: &'a TestError,
85}
86
87fn is_zero(n: &u32) -> bool {
88    *n == 0
89}
90
91fn line<T: Serialize>(event: &T) -> String {
92    // The structs hold only strings and numbers, which always serialise.
93    let mut text = serde_json::to_string(event).unwrap_or_default();
94    text.push('\n');
95    text
96}
97
98fn run_begin(suites: usize) -> String {
99    line(&RunBegin {
100        r#type: "run-begin",
101        suites,
102    })
103}
104
105fn suite_begin(name: &str, kind: &str, tests: usize) -> String {
106    line(&SuiteBegin {
107        r#type: "suite-begin",
108        name,
109        kind,
110        tests,
111    })
112}
113
114fn case_event(suite: &str, case: &Case) -> String {
115    line(&CaseEvent {
116        r#type: "case",
117        suite,
118        case,
119        duration_ms: case.duration_ms,
120    })
121}
122
123fn suite_end(name: &str) -> String {
124    line(&SuiteEnd {
125        r#type: "suite-end",
126        name,
127    })
128}
129
130fn run_end(passed: u32, failed: u32, skipped: u32, transport: Option<&str>) -> String {
131    line(&RunEnd {
132        r#type: "run-end",
133        passed,
134        failed,
135        skipped,
136        transport,
137    })
138}
139
140fn error_event(error: &TestError) -> String {
141    line(&ErrorEvent {
142        r#type: "error",
143        error,
144    })
145}
146
147/// The stream for a document held in full: a run that could not start (a
148/// compile failure, a launch error) is one `error` event, and a normal one is
149/// its suites' events then `run-end`. Used where nothing is read live.
150pub fn events_for(run: &TestRun) -> String {
151    let mut out = String::new();
152    let Some(suites) = &run.suites else {
153        if let Some(error) = &run.error {
154            out.push_str(&error_event(error));
155        }
156        return out;
157    };
158    // The live stream never sees a skipped case: the runner did not run it. So a
159    // suite's `tests` counts the cases between its `suite-begin` and `suite-end`,
160    // and a suite made only of skipped cases (which `with_remote` adds for them)
161    // has no such block, only its skipped cases after the last `suite-end`.
162    let ran = |suite: &&crate::test_json::Suite| {
163        suite.cases.is_empty() || suite.cases.iter().any(|c| c.outcome != "skipped")
164    };
165    out.push_str(&run_begin(suites.iter().filter(ran).count()));
166    for suite in suites.iter().filter(ran) {
167        let cases: Vec<&Case> = suite
168            .cases
169            .iter()
170            .filter(|c| c.outcome != "skipped")
171            .collect();
172        out.push_str(&suite_begin(&suite.name, &suite.kind, cases.len()));
173        for case in cases {
174            out.push_str(&case_event(&suite.name, case));
175        }
176        out.push_str(&suite_end(&suite.name));
177    }
178    for suite in suites {
179        for case in suite.cases.iter().filter(|c| c.outcome == "skipped") {
180            out.push_str(&case_event(&suite.name, case));
181        }
182    }
183    match &run.error {
184        Some(error) => out.push_str(&error_event(error)),
185        None => out.push_str(&run_end(
186            run.passed,
187            run.failed,
188            run.skipped,
189            run.transport.as_deref(),
190        )),
191    }
192    out
193}
194
195/// Print a document in `--format json`, or its events in `--format ndjson`.
196pub fn print_run(stream: bool, run: &TestRun) {
197    if stream {
198        print!("{}", events_for(run));
199    } else {
200        print!("{}", run.render());
201    }
202}
203
204/// A remote run's skipped case, as it goes into the stream: suite, case, reason.
205pub type Skipped = (String, String, String);
206
207/// What the stream knows about the run besides the runner's own events.
208pub struct StreamContext<'a> {
209    /// The compile's case table, for each case's tier.
210    pub discovered: &'a [bynk_emit::project::DiscoveredSuite],
211    /// The cases a remote run skipped, which the runner never saw.
212    pub skipped: Vec<Skipped>,
213    /// `remote:<target>` under `--against`.
214    pub transport: Option<String>,
215}
216
217/// Turns the runner's event lines into the stream's, one at a time.
218pub struct Stream<'a> {
219    context: StreamContext<'a>,
220    ended: bool,
221    unmatched: Unmatched,
222}
223
224impl<'a> Stream<'a> {
225    pub fn new(context: StreamContext<'a>) -> Self {
226        Self {
227            context,
228            ended: false,
229            unmatched: Unmatched::default(),
230        }
231    }
232
233    /// The `--suite`/`--case` values the runner reported as selecting nothing:
234    /// not part of the stream, but the driver turns them into an error (#2273).
235    pub fn unmatched(&self) -> &Unmatched {
236        &self.unmatched
237    }
238
239    fn tier_of(&self, suite: &str, case: &str) -> Option<&'static str> {
240        self.context
241            .discovered
242            .iter()
243            .find(|d| d.name == suite)?
244            .cases
245            .iter()
246            .find(|c| c.name == case)?
247            .tier
248    }
249
250    /// The stream's text for one line of the runner's output: nothing for a line
251    /// that is not a recognised event.
252    pub fn push(&mut self, raw: &str) -> String {
253        let raw = raw.trim();
254        let Ok(value) = serde_json::from_str::<serde_json::Value>(raw) else {
255            return String::new();
256        };
257        let text = |key: &str| value.get(key).and_then(|v| v.as_str()).unwrap_or_default();
258        let count = |key: &str| {
259            value
260                .get(key)
261                .and_then(serde_json::Value::as_u64)
262                .map_or(0, |n| usize::try_from(n).unwrap_or(usize::MAX))
263        };
264        match text("type") {
265            "run-begin" => run_begin(count("suites")),
266            "suite-begin" => suite_begin(text("name"), text("kind"), count("tests")),
267            "case" => {
268                let mut case = case_from_event(&value);
269                case.tier = self.tier_of(text("suite"), &case.name).map(str::to_string);
270                case_event(text("suite"), &case)
271            }
272            "suite-end" => suite_end(text("name")),
273            "run-end" => {
274                self.ended = true;
275                self.unmatched = unmatched_from_event(&value);
276                let mut out = String::new();
277                for (suite, name, reason) in &self.context.skipped {
278                    out.push_str(&case_event(
279                        suite,
280                        &Case {
281                            name: name.clone(),
282                            outcome: "skipped".to_string(),
283                            message: Some(reason.clone()),
284                            location: None,
285                            retries: None,
286                            tier: self.tier_of(suite, name).map(str::to_string),
287                            kind: None,
288                            seed: None,
289                            counterexample: None,
290                            violation: None,
291                            refusals: None,
292                            duration_ms: None,
293                        },
294                    ));
295                }
296                let passed = u32::try_from(count("passed")).unwrap_or(u32::MAX);
297                let failed = u32::try_from(count("failed")).unwrap_or(u32::MAX);
298                let skipped = u32::try_from(self.context.skipped.len()).unwrap_or(u32::MAX);
299                out.push_str(&run_end(
300                    passed,
301                    failed,
302                    skipped,
303                    self.context.transport.as_deref(),
304                ));
305                out
306            }
307            _ => String::new(),
308        }
309    }
310
311    /// What closes the stream when the runner's output has ended: nothing after
312    /// a `run-end`, else an `error` event, the runner having died mid-run.
313    pub fn finish(&self, stderr: &str) -> String {
314        if self.ended {
315            return String::new();
316        }
317        let trimmed = stderr.trim();
318        error_event(&TestError {
319            kind: "runtime".to_string(),
320            message: Some("the test runner exited before completing".to_string()),
321            diagnostics: Vec::new(),
322            stderr: (!trimmed.is_empty()).then(|| trimmed.to_string()),
323        })
324    }
325}
326
327/// Run `cmd`, forwarding each event to `out` as the runner writes it, and return
328/// its exit status. The runner's stderr is collected on a thread, so a full pipe
329/// cannot stall it. If `out` stops accepting output (a consumer that went away),
330/// the runner is killed rather than left running.
331pub fn forward(
332    mut cmd: Command,
333    stream: &mut Stream<'_>,
334    out: &mut impl Write,
335) -> std::io::Result<ExitStatus> {
336    let mut child = cmd.stdout(Stdio::piped()).stderr(Stdio::piped()).spawn()?;
337    let stderr_pipe = child.stderr.take();
338    let collector = std::thread::spawn(move || {
339        let mut text = String::new();
340        if let Some(pipe) = stderr_pipe {
341            for line in BufReader::new(pipe).lines().map_while(Result::ok) {
342                text.push_str(&line);
343                text.push('\n');
344            }
345        }
346        text
347    });
348    let mut gone = false;
349    if let Some(pipe) = child.stdout.take() {
350        // Lossy, so a stray byte in a line cannot end the stream early.
351        let mut reader = BufReader::new(pipe);
352        let mut buffer = Vec::new();
353        loop {
354            buffer.clear();
355            match reader.read_until(b'\n', &mut buffer) {
356                Ok(0) | Err(_) => break,
357                Ok(_) => {}
358            }
359            let text = stream.push(&String::from_utf8_lossy(&buffer));
360            if !text.is_empty() && write_and_flush(out, &text).is_err() {
361                gone = true;
362                break;
363            }
364        }
365    }
366    if gone {
367        kill(&mut child);
368    }
369    let status = child.wait()?;
370    let stderr = collector.join().unwrap_or_default();
371    if !gone {
372        let closing = stream.finish(&stderr);
373        if !closing.is_empty() {
374            let _ = write_and_flush(out, &closing);
375        }
376    }
377    Ok(status)
378}
379
380fn write_and_flush(out: &mut impl Write, text: &str) -> std::io::Result<()> {
381    out.write_all(text.as_bytes())?;
382    out.flush()
383}
384
385fn kill(child: &mut Child) {
386    let _ = child.kill();
387}
388
389#[cfg(test)]
390mod tests {
391    use super::*;
392    use crate::test_json::{Location, Suite};
393
394    fn context<'a>() -> StreamContext<'a> {
395        StreamContext {
396            discovered: &[],
397            skipped: Vec::new(),
398            transport: None,
399        }
400    }
401
402    fn case(name: &str, outcome: &str) -> Case {
403        Case {
404            name: name.to_string(),
405            outcome: outcome.to_string(),
406            message: None,
407            location: None,
408            retries: None,
409            tier: None,
410            kind: None,
411            seed: None,
412            counterexample: None,
413            violation: None,
414            refusals: None,
415            duration_ms: None,
416        }
417    }
418
419    #[test]
420    fn a_case_event_is_the_documents_case_behind_type_and_suite() {
421        let mut failed = case("never seventy three", "fail");
422        failed.message = Some("property failed".to_string());
423        failed.location = Some(Location {
424            path: "tests/m.test.bynk".to_string(),
425            line: 3,
426            col: 4,
427        });
428        failed.tier = Some("unit".to_string());
429        failed.kind = Some("property".to_string());
430        failed.seed = Some("0x5f3a".to_string());
431        failed.counterexample = Some("p = 73".to_string());
432        assert_eq!(
433            case_event("demo.m", &failed),
434            concat!(
435                r#"{"type":"case","suite":"demo.m","name":"never seventy three","outcome":"fail","#,
436                r#""message":"property failed","location":{"path":"tests/m.test.bynk","line":3,"col":4},"#,
437                r#""tier":"unit","kind":"property","seed":"0x5f3a","counterexample":"p = 73"}"#,
438                "\n"
439            )
440        );
441    }
442
443    #[test]
444    fn a_case_event_names_the_refused_steps_before_its_duration() {
445        let mut failed = case("every call is accepted", "fail");
446        failed.refusals = Some(vec![crate::test_json::Refusal {
447            index: 0,
448            call: "Spend".to_string(),
449            kind: "invariant".to_string(),
450            agent: "Wallet".to_string(),
451            name: "nonneg".to_string(),
452            location: Some(Location {
453                path: "src/demo/wallet.bynk".to_string(),
454                line: 9,
455                col: 13,
456            }),
457        }]);
458        failed.duration_ms = Some(3.0);
459        assert_eq!(
460            case_event("demo.wallet", &failed),
461            concat!(
462                r#"{"type":"case","suite":"demo.wallet","name":"every call is accepted","outcome":"fail","#,
463                r#""refusals":[{"index":0,"call":"Spend","kind":"invariant","agent":"Wallet","name":"nonneg","#,
464                r#""location":{"path":"src/demo/wallet.bynk","line":9,"col":13}}],"durationMs":3.0}"#,
465                "\n"
466            )
467        );
468    }
469
470    #[test]
471    fn a_case_event_ends_with_its_duration_when_the_runner_measured_one() {
472        let mut ran = case("one", "pass");
473        ran.tier = Some("unit".to_string());
474        ran.duration_ms = Some(12.5);
475        assert_eq!(
476            case_event("demo.m", &ran),
477            "{\"type\":\"case\",\"suite\":\"demo.m\",\"name\":\"one\",\"outcome\":\"pass\",\"tier\":\"unit\",\"durationMs\":12.5}\n"
478        );
479        // A case without one (a skipped case, a replayed document) has no field.
480        assert!(!case_event("demo.m", &case("two", "skipped")).contains("durationMs"));
481    }
482
483    #[test]
484    fn the_runners_events_become_the_streams_with_tiers() {
485        let discovered = [bynk_emit::project::DiscoveredSuite {
486            name: "demo.m".to_string(),
487            kind: "unit",
488            cases: vec![bynk_emit::project::DiscoveredCase {
489                name: "one".to_string(),
490                location: None,
491                tier: Some("unit"),
492                generative: None,
493            }],
494            skipped: Vec::new(),
495            envs: Vec::new(),
496        }];
497        let mut stream = Stream::new(StreamContext {
498            discovered: &discovered,
499            ..context()
500        });
501        let out: String = [
502            r#"{"type":"run-begin","suites":1}"#,
503            r#"{"type":"suite-begin","name":"demo.m","kind":"unit","tests":1}"#,
504            r#"{"type":"case","suite":"demo.m","name":"one","outcome":"pass","durationMs":0.25}"#,
505            r#"{"type":"suite-end","name":"demo.m"}"#,
506            "not an event",
507            r#"{"type":"something-new","x":1}"#,
508            r#"{"type":"run-end","passed":1,"failed":0,"unmatched":{"cases":["x"]}}"#,
509        ]
510        .iter()
511        .map(|l| stream.push(l))
512        .collect();
513        assert_eq!(
514            out,
515            concat!(
516                "{\"type\":\"run-begin\",\"suites\":1}\n",
517                "{\"type\":\"suite-begin\",\"name\":\"demo.m\",\"kind\":\"unit\",\"tests\":1}\n",
518                "{\"type\":\"case\",\"suite\":\"demo.m\",\"name\":\"one\",\"outcome\":\"pass\",\"tier\":\"unit\",\"durationMs\":0.25}\n",
519                "{\"type\":\"suite-end\",\"name\":\"demo.m\"}\n",
520                "{\"type\":\"run-end\",\"passed\":1,\"failed\":0}\n",
521            ),
522            "an unrecognised line is dropped, and `unmatched` is not part of the stream"
523        );
524        assert_eq!(stream.finish(""), "", "nothing closes a stream that ended");
525    }
526
527    #[test]
528    fn skipped_cases_and_the_transport_close_a_remote_run() {
529        let mut stream = Stream::new(StreamContext {
530            skipped: vec![(
531                "demo.s".to_string(),
532                "advances".to_string(),
533                "advance Clock".to_string(),
534            )],
535            transport: Some("remote:staging".to_string()),
536            ..context()
537        });
538        assert_eq!(
539            stream.push(r#"{"type":"run-end","passed":2,"failed":0}"#),
540            concat!(
541                "{\"type\":\"case\",\"suite\":\"demo.s\",\"name\":\"advances\",\"outcome\":\"skipped\",\"message\":\"advance Clock\"}\n",
542                "{\"type\":\"run-end\",\"passed\":2,\"failed\":0,\"skipped\":1,\"transport\":\"remote:staging\"}\n",
543            )
544        );
545    }
546
547    #[test]
548    fn a_runner_that_died_is_an_error_event() {
549        let mut stream = Stream::new(context());
550        stream.push(r#"{"type":"run-begin","suites":1}"#);
551        assert_eq!(
552            stream.finish("boom\n"),
553            "{\"type\":\"error\",\"kind\":\"runtime\",\"message\":\"the test runner exited before completing\",\"stderr\":\"boom\"}\n"
554        );
555    }
556
557    #[test]
558    fn a_document_held_in_full_becomes_its_events() {
559        let compile = TestRun::compile_error(vec!["a.bynk:1:1: error[x]: no".to_string()]);
560        assert_eq!(
561            events_for(&compile),
562            "{\"type\":\"error\",\"kind\":\"compile\",\"diagnostics\":[\"a.bynk:1:1: error[x]: no\"]}\n"
563        );
564
565        let mut run = TestRun::empty();
566        run.suites = Some(vec![
567            Suite {
568                name: "demo.m".to_string(),
569                kind: "unit".to_string(),
570                cases: vec![case("one", "pass"), case("two", "skipped")],
571            },
572            Suite {
573                name: "demo.s".to_string(),
574                kind: "integration".to_string(),
575                cases: vec![case("advances", "skipped")],
576            },
577            Suite {
578                name: "demo.empty".to_string(),
579                kind: "unit".to_string(),
580                cases: Vec::new(),
581            },
582        ]);
583        run.passed = 1;
584        run.skipped = 2;
585        assert_eq!(
586            events_for(&run),
587            concat!(
588                "{\"type\":\"run-begin\",\"suites\":2}\n",
589                "{\"type\":\"suite-begin\",\"name\":\"demo.m\",\"kind\":\"unit\",\"tests\":1}\n",
590                "{\"type\":\"case\",\"suite\":\"demo.m\",\"name\":\"one\",\"outcome\":\"pass\"}\n",
591                "{\"type\":\"suite-end\",\"name\":\"demo.m\"}\n",
592                "{\"type\":\"suite-begin\",\"name\":\"demo.empty\",\"kind\":\"unit\",\"tests\":0}\n",
593                "{\"type\":\"suite-end\",\"name\":\"demo.empty\"}\n",
594                "{\"type\":\"case\",\"suite\":\"demo.m\",\"name\":\"two\",\"outcome\":\"skipped\"}\n",
595                "{\"type\":\"case\",\"suite\":\"demo.s\",\"name\":\"advances\",\"outcome\":\"skipped\"}\n",
596                "{\"type\":\"run-end\",\"passed\":1,\"failed\":0,\"skipped\":2}\n",
597            ),
598            "`tests` counts what the suite reports; a suite of only skipped cases has no block"
599        );
600    }
601
602    /// A sink that records when each write arrived.
603    struct Timed(Vec<(std::time::Instant, String)>);
604
605    impl Write for Timed {
606        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
607            self.0.push((
608                std::time::Instant::now(),
609                String::from_utf8_lossy(buf).into_owned(),
610            ));
611            Ok(buf.len())
612        }
613        fn flush(&mut self) -> std::io::Result<()> {
614            Ok(())
615        }
616    }
617
618    /// An event reaches the consumer while the runner is still running, not when
619    /// it exits; an event split across two writes is still one event; a runner
620    /// that never says `run-end` ends the stream with an `error`.
621    #[cfg(unix)]
622    #[test]
623    fn events_are_forwarded_as_they_are_written() {
624        let mut cmd = Command::new("sh");
625        cmd.arg("-c").arg(
626            r#"printf '{"type":"run-begin","suites":1}\n{"type":"suite-be'; sleep 0.4;
627               printf 'gin","name":"s","kind":"unit","tests":0}\n'; sleep 0.4; echo oops >&2"#,
628        );
629        let mut stream = Stream::new(context());
630        let mut sink = Timed(Vec::new());
631        let started = std::time::Instant::now();
632        let status = forward(cmd, &mut stream, &mut sink).unwrap();
633        let finished = started.elapsed();
634        assert!(status.success());
635        let first = sink.0.first().expect("an event");
636        assert_eq!(first.1, "{\"type\":\"run-begin\",\"suites\":1}\n");
637        assert!(
638            first.0.duration_since(started) < finished - std::time::Duration::from_millis(500),
639            "the first event arrived well before the runner exited"
640        );
641        let all: String = sink.0.iter().map(|(_, t)| t.as_str()).collect();
642        assert_eq!(
643            all,
644            concat!(
645                "{\"type\":\"run-begin\",\"suites\":1}\n",
646                "{\"type\":\"suite-begin\",\"name\":\"s\",\"kind\":\"unit\",\"tests\":0}\n",
647                "{\"type\":\"error\",\"kind\":\"runtime\",\"message\":\"the test runner exited before completing\",\"stderr\":\"oops\"}\n",
648            ),
649            "a split event is one event, and the missing run-end is an error"
650        );
651    }
652}
bynk_driver/test_stream.rs

bynk_driver/
test_stream.rs
