fix(cli): ensure that pre- and post-test output is flushed at the appropriate times (#22611)

Some `deno_std` tests were failing to print output that was resolved
after the last test finished. In addition, output printed before tests
began would sometimes appear above the "running X tests ..." line, and
sometimes below it depending on timing.

We now guarantee that all output is flushed before and after tests run,
making the output consistent.

Pre-test and post-test output are captured in `------ pre-test output
------` and `------ post-test output ------` blocks to differentiate
them from the regular output blocks.

Here's an example of a test (that is much noisier than normal, but an
example of what the output will look like):

```
Check ./load_unload.ts
------- pre-test output -------
load
----- output end -----
running 1 test from ./load_unload.ts
test ...
------- output -------
test
----- output end -----
test ... ok ([WILDCARD])
------- post-test output -------
unload
----- output end -----
```

cli/lsp/testing/execution.rs

@@ -397,6 +397,9 @@ impl TestRun {
                 );
               }
             }
+            test::TestEvent::Completed => {
+              reporter.report_completed();
+            }
             test::TestEvent::ForceEndReport => {}
             test::TestEvent::Sigint => {}
           }
@@ -742,6 +745,10 @@ impl LspTestReporter {
     }
   }
 
+  fn report_completed(&mut self) {
+    // there is nothing to do on report_completed
+  }
+
   fn report_summary(
     &mut self,
     _summary: &test::TestSummary,

cli/tools/test/channel.rs

@@ -384,6 +384,8 @@ impl TestEventSender {
 
 #[cfg(test)]
 mod tests {
+  use crate::tools::test::TestResult;
+
   use super::*;
   use deno_core::unsync::spawn;
   use deno_core::unsync::spawn_blocking;
@@ -466,6 +468,75 @@ mod tests {
     assert_eq!(messages.len(), 100000);
   }
 
+  /// Test that large numbers of interleaved steps are routed properly.
+  #[tokio::test]
+  async fn test_interleave() {
+    test_util::timeout!(60);
+    const MESSAGE_COUNT: usize = 10_000;
+    let (mut worker, mut receiver) = create_single_test_event_channel();
+    let recv_handle = spawn(async move {
+      let mut i = 0;
+      while let Some((_, message)) = receiver.recv().await {
+        if i % 2 == 0 {
+          let expected_text = format!("{:08x}", i / 2).into_bytes();
+          let TestEvent::Output(TestStdioStream::Stderr, text) = message else {
+            panic!("Incorrect message: {message:?}");
+          };
+          assert_eq!(text, expected_text);
+        } else {
+          let TestEvent::Result(index, TestResult::Ok, 0) = message else {
+            panic!("Incorrect message: {message:?}");
+          };
+          assert_eq!(index, i / 2);
+        }
+        i += 1;
+      }
+      eprintln!("Receiver closed");
+      i
+    });
+    let send_handle: deno_core::unsync::JoinHandle<()> =
+      spawn_blocking(move || {
+        for i in 0..MESSAGE_COUNT {
+          worker
+            .stderr
+            .write_all(format!("{i:08x}").as_str().as_bytes())
+            .unwrap();
+          worker
+            .sender
+            .send(TestEvent::Result(i, TestResult::Ok, 0))
+            .unwrap();
+        }
+        eprintln!("Sent all messages");
+      });
+    send_handle.await.unwrap();
+    let messages = recv_handle.await.unwrap();
+    assert_eq!(messages, MESSAGE_COUNT * 2);
+  }
+
+  #[tokio::test]
+  async fn test_sender_shutdown_before_receive() {
+    test_util::timeout!(60);
+    for _ in 0..10 {
+      let (mut worker, mut receiver) = create_single_test_event_channel();
+      worker.stderr.write_all(b"hello").unwrap();
+      worker
+        .sender
+        .send(TestEvent::Result(0, TestResult::Ok, 0))
+        .unwrap();
+      drop(worker);
+      let (_, message) = receiver.recv().await.unwrap();
+      let TestEvent::Output(TestStdioStream::Stderr, text) = message else {
+        panic!("Incorrect message: {message:?}");
+      };
+      assert_eq!(text.as_slice(), b"hello");
+      let (_, message) = receiver.recv().await.unwrap();
+      let TestEvent::Result(..) = message else {
+        panic!("Incorrect message: {message:?}");
+      };
+      assert!(receiver.recv().await.is_none());
+    }
+  }
+
   /// Ensure nothing panics if we're racing the runtime shutdown.
   #[test]
   fn test_runtime_shutdown() {

cli/tools/test/mod.rs

@@ -391,8 +391,14 @@ pub enum TestEvent {
   StepRegister(TestStepDescription),
   StepWait(usize),
   StepResult(usize, TestStepResult, u64),
-  ForceEndReport,
+  /// Indicates that this worker has completed.
+  Completed,
+  /// Indicates that the user has cancelled the test run with Ctrl+C and
+  /// the run should be aborted.
   Sigint,
+  /// Used by the REPL to force a report to end without closing the worker
+  /// or receiver.
+  ForceEndReport,
 }
 
 impl TestEvent {
@@ -401,11 +407,13 @@ impl TestEvent {
   pub fn requires_stdio_sync(&self) -> bool {
     matches!(
       self,
-      TestEvent::Result(..)
+      TestEvent::Plan(..)
+        | TestEvent::Result(..)
         | TestEvent::StepWait(..)
         | TestEvent::StepResult(..)
         | TestEvent::UncaughtError(..)
         | TestEvent::ForceEndReport
+        | TestEvent::Completed
     )
   }
 }
@@ -641,24 +649,64 @@ pub async fn run_tests_for_worker(
   };
   let tests: Arc<TestDescriptions> = tests.into();
   sender.send(TestEvent::Register(tests.clone()))?;
+  let res = run_tests_for_worker_inner(
+    worker,
+    specifier,
+    tests,
+    test_functions,
+    &mut sender,
+    options,
+    fail_fast_tracker,
+  )
+  .await;
+  _ = sender.send(TestEvent::Completed);
+  res
+}
+
+async fn run_tests_for_worker_inner(
+  worker: &mut MainWorker,
+  specifier: &ModuleSpecifier,
+  tests: Arc<TestDescriptions>,
+  test_functions: Vec<v8::Global<v8::Function>>,
+  sender: &mut TestEventSender,
+  options: &TestSpecifierOptions,
+  fail_fast_tracker: &FailFastTracker,
+) -> Result<(), AnyError> {
   let unfiltered = tests.len();
-  let tests = tests
-    .into_iter()
-    .filter(|(_, d)| options.filter.includes(&d.name))
-    .collect::<Vec<_>>();
-  let (only, no_only): (Vec<_>, Vec<_>) =
-    tests.into_iter().partition(|(_, d)| d.only);
-  let used_only = !only.is_empty();
-  let mut tests = if used_only { only } else { no_only };
-  if let Some(seed) = options.shuffle {
-    tests.shuffle(&mut SmallRng::seed_from_u64(seed));
+
+  // Build the test plan in a single pass
+  let mut tests_to_run = Vec::with_capacity(tests.len());
+  let mut used_only = false;
+  for ((_, d), f) in tests.tests.iter().zip(test_functions) {
+    if !options.filter.includes(&d.name) {
+      continue;
+    }
+
+    // If we've seen an "only: true" test, the remaining tests must be "only: true" to be added
+    if used_only && !d.only {
+      continue;
+    }
+
+    // If this is the first "only: true" test we've seen, clear the other tests since they were
+    // only: false.
+    if d.only && !used_only {
+      used_only = true;
+      tests_to_run.clear();
+    }
+    tests_to_run.push((d, f));
   }
+
+  if let Some(seed) = options.shuffle {
+    tests_to_run.shuffle(&mut SmallRng::seed_from_u64(seed));
+  }
+
   sender.send(TestEvent::Plan(TestPlan {
     origin: specifier.to_string(),
-    total: tests.len(),
-    filtered_out: unfiltered - tests.len(),
+    total: tests_to_run.len(),
+    filtered_out: unfiltered - tests_to_run.len(),
     used_only,
   }))?;
+
   let mut had_uncaught_error = false;
   let stats = worker.js_runtime.runtime_activity_stats_factory();
   let ops = worker.js_runtime.op_names();
@@ -683,7 +731,7 @@ pub async fn run_tests_for_worker(
   filter = filter.omit_op(op_id_host_recv_ctrl as _);
   filter = filter.omit_op(op_id_host_recv_message as _);
 
-  for ((_, desc), function) in tests.into_iter().zip(test_functions) {
+  for (desc, function) in tests_to_run.into_iter() {
     if fail_fast_tracker.should_stop() {
       break;
     }
@@ -1254,6 +1302,9 @@ pub async fn report_tests(
       TestEvent::ForceEndReport => {
         break;
       }
+      TestEvent::Completed => {
+        reporter.report_completed();
+      }
       TestEvent::Sigint => {
         let elapsed = start_time
           .map(|t| Instant::now().duration_since(t))

cli/tools/test/reporters/compound.rs

@@ -101,6 +101,12 @@ impl TestReporter for CompoundTestReporter {
     }
   }
 
+  fn report_completed(&mut self) {
+    for reporter in &mut self.test_reporters {
+      reporter.report_completed();
+    }
+  }
+
   fn flush_report(
     &mut self,
     elapsed: &Duration,

cli/tools/test/reporters/dot.rs

@@ -206,6 +206,8 @@ impl TestReporter for DotTestReporter {
     );
   }
 
+  fn report_completed(&mut self) {}
+
   fn flush_report(
     &mut self,
     _elapsed: &Duration,

cli/tools/test/reporters/junit.rs

@@ -158,6 +158,11 @@ impl TestReporter for JunitTestReporter {
     }
   }
 
+  fn report_completed(&mut self) {
+    // TODO(mmastrac): This reporter does not handle stdout/stderr yet, and when we do, we may need to redirect
+    // pre-and-post-test output somewhere.
+  }
+
   fn flush_report(
     &mut self,
     elapsed: &Duration,

cli/tools/test/reporters/mod.rs

@@ -49,6 +49,7 @@ pub trait TestReporter {
     tests: &IndexMap<usize, TestDescription>,
     test_steps: &IndexMap<usize, TestStepDescription>,
   );
+  fn report_completed(&mut self);
   fn flush_report(
     &mut self,
     elapsed: &Duration,

cli/tools/test/reporters/pretty.rs

@@ -8,12 +8,14 @@ pub struct PrettyTestReporter {
   parallel: bool,
   echo_output: bool,
   in_new_line: bool,
+  phase: &'static str,
   filter: bool,
   repl: bool,
   scope_test_id: Option<usize>,
   cwd: Url,
   did_have_user_output: bool,
   started_tests: bool,
+  ended_tests: bool,
   child_results_buffer:
     HashMap<usize, IndexMap<usize, (TestStepDescription, TestStepResult, u64)>>,
   summary: TestSummary,
@@ -31,12 +33,14 @@ impl PrettyTestReporter {
       parallel,
       echo_output,
       in_new_line: true,
+      phase: "",
       filter,
       repl,
       scope_test_id: None,
       cwd: Url::from_directory_path(std::env::current_dir().unwrap()).unwrap(),
       did_have_user_output: false,
       started_tests: false,
+      ended_tests: false,
       child_results_buffer: Default::default(),
       summary: TestSummary::new(),
       writer: Box::new(std::io::stdout()),
@@ -149,7 +153,7 @@ impl PrettyTestReporter {
       writeln!(
         &mut self.writer,
         "{}",
-        colors::gray("----- output end -----")
+        colors::gray(format!("----- {}output end -----", self.phase))
       )
       .unwrap();
       self.in_new_line = true;
@@ -161,6 +165,7 @@ impl PrettyTestReporter {
 impl TestReporter for PrettyTestReporter {
   fn report_register(&mut self, _description: &TestDescription) {}
   fn report_plan(&mut self, plan: &TestPlan) {
+    self.write_output_end();
     self.summary.total += plan.total;
     self.summary.filtered_out += plan.filtered_out;
     if self.repl {
@@ -196,15 +201,22 @@ impl TestReporter for PrettyTestReporter {
       return;
     }
 
-    if !self.did_have_user_output && self.started_tests {
+    if !self.did_have_user_output {
       self.did_have_user_output = true;
       if !self.in_new_line {
         writeln!(&mut self.writer).unwrap();
       }
+      self.phase = if !self.started_tests {
+        "pre-test "
+      } else if self.ended_tests {
+        "post-test "
+      } else {
+        ""
+      };
       writeln!(
         &mut self.writer,
         "{}",
-        colors::gray("------- output -------")
+        colors::gray(format!("------- {}output -------", self.phase))
       )
       .unwrap();
       self.in_new_line = true;
@@ -369,6 +381,7 @@ impl TestReporter for PrettyTestReporter {
     _tests: &IndexMap<usize, TestDescription>,
     _test_steps: &IndexMap<usize, TestStepDescription>,
   ) {
+    self.write_output_end();
     common::report_summary(&mut self.writer, &self.cwd, &self.summary, elapsed);
     if !self.repl {
       writeln!(&mut self.writer).unwrap();
@@ -392,6 +405,11 @@ impl TestReporter for PrettyTestReporter {
     self.in_new_line = true;
   }
 
+  fn report_completed(&mut self) {
+    self.write_output_end();
+    self.ended_tests = true;
+  }
+
   fn flush_report(
     &mut self,
     _elapsed: &Duration,

cli/tools/test/reporters/tap.rs

@@ -236,6 +236,8 @@ impl TestReporter for TapTestReporter {
     );
   }
 
+  fn report_completed(&mut self) {}
+
   fn flush_report(
     &mut self,
     _elapsed: &Duration,

tests/integration/run_tests.rs

@@ -4560,15 +4560,14 @@ async fn websocket_server_multi_field_connection_header() {
   assert!(child.wait().unwrap().success());
 }
 
-// TODO(bartlomieju): this should use `deno run`, not `deno test`; but the
-// test hangs then. https://github.com/denoland/deno/issues/14283
 #[tokio::test]
 async fn websocket_server_idletimeout() {
+  test_util::timeout!(60);
   let script =
     util::testdata_path().join("run/websocket_server_idletimeout.ts");
   let root_ca = util::testdata_path().join("tls/RootCA.pem");
   let mut child = util::deno_cmd()
-    .arg("test")
+    .arg("run")
     .arg("--unstable")
     .arg("--allow-net")
     .arg("--cert")
@@ -4579,11 +4578,13 @@ async fn websocket_server_idletimeout() {
     .unwrap();
 
   let stdout = child.stdout.as_mut().unwrap();
-  let mut buffer = [0; 5];
-  let read = stdout.read(&mut buffer).unwrap();
-  assert_eq!(read, 5);
-  let msg = std::str::from_utf8(&buffer).unwrap();
-  assert_eq!(msg, "READY");
+  let mut buf: Vec<u8> = vec![];
+  while !String::from_utf8(buf.clone()).unwrap().contains("READY") {
+    let mut buffer = [0; 64];
+    let read = stdout.read(&mut buffer).unwrap();
+    buf.extend_from_slice(&buffer[0..read]);
+    eprintln!("buf = {buf:?}");
+  }
 
   let stream = tokio::net::TcpStream::connect("localhost:4509")
     .await
@@ -4604,8 +4605,7 @@ async fn websocket_server_idletimeout() {
     fastwebsockets::handshake::client(&SpawnExecutor, req, stream)
       .await
       .unwrap();
-
-  assert!(child.wait().unwrap().success());
+  assert_eq!(child.wait().unwrap().code(), Some(123));
 }
 
 itest!(auto_discover_lockfile {

tests/testdata/run/websocket_server_idletimeout.ts

@@ -11,10 +11,12 @@ const { response, socket } = Deno.upgradeWebSocket(request, {
   idleTimeout: 1,
 });
 socket.onerror = (e) => {
+  console.log(e);
   assertEquals((e as ErrorEvent).message, "No response from ping frame.");
   errorDeferred.resolve();
 };
 socket.onclose = (e) => {
+  console.log(e);
   assertEquals(e.reason, "No response from ping frame.");
   closeDeferred.resolve();
 };
@@ -22,4 +24,6 @@ await respondWith(response);
 
 await errorDeferred.promise;
 await closeDeferred.promise;
-listener.close();
+
+// TODO(mmastrac): this doesn't exit on its own. Why?
+Deno.exit(123);

tests/testdata/test/load_unload.out

@@ -1,6 +1,16 @@
-Check [WILDCARD]/test/load_unload.ts
-running 1 test from ./test/load_unload.ts
+Check [WILDCARD]/load_unload.ts
+------- pre-test output -------
+load
+----- pre-test output end -----
+running 1 test from [WILDCARD]/load_unload.ts
+test ...
+------- output -------
+test
+----- output end -----
 test ... ok ([WILDCARD])
+------- post-test output -------
+unload
+----- post-test output end -----
 
 ok | 1 passed | 0 failed ([WILDCARD])
 

tests/testdata/test/load_unload.ts

@@ -4,6 +4,7 @@ addEventListener("load", () => {
     throw new Error("Interval is already set");
   }
 
+  console.log("load");
   interval = setInterval(() => {}, 0);
 });
 
@@ -12,10 +13,12 @@ addEventListener("unload", () => {
     throw new Error("Interval was not set");
   }
 
+  console.log("unload");
   clearInterval(interval);
 });
 
 Deno.test("test", () => {
+  console.log("test");
   if (!interval) {
     throw new Error("Interval was not set");
   }

tests/util/server/src/macros.rs

@@ -30,7 +30,9 @@ macro_rules! timeout {
       let timeout = *timeout.get(0).unwrap_or(&300);
       ::std::thread::spawn(move || {
         if rx.recv_timeout(::std::time::Duration::from_secs(timeout)) == Err(::std::sync::mpsc::RecvTimeoutError::Timeout) {
+          use std::io::Write;
           eprintln!("Test {function} timed out after {timeout} seconds, aborting");
+          _ = std::io::stderr().flush();
           ::std::process::exit(1);
         }
       });