fix(test): actually capture stdout and stderr in workers (#14435)

cli/lsp/testing/execution.rs

@@ -16,6 +16,7 @@ use crate::lsp::logging::lsp_log;
 use crate::ops;
 use crate::proc_state;
 use crate::tools::test;
+use crate::tools::test::TestEventSender;
 
 use deno_core::anyhow::anyhow;
 use deno_core::error::AnyError;
@@ -180,21 +181,20 @@ async fn test_specifier(
   permissions: Permissions,
   specifier: ModuleSpecifier,
   mode: test::TestMode,
-  channel: mpsc::UnboundedSender<test::TestEvent>,
+  sender: TestEventSender,
   token: CancellationToken,
   options: Option<Value>,
 ) -> Result<(), AnyError> {
   if !token.is_cancelled() {
-    let (stdout, stderr) = test::create_stdout_stderr_pipes(channel.clone());
     let mut worker = create_main_worker(
       &ps,
       specifier.clone(),
       permissions,
-      vec![ops::testing::init(channel.clone())],
+      vec![ops::testing::init(sender.clone())],
       Stdio {
         stdin: StdioPipe::Inherit,
-        stdout: StdioPipe::File(stdout),
+        stdout: StdioPipe::File(sender.stdout()),
-        stderr: StdioPipe::File(stderr),
+        stderr: StdioPipe::File(sender.stderr()),
       },
     );
 
@@ -318,6 +318,7 @@ impl TestRun {
     .await?;
 
     let (sender, mut receiver) = mpsc::unbounded_channel::<test::TestEvent>();
+    let sender = TestEventSender::new(sender);
 
     let (concurrent_jobs, fail_fast) =
       if let flags::DenoSubcommand::Test(test_flags) = &ps.flags.subcommand {
@@ -754,19 +755,14 @@ impl test::TestReporter for LspTestReporter {
     self.progress(lsp_custom::TestRunProgressMessage::Started { test });
   }
 
-  fn report_output(&mut self, output: &test::TestOutput) {
+  fn report_output(&mut self, output: &[u8]) {
     let test = self.current_origin.as_ref().and_then(|origin| {
       self
         .stack
         .get(origin)
         .and_then(|v| v.last().map(|td| td.into()))
     });
-    let value = match output {
+    let value = String::from_utf8_lossy(output).replace('\n', "\r\n");
-      test::TestOutput::String(value) => value.replace('\n', "\r\n"),
-      test::TestOutput::Bytes(bytes) => {
-        String::from_utf8_lossy(bytes).replace('\n', "\r\n")
-      }
-    };
 
     self.progress(lsp_custom::TestRunProgressMessage::Output {
       value,

cli/ops/testing.rs

@@ -1,7 +1,7 @@
 // Copyright 2018-2022 the Deno authors. All rights reserved. MIT license.
 
 use crate::tools::test::TestEvent;
-use crate::tools::test::TestOutput;
+use crate::tools::test::TestEventSender;
 
 use deno_core::error::generic_error;
 use deno_core::error::AnyError;
@@ -12,10 +12,9 @@ use deno_core::OpState;
 use deno_runtime::permissions::create_child_permissions;
 use deno_runtime::permissions::ChildPermissionsArg;
 use deno_runtime::permissions::Permissions;
-use tokio::sync::mpsc::UnboundedSender;
 use uuid::Uuid;
 
-pub fn init(sender: UnboundedSender<TestEvent>) -> Extension {
+pub fn init(sender: TestEventSender) -> Extension {
   Extension::builder()
     .ops(vec![
       op_pledge_test_permissions::decl(),
@@ -23,10 +22,6 @@ pub fn init(sender: UnboundedSender<TestEvent>) -> Extension {
       op_get_test_origin::decl(),
       op_dispatch_test_event::decl(),
     ])
-    .middleware(|op| match op.name {
-      "op_print" => op_print::decl(),
-      _ => op,
-    })
     .state(move |state| {
       state.put(sender.clone());
       Ok(())
@@ -86,19 +81,7 @@ fn op_dispatch_test_event(
   state: &mut OpState,
   event: TestEvent,
 ) -> Result<(), AnyError> {
-  let sender = state.borrow::<UnboundedSender<TestEvent>>().clone();
+  let mut sender = state.borrow::<TestEventSender>().clone();
   sender.send(event).ok();
   Ok(())
 }
-
-#[op]
-pub fn op_print(
-  state: &mut OpState,
-  msg: String,
-  _is_err: bool,
-) -> Result<(), AnyError> {
-  let sender = state.borrow::<UnboundedSender<TestEvent>>().clone();
-  let msg = TestOutput::String(msg);
-  sender.send(TestEvent::Output(msg)).ok();
-  Ok(())
-}

cli/tests/integration/test_tests.rs

@@ -302,11 +302,37 @@ itest!(no_prompt_with_denied_perms {
   output: "test/no_prompt_with_denied_perms.out",
 });
 
-itest!(captured_output {
+#[test]
-  args: "test --allow-run --allow-read --unstable test/captured_output.ts",
+fn captured_output() {
-  exit_code: 0,
+  let output = util::deno_cmd()
-  output: "test/captured_output.out",
+    .current_dir(util::testdata_path())
-});
+    .arg("test")
+    .arg("--allow-run")
+    .arg("--allow-read")
+    .arg("--unstable")
+    .arg("test/captured_output.ts")
+    .env("NO_COLOR", "1")
+    .stdout(std::process::Stdio::piped())
+    .spawn()
+    .unwrap()
+    .wait_with_output()
+    .unwrap();
+
+  let output_start = "------- output -------";
+  let output_end = "----- output end -----";
+  assert!(output.status.success());
+  let output_text = String::from_utf8(output.stdout).unwrap();
+  let start = output_text.find(output_start).unwrap() + output_start.len();
+  let end = output_text.find(output_end).unwrap();
+  let output_text = output_text[start..end].trim();
+  let mut lines = output_text.lines().collect::<Vec<_>>();
+  // the output is racy on either stdout or stderr being flushed
+  // from the runtime into the rust code, so sort it... the main
+  // thing here to ensure is that we're capturing the output in
+  // this block on stdout
+  lines.sort_unstable();
+  assert_eq!(lines.join(" "), "0 1 2 3 4 5 6 7 8 9");
+}
 
 #[test]
 fn recursive_permissions_pledge() {

cli/tests/testdata/test/captured_output.out

@@ -1,19 +0,0 @@
-[WILDCARD]
-running 1 test from [WILDCARD]/captured_output.ts
-output ...
-------- output -------
-1
-2
-3
-4
-5
-6
-7
-8
-9
-10
------ output end -----
-ok ([WILDCARD]s)
-
-test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out ([WILDCARD]s)
-[WILDCARD]

cli/tests/testdata/test/captured_output.ts

@@ -1,21 +1,21 @@
 Deno.test("output", async () => {
   const p = Deno.run({
-    cmd: [Deno.execPath(), "eval", "console.log(1); console.error(2);"],
+    cmd: [Deno.execPath(), "eval", "console.log(0); console.error(1);"],
   });
   await p.status();
   await p.close();
   Deno.spawnSync(Deno.execPath(), {
-    args: ["eval", "console.log(3); console.error(4);"],
+    args: ["eval", "console.log(2); console.error(3);"],
     stdout: "inherit",
     stderr: "inherit",
   });
   await Deno.spawn(Deno.execPath(), {
-    args: ["eval", "console.log(5); console.error(6);"],
+    args: ["eval", "console.log(4); console.error(5);"],
     stdout: "inherit",
     stderr: "inherit",
   });
   const c = await Deno.spawnChild(Deno.execPath(), {
-    args: ["eval", "console.log(7); console.error(8);"],
+    args: ["eval", "console.log(6); console.error(7);"],
     stdout: "inherit",
     stderr: "inherit",
   });

cli/tests/testdata/test/captured_output.worker.js

@@ -1,6 +1,6 @@
 self.onmessage = () => {
-  console.log(9);
+  console.log(8);
-  console.error(10);
+  console.error(9);
   self.postMessage({});
   self.close();
 };

cli/tools/test.rs

@@ -138,7 +138,7 @@ pub struct TestPlan {
 pub enum TestEvent {
   Plan(TestPlan),
   Wait(TestDescription),
-  Output(TestOutput),
+  Output(Vec<u8>),
   Result(TestDescription, TestResult, u64),
   StepWait(TestStepDescription),
   StepResult(TestStepDescription, TestStepResult, u64),
@@ -198,7 +198,7 @@ impl TestSummary {
 pub trait TestReporter {
   fn report_plan(&mut self, plan: &TestPlan);
   fn report_wait(&mut self, description: &TestDescription);
-  fn report_output(&mut self, output: &TestOutput);
+  fn report_output(&mut self, output: &[u8]);
   fn report_result(
     &mut self,
     description: &TestDescription,
@@ -340,7 +340,7 @@ impl TestReporter for PrettyTestReporter {
     self.in_test_count += 1;
   }
 
-  fn report_output(&mut self, output: &TestOutput) {
+  fn report_output(&mut self, output: &[u8]) {
     if !self.echo_output {
       return;
     }
@@ -350,16 +350,10 @@ impl TestReporter for PrettyTestReporter {
       println!();
       println!("{}", colors::gray("------- output -------"));
     }
-    match output {
+
-      TestOutput::String(line) => {
     // output everything to stdout in order to prevent
     // stdout and stderr racing
-        print!("{}", line)
+    std::io::stdout().write_all(output).unwrap();
-      }
-      TestOutput::Bytes(bytes) => {
-        std::io::stdout().write_all(bytes).unwrap();
-      }
-    }
   }
 
   fn report_result(
@@ -587,19 +581,18 @@ async fn test_specifier(
   permissions: Permissions,
   specifier: ModuleSpecifier,
   mode: TestMode,
-  channel: UnboundedSender<TestEvent>,
+  sender: TestEventSender,
   options: TestSpecifierOptions,
 ) -> Result<(), AnyError> {
-  let (stdout, stderr) = create_stdout_stderr_pipes(channel.clone());
   let mut worker = create_main_worker(
     &ps,
     specifier.clone(),
     permissions,
-    vec![ops::testing::init(channel.clone())],
+    vec![ops::testing::init(sender.clone())],
     Stdio {
       stdin: StdioPipe::Inherit,
-      stdout: StdioPipe::File(stdout),
+      stdout: StdioPipe::File(sender.stdout()),
-      stderr: StdioPipe::File(stderr),
+      stderr: StdioPipe::File(sender.stderr()),
     },
   );
 
@@ -951,6 +944,7 @@ async fn test_specifiers(
   };
 
   let (sender, mut receiver) = unbounded_channel::<TestEvent>();
+  let sender = TestEventSender::new(sender);
   let concurrent_jobs = options.concurrent_jobs;
   let fail_fast = options.fail_fast;
 
@@ -1455,20 +1449,61 @@ pub async fn run_tests_with_watch(
   Ok(())
 }
 
-/// Creates the stdout and stderr pipes and returns the writers for stdout and stderr.
+pub struct TestEventSender {
-pub fn create_stdout_stderr_pipes(
   sender: UnboundedSender<TestEvent>,
-) -> (std::fs::File, std::fs::File) {
+  stdout_writer: os_pipe::PipeWriter,
+  stderr_writer: os_pipe::PipeWriter,
+}
+
+impl Clone for TestEventSender {
+  fn clone(&self) -> Self {
+    Self {
+      sender: self.sender.clone(),
+      stdout_writer: self.stdout_writer.try_clone().unwrap(),
+      stderr_writer: self.stderr_writer.try_clone().unwrap(),
+    }
+  }
+}
+
+impl TestEventSender {
+  pub fn new(sender: UnboundedSender<TestEvent>) -> Self {
     let (stdout_reader, stdout_writer) = os_pipe::pipe().unwrap();
     let (stderr_reader, stderr_writer) = os_pipe::pipe().unwrap();
 
     start_output_redirect_thread(stdout_reader, sender.clone());
-  start_output_redirect_thread(stderr_reader, sender);
+    start_output_redirect_thread(stderr_reader, sender.clone());
 
-  (
+    Self {
-    pipe_writer_to_file(stdout_writer),
+      sender,
-    pipe_writer_to_file(stderr_writer),
+      stdout_writer,
-  )
+      stderr_writer,
+    }
+  }
+
+  pub fn stdout(&self) -> std::fs::File {
+    pipe_writer_to_file(self.stdout_writer.try_clone().unwrap())
+  }
+
+  pub fn stderr(&self) -> std::fs::File {
+    pipe_writer_to_file(self.stderr_writer.try_clone().unwrap())
+  }
+
+  pub fn send(&mut self, message: TestEvent) -> Result<(), AnyError> {
+    // for any event that finishes collecting output, we need to
+    // ensure that the collected stdout and stderr pipes are flushed
+    if matches!(
+      message,
+      TestEvent::Result(_, _, _)
+        | TestEvent::StepWait(_)
+        | TestEvent::StepResult(_, _, _)
+    ) {
+      self.stdout_writer.flush().unwrap();
+      self.stderr_writer.flush().unwrap();
+    }
+
+    self.sender.send(message)?;
+    Ok(())
+  }
 }
 
 #[cfg(windows)]
@@ -1497,10 +1532,9 @@ fn start_output_redirect_thread(
       Ok(0) | Err(_) => break,
       Ok(size) => size,
     };
+
     if sender
-      .send(TestEvent::Output(TestOutput::Bytes(
+      .send(TestEvent::Output(buffer[0..size].to_vec()))
-        buffer[0..size].to_vec(),
-      )))
       .is_err()
     {
       break;

runtime/ops/io.rs

@@ -111,6 +111,10 @@ pub fn init_stdio(stdio: Stdio) -> Extension {
   let stdio = Rc::new(RefCell::new(Some(stdio)));
 
   Extension::builder()
+    .middleware(|op| match op.name {
+      "op_print" => op_print::decl(),
+      _ => op,
+    })
     .state(move |state| {
       let stdio = stdio
         .borrow_mut()
@@ -419,6 +423,24 @@ impl Resource for StdFileResource {
   }
 }
 
+// override op_print to use the stdout and stderr in the resource table
+#[op]
+pub fn op_print(
+  state: &mut OpState,
+  msg: String,
+  is_err: bool,
+) -> Result<(), AnyError> {
+  let rid = if is_err { 2 } else { 1 };
+  StdFileResource::with(state, rid, move |r| match r {
+    Ok(std_file) => {
+      std_file.write_all(msg.as_bytes())?;
+      std_file.flush().unwrap();
+      Ok(())
+    }
+    Err(_) => Err(not_supported()),
+  })
+}
+
 #[op]
 fn op_read_sync(
   state: &mut OpState,