feat(test): run test modules in parallel (#9815)

This commit adds support for running test in parallel.

Entire test runner functionality has been rewritten
from JavaScript to Rust and a set of ops was added to support reporting in Rust.

A new "--jobs" flag was added to "deno test" that allows to configure 
how many threads will be used. When given no value it defaults to 2.

cli/flags.rs

@@ -102,6 +102,7 @@ pub enum DenoSubcommand {
     allow_none: bool,
     include: Option<Vec<String>>,
     filter: Option<String>,
+    concurrent_jobs: usize,
   },
   Types,
   Upgrade {
@@ -1012,6 +1013,18 @@ fn test_subcommand<'a, 'b>() -> App<'a, 'b> {
         .conflicts_with("inspect-brk")
         .help("UNSTABLE: Collect coverage profile data"),
     )
+    .arg(
+      Arg::with_name("jobs")
+        .short("j")
+        .long("jobs")
+        .min_values(0)
+        .max_values(1)
+        .takes_value(true)
+        .validator(|val: String| match val.parse::<usize>() {
+          Ok(_) => Ok(()),
+          Err(_) => Err("jobs should be a number".to_string()),
+        }),
+    )
     .arg(
       Arg::with_name("files")
         .help("List of file names to run")
@@ -1666,6 +1679,18 @@ fn test_parse(flags: &mut Flags, matches: &clap::ArgMatches) {
     }
   }
 
+  let concurrent_jobs = if matches.is_present("jobs") {
+    if let Some(value) = matches.value_of("jobs") {
+      value.parse().unwrap()
+    } else {
+      // TODO(caspervonb) when no value is given use
+      // https://doc.rust-lang.org/std/thread/fn.available_concurrency.html
+      2
+    }
+  } else {
+    1
+  };
+
   let include = if matches.is_present("files") {
     let files: Vec<String> = matches
       .values_of("files")
@@ -1685,6 +1710,7 @@ fn test_parse(flags: &mut Flags, matches: &clap::ArgMatches) {
     include,
     filter,
     allow_none,
+    concurrent_jobs,
   };
 }
 
@@ -3330,6 +3356,7 @@ mod tests {
           allow_none: true,
           quiet: false,
           include: Some(svec!["dir1/", "dir2/"]),
+          concurrent_jobs: 1,
         },
         unstable: true,
         coverage_dir: Some("cov".to_string()),

cli/main.rs

@@ -219,7 +219,7 @@ pub fn create_main_worker(
     ops::runtime_compiler::init(js_runtime);
 
     if enable_testing {
-      ops::test_runner::init(js_runtime);
+      ops::testing::init(js_runtime);
     }
 
     js_runtime.sync_ops_cache();
@@ -902,6 +902,7 @@ async fn coverage_command(
   .await
 }
 
+#[allow(clippy::too_many_arguments)]
 async fn test_command(
   flags: Flags,
   include: Option<Vec<String>>,
@@ -910,87 +911,23 @@ async fn test_command(
   quiet: bool,
   allow_none: bool,
   filter: Option<String>,
+  concurrent_jobs: usize,
 ) -> Result<(), AnyError> {
-  let program_state = ProgramState::build(flags.clone()).await?;
-  let permissions = Permissions::from_options(&flags.clone().into());
-  let cwd = std::env::current_dir().expect("No current directory");
-  let include = include.unwrap_or_else(|| vec![".".to_string()]);
-  let test_modules =
-    tools::test_runner::prepare_test_modules_urls(include, &cwd)?;
-
-  if test_modules.is_empty() {
-    println!("No matching test modules found");
-    if !allow_none {
-      std::process::exit(1);
-    }
-    return Ok(());
-  }
-  let main_module = deno_core::resolve_path("$deno$test.ts")?;
-  // Create a dummy source file.
-  let source_file = File {
-    local: main_module.to_file_path().unwrap(),
-    maybe_types: None,
-    media_type: MediaType::TypeScript,
-    source: tools::test_runner::render_test_file(
-      test_modules.clone(),
-      fail_fast,
-      quiet,
-      filter,
-    ),
-    specifier: main_module.clone(),
-  };
-  // Save our fake file into file fetcher cache
-  // to allow module access by TS compiler
-  program_state.file_fetcher.insert_cached(source_file);
-
-  if no_run {
-    let lib = if flags.unstable {
-      module_graph::TypeLib::UnstableDenoWindow
-    } else {
-      module_graph::TypeLib::DenoWindow
-    };
-    program_state
-      .prepare_module_load(
-        main_module.clone(),
-        lib,
-        Permissions::allow_all(),
-        false,
-        program_state.maybe_import_map.clone(),
-      )
-      .await?;
-    return Ok(());
-  }
-
-  let mut worker =
-    create_main_worker(&program_state, main_module.clone(), permissions, true);
-
   if let Some(ref coverage_dir) = flags.coverage_dir {
     env::set_var("DENO_UNSTABLE_COVERAGE_DIR", coverage_dir);
   }
 
-  let mut maybe_coverage_collector =
-    if let Some(ref coverage_dir) = program_state.coverage_dir {
-      let session = worker.create_inspector_session();
-      let coverage_dir = PathBuf::from(coverage_dir);
-      let mut coverage_collector =
-        tools::coverage::CoverageCollector::new(coverage_dir, session);
-      coverage_collector.start_collecting().await?;
-
-      Some(coverage_collector)
-    } else {
-      None
-    };
-
-  let execute_result = worker.execute_module(&main_module).await;
-  execute_result?;
-  worker.execute("window.dispatchEvent(new Event('load'))")?;
-  worker.run_event_loop().await?;
-  worker.execute("window.dispatchEvent(new Event('unload'))")?;
-  worker.run_event_loop().await?;
-
-  if let Some(coverage_collector) = maybe_coverage_collector.as_mut() {
-    coverage_collector.stop_collecting().await?;
-  }
+  tools::test_runner::run_tests(
+    flags,
+    include,
+    no_run,
+    fail_fast,
+    quiet,
+    allow_none,
+    filter,
+    concurrent_jobs,
+  )
+  .await?;
 
   Ok(())
 }
@@ -1125,10 +1062,18 @@ fn get_subcommand(
       include,
       allow_none,
       filter,
-    } => {
-      test_command(flags, include, no_run, fail_fast, quiet, allow_none, filter)
-        .boxed_local()
-    }
+      concurrent_jobs,
+    } => test_command(
+      flags,
+      include,
+      no_run,
+      fail_fast,
+      quiet,
+      allow_none,
+      filter,
+      concurrent_jobs,
+    )
+    .boxed_local(),
     DenoSubcommand::Completions { buf } => {
       if let Err(e) = write_to_stdout_ignore_sigpipe(&buf) {
         eprintln!("{}", e);

cli/ops/mod.rs

@@ -2,6 +2,6 @@
 
 pub mod errors;
 pub mod runtime_compiler;
-pub mod test_runner;
+pub mod testing;
 
 pub use deno_runtime::ops::{reg_async, reg_sync};

cli/ops/test_runner.rs → cli/ops/testing.rs

@@ -1,23 +1,27 @@
-// Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
-
+use crate::tools::test_runner::TestMessage;
 use deno_core::error::generic_error;
 use deno_core::error::AnyError;
 use deno_core::serde_json;
+use deno_core::serde_json::json;
 use deno_core::serde_json::Value;
+use deno_core::JsRuntime;
 use deno_core::OpState;
 use deno_core::ZeroCopyBuf;
 use deno_runtime::ops::worker_host::create_worker_permissions;
 use deno_runtime::ops::worker_host::PermissionsArg;
 use deno_runtime::permissions::Permissions;
+use serde::Deserialize;
+use std::sync::mpsc::Sender;
 use uuid::Uuid;
 
-pub fn init(rt: &mut deno_core::JsRuntime) {
+pub fn init(rt: &mut JsRuntime) {
   super::reg_sync(rt, "op_pledge_test_permissions", op_pledge_test_permissions);
   super::reg_sync(
     rt,
     "op_restore_test_permissions",
     op_restore_test_permissions,
   );
+  super::reg_sync(rt, "op_post_test_message", op_post_test_message);
 }
 
 #[derive(Clone)]
@@ -64,3 +68,24 @@ pub fn op_restore_test_permissions(
     Err(generic_error("no permissions to restore"))
   }
 }
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct PostTestMessageArgs {
+  message: TestMessage,
+}
+
+fn op_post_test_message(
+  state: &mut OpState,
+  args: Value,
+  _zero_copy: Option<ZeroCopyBuf>,
+) -> Result<Value, AnyError> {
+  let args: PostTestMessageArgs = serde_json::from_value(args)?;
+  let sender = state.borrow::<Sender<TestMessage>>().clone();
+
+  if sender.send(args.message).is_err() {
+    Ok(json!(false))
+  } else {
+    Ok(json!(true))
+  }
+}

cli/program_state.rs

@@ -133,6 +133,72 @@ impl ProgramState {
     Ok(Arc::new(program_state))
   }
 
+  /// Prepares a set of module specifiers for loading in one shot.
+  ///
+  pub async fn prepare_module_graph(
+    self: &Arc<Self>,
+    specifiers: Vec<ModuleSpecifier>,
+    lib: TypeLib,
+    runtime_permissions: Permissions,
+    maybe_import_map: Option<ImportMap>,
+  ) -> Result<(), AnyError> {
+    let handler = Arc::new(Mutex::new(FetchHandler::new(
+      self,
+      runtime_permissions.clone(),
+    )?));
+
+    let mut builder =
+      GraphBuilder::new(handler, maybe_import_map, self.lockfile.clone());
+
+    for specifier in specifiers {
+      builder.add(&specifier, false).await?;
+    }
+
+    let mut graph = builder.get_graph();
+    let debug = self.flags.log_level == Some(log::Level::Debug);
+    let maybe_config_path = self.flags.config_path.clone();
+
+    let result_modules = if self.flags.no_check {
+      let result_info = graph.transpile(TranspileOptions {
+        debug,
+        maybe_config_path,
+        reload: self.flags.reload,
+      })?;
+      debug!("{}", result_info.stats);
+      if let Some(ignored_options) = result_info.maybe_ignored_options {
+        warn!("{}", ignored_options);
+      }
+      result_info.loadable_modules
+    } else {
+      let result_info = graph.check(CheckOptions {
+        debug,
+        emit: true,
+        lib,
+        maybe_config_path,
+        reload: self.flags.reload,
+      })?;
+
+      debug!("{}", result_info.stats);
+      if let Some(ignored_options) = result_info.maybe_ignored_options {
+        eprintln!("{}", ignored_options);
+      }
+      if !result_info.diagnostics.is_empty() {
+        return Err(anyhow!(result_info.diagnostics));
+      }
+      result_info.loadable_modules
+    };
+
+    let mut loadable_modules = self.modules.lock().unwrap();
+    loadable_modules.extend(result_modules);
+
+    if let Some(ref lockfile) = self.lockfile {
+      let g = lockfile.lock().unwrap();
+      g.write()?;
+    }
+
+    Ok(())
+  }
+
   /// This function is called when new module load is
   /// initialized by the JsRuntime. Its resposibility is to collect
   /// all dependencies and if it is required then also perform TS typecheck

cli/tests/integration_tests.rs

@@ -34,6 +34,9 @@ fn js_unit_tests_lint() {
 fn js_unit_tests() {
   let _g = util::http_server();
 
+  // Note that the unit tests are not safe for concurrency and must be run with a concurrency limit
+  // of one because there are some chdir tests in there.
+  // TODO(caspervonb) split these tests into two groups: parallel and serial.
   let mut deno = util::deno_cmd()
     .current_dir(util::root_path())
     .arg("test")
@@ -2438,11 +2441,23 @@ mod integration {
       output: "test/deno_test_unresolved_promise.out",
     });
 
+    itest!(unhandled_rejection {
+      args: "test test/unhandled_rejection.ts",
+      exit_code: 1,
+      output: "test/unhandled_rejection.out",
+    });
+
     itest!(exit_sanitizer {
       args: "test test/exit_sanitizer_test.ts",
       output: "test/exit_sanitizer_test.out",
       exit_code: 1,
     });
+
+    itest!(quiet {
+      args: "test --quiet test/quiet_test.ts",
+      exit_code: 0,
+      output: "test/quiet_test.out",
+    });
   }
 
   #[test]

cli/tests/test/deno_test_only.ts.out

@@ -1,7 +1,8 @@
-[WILDCARD]running 1 tests
+[WILDCARD]
+running 1 tests
 test def ... ok ([WILDCARD])
 
-test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out ([WILDCARD])
+test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 2 filtered out ([WILDCARD])
 
 FAILED because the "only" option was used
 

cli/tests/test/deno_test_unresolved_promise.out

@@ -1,4 +1,5 @@
-Check [WILDCARD]
 running 2 tests
-test unresolved promise ... in promise
+test unresolved promise ...
+test result: FAILED. 0 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out [WILDCARD]
+
 error: Module evaluation is still pending but there are no pending ops or dynamic imports. This situation is often caused by unresolved promise.

cli/tests/test/exit_sanitizer_test.out

@@ -1,4 +1,4 @@
-Check [WILDCARD]/$deno$test.ts
+Check [WILDCARD]/exit_sanitizer_test.ts
 running 3 tests
 test exit(0) ... FAILED ([WILDCARD])
 test exit(1) ... FAILED ([WILDCARD])