refactor(test): support custom writer in PrettyTestReporter (#20783)

1 files changed, 158 insertions, 135 deletions

diff --git a/cli/tools/test/mod.rs b/cli/tools/test/mod.rs
index bf466579f..66e3a5870 100644
--- a/cli/tools/test/mod.rs
+++ b/cli/tools/test/mod.rs
@@ -50,6 +50,7 @@ use deno_runtime::fmt_errors::format_js_error;
 use deno_runtime::permissions::Permissions;
 use deno_runtime::permissions::PermissionsContainer;
 use deno_runtime::tokio_util::create_and_run_current_thread;
+use deno_runtime::worker::MainWorker;
 use indexmap::IndexMap;
 use indexmap::IndexSet;
 use log::Level;
@@ -77,11 +78,12 @@ use std::time::Instant;
 use std::time::SystemTime;
 use tokio::signal;
 use tokio::sync::mpsc::unbounded_channel;
+use tokio::sync::mpsc::UnboundedReceiver;
 use tokio::sync::mpsc::UnboundedSender;
 use tokio::sync::mpsc::WeakUnboundedSender;
 
 pub mod fmt;
-mod reporters;
+pub mod reporters;
 
 pub use fmt::format_test_error;
 use reporters::CompoundTestReporter;
@@ -313,6 +315,7 @@ pub enum TestEvent {
   StepRegister(TestStepDescription),
   StepWait(usize),
   StepResult(usize, TestStepResult, u64),
+  ForceEndReport,
   Sigint,
 }
 
@@ -342,7 +345,7 @@ struct TestSpecifiersOptions {
   junit_path: Option<String>,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Default, Clone)]
 pub struct TestSpecifierOptions {
   pub shuffle: Option<u64>,
   pub filter: TestFilter,
@@ -379,6 +382,7 @@ fn get_test_reporter(options: &TestSpecifiersOptions) -> Box<dyn TestReporter> {
       parallel,
       options.log_level != Some(Level::Error),
       options.filter,
+      false,
     )),
     TestReporterConfig::Junit => {
       Box::new(JunitTestReporter::new("-".to_string()))
@@ -453,10 +457,35 @@ pub async fn test_specifier(
 
   worker.dispatch_load_event(located_script_name!())?;
 
-  let tests = {
+  run_tests_for_worker(&mut worker, &specifier, &options, &fail_fast_tracker)
+    .await?;
+
+  // Ignore `defaultPrevented` of the `beforeunload` event. We don't allow the
+  // event loop to continue beyond what's needed to await results.
+  worker.dispatch_beforeunload_event(located_script_name!())?;
+  worker.dispatch_unload_event(located_script_name!())?;
+
+  if let Some(coverage_collector) = coverage_collector.as_mut() {
+    worker
+      .with_event_loop(coverage_collector.stop_collecting().boxed_local())
+      .await?;
+  }
+  Ok(())
+}
+
+pub async fn run_tests_for_worker(
+  worker: &mut MainWorker,
+  specifier: &ModuleSpecifier,
+  options: &TestSpecifierOptions,
+  fail_fast_tracker: &FailFastTracker,
+) -> Result<(), AnyError> {
+  let (tests, mut sender) = {
     let state_rc = worker.js_runtime.op_state();
     let mut state = state_rc.borrow_mut();
-    std::mem::take(&mut state.borrow_mut::<ops::testing::TestContainer>().0)
+    (
+      std::mem::take(&mut state.borrow_mut::<ops::testing::TestContainer>().0),
+      state.borrow::<TestEventSender>().clone(),
+    )
   };
   let unfiltered = tests.len();
   let tests = tests
@@ -532,17 +561,6 @@ pub async fn test_specifier(
     let elapsed = SystemTime::now().duration_since(earlier)?.as_millis();
     sender.send(TestEvent::Result(desc.id, result, elapsed as u64))?;
   }
-
-  // Ignore `defaultPrevented` of the `beforeunload` event. We don't allow the
-  // event loop to continue beyond what's needed to await results.
-  worker.dispatch_beforeunload_event(located_script_name!())?;
-  worker.dispatch_unload_event(located_script_name!())?;
-
-  if let Some(coverage_collector) = coverage_collector.as_mut() {
-    worker
-      .with_event_loop(coverage_collector.stop_collecting().boxed_local())
-      .await?;
-  }
   Ok(())
 }
 
@@ -810,7 +828,7 @@ async fn test_specifiers(
     specifiers
   };
 
-  let (sender, mut receiver) = unbounded_channel::<TestEvent>();
+  let (sender, receiver) = unbounded_channel::<TestEvent>();
   let sender = TestEventSender::new(sender);
   let concurrent_jobs = options.concurrent_jobs;
 
@@ -820,7 +838,7 @@ async fn test_specifiers(
     sender_.upgrade().map(|s| s.send(TestEvent::Sigint).ok());
   });
   HAS_TEST_RUN_SIGINT_HANDLER.store(true, Ordering::Relaxed);
-  let mut reporter = get_test_reporter(&options);
+  let reporter = get_test_reporter(&options);
   let fail_fast_tracker = FailFastTracker::new(options.fail_fast);
 
   let join_handles = specifiers.into_iter().map(move |specifier| {
@@ -840,142 +858,147 @@ async fn test_specifiers(
       ))
     })
   });
-
   let join_stream = stream::iter(join_handles)
     .buffer_unordered(concurrent_jobs.get())
     .collect::<Vec<Result<Result<(), AnyError>, tokio::task::JoinError>>>();
 
-  let handler = {
-    spawn(async move {
-      let earlier = Instant::now();
-      let mut tests = IndexMap::new();
-      let mut test_steps = IndexMap::new();
-      let mut tests_started = HashSet::new();
-      let mut tests_with_result = HashSet::new();
-      let mut used_only = false;
-      let mut failed = false;
-
-      while let Some(event) = receiver.recv().await {
-        match event {
-          TestEvent::Register(description) => {
-            reporter.report_register(&description);
-            tests.insert(description.id, description);
-          }
-
-          TestEvent::Plan(plan) => {
-            if plan.used_only {
-              used_only = true;
-            }
-
-            reporter.report_plan(&plan);
-          }
-
-          TestEvent::Wait(id) => {
-            if tests_started.insert(id) {
-              reporter.report_wait(tests.get(&id).unwrap());
-            }
-          }
-
-          TestEvent::Output(output) => {
-            reporter.report_output(&output);
-          }
-
-          TestEvent::Result(id, result, elapsed) => {
-            if tests_with_result.insert(id) {
-              match result {
-                TestResult::Failed(_) | TestResult::Cancelled => {
-                  failed = true;
-                }
-                _ => (),
-              }
-              reporter.report_result(tests.get(&id).unwrap(), &result, elapsed);
-            }
-          }
-
-          TestEvent::UncaughtError(origin, error) => {
-            failed = true;
-            reporter.report_uncaught_error(&origin, error);
-          }
-
-          TestEvent::StepRegister(description) => {
-            reporter.report_step_register(&description);
-            test_steps.insert(description.id, description);
-          }
+  let handler = spawn(async move { report_tests(receiver, reporter).await.0 });
 
-          TestEvent::StepWait(id) => {
-            if tests_started.insert(id) {
-              reporter.report_step_wait(test_steps.get(&id).unwrap());
-            }
-          }
+  let (join_results, result) = future::join(join_stream, handler).await;
+  sigint_handler_handle.abort();
+  HAS_TEST_RUN_SIGINT_HANDLER.store(false, Ordering::Relaxed);
+  for join_result in join_results {
+    join_result??;
+  }
+  result??;
 
-          TestEvent::StepResult(id, result, duration) => {
-            if tests_with_result.insert(id) {
-              reporter.report_step_result(
-                test_steps.get(&id).unwrap(),
-                &result,
-                duration,
-                &tests,
-                &test_steps,
-              );
-            }
-          }
+  Ok(())
+}
 
-          TestEvent::Sigint => {
-            let elapsed = Instant::now().duration_since(earlier);
-            reporter.report_sigint(
-              &tests_started
-                .difference(&tests_with_result)
-                .copied()
-                .collect(),
-              &tests,
-              &test_steps,
-            );
-            if let Err(err) =
-              reporter.flush_report(&elapsed, &tests, &test_steps)
-            {
-              eprint!("Test reporter failed to flush: {}", err)
+/// Gives receiver back in case it was ended with `TestEvent::ForceEndReport`.
+pub async fn report_tests(
+  mut receiver: UnboundedReceiver<TestEvent>,
+  mut reporter: Box<dyn TestReporter>,
+) -> (Result<(), AnyError>, UnboundedReceiver<TestEvent>) {
+  let mut tests = IndexMap::new();
+  let mut test_steps = IndexMap::new();
+  let mut tests_started = HashSet::new();
+  let mut tests_with_result = HashSet::new();
+  let mut start_time = None;
+  let mut had_plan = false;
+  let mut used_only = false;
+  let mut failed = false;
+
+  while let Some(event) = receiver.recv().await {
+    match event {
+      TestEvent::Register(description) => {
+        reporter.report_register(&description);
+        tests.insert(description.id, description);
+      }
+      TestEvent::Plan(plan) => {
+        if !had_plan {
+          start_time = Some(Instant::now());
+          had_plan = true;
+        }
+        if plan.used_only {
+          used_only = true;
+        }
+        reporter.report_plan(&plan);
+      }
+      TestEvent::Wait(id) => {
+        if tests_started.insert(id) {
+          reporter.report_wait(tests.get(&id).unwrap());
+        }
+      }
+      TestEvent::Output(output) => {
+        reporter.report_output(&output);
+      }
+      TestEvent::Result(id, result, elapsed) => {
+        if tests_with_result.insert(id) {
+          match result {
+            TestResult::Failed(_) | TestResult::Cancelled => {
+              failed = true;
             }
-            std::process::exit(130);
+            _ => (),
           }
+          reporter.report_result(tests.get(&id).unwrap(), &result, elapsed);
         }
       }
-
-      sigint_handler_handle.abort();
-      HAS_TEST_RUN_SIGINT_HANDLER.store(false, Ordering::Relaxed);
-
-      let elapsed = Instant::now().duration_since(earlier);
-      reporter.report_summary(&elapsed, &tests, &test_steps);
-      if let Err(err) = reporter.flush_report(&elapsed, &tests, &test_steps) {
-        return Err(generic_error(format!(
-          "Test reporter failed to flush: {}",
-          err
-        )));
+      TestEvent::UncaughtError(origin, error) => {
+        failed = true;
+        reporter.report_uncaught_error(&origin, error);
       }
-
-      if used_only {
-        return Err(generic_error(
-          "Test failed because the \"only\" option was used",
-        ));
+      TestEvent::StepRegister(description) => {
+        reporter.report_step_register(&description);
+        test_steps.insert(description.id, description);
       }
-
-      if failed {
-        return Err(generic_error("Test failed"));
+      TestEvent::StepWait(id) => {
+        if tests_started.insert(id) {
+          reporter.report_step_wait(test_steps.get(&id).unwrap());
+        }
       }
+      TestEvent::StepResult(id, result, duration) => {
+        if tests_with_result.insert(id) {
+          reporter.report_step_result(
+            test_steps.get(&id).unwrap(),
+            &result,
+            duration,
+            &tests,
+            &test_steps,
+          );
+        }
+      }
+      TestEvent::ForceEndReport => {
+        break;
+      }
+      TestEvent::Sigint => {
+        let elapsed = start_time
+          .map(|t| Instant::now().duration_since(t))
+          .unwrap_or_default();
+        reporter.report_sigint(
+          &tests_started
+            .difference(&tests_with_result)
+            .copied()
+            .collect(),
+          &tests,
+          &test_steps,
+        );
+        if let Err(err) = reporter.flush_report(&elapsed, &tests, &test_steps) {
+          eprint!("Test reporter failed to flush: {}", err)
+        }
+        std::process::exit(130);
+      }
+    }
+  }
 
-      Ok(())
-    })
-  };
-
-  let (join_results, result) = future::join(join_stream, handler).await;
+  let elapsed = start_time
+    .map(|t| Instant::now().duration_since(t))
+    .unwrap_or_default();
+  reporter.report_summary(&elapsed, &tests, &test_steps);
+  if let Err(err) = reporter.flush_report(&elapsed, &tests, &test_steps) {
+    return (
+      Err(generic_error(format!(
+        "Test reporter failed to flush: {}",
+        err
+      ))),
+      receiver,
+    );
+  }
 
-  // propagate any errors
-  for join_result in join_results {
-    join_result??;
+  if used_only {
+    return (
+      Err(generic_error(
+        "Test failed because the \"only\" option was used",
+      )),
+      receiver,
+    );
   }
 
-  result??;
+  if failed {
+    return (Err(generic_error("Test failed")), receiver);
+  }
 
-  Ok(())
+  (Ok(()), receiver)
 }
 
 /// Checks if the path has a basename and extension Deno supports for tests.
@@ -1300,7 +1323,7 @@ pub async fn run_tests_with_watch(
 
 /// Tracks failures for the `--fail-fast` argument in
 /// order to tell when to stop running tests.
-#[derive(Clone)]
+#[derive(Clone, Default)]
 pub struct FailFastTracker {
   max_count: Option<usize>,
   failure_count: Arc<AtomicUsize>,