fix(test): capture worker stdout and stderr in test output (#14410)

10 files changed, 154 insertions, 109 deletions

diff --git a/cli/lsp/testing/execution.rs b/cli/lsp/testing/execution.rs
index 6f90e60a3..92964de87 100644
--- a/cli/lsp/testing/execution.rs
+++ b/cli/lsp/testing/execution.rs
@@ -14,7 +14,6 @@ use crate::lsp::client::TestingNotification;
 use crate::lsp::config;
 use crate::lsp::logging::lsp_log;
 use crate::ops;
-use crate::ops::testing::create_stdout_stderr_pipes;
 use crate::proc_state;
 use crate::tools::test;
 
@@ -27,6 +26,8 @@ use deno_core::parking_lot::Mutex;
 use deno_core::serde_json::json;
 use deno_core::serde_json::Value;
 use deno_core::ModuleSpecifier;
+use deno_runtime::ops::io::Stdio;
+use deno_runtime::ops::io::StdioPipe;
 use deno_runtime::permissions::Permissions;
 use deno_runtime::tokio_util::run_basic;
 use std::collections::HashMap;
@@ -184,17 +185,17 @@ async fn test_specifier(
   options: Option<Value>,
 ) -> Result<(), AnyError> {
   if !token.is_cancelled() {
-    let (stdout_writer, stderr_writer) =
-      create_stdout_stderr_pipes(channel.clone());
+    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(),
-        stdout_writer,
-        stderr_writer,
-      )],
+      vec![ops::testing::init(channel.clone())],
+      Stdio {
+        stdin: StdioPipe::Inherit,
+        stdout: StdioPipe::File(stdout),
+        stderr: StdioPipe::File(stderr),
+      },
     );
 
     worker
diff --git a/cli/main.rs b/cli/main.rs
index 218bc70f5..046d66a24 100644
--- a/cli/main.rs
+++ b/cli/main.rs
@@ -123,7 +123,10 @@ fn create_web_worker_preload_module_callback(
   })
 }
 
-fn create_web_worker_callback(ps: ProcState) -> Arc<CreateWebWorkerCb> {
+fn create_web_worker_callback(
+  ps: ProcState,
+  stdio: deno_runtime::ops::io::Stdio,
+) -> Arc<CreateWebWorkerCb> {
   Arc::new(move |args| {
     let maybe_inspector_server = ps.maybe_inspector_server.clone();
 
@@ -131,7 +134,8 @@ fn create_web_worker_callback(ps: ProcState) -> Arc<CreateWebWorkerCb> {
       ps.clone(),
       args.parent_permissions.clone(),
     );
-    let create_web_worker_cb = create_web_worker_callback(ps.clone());
+    let create_web_worker_cb =
+      create_web_worker_callback(ps.clone(), stdio.clone());
     let preload_module_cb =
       create_web_worker_preload_module_callback(ps.clone());
 
@@ -177,6 +181,7 @@ fn create_web_worker_callback(ps: ProcState) -> Arc<CreateWebWorkerCb> {
       shared_array_buffer_store: Some(ps.shared_array_buffer_store.clone()),
       compiled_wasm_module_store: Some(ps.compiled_wasm_module_store.clone()),
       maybe_exit_code: args.maybe_exit_code,
+      stdio: stdio.clone(),
     };
 
     WebWorker::bootstrap_from_options(
@@ -194,13 +199,15 @@ pub fn create_main_worker(
   main_module: ModuleSpecifier,
   permissions: Permissions,
   mut custom_extensions: Vec<Extension>,
+  stdio: deno_runtime::ops::io::Stdio,
 ) -> MainWorker {
   let module_loader = CliModuleLoader::new(ps.clone());
 
   let maybe_inspector_server = ps.maybe_inspector_server.clone();
   let should_break_on_first_statement = ps.flags.inspect_brk.is_some();
 
-  let create_web_worker_cb = create_web_worker_callback(ps.clone());
+  let create_web_worker_cb =
+    create_web_worker_callback(ps.clone(), stdio.clone());
   let web_worker_preload_module_cb =
     create_web_worker_preload_module_callback(ps.clone());
 
@@ -269,6 +276,7 @@ pub fn create_main_worker(
     broadcast_channel: ps.broadcast_channel.clone(),
     shared_array_buffer_store: Some(ps.shared_array_buffer_store.clone()),
     compiled_wasm_module_store: Some(ps.compiled_wasm_module_store.clone()),
+    stdio,
   };
 
   MainWorker::bootstrap_from_options(main_module, permissions, options)
@@ -510,8 +518,13 @@ async fn install_command(
     Permissions::from_options(&preload_flags.permissions_options());
   let ps = ProcState::build(Arc::new(preload_flags)).await?;
   let main_module = resolve_url_or_path(&install_flags.module_url)?;
-  let mut worker =
-    create_main_worker(&ps, main_module.clone(), permissions, vec![]);
+  let mut worker = create_main_worker(
+    &ps,
+    main_module.clone(),
+    permissions,
+    vec![],
+    Default::default(),
+  );
   // First, fetch and compile the module; this step ensures that the module exists.
   worker.preload_module(&main_module, true).await?;
   tools::installer::install(flags, install_flags)?;
@@ -605,8 +618,13 @@ async fn eval_command(
     resolve_url_or_path(&format!("./$deno$eval.{}", eval_flags.ext)).unwrap();
   let permissions = Permissions::from_options(&flags.permissions_options());
   let ps = ProcState::build(Arc::new(flags)).await?;
-  let mut worker =
-    create_main_worker(&ps, main_module.clone(), permissions, vec![]);
+  let mut worker = create_main_worker(
+    &ps,
+    main_module.clone(),
+    permissions,
+    vec![],
+    Default::default(),
+  );
   // Create a dummy source file.
   let source_code = if eval_flags.print {
     format!("console.log({})", eval_flags.code)
@@ -920,8 +938,13 @@ async fn repl_command(
   let main_module = resolve_url_or_path("./$deno$repl.ts").unwrap();
   let permissions = Permissions::from_options(&flags.permissions_options());
   let ps = ProcState::build(Arc::new(flags)).await?;
-  let mut worker =
-    create_main_worker(&ps, main_module.clone(), permissions, vec![]);
+  let mut worker = create_main_worker(
+    &ps,
+    main_module.clone(),
+    permissions,
+    vec![],
+    Default::default(),
+  );
   if ps.flags.compat {
     worker.execute_side_module(&compat::GLOBAL_URL).await?;
     compat::add_global_require(&mut worker.js_runtime, main_module.as_str())?;
@@ -937,8 +960,13 @@ async fn run_from_stdin(flags: Flags) -> Result<i32, AnyError> {
   let ps = ProcState::build(Arc::new(flags)).await?;
   let permissions = Permissions::from_options(&ps.flags.permissions_options());
   let main_module = resolve_url_or_path("./$deno$stdin.ts").unwrap();
-  let mut worker =
-    create_main_worker(&ps.clone(), main_module.clone(), permissions, vec![]);
+  let mut worker = create_main_worker(
+    &ps.clone(),
+    main_module.clone(),
+    permissions,
+    vec![],
+    Default::default(),
+  );
 
   let mut source = Vec::new();
   std::io::stdin().read_to_end(&mut source)?;
@@ -1125,7 +1153,13 @@ async fn run_with_watch(flags: Flags, script: String) -> Result<i32, AnyError> {
       // We make use an module executor guard to ensure that unload is always fired when an
       // operation is called.
       let mut executor = FileWatcherModuleExecutor::new(
-        create_main_worker(&ps, main_module.clone(), permissions, vec![]),
+        create_main_worker(
+          &ps,
+          main_module.clone(),
+          permissions,
+          vec![],
+          Default::default(),
+        ),
         flags.compat,
       );
 
@@ -1168,8 +1202,13 @@ async fn run_command(
   let main_module = resolve_url_or_path(&run_flags.script)?;
   let ps = ProcState::build(Arc::new(flags)).await?;
   let permissions = Permissions::from_options(&ps.flags.permissions_options());
-  let mut worker =
-    create_main_worker(&ps, main_module.clone(), permissions, vec![]);
+  let mut worker = create_main_worker(
+    &ps,
+    main_module.clone(),
+    permissions,
+    vec![],
+    Default::default(),
+  );
 
   let mut maybe_coverage_collector =
     if let Some(ref coverage_dir) = ps.coverage_dir {
diff --git a/cli/ops/testing.rs b/cli/ops/testing.rs
index 008e4d113..8bb16ccf3 100644
--- a/cli/ops/testing.rs
+++ b/cli/ops/testing.rs
@@ -1,31 +1,21 @@
-use std::cell::RefCell;
-use std::io::Read;
-use std::rc::Rc;
+// Copyright 2018-2022 the Deno authors. All rights reserved. MIT license.
 
 use crate::tools::test::TestEvent;
 use crate::tools::test::TestOutput;
+
 use deno_core::error::generic_error;
 use deno_core::error::AnyError;
 use deno_core::op;
 use deno_core::Extension;
 use deno_core::ModuleSpecifier;
 use deno_core::OpState;
-use deno_runtime::ops::io::StdFileResource;
 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>,
-  stdout_writer: os_pipe::PipeWriter,
-  stderr_writer: os_pipe::PipeWriter,
-) -> Extension {
-  // todo(dsheret): don't do this? Taking out the writers was necessary to prevent invalid handle panics
-  let stdout_writer = Rc::new(RefCell::new(Some(stdout_writer)));
-  let stderr_writer = Rc::new(RefCell::new(Some(stderr_writer)));
-
+pub fn init(sender: UnboundedSender<TestEvent>) -> Extension {
   Extension::builder()
     .ops(vec![
       op_pledge_test_permissions::decl(),
@@ -38,74 +28,12 @@ pub fn init(
       _ => op,
     })
     .state(move |state| {
-      state.resource_table.replace(
-        1,
-        StdFileResource::stdio(
-          &pipe_writer_to_file(&stdout_writer.borrow_mut().take().unwrap()),
-          "stdout",
-        ),
-      );
-      state.resource_table.replace(
-        2,
-        StdFileResource::stdio(
-          &pipe_writer_to_file(&stderr_writer.borrow_mut().take().unwrap()),
-          "stderr",
-        ),
-      );
       state.put(sender.clone());
       Ok(())
     })
     .build()
 }
 
-#[cfg(windows)]
-fn pipe_writer_to_file(writer: &os_pipe::PipeWriter) -> std::fs::File {
-  use std::os::windows::prelude::AsRawHandle;
-  use std::os::windows::prelude::FromRawHandle;
-  unsafe { std::fs::File::from_raw_handle(writer.as_raw_handle()) }
-}
-
-#[cfg(unix)]
-fn pipe_writer_to_file(writer: &os_pipe::PipeWriter) -> std::fs::File {
-  use std::os::unix::io::AsRawFd;
-  use std::os::unix::io::FromRawFd;
-  unsafe { std::fs::File::from_raw_fd(writer.as_raw_fd()) }
-}
-
-/// Creates the stdout and stderr pipes and returns the writers for stdout and stderr.
-pub fn create_stdout_stderr_pipes(
-  sender: UnboundedSender<TestEvent>,
-) -> (os_pipe::PipeWriter, os_pipe::PipeWriter) {
-  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);
-
-  (stdout_writer, stderr_writer)
-}
-
-fn start_output_redirect_thread(
-  mut pipe_reader: os_pipe::PipeReader,
-  sender: UnboundedSender<TestEvent>,
-) {
-  tokio::task::spawn_blocking(move || loop {
-    let mut buffer = [0; 512];
-    let size = match pipe_reader.read(&mut buffer) {
-      Ok(0) | Err(_) => break,
-      Ok(size) => size,
-    };
-    if sender
-      .send(TestEvent::Output(TestOutput::Bytes(
-        buffer[0..size].to_vec(),
-      )))
-      .is_err()
-    {
-      break;
-    }
-  });
-}
-
 #[derive(Clone)]
 struct PermissionsHolder(Uuid, Permissions);
 
diff --git a/cli/standalone.rs b/cli/standalone.rs
index 9f2aba9bd..a532872ca 100644
--- a/cli/standalone.rs
+++ b/cli/standalone.rs
@@ -305,6 +305,7 @@ pub async fn run(
     broadcast_channel,
     shared_array_buffer_store: None,
     compiled_wasm_module_store: None,
+    stdio: Default::default(),
   };
   let mut worker = MainWorker::bootstrap_from_options(
     main_module.clone(),
diff --git a/cli/tests/integration/test_tests.rs b/cli/tests/integration/test_tests.rs
index 1e8db52fd..328d9b494 100644
--- a/cli/tests/integration/test_tests.rs
+++ b/cli/tests/integration/test_tests.rs
@@ -302,10 +302,10 @@ itest!(no_prompt_with_denied_perms {
   output: "test/no_prompt_with_denied_perms.out",
 });
 
-itest!(captured_subprocess_output {
-  args: "test --allow-run --allow-read --unstable test/captured_subprocess_output.ts",
+itest!(captured_output {
+  args: "test --allow-run --allow-read --unstable test/captured_output.ts",
   exit_code: 0,
-  output: "test/captured_subprocess_output.out",
+  output: "test/captured_output.out",
 });
 
 #[test]
diff --git a/cli/tests/testdata/test/captured_subprocess_output.out b/cli/tests/testdata/test/captured_output.out
index 2a40170af..5ac367561 100644
--- a/cli/tests/testdata/test/captured_subprocess_output.out
+++ b/cli/tests/testdata/test/captured_output.out
@@ -1,5 +1,5 @@
 [WILDCARD]
-running 1 test from [WILDCARD]/captured_subprocess_output.ts
+running 1 test from [WILDCARD]/captured_output.ts
 output ...
 ------- output -------
 1
@@ -10,6 +10,8 @@ output ...
 6
 7
 8
+9
+10
 ----- output end -----
 ok ([WILDCARD]s)
 
diff --git a/cli/tests/testdata/test/captured_subprocess_output.ts b/cli/tests/testdata/test/captured_output.ts
index 277ac340e..3710c27b0 100644
--- a/cli/tests/testdata/test/captured_subprocess_output.ts
+++ b/cli/tests/testdata/test/captured_output.ts
@@ -20,4 +20,16 @@ Deno.test("output", async () => {
     stderr: "inherit",
   });
   await c.status;
+  const worker = new Worker(
+    new URL("./captured_output.worker.js", import.meta.url).href,
+    { type: "module" },
+  );
+
+  // ensure worker output is captured
+  const response = new Promise<void>((resolve) =>
+    worker.onmessage = () => resolve()
+  );
+  worker.postMessage({});
+  await response;
+  worker.terminate();
 });
diff --git a/cli/tests/testdata/test/captured_output.worker.js b/cli/tests/testdata/test/captured_output.worker.js
new file mode 100644
index 000000000..b674bce56
--- /dev/null
+++ b/cli/tests/testdata/test/captured_output.worker.js
@@ -0,0 +1,6 @@
+self.onmessage = () => {
+  console.log(9);
+  console.error(10);
+  self.postMessage({});
+  self.close();
+};
diff --git a/cli/tools/bench.rs b/cli/tools/bench.rs
index 3967abf2c..afd85a8d0 100644
--- a/cli/tools/bench.rs
+++ b/cli/tools/bench.rs
@@ -385,6 +385,7 @@ async fn bench_specifier(
     specifier.clone(),
     permissions,
     vec![ops::bench::init(channel.clone(), ps.flags.unstable)],
+    Default::default(),
   );
 
   if options.compat_mode {
diff --git a/cli/tools/test.rs b/cli/tools/test.rs
index d7817eb1a..5eb3552ec 100644
--- a/cli/tools/test.rs
+++ b/cli/tools/test.rs
@@ -22,7 +22,6 @@ use crate::graph_util::graph_valid;
 use crate::located_script_name;
 use crate::lockfile;
 use crate::ops;
-use crate::ops::testing::create_stdout_stderr_pipes;
 use crate::proc_state::ProcState;
 use crate::resolver::ImportMapResolver;
 use crate::resolver::JsxResolver;
@@ -41,6 +40,8 @@ use deno_core::serde_json::json;
 use deno_core::url::Url;
 use deno_core::ModuleSpecifier;
 use deno_graph::ModuleKind;
+use deno_runtime::ops::io::Stdio;
+use deno_runtime::ops::io::StdioPipe;
 use deno_runtime::permissions::Permissions;
 use deno_runtime::tokio_util::run_basic;
 use log::Level;
@@ -52,6 +53,7 @@ use serde::Deserialize;
 use std::collections::BTreeMap;
 use std::collections::HashMap;
 use std::collections::HashSet;
+use std::io::Read;
 use std::io::Write;
 use std::num::NonZeroUsize;
 use std::path::PathBuf;
@@ -588,17 +590,17 @@ async fn test_specifier(
   channel: UnboundedSender<TestEvent>,
   options: TestSpecifierOptions,
 ) -> Result<(), AnyError> {
-  let (stdout_writer, stderr_writer) =
-    create_stdout_stderr_pipes(channel.clone());
+  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(),
-      stdout_writer,
-      stderr_writer,
-    )],
+    vec![ops::testing::init(channel.clone())],
+    Stdio {
+      stdin: StdioPipe::Inherit,
+      stdout: StdioPipe::File(stdout),
+      stderr: StdioPipe::File(stderr),
+    },
   );
 
   let mut maybe_coverage_collector = if let Some(ref coverage_dir) =
@@ -1452,3 +1454,56 @@ pub async fn run_tests_with_watch(
 
   Ok(())
 }
+
+/// Creates the stdout and stderr pipes and returns the writers for stdout and stderr.
+pub fn create_stdout_stderr_pipes(
+  sender: UnboundedSender<TestEvent>,
+) -> (std::fs::File, std::fs::File) {
+  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);
+
+  (
+    pipe_writer_to_file(stdout_writer),
+    pipe_writer_to_file(stderr_writer),
+  )
+}
+
+#[cfg(windows)]
+fn pipe_writer_to_file(writer: os_pipe::PipeWriter) -> std::fs::File {
+  use std::os::windows::prelude::FromRawHandle;
+  use std::os::windows::prelude::IntoRawHandle;
+  // SAFETY: Requires consuming ownership of the provided handle
+  unsafe { std::fs::File::from_raw_handle(writer.into_raw_handle()) }
+}
+
+#[cfg(unix)]
+fn pipe_writer_to_file(writer: os_pipe::PipeWriter) -> std::fs::File {
+  use std::os::unix::io::FromRawFd;
+  use std::os::unix::io::IntoRawFd;
+  // SAFETY: Requires consuming ownership of the provided handle
+  unsafe { std::fs::File::from_raw_fd(writer.into_raw_fd()) }
+}
+
+fn start_output_redirect_thread(
+  mut pipe_reader: os_pipe::PipeReader,
+  sender: UnboundedSender<TestEvent>,
+) {
+  tokio::task::spawn_blocking(move || loop {
+    let mut buffer = [0; 512];
+    let size = match pipe_reader.read(&mut buffer) {
+      Ok(0) | Err(_) => break,
+      Ok(size) => size,
+    };
+    if sender
+      .send(TestEvent::Output(TestOutput::Bytes(
+        buffer[0..size].to_vec(),
+      )))
+      .is_err()
+    {
+      break;
+    }
+  });
+}