tests: Refactor inspector tests (#16967)

This commit adds "InspectorTester" struct which is used in
inspector tests - it encapsulated various functionalities that
we need (like reading/writing to WebSocket), but also adds
better error handling which should help with debugging flaky
tests.

1 files changed, 718 insertions, 788 deletions

diff --git a/cli/tests/inspector_tests.rs b/cli/tests/inspector_tests.rs
index 8a1bc141f..8e5a8d1f6 100644
--- a/cli/tests/inspector_tests.rs
+++ b/cli/tests/inspector_tests.rs
@@ -1,15 +1,17 @@
 // Copyright 2018-2022 the Deno authors. All rights reserved. MIT license.
 
-use deno_core::futures;
+use deno_core::error::AnyError;
 use deno_core::futures::prelude::*;
 use deno_core::futures::stream::SplitSink;
+use deno_core::futures::stream::SplitStream;
 use deno_core::serde_json;
+use deno_core::serde_json::json;
 use deno_core::url;
 use deno_runtime::deno_fetch::reqwest;
 use deno_runtime::deno_websocket::tokio_tungstenite;
 use deno_runtime::deno_websocket::tokio_tungstenite::tungstenite;
 use std::io::BufRead;
-use std::pin::Pin;
+use std::process::Child;
 use test_util as util;
 use tokio::net::TcpStream;
 use util::http_server;
@@ -17,46 +19,178 @@ use util::http_server;
 mod inspector {
   use super::*;
 
-  macro_rules! assert_starts_with {
-  ($string:expr, $($test:expr),+) => {
-    let string = $string; // This might be a function call or something
-    if !($(string.starts_with($test))||+) {
-      panic!("{:?} does not start with {:?}", string, [$($test),+]);
-    }
+  struct InspectorTester {
+    socket_tx: SplitSink<
+      tokio_tungstenite::WebSocketStream<
+        tokio_tungstenite::MaybeTlsStream<TcpStream>,
+      >,
+      tungstenite::Message,
+    >,
+    socket_rx: SplitStream<
+      tokio_tungstenite::WebSocketStream<
+        tokio_tungstenite::MaybeTlsStream<TcpStream>,
+      >,
+    >,
+    notification_filter: Box<dyn FnMut(&str) -> bool + 'static>,
+    child: Child,
+    stderr_lines: Box<dyn Iterator<Item = String>>,
+    stdout_lines: Box<dyn Iterator<Item = String>>,
   }
-}
 
-  fn inspect_flag_with_unique_port(flag_prefix: &str) -> String {
-    use std::sync::atomic::{AtomicU16, Ordering};
-    static PORT: AtomicU16 = AtomicU16::new(9229);
-    let port = PORT.fetch_add(1, Ordering::Relaxed);
-    format!("{}=127.0.0.1:{}", flag_prefix, port)
+  fn ignore_script_parsed(msg: &str) -> bool {
+    !msg.starts_with(r#"{"method":"Debugger.scriptParsed","#)
   }
 
-  fn extract_ws_url_from_stderr(
-    stderr_lines: &mut impl std::iter::Iterator<Item = String>,
-  ) -> url::Url {
-    let stderr_first_line = skip_check_line(stderr_lines);
-    assert_starts_with!(&stderr_first_line, "Debugger listening on ");
-    let v: Vec<_> = stderr_first_line.match_indices("ws:").collect();
-    assert_eq!(v.len(), 1);
-    let ws_url_index = v[0].0;
-    let ws_url = &stderr_first_line[ws_url_index..];
-    url::Url::parse(ws_url).unwrap()
-  }
+  impl InspectorTester {
+    async fn create<F>(mut child: Child, notification_filter: F) -> Self
+    where
+      F: FnMut(&str) -> bool + 'static,
+    {
+      let stdout = child.stdout.take().unwrap();
+      let stdout_lines =
+        std::io::BufReader::new(stdout).lines().map(|r| r.unwrap());
 
-  fn skip_check_line(
-    stderr_lines: &mut impl std::iter::Iterator<Item = String>,
-  ) -> String {
-    loop {
-      let mut line = stderr_lines.next().unwrap();
-      line = util::strip_ansi_codes(&line).to_string();
+      let stderr = child.stderr.take().unwrap();
+      let mut stderr_lines =
+        std::io::BufReader::new(stderr).lines().map(|r| r.unwrap());
 
-      if line.starts_with("Check") {
-        continue;
+      let ws_url = extract_ws_url_from_stderr(&mut stderr_lines);
+
+      let (socket, response) =
+        tokio_tungstenite::connect_async(ws_url).await.unwrap();
+      assert_eq!(response.status(), 101); // Switching protocols.
+
+      let (socket_tx, socket_rx) = socket.split();
+
+      Self {
+        socket_tx,
+        socket_rx,
+        notification_filter: Box::new(notification_filter),
+        child,
+        stderr_lines: Box::new(stderr_lines),
+        stdout_lines: Box::new(stdout_lines),
       }
+    }
 
-      return line;
+    async fn send_many(&mut self, messages: &[serde_json::Value]) {
+      // TODO(bartlomieju): add graceful error handling
+      for msg in messages {
+        let result = self
+          .socket_tx
+          .send(msg.to_string().into())
+          .await
+          .map_err(|e| e.into());
+        self.handle_error(result);
+      }
+    }
+
+    async fn send(&mut self, message: serde_json::Value) {
+      self.send_many(&[message]).await;
+    }
+
+    fn handle_error<T>(&mut self, result: Result<T, AnyError>) -> T {
+      match result {
+        Ok(result) => result,
+        Err(err) => {
+          let mut stdout = vec![];
+          for line in self.stdout_lines.by_ref() {
+            stdout.push(line);
+          }
+          let mut stderr = vec![];
+          for line in self.stderr_lines.by_ref() {
+            stderr.push(line);
+          }
+          let stdout = stdout.join("\n");
+          let stderr = stderr.join("\n");
+          self.child.kill().unwrap();
+
+          panic!(
+            "Inspector test failed with error: {:?}.\nstdout:\n{}\nstderr:\n{}",
+            err, stdout, stderr
+          );
+        }
+      }
+    }
+
+    async fn recv(&mut self) -> String {
+      loop {
+        let result = self.socket_rx.next().await.unwrap().map_err(|e| e.into());
+        let message = self.handle_error(result).to_string();
+        if (self.notification_filter)(&message) {
+          return message;
+        }
+      }
+    }
+
+    async fn recv_as_json(&mut self) -> serde_json::Value {
+      let msg = self.recv().await;
+      serde_json::from_str(&msg).unwrap()
+    }
+
+    async fn assert_received_messages(
+      &mut self,
+      responses: &[&str],
+      notifications: &[&str],
+    ) {
+      let expected_messages = responses.len() + notifications.len();
+      let mut responses_idx = 0;
+      let mut notifications_idx = 0;
+
+      for _ in 0..expected_messages {
+        let msg = self.recv().await;
+
+        if msg.starts_with(r#"{"id":"#) {
+          assert!(
+            msg.starts_with(responses[responses_idx]),
+            "Doesn't start with {}, instead received {}",
+            responses[responses_idx],
+            msg
+          );
+          responses_idx += 1;
+        } else {
+          assert!(
+            msg.starts_with(notifications[notifications_idx]),
+            "Doesn't start with {}, instead received {}",
+            notifications[notifications_idx],
+            msg
+          );
+          notifications_idx += 1;
+        }
+      }
+    }
+
+    fn stderr_line(&mut self) -> String {
+      self.stderr_lines.next().unwrap()
+    }
+
+    fn stdout_line(&mut self) -> String {
+      self.stdout_lines.next().unwrap()
+    }
+
+    fn assert_stderr_for_inspect(&mut self) {
+      assert_stderr(
+        &mut self.stderr_lines,
+        &["Visit chrome://inspect to connect to the debugger."],
+      );
+    }
+
+    fn assert_stderr_for_inspect_brk(&mut self) {
+      assert_stderr(
+        &mut self.stderr_lines,
+        &[
+          "Visit chrome://inspect to connect to the debugger.",
+          "Deno is waiting for debugger to connect.",
+        ],
+      );
+    }
+  }
+
+  macro_rules! assert_starts_with {
+    ($string:expr, $($test:expr),+) => {
+      let string = $string; // This might be a function call or something
+      if !($(string.starts_with($test))||+) {
+        panic!("{:?} does not start with {:?}", string, [$($test),+]);
+      }
     }
   }
 
@@ -78,67 +212,37 @@ mod inspector {
     }
   }
 
-  fn assert_stderr_for_inspect(
-    stderr_lines: &mut impl std::iter::Iterator<Item = String>,
-  ) {
-    assert_stderr(
-      stderr_lines,
-      &["Visit chrome://inspect to connect to the debugger."],
-    );
+  fn inspect_flag_with_unique_port(flag_prefix: &str) -> String {
+    use std::sync::atomic::{AtomicU16, Ordering};
+    static PORT: AtomicU16 = AtomicU16::new(9229);
+    let port = PORT.fetch_add(1, Ordering::Relaxed);
+    format!("{}=127.0.0.1:{}", flag_prefix, port)
   }
 
-  fn assert_stderr_for_inspect_brk(
+  fn extract_ws_url_from_stderr(
     stderr_lines: &mut impl std::iter::Iterator<Item = String>,
-  ) {
-    assert_stderr(
-      stderr_lines,
-      &[
-        "Visit chrome://inspect to connect to the debugger.",
-        "Deno is waiting for debugger to connect.",
-      ],
-    );
+  ) -> url::Url {
+    let stderr_first_line = skip_check_line(stderr_lines);
+    assert_starts_with!(&stderr_first_line, "Debugger listening on ");
+    let v: Vec<_> = stderr_first_line.match_indices("ws:").collect();
+    assert_eq!(v.len(), 1);
+    let ws_url_index = v[0].0;
+    let ws_url = &stderr_first_line[ws_url_index..];
+    url::Url::parse(ws_url).unwrap()
   }
 
-  async fn assert_inspector_messages(
-    socket_tx: &mut SplitSink<
-      tokio_tungstenite::WebSocketStream<
-        tokio_tungstenite::MaybeTlsStream<TcpStream>,
-      >,
-      tungstenite::Message,
-    >,
-    messages: &[&str],
-    socket_rx: &mut Pin<Box<dyn Stream<Item = String>>>,
-    responses: &[&str],
-    notifications: &[&str],
-  ) {
-    for msg in messages {
-      socket_tx.send(msg.to_string().into()).await.unwrap();
-    }
+  fn skip_check_line(
+    stderr_lines: &mut impl std::iter::Iterator<Item = String>,
+  ) -> String {
+    loop {
+      let mut line = stderr_lines.next().unwrap();
+      line = util::strip_ansi_codes(&line).to_string();
 
-    let expected_messages = responses.len() + notifications.len();
-    let mut responses_idx = 0;
-    let mut notifications_idx = 0;
-
-    for _ in 0..expected_messages {
-      let msg = socket_rx.next().await.unwrap();
-
-      if msg.starts_with(r#"{"id":"#) {
-        assert!(
-          msg.starts_with(responses[responses_idx]),
-          "Doesn't start with {}, instead received {}",
-          responses[responses_idx],
-          msg
-        );
-        responses_idx += 1;
-      } else {
-        assert!(
-          msg.starts_with(notifications[notifications_idx]),
-          "Doesn't start with {}, instead received {}",
-          notifications[notifications_idx],
-          msg
-        );
-        notifications_idx += 1;
+      if line.starts_with("Check") {
+        continue;
       }
+
+      return line;
     }
   }
 
@@ -170,7 +274,7 @@ mod inspector {
   #[tokio::test]
   async fn inspector_break_on_first_line() {
     let script = util::testdata_path().join("inspector/inspector2.js");
-    let mut child = util::deno_cmd()
+    let child = util::deno_cmd()
       .arg("run")
       .arg(inspect_flag_with_unique_port("--inspect-brk"))
       .arg(script)
@@ -179,142 +283,108 @@ mod inspector {
       .spawn()
       .unwrap();
 
-    let stderr = child.stderr.as_mut().unwrap();
-    let mut stderr_lines =
-      std::io::BufReader::new(stderr).lines().map(|r| r.unwrap());
-    let ws_url = extract_ws_url_from_stderr(&mut stderr_lines);
+    let mut tester = InspectorTester::create(child, ignore_script_parsed).await;
 
-    let (socket, response) =
-      tokio_tungstenite::connect_async(ws_url).await.unwrap();
-    assert_eq!(response.status(), 101); // Switching protocols.
+    tester.assert_stderr_for_inspect_brk();
 
-    let (mut socket_tx, socket_rx) = socket.split();
-    let mut socket_rx = socket_rx
-      .map(|msg| msg.unwrap().to_string())
-      .filter(|msg| {
-        let pass = !msg.starts_with(r#"{"method":"Debugger.scriptParsed","#);
-        futures::future::ready(pass)
-      })
-      .boxed_local();
+    tester
+      .send_many(&[
+        json!({"id":1,"method":"Runtime.enable"}),
+        json!({"id":2,"method":"Debugger.enable"}),
+      ])
+      .await;
+    tester.assert_received_messages(
+      &[
+        r#"{"id":1,"result":{}}"#,
+        r#"{"id":2,"result":{"debuggerId":"#,
+      ],
+      &[
+        r#"{"method":"Runtime.executionContextCreated","params":{"context":{"id":1,"#,
+      ],
+    )
+    .await;
 
-    let stdout = child.stdout.as_mut().unwrap();
-    let mut stdout_lines =
-      std::io::BufReader::new(stdout).lines().map(|r| r.unwrap());
+    tester
+      .send(json!({"id":3,"method":"Runtime.runIfWaitingForDebugger"}))
+      .await;
+    tester
+      .assert_received_messages(
+        &[r#"{"id":3,"result":{}}"#],
+        &[r#"{"method":"Debugger.paused","#],
+      )
+      .await;
 
-    assert_stderr_for_inspect_brk(&mut stderr_lines);
+    tester
+      .send(json!({
+        "id":4,
+        "method":"Runtime.evaluate",
+        "params":{
+          "expression":"Deno.core.print(\"hello from the inspector\\n\")",
+          "contextId":1,
+          "includeCommandLineAPI":true,
+          "silent":false,
+          "returnByValue":true
+        }
+      }))
+      .await;
+    tester
+      .assert_received_messages(
+        &[r#"{"id":4,"result":{"result":{"type":"undefined"}}}"#],
+        &[],
+      )
+      .await;
 
-    assert_inspector_messages(
-    &mut socket_tx,
-    &[
-      r#"{"id":1,"method":"Runtime.enable"}"#,
-      r#"{"id":2,"method":"Debugger.enable"}"#,
-    ],
-    &mut socket_rx,
-    &[
-      r#"{"id":1,"result":{}}"#,
-      r#"{"id":2,"result":{"debuggerId":"#,
-    ],
-    &[
-      r#"{"method":"Runtime.executionContextCreated","params":{"context":{"id":1,"#,
-    ],
-  )
-  .await;
-
-    assert_inspector_messages(
-      &mut socket_tx,
-      &[r#"{"id":3,"method":"Runtime.runIfWaitingForDebugger"}"#],
-      &mut socket_rx,
-      &[r#"{"id":3,"result":{}}"#],
-      &[r#"{"method":"Debugger.paused","#],
-    )
-    .await;
+    assert_eq!(
+      &tester.stdout_lines.next().unwrap(),
+      "hello from the inspector"
+    );
 
-    assert_inspector_messages(
-    &mut socket_tx,
-    &[
-      r#"{"id":4,"method":"Runtime.evaluate","params":{"expression":"Deno.core.print(\"hello from the inspector\\n\")","contextId":1,"includeCommandLineAPI":true,"silent":false,"returnByValue":true}}"#,
-    ],
-    &mut socket_rx,
-    &[r#"{"id":4,"result":{"result":{"type":"undefined"}}}"#],
-    &[],
-  )
-  .await;
-
-    assert_eq!(&stdout_lines.next().unwrap(), "hello from the inspector");
-
-    assert_inspector_messages(
-      &mut socket_tx,
-      &[r#"{"id":5,"method":"Debugger.resume"}"#],
-      &mut socket_rx,
-      &[r#"{"id":5,"result":{}}"#],
-      &[],
-    )
-    .await;
+    tester
+      .send(json!({"id":5,"method":"Debugger.resume"}))
+      .await;
+    tester
+      .assert_received_messages(&[r#"{"id":5,"result":{}}"#], &[])
+      .await;
 
-    assert_eq!(&stdout_lines.next().unwrap(), "hello from the script");
+    assert_eq!(
+      &tester.stdout_lines.next().unwrap(),
+      "hello from the script"
+    );
 
-    child.kill().unwrap();
-    child.wait().unwrap();
+    tester.child.kill().unwrap();
+    tester.child.wait().unwrap();
   }
 
   #[tokio::test]
   async fn inspector_pause() {
     let script = util::testdata_path().join("inspector/inspector1.js");
-    let mut child = util::deno_cmd()
+    let child = util::deno_cmd()
       .arg("run")
       .arg(inspect_flag_with_unique_port("--inspect"))
       .arg(script)
+      .stdout(std::process::Stdio::piped())
       .stderr(std::process::Stdio::piped())
       .spawn()
       .unwrap();
 
-    let stderr = child.stderr.as_mut().unwrap();
-    let mut stderr_lines =
-      std::io::BufReader::new(stderr).lines().map(|r| r.unwrap());
-    let ws_url = extract_ws_url_from_stderr(&mut stderr_lines);
-
-    // We use tokio_tungstenite as a websocket client because warp (which is
-    // a dependency of Deno) uses it.
-    let (mut socket, _) =
-      tokio_tungstenite::connect_async(ws_url).await.unwrap();
-
-    /// Returns the next websocket message as a string ignoring
-    /// Debugger.scriptParsed messages.
-    async fn ws_read_msg(
-      socket: &mut tokio_tungstenite::WebSocketStream<
-        tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
-      >,
-    ) -> String {
-      use deno_core::futures::stream::StreamExt;
-      while let Some(msg) = socket.next().await {
-        let msg = msg.unwrap().to_string();
-        // FIXME(bartlomieju): fails because there's a file loaded
-        // called 150_errors.js
-        // assert!(!msg.contains("error"));
-        if !msg.contains("Debugger.scriptParsed") {
-          return msg;
-        }
-      }
-      unreachable!()
-    }
-
-    socket
-      .send(r#"{"id":6,"method":"Debugger.enable"}"#.into())
-      .await
-      .unwrap();
+    let mut tester = InspectorTester::create(child, ignore_script_parsed).await;
 
-    let msg = ws_read_msg(&mut socket).await;
-    assert_starts_with!(msg, r#"{"id":6,"result":{"debuggerId":"#);
+    tester
+      .send(json!({"id":6,"method":"Debugger.enable"}))
+      .await;
+    tester
+      .assert_received_messages(&[r#"{"id":6,"result":{"debuggerId":"#], &[])
+      .await;
 
-    socket
-      .send(r#"{"id":31,"method":"Debugger.pause"}"#.into())
-      .await
-      .unwrap();
+    tester
+      .send(json!({"id":31,"method":"Debugger.pause"}))
+      .await;
 
-    let msg = ws_read_msg(&mut socket).await;
-    assert_eq!(msg, r#"{"id":31,"result":{}}"#);
+    tester
+      .assert_received_messages(&[r#"{"id":31,"result":{}}"#], &[])
+      .await;
 
-    child.kill().unwrap();
+    tester.child.kill().unwrap();
   }
 
   #[tokio::test]
@@ -368,7 +438,7 @@ mod inspector {
   #[tokio::test]
   async fn inspector_does_not_hang() {
     let script = util::testdata_path().join("inspector/inspector3.js");
-    let mut child = util::deno_cmd()
+    let child = util::deno_cmd()
       .arg("run")
       .arg(inspect_flag_with_unique_port("--inspect-brk"))
       .env("NO_COLOR", "1")
@@ -378,102 +448,80 @@ mod inspector {
       .spawn()
       .unwrap();
 
-    let stderr = child.stderr.as_mut().unwrap();
-    let mut stderr_lines =
-      std::io::BufReader::new(stderr).lines().map(|r| r.unwrap());
-    let ws_url = extract_ws_url_from_stderr(&mut stderr_lines);
-
-    let (socket, response) =
-      tokio_tungstenite::connect_async(ws_url).await.unwrap();
-    assert_eq!(response.status(), 101); // Switching protocols.
-
-    let (mut socket_tx, socket_rx) = socket.split();
-    let mut socket_rx = socket_rx
-      .map(|msg| msg.unwrap().to_string())
-      .filter(|msg| {
-        let pass = !msg.starts_with(r#"{"method":"Debugger.scriptParsed","#);
-        futures::future::ready(pass)
-      })
-      .boxed_local();
-
-    let stdout = child.stdout.as_mut().unwrap();
-    let mut stdout_lines =
-      std::io::BufReader::new(stdout).lines().map(|r| r.unwrap());
+    let mut tester = InspectorTester::create(child, ignore_script_parsed).await;
 
-    assert_stderr_for_inspect_brk(&mut stderr_lines);
+    tester.assert_stderr_for_inspect_brk();
 
-    assert_inspector_messages(
-    &mut socket_tx,
-    &[
-      r#"{"id":1,"method":"Runtime.enable"}"#,
-      r#"{"id":2,"method":"Debugger.enable"}"#,
-    ],
-    &mut socket_rx,
-    &[
-      r#"{"id":1,"result":{}}"#,
-      r#"{"id":2,"result":{"debuggerId":"#
-    ],
-    &[
-      r#"{"method":"Runtime.executionContextCreated","params":{"context":{"id":1,"#
-    ],
-  )
-  .await;
-
-    assert_inspector_messages(
-      &mut socket_tx,
-      &[r#"{"id":3,"method":"Runtime.runIfWaitingForDebugger"}"#],
-      &mut socket_rx,
-      &[r#"{"id":3,"result":{}}"#],
-      &[r#"{"method":"Debugger.paused","#],
-    )
-    .await;
-
-    assert_inspector_messages(
-      &mut socket_tx,
-      &[r#"{"id":4,"method":"Debugger.resume"}"#],
-      &mut socket_rx,
-      &[r#"{"id":4,"result":{}}"#],
-      &[r#"{"method":"Debugger.resumed","params":{}}"#],
+    tester
+      .send_many(&[
+        json!({"id":1,"method":"Runtime.enable"}),
+        json!({"id":2,"method":"Debugger.enable"}),
+      ])
+      .await;
+    tester.assert_received_messages(
+      &[
+        r#"{"id":1,"result":{}}"#,
+        r#"{"id":2,"result":{"debuggerId":"#
+      ],
+      &[
+        r#"{"method":"Runtime.executionContextCreated","params":{"context":{"id":1,"#
+      ],
     )
     .await;
 
-    for i in 0..128u32 {
-      let request_id = i + 10;
-      // Expect the number {i} on stdout.
-      let s = i.to_string();
-      assert_eq!(stdout_lines.next().unwrap(), s);
-
-      assert_inspector_messages(
-        &mut socket_tx,
-        &[],
-        &mut socket_rx,
-        &[],
-        &[
-          r#"{"method":"Runtime.consoleAPICalled","#,
-          r#"{"method":"Debugger.paused","#,
-        ],
+    tester
+      .send(json!({"id":3,"method":"Runtime.runIfWaitingForDebugger"}))
+      .await;
+    tester
+      .assert_received_messages(
+        &[r#"{"id":3,"result":{}}"#],
+        &[r#"{"method":"Debugger.paused","#],
       )
       .await;
 
-      assert_inspector_messages(
-        &mut socket_tx,
-        &[&format!(
-          r#"{{"id":{},"method":"Debugger.resume"}}"#,
-          request_id
-        )],
-        &mut socket_rx,
-        &[&format!(r#"{{"id":{},"result":{{}}}}"#, request_id)],
+    tester
+      .send(json!({"id":4,"method":"Debugger.resume"}))
+      .await;
+    tester
+      .assert_received_messages(
+        &[r#"{"id":4,"result":{}}"#],
         &[r#"{"method":"Debugger.resumed","params":{}}"#],
       )
       .await;
+
+    for i in 0..128u32 {
+      let request_id = i + 10;
+      // Expect the number {i} on stdout.
+      let s = i.to_string();
+      assert_eq!(tester.stdout_lines.next().unwrap(), s);
+
+      tester
+        .assert_received_messages(
+          &[],
+          &[
+            r#"{"method":"Runtime.consoleAPICalled","#,
+            r#"{"method":"Debugger.paused","#,
+          ],
+        )
+        .await;
+
+      tester
+        .send(json!({"id":request_id,"method":"Debugger.resume"}))
+        .await;
+      tester
+        .assert_received_messages(
+          &[&format!(r#"{{"id":{},"result":{{}}}}"#, request_id)],
+          &[r#"{"method":"Debugger.resumed","params":{}}"#],
+        )
+        .await;
     }
 
     // Check that we can gracefully close the websocket connection.
-    socket_tx.close().await.unwrap();
-    socket_rx.for_each(|_| async {}).await;
+    tester.socket_tx.close().await.unwrap();
+    tester.socket_rx.for_each(|_| async {}).await;
 
-    assert_eq!(&stdout_lines.next().unwrap(), "done");
-    assert!(child.wait().unwrap().success());
+    assert_eq!(&tester.stdout_lines.next().unwrap(), "done");
+    assert!(tester.child.wait().unwrap().success());
   }
 
   #[tokio::test]
@@ -507,7 +555,7 @@ mod inspector {
 
   #[tokio::test]
   async fn inspector_runtime_evaluate_does_not_crash() {
-    let mut child = util::deno_cmd()
+    let child = util::deno_cmd()
       .arg("repl")
       .arg(inspect_flag_with_unique_port("--inspect"))
       .stdin(std::process::Stdio::piped())
@@ -516,86 +564,101 @@ mod inspector {
       .spawn()
       .unwrap();
 
-    let stderr = child.stderr.as_mut().unwrap();
-    let mut stderr_lines = std::io::BufReader::new(stderr)
-      .lines()
-      .map(|r| r.unwrap())
-      .filter(|s| s.as_str() != "Debugger session started.");
-    let ws_url = extract_ws_url_from_stderr(&mut stderr_lines);
-
-    let (socket, response) =
-      tokio_tungstenite::connect_async(ws_url).await.unwrap();
-    assert_eq!(response.status(), 101); // Switching protocols.
-
-    let (mut socket_tx, socket_rx) = socket.split();
-    let mut socket_rx = socket_rx
-      .map(|msg| msg.unwrap().to_string())
-      .filter(|msg| {
-        let pass = !msg.starts_with(r#"{"method":"Debugger.scriptParsed","#);
-        futures::future::ready(pass)
-      })
-      .boxed_local();
-
-    let stdin = child.stdin.take().unwrap();
+    let mut tester = InspectorTester::create(child, ignore_script_parsed).await;
 
-    let stdout = child.stdout.as_mut().unwrap();
-    let mut stdout_lines = std::io::BufReader::new(stdout)
-      .lines()
-      .map(|r| r.unwrap())
-      .filter(|s| !s.starts_with("Deno "));
+    let stdin = tester.child.stdin.take().unwrap();
 
-    assert_stderr_for_inspect(&mut stderr_lines);
+    tester.assert_stderr_for_inspect();
+    assert_starts_with!(&tester.stdout_line(), "Deno");
     assert_eq!(
-      &stdout_lines.next().unwrap(),
+      &tester.stdout_line(),
       "exit using ctrl+d, ctrl+c, or close()"
     );
+    assert_eq!(&tester.stderr_line(), "Debugger session started.");
 
-    assert_inspector_messages(
-    &mut socket_tx,
-    &[
-      r#"{"id":1,"method":"Runtime.enable"}"#,
-      r#"{"id":2,"method":"Debugger.enable"}"#,
-    ],
-    &mut socket_rx,
-    &[
-      r#"{"id":1,"result":{}}"#,
-      r#"{"id":2,"result":{"debuggerId":"#,
-    ],
-    &[
-      r#"{"method":"Runtime.executionContextCreated","params":{"context":{"id":1,"#,
-    ],
-  )
-  .await;
+    tester
+      .send_many(&[
+        json!({"id":1,"method":"Runtime.enable"}),
+        json!({"id":2,"method":"Debugger.enable"}),
+      ])
+      .await;
+    tester.assert_received_messages(
+      &[
+        r#"{"id":1,"result":{}}"#,
+        r#"{"id":2,"result":{"debuggerId":"#,
+      ],
+      &[
+        r#"{"method":"Runtime.executionContextCreated","params":{"context":{"id":1,"#,
+      ],
+    )
+    .await;
 
-    assert_inspector_messages(
-    &mut socket_tx,
-    &[
-      r#"{"id":3,"method":"Runtime.compileScript","params":{"expression":"Deno.cwd()","sourceURL":"","persistScript":false,"executionContextId":1}}"#,
-    ],
-    &mut socket_rx,
-    &[r#"{"id":3,"result":{}}"#], &[]
-  ).await;
-    assert_inspector_messages(
-    &mut socket_tx,
-    &[
-      r#"{"id":4,"method":"Runtime.evaluate","params":{"expression":"Deno.cwd()","objectGroup":"console","includeCommandLineAPI":true,"silent":false,"contextId":1,"returnByValue":true,"generatePreview":true,"userGesture":true,"awaitPromise":false,"replMode":true}}"#,
-    ],
-    &mut socket_rx,
-    &[r#"{"id":4,"result":{"result":{"type":"string","value":""#],
-    &[],
-  ).await;
-    assert_inspector_messages(
-    &mut socket_tx,
-    &[
-      r#"{"id":5,"method":"Runtime.evaluate","params":{"expression":"console.error('done');","objectGroup":"console","includeCommandLineAPI":true,"silent":false,"contextId":1,"returnByValue":true,"generatePreview":true,"userGesture":true,"awaitPromise":false,"replMode":true}}"#,
-    ],
-    &mut socket_rx,
-    &[r#"{"id":5,"result":{"result":{"type":"undefined"}}}"#],
-    &[r#"{"method":"Runtime.consoleAPICalled"#],
-  ).await;
-    assert_eq!(&stderr_lines.next().unwrap(), "done");
+    tester
+      .send(json!({
+        "id":3,
+        "method":"Runtime.compileScript",
+        "params":{
+          "expression":"Deno.cwd()",
+          "sourceURL":"",
+          "persistScript":false,
+          "executionContextId":1
+        }
+      }))
+      .await;
+    tester
+      .assert_received_messages(&[r#"{"id":3,"result":{}}"#], &[])
+      .await;
+    tester
+      .send(json!({
+        "id":4,
+        "method":"Runtime.evaluate",
+        "params":{
+          "expression":"Deno.cwd()",
+          "objectGroup":"console",
+          "includeCommandLineAPI":true,
+          "silent":false,
+          "contextId":1,
+          "returnByValue":true,
+          "generatePreview":true,
+          "userGesture":true,
+          "awaitPromise":false,
+          "replMode":true
+        }
+      }))
+      .await;
+    tester
+      .assert_received_messages(
+        &[r#"{"id":4,"result":{"result":{"type":"string","value":""#],
+        &[],
+      )
+      .await;
+    tester
+      .send(json!({
+        "id":5,
+        "method":"Runtime.evaluate",
+        "params":{
+          "expression":"console.error('done');",
+          "objectGroup":"console",
+          "includeCommandLineAPI":true,
+          "silent":false,
+          "contextId":1,
+          "returnByValue":true,
+          "generatePreview":true,
+          "userGesture":true,
+          "awaitPromise":false,
+          "replMode":true
+        }
+      }))
+      .await;
+    tester
+      .assert_received_messages(
+        &[r#"{"id":5,"result":{"result":{"type":"undefined"}}}"#],
+        &[r#"{"method":"Runtime.consoleAPICalled"#],
+      )
+      .await;
+    assert_eq!(&tester.stderr_line(), "done");
     drop(stdin);
-    child.wait().unwrap();
+    tester.child.wait().unwrap();
   }
 
   #[tokio::test]
@@ -685,7 +748,7 @@ mod inspector {
   #[tokio::test]
   async fn inspector_break_on_first_line_in_test() {
     let script = util::testdata_path().join("inspector/inspector_test.js");
-    let mut child = util::deno_cmd()
+    let child = util::deno_cmd()
       .arg("test")
       .arg("--quiet")
       .arg(inspect_flag_with_unique_port("--inspect-brk"))
@@ -696,87 +759,74 @@ mod inspector {
       .spawn()
       .unwrap();
 
-    let stderr = child.stderr.as_mut().unwrap();
-    let mut stderr_lines =
-      std::io::BufReader::new(stderr).lines().map(|r| r.unwrap());
-    let ws_url = extract_ws_url_from_stderr(&mut stderr_lines);
-
-    let (socket, response) =
-      tokio_tungstenite::connect_async(ws_url).await.unwrap();
-    assert_eq!(response.status(), 101); // Switching protocols.
-
-    let (mut socket_tx, socket_rx) = socket.split();
-    let mut socket_rx = socket_rx
-      .map(|msg| msg.unwrap().to_string())
-      .filter(|msg| {
-        let pass = !msg.starts_with(r#"{"method":"Debugger.scriptParsed","#);
-        futures::future::ready(pass)
-      })
-      .boxed_local();
-
-    let stdout = child.stdout.as_mut().unwrap();
-    let mut stdout_lines =
-      std::io::BufReader::new(stdout).lines().map(|r| r.unwrap());
+    let mut tester = InspectorTester::create(child, ignore_script_parsed).await;
 
-    assert_stderr_for_inspect_brk(&mut stderr_lines);
+    tester.assert_stderr_for_inspect_brk();
 
-    assert_inspector_messages(
-    &mut socket_tx,
-    &[
-      r#"{"id":1,"method":"Runtime.enable"}"#,
-      r#"{"id":2,"method":"Debugger.enable"}"#,
-    ],
-    &mut socket_rx,
-    &[
-      r#"{"id":1,"result":{}}"#,
-      r#"{"id":2,"result":{"debuggerId":"#,
-    ],
-    &[
-      r#"{"method":"Runtime.executionContextCreated","params":{"context":{"id":1,"#,
-    ],
-  )
-  .await;
-
-    assert_inspector_messages(
-      &mut socket_tx,
-      &[r#"{"id":3,"method":"Runtime.runIfWaitingForDebugger"}"#],
-      &mut socket_rx,
-      &[r#"{"id":3,"result":{}}"#],
-      &[r#"{"method":"Debugger.paused","#],
-    )
-    .await;
-
-    assert_inspector_messages(
-      &mut socket_tx,
+    tester
+      .send_many(&[
+        json!({"id":1,"method":"Runtime.enable"}),
+        json!({"id":2,"method":"Debugger.enable"}),
+      ])
+      .await;
+    tester.assert_received_messages(
       &[
-        r#"{"id":4,"method":"Runtime.evaluate","params":{"expression":"1 + 1","contextId":1,"includeCommandLineAPI":true,"silent":false,"returnByValue":true}}"#,
+        r#"{"id":1,"result":{}}"#,
+        r#"{"id":2,"result":{"debuggerId":"#,
+      ],
+      &[
+        r#"{"method":"Runtime.executionContextCreated","params":{"context":{"id":1,"#,
       ],
-      &mut socket_rx,
-      &[r#"{"id":4,"result":{"result":{"type":"number","value":2,"description":"2"}}}"#],
-      &[],
     )
     .await;
 
-    assert_inspector_messages(
-      &mut socket_tx,
-      &[r#"{"id":5,"method":"Debugger.resume"}"#],
-      &mut socket_rx,
-      &[r#"{"id":5,"result":{}}"#],
+    tester
+      .send(json!({"id":3,"method":"Runtime.runIfWaitingForDebugger"}))
+      .await;
+    tester
+      .assert_received_messages(
+        &[r#"{"id":3,"result":{}}"#],
+        &[r#"{"method":"Debugger.paused","#],
+      )
+      .await;
+
+    tester
+      .send(json!({
+        "id":4,
+        "method":"Runtime.evaluate",
+        "params":{
+          "expression":"1 + 1",
+          "contextId":1,
+          "includeCommandLineAPI":true,
+          "silent":false,
+          "returnByValue":true
+        }
+      }))
+      .await;
+    tester.assert_received_messages(
+      &[r#"{"id":4,"result":{"result":{"type":"number","value":2,"description":"2"}}}"#],
       &[],
     )
     .await;
 
-    assert_starts_with!(&stdout_lines.next().unwrap(), "running 1 test from");
-    assert!(&stdout_lines.next().unwrap().contains("basic test ... ok"));
+    tester
+      .send(json!({"id":5,"method":"Debugger.resume"}))
+      .await;
+    tester
+      .assert_received_messages(&[r#"{"id":5,"result":{}}"#], &[])
+      .await;
 
-    child.kill().unwrap();
-    child.wait().unwrap();
+    assert_starts_with!(&tester.stdout_line(), "running 1 test from");
+    assert!(&tester.stdout_line().contains("basic test ... ok"));
+
+    tester.child.kill().unwrap();
+    tester.child.wait().unwrap();
   }
 
   #[tokio::test]
   async fn inspector_with_ts_files() {
     let script = util::testdata_path().join("inspector/test.ts");
-    let mut child = util::deno_cmd()
+    let child = util::deno_cmd()
       .arg("run")
       .arg("--check")
       .arg(inspect_flag_with_unique_port("--inspect-brk"))
@@ -786,119 +836,94 @@ mod inspector {
       .spawn()
       .unwrap();
 
-    let stderr = child.stderr.as_mut().unwrap();
-    let mut stderr_lines =
-      std::io::BufReader::new(stderr).lines().map(|r| r.unwrap());
-    let ws_url = extract_ws_url_from_stderr(&mut stderr_lines);
-
-    let (socket, response) =
-      tokio_tungstenite::connect_async(ws_url).await.unwrap();
-    assert_eq!(response.status(), 101); // Switching protocols.
-
-    let (mut socket_tx, socket_rx) = socket.split();
-    let mut socket_rx = socket_rx
-      .map(|msg| msg.unwrap().to_string())
-      .filter(|msg| {
-        let pass = (msg.starts_with(r#"{"method":"Debugger.scriptParsed","#)
-          && msg.contains("testdata/inspector"))
-          || !msg.starts_with(r#"{"method":"Debugger.scriptParsed","#);
-        futures::future::ready(pass)
-      })
-      .boxed_local();
+    fn notification_filter(msg: &str) -> bool {
+      (msg.starts_with(r#"{"method":"Debugger.scriptParsed","#)
+        && msg.contains("testdata/inspector"))
+        || !msg.starts_with(r#"{"method":"Debugger.scriptParsed","#)
+    }
 
-    let stdout = child.stdout.as_mut().unwrap();
-    let mut stdout_lines =
-      std::io::BufReader::new(stdout).lines().map(|r| r.unwrap());
+    let mut tester = InspectorTester::create(child, notification_filter).await;
 
-    assert_stderr_for_inspect_brk(&mut stderr_lines);
+    tester.assert_stderr_for_inspect_brk();
 
-    assert_inspector_messages(
-    &mut socket_tx,
-    &[
-      r#"{"id":1,"method":"Runtime.enable"}"#,
-      r#"{"id":2,"method":"Debugger.enable"}"#,
-    ],
-    &mut socket_rx,
-    &[
-      r#"{"id":1,"result":{}}"#,
-    ],
-    &[
-      r#"{"method":"Runtime.executionContextCreated","params":{"context":{"id":1,"#,
-    ],
-  )
-  .await;
+    tester
+      .send_many(&[
+        json!({"id":1,"method":"Runtime.enable"}),
+        json!({"id":2,"method":"Debugger.enable"}),
+      ])
+      .await;
+    tester.assert_received_messages(
+      &[
+        r#"{"id":1,"result":{}}"#,
+      ],
+      &[
+        r#"{"method":"Runtime.executionContextCreated","params":{"context":{"id":1,"#,
+      ],
+    )
+    .await;
 
     // receive messages with sources from this test
-    let script1 = socket_rx.next().await.unwrap();
+    let script1 = tester.recv().await;
     assert!(script1.contains("testdata/inspector/test.ts"));
     let script1_id = {
       let v: serde_json::Value = serde_json::from_str(&script1).unwrap();
       v["params"]["scriptId"].as_str().unwrap().to_string()
     };
-    let script2 = socket_rx.next().await.unwrap();
+    let script2 = tester.recv().await;
     assert!(script2.contains("testdata/inspector/foo.ts"));
     let script2_id = {
       let v: serde_json::Value = serde_json::from_str(&script2).unwrap();
       v["params"]["scriptId"].as_str().unwrap().to_string()
     };
-    let script3 = socket_rx.next().await.unwrap();
+    let script3 = tester.recv().await;
     assert!(script3.contains("testdata/inspector/bar.js"));
     let script3_id = {
       let v: serde_json::Value = serde_json::from_str(&script3).unwrap();
       v["params"]["scriptId"].as_str().unwrap().to_string()
     };
 
-    assert_inspector_messages(
-      &mut socket_tx,
-      &[],
-      &mut socket_rx,
-      &[r#"{"id":2,"result":{"debuggerId":"#],
-      &[],
-    )
-    .await;
+    tester
+      .assert_received_messages(&[r#"{"id":2,"result":{"debuggerId":"#], &[])
+      .await;
 
-    assert_inspector_messages(
-      &mut socket_tx,
-      &[r#"{"id":3,"method":"Runtime.runIfWaitingForDebugger"}"#],
-      &mut socket_rx,
-      &[r#"{"id":3,"result":{}}"#],
-      &[r#"{"method":"Debugger.paused","#],
-    )
-    .await;
+    tester
+      .send(json!({"id":3,"method":"Runtime.runIfWaitingForDebugger"}))
+      .await;
+    tester
+      .assert_received_messages(
+        &[r#"{"id":3,"result":{}}"#],
+        &[r#"{"method":"Debugger.paused","#],
+      )
+      .await;
 
-    assert_inspector_messages(
-    &mut socket_tx,
-    &[
-      &format!(r#"{{"id":4,"method":"Debugger.getScriptSource","params":{{"scriptId":"{}"}}}}"#, script1_id),
-      &format!(r#"{{"id":5,"method":"Debugger.getScriptSource","params":{{"scriptId":"{}"}}}}"#, script2_id),
-      &format!(r#"{{"id":6,"method":"Debugger.getScriptSource","params":{{"scriptId":"{}"}}}}"#, script3_id),
-    ],
-    &mut socket_rx,
+    tester.send_many(
     &[
-      r#"{"id":4,"result":{"scriptSource":"import { foo } from \"./foo.ts\";\nimport { bar } from \"./bar.js\";\nconsole.log(foo());\nconsole.log(bar());\n//# sourceMappingURL=data:application/json;base64,"#,
-      r#"{"id":5,"result":{"scriptSource":"class Foo {\n    hello() {\n        return \"hello\";\n    }\n}\nexport function foo() {\n    const f = new Foo();\n    return f.hello();\n}\n//# sourceMappingURL=data:application/json;base64,"#,
-      r#"{"id":6,"result":{"scriptSource":"export function bar() {\n  return \"world\";\n}\n"#,
-    ],
-    &[],
-  )
-  .await;
-
-    assert_inspector_messages(
-      &mut socket_tx,
-      &[r#"{"id":7,"method":"Debugger.resume"}"#],
-      &mut socket_rx,
-      &[r#"{"id":7,"result":{}}"#],
+      json!({"id":4,"method":"Debugger.getScriptSource","params":{"scriptId":script1_id.as_str()}}),
+      json!({"id":5,"method":"Debugger.getScriptSource","params":{"scriptId":script2_id.as_str()}}),
+      json!({"id":6,"method":"Debugger.getScriptSource","params":{"scriptId":script3_id.as_str()}}),
+    ])
+    .await;
+    tester.assert_received_messages(
+      &[
+        r#"{"id":4,"result":{"scriptSource":"import { foo } from \"./foo.ts\";\nimport { bar } from \"./bar.js\";\nconsole.log(foo());\nconsole.log(bar());\n//# sourceMappingURL=data:application/json;base64,"#,
+        r#"{"id":5,"result":{"scriptSource":"class Foo {\n    hello() {\n        return \"hello\";\n    }\n}\nexport function foo() {\n    const f = new Foo();\n    return f.hello();\n}\n//# sourceMappingURL=data:application/json;base64,"#,
+        r#"{"id":6,"result":{"scriptSource":"export function bar() {\n  return \"world\";\n}\n"#,
+      ],
       &[],
     )
     .await;
 
-    assert_eq!(&stdout_lines.next().unwrap(), "hello");
-    assert_eq!(&stdout_lines.next().unwrap(), "world");
+    tester
+      .send(json!({"id":7,"method":"Debugger.resume"}))
+      .await;
+    tester
+      .assert_received_messages(&[r#"{"id":7,"result":{}}"#], &[])
+      .await;
+
+    assert_eq!(&tester.stdout_line(), "hello");
+    assert_eq!(&tester.stdout_line(), "world");
 
-    assert_inspector_messages(
-      &mut socket_tx,
-      &[],
-      &mut socket_rx,
+    tester.assert_received_messages(
       &[],
       &[
         r#"{"method":"Debugger.resumed","params":{}}"#,
@@ -908,21 +933,20 @@ mod inspector {
       ],
     )
     .await;
-    let line = &stdout_lines.next().unwrap();
 
     assert_eq!(
-      line,
+      &tester.stdout_line(),
       "Program finished. Waiting for inspector to disconnect to exit the process..."
     );
 
-    child.kill().unwrap();
-    child.wait().unwrap();
+    tester.child.kill().unwrap();
+    tester.child.wait().unwrap();
   }
 
   #[tokio::test]
   async fn inspector_memory() {
     let script = util::testdata_path().join("inspector/memory.js");
-    let mut child = util::deno_cmd()
+    let child = util::deno_cmd()
       .arg("run")
       .arg(inspect_flag_with_unique_port("--inspect-brk"))
       .arg(script)
@@ -931,66 +955,45 @@ mod inspector {
       .spawn()
       .unwrap();
 
-    let stderr = child.stderr.as_mut().unwrap();
-    let mut stderr_lines =
-      std::io::BufReader::new(stderr).lines().map(|r| r.unwrap());
-    let ws_url = extract_ws_url_from_stderr(&mut stderr_lines);
-
-    let (socket, response) =
-      tokio_tungstenite::connect_async(ws_url).await.unwrap();
-    assert_eq!(response.status(), 101); // Switching protocols.
-
-    let (mut socket_tx, socket_rx) = socket.split();
-    let mut socket_rx = socket_rx
-      .map(|msg| msg.unwrap().to_string())
-      .filter(|msg| {
-        let pass = !msg.starts_with(r#"{"method":"Debugger.scriptParsed","#);
-        futures::future::ready(pass)
-      })
-      .boxed_local();
-
-    assert_stderr_for_inspect_brk(&mut stderr_lines);
-
-    assert_inspector_messages(
-    &mut socket_tx,
-    &[
-      r#"{"id":1,"method":"Runtime.enable"}"#,
-      r#"{"id":2,"method":"Debugger.enable"}"#,
+    let mut tester = InspectorTester::create(child, ignore_script_parsed).await;
 
-    ],
-    &mut socket_rx,
-    &[
-      r#"{"id":1,"result":{}}"#,
-      r#"{"id":2,"result":{"debuggerId":"#,
-    ],
-    &[
-      r#"{"method":"Runtime.executionContextCreated","params":{"context":{"id":1,"#,
-    ],
-  )
-  .await;
+    tester.assert_stderr_for_inspect_brk();
 
-    assert_inspector_messages(
-      &mut socket_tx,
+    tester
+      .send_many(&[
+        json!({"id":1,"method":"Runtime.enable"}),
+        json!({"id":2,"method":"Debugger.enable"}),
+      ])
+      .await;
+    tester.assert_received_messages(
       &[
-        r#"{"id":3,"method":"Runtime.runIfWaitingForDebugger"}"#,
-        r#"{"id":4,"method":"HeapProfiler.enable"}"#,
+        r#"{"id":1,"result":{}}"#,
+        r#"{"id":2,"result":{"debuggerId":"#,
+      ],
+      &[
+        r#"{"method":"Runtime.executionContextCreated","params":{"context":{"id":1,"#,
       ],
-      &mut socket_rx,
-      &[r#"{"id":3,"result":{}}"#, r#"{"id":4,"result":{}}"#],
-      &[r#"{"method":"Debugger.paused","#],
     )
     .await;
 
-    socket_tx
-      .send(
-        r#"{"id":5,"method":"Runtime.getHeapUsage", "params": {}}"#
-          .to_string()
-          .into(),
+    tester
+      .send_many(&[
+        json!({"id":3,"method":"Runtime.runIfWaitingForDebugger"}),
+        json!({"id":4,"method":"HeapProfiler.enable"}),
+      ])
+      .await;
+    tester
+      .assert_received_messages(
+        &[r#"{"id":3,"result":{}}"#, r#"{"id":4,"result":{}}"#],
+        &[r#"{"method":"Debugger.paused","#],
       )
-      .await
-      .unwrap();
-    let msg = socket_rx.next().await.unwrap();
-    let json_msg: serde_json::Value = serde_json::from_str(&msg).unwrap();
+      .await;
+
+    tester
+      .send(json!({"id":5,"method":"Runtime.getHeapUsage", "params": {}}))
+      .await;
+
+    let json_msg = tester.recv_as_json().await;
     assert_eq!(json_msg["id"].as_i64().unwrap(), 5);
     let result = &json_msg["result"];
     assert!(
@@ -998,15 +1001,23 @@ mod inspector {
         <= result["totalSize"].as_i64().unwrap()
     );
 
-    socket_tx.send(
-    r#"{"id":6,"method":"HeapProfiler.takeHeapSnapshot","params": {"reportProgress": true, "treatGlobalObjectsAsRoots": true, "captureNumberValue": false}}"#
-      .to_string().into()
-  ).await.unwrap();
+    tester
+      .send(json!({
+        "id":6,
+        "method":"HeapProfiler.takeHeapSnapshot",
+        "params": {
+          "reportProgress": true,
+          "treatGlobalObjectsAsRoots": true,
+          "captureNumberValue": false
+        }
+      }))
+      .await;
 
     let mut progress_report_completed = false;
     loop {
-      let msg = socket_rx.next().await.unwrap();
+      let msg = tester.recv().await;
 
+      // TODO(bartlomieju): can be abstracted
       if !progress_report_completed
         && msg.starts_with(
           r#"{"method":"HeapProfiler.reportHeapSnapshotProgress","params""#,
@@ -1020,8 +1031,8 @@ mod inspector {
       }
 
       if msg.starts_with(r#"{"method":"HeapProfiler.reportHeapSnapshotProgress","params":{"done":"#,) {
-      continue;
-    }
+        continue;
+      }
 
       if msg.starts_with(r#"{"id":6,"result":{}}"#) {
         assert!(progress_report_completed);
@@ -1029,14 +1040,14 @@ mod inspector {
       }
     }
 
-    child.kill().unwrap();
-    child.wait().unwrap();
+    tester.child.kill().unwrap();
+    tester.child.wait().unwrap();
   }
 
   #[tokio::test]
   async fn inspector_profile() {
     let script = util::testdata_path().join("inspector/memory.js");
-    let mut child = util::deno_cmd()
+    let child = util::deno_cmd()
       .arg("run")
       .arg(inspect_flag_with_unique_port("--inspect-brk"))
       .arg(script)
@@ -1045,78 +1056,59 @@ mod inspector {
       .spawn()
       .unwrap();
 
-    let stderr = child.stderr.as_mut().unwrap();
-    let mut stderr_lines =
-      std::io::BufReader::new(stderr).lines().map(|r| r.unwrap());
-    let ws_url = extract_ws_url_from_stderr(&mut stderr_lines);
+    let mut tester = InspectorTester::create(child, ignore_script_parsed).await;
 
-    let (socket, response) =
-      tokio_tungstenite::connect_async(ws_url).await.unwrap();
-    assert_eq!(response.status(), 101); // Switching protocols.
+    tester.assert_stderr_for_inspect_brk();
 
-    let (mut socket_tx, socket_rx) = socket.split();
-    let mut socket_rx = socket_rx
-      .map(|msg| msg.unwrap().to_string())
-      .filter(|msg| {
-        let pass = !msg.starts_with(r#"{"method":"Debugger.scriptParsed","#);
-        futures::future::ready(pass)
-      })
-      .boxed_local();
-
-    assert_stderr_for_inspect_brk(&mut stderr_lines);
-
-    assert_inspector_messages(
-    &mut socket_tx,
-    &[
-      r#"{"id":1,"method":"Runtime.enable"}"#,
-      r#"{"id":2,"method":"Debugger.enable"}"#,
-
-    ],
-    &mut socket_rx,
-    &[
-      r#"{"id":1,"result":{}}"#,
-      r#"{"id":2,"result":{"debuggerId":"#,
-    ],
-    &[
-      r#"{"method":"Runtime.executionContextCreated","params":{"context":{"id":1,"#,
-    ],
-  )
-  .await;
-
-    assert_inspector_messages(
-      &mut socket_tx,
+    tester
+      .send_many(&[
+        json!({"id":1,"method":"Runtime.enable"}),
+        json!({"id":2,"method":"Debugger.enable"}),
+      ])
+      .await;
+    tester.assert_received_messages(
       &[
-        r#"{"id":3,"method":"Runtime.runIfWaitingForDebugger"}"#,
-        r#"{"id":4,"method":"Profiler.enable"}"#,
+        r#"{"id":1,"result":{}}"#,
+        r#"{"id":2,"result":{"debuggerId":"#,
+      ],
+      &[
+        r#"{"method":"Runtime.executionContextCreated","params":{"context":{"id":1,"#,
       ],
-      &mut socket_rx,
-      &[r#"{"id":3,"result":{}}"#, r#"{"id":4,"result":{}}"#],
-      &[r#"{"method":"Debugger.paused","#],
     )
     .await;
 
-    assert_inspector_messages(
-    &mut socket_tx,
-    &[
-      r#"{"id":5,"method":"Profiler.setSamplingInterval","params":{"interval": 100}}"#,
-      r#"{"id":6,"method":"Profiler.start","params":{}}"#,
-    ],
-    &mut socket_rx,
-    &[r#"{"id":5,"result":{}}"#, r#"{"id":6,"result":{}}"#],
-    &[],
-  )
-  .await;
+    tester
+      .send_many(&[
+        json!({"id":3,"method":"Runtime.runIfWaitingForDebugger"}),
+        json!({"id":4,"method":"Profiler.enable"}),
+      ])
+      .await;
+    tester
+      .assert_received_messages(
+        &[r#"{"id":3,"result":{}}"#, r#"{"id":4,"result":{}}"#],
+        &[r#"{"method":"Debugger.paused","#],
+      )
+      .await;
+
+    tester.send_many(
+      &[
+        json!({"id":5,"method":"Profiler.setSamplingInterval","params":{"interval": 100}}),
+        json!({"id":6,"method":"Profiler.start","params":{}}),
+      ],
+    ).await;
+    tester
+      .assert_received_messages(
+        &[r#"{"id":5,"result":{}}"#, r#"{"id":6,"result":{}}"#],
+        &[],
+      )
+      .await;
 
     tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
 
-    socket_tx
-      .send(
-        r#"{"id":7,"method":"Profiler.stop", "params": {}}"#.to_string().into(),
-      )
-      .await
-      .unwrap();
-    let msg = socket_rx.next().await.unwrap();
-    let json_msg: serde_json::Value = serde_json::from_str(&msg).unwrap();
+    tester
+      .send(json!({"id":7,"method":"Profiler.stop", "params": {}}))
+      .await;
+    let json_msg = tester.recv_as_json().await;
     assert_eq!(json_msg["id"].as_i64().unwrap(), 7);
     let result = &json_msg["result"];
     let profile = &result["profile"];
@@ -1127,15 +1119,15 @@ mod inspector {
     profile["samples"].as_array().unwrap();
     profile["nodes"].as_array().unwrap();
 
-    child.kill().unwrap();
-    child.wait().unwrap();
+    tester.child.kill().unwrap();
+    tester.child.wait().unwrap();
   }
 
   #[tokio::test]
   async fn inspector_break_on_first_line_npm_esm() {
     let _server = http_server();
 
-    let mut child = util::deno_cmd()
+    let child = util::deno_cmd()
       .arg("run")
       .arg("--quiet")
       .arg(inspect_flag_with_unique_port("--inspect-brk"))
@@ -1150,37 +1142,17 @@ mod inspector {
       .spawn()
       .unwrap();
 
-    let stderr = child.stderr.as_mut().unwrap();
-    let mut stderr_lines =
-      std::io::BufReader::new(stderr).lines().map(|r| r.unwrap());
-    let ws_url = extract_ws_url_from_stderr(&mut stderr_lines);
-
-    let (socket, response) =
-      tokio_tungstenite::connect_async(ws_url).await.unwrap();
-    assert_eq!(response.status(), 101); // Switching protocols.
+    let mut tester = InspectorTester::create(child, ignore_script_parsed).await;
 
-    let (mut socket_tx, socket_rx) = socket.split();
-    let mut socket_rx = socket_rx
-      .map(|msg| msg.unwrap().to_string())
-      .filter(|msg| {
-        let pass = !msg.starts_with(r#"{"method":"Debugger.scriptParsed","#);
-        futures::future::ready(pass)
-      })
-      .boxed_local();
+    tester.assert_stderr_for_inspect_brk();
 
-    let stdout = child.stdout.as_mut().unwrap();
-    let mut stdout_lines =
-      std::io::BufReader::new(stdout).lines().map(|r| r.unwrap());
-
-    assert_stderr_for_inspect_brk(&mut stderr_lines);
-
-    assert_inspector_messages(
-      &mut socket_tx,
-      &[
-        r#"{"id":1,"method":"Runtime.enable"}"#,
-        r#"{"id":2,"method":"Debugger.enable"}"#,
-      ],
-      &mut socket_rx,
+    tester
+      .send_many(&[
+        json!({"id":1,"method":"Runtime.enable"}),
+        json!({"id":2,"method":"Debugger.enable"}),
+      ])
+      .await;
+    tester.assert_received_messages(
       &[
         r#"{"id":1,"result":{}}"#,
         r#"{"id":2,"result":{"debuggerId":"#,
@@ -1191,37 +1163,36 @@ mod inspector {
     )
     .await;
 
-    assert_inspector_messages(
-      &mut socket_tx,
-      &[r#"{"id":3,"method":"Runtime.runIfWaitingForDebugger"}"#],
-      &mut socket_rx,
-      &[r#"{"id":3,"result":{}}"#],
-      &[r#"{"method":"Debugger.paused","#],
-    )
-    .await;
+    tester
+      .send(json!({"id":3,"method":"Runtime.runIfWaitingForDebugger"}))
+      .await;
+    tester
+      .assert_received_messages(
+        &[r#"{"id":3,"result":{}}"#],
+        &[r#"{"method":"Debugger.paused","#],
+      )
+      .await;
 
-    assert_inspector_messages(
-      &mut socket_tx,
-      &[r#"{"id":4,"method":"Debugger.resume"}"#],
-      &mut socket_rx,
-      &[r#"{"id":4,"result":{}}"#],
-      &[],
-    )
-    .await;
+    tester
+      .send(json!({"id":4,"method":"Debugger.resume"}))
+      .await;
+    tester
+      .assert_received_messages(&[r#"{"id":4,"result":{}}"#], &[])
+      .await;
 
-    assert_eq!(&stdout_lines.next().unwrap(), "this");
-    assert_eq!(&stdout_lines.next().unwrap(), "is");
-    assert_eq!(&stdout_lines.next().unwrap(), "a");
-    assert_eq!(&stdout_lines.next().unwrap(), "test");
+    assert_eq!(&tester.stdout_line(), "this");
+    assert_eq!(&tester.stdout_line(), "is");
+    assert_eq!(&tester.stdout_line(), "a");
+    assert_eq!(&tester.stdout_line(), "test");
 
-    child.kill().unwrap();
-    child.wait().unwrap();
+    tester.child.kill().unwrap();
+    tester.child.wait().unwrap();
   }
 
   #[tokio::test]
   async fn inspector_break_on_first_line_npm_cjs() {
     let _server = http_server();
-    let mut child = util::deno_cmd()
+    let child = util::deno_cmd()
       .arg("run")
       .arg("--quiet")
       .arg(inspect_flag_with_unique_port("--inspect-brk"))
@@ -1236,37 +1207,17 @@ mod inspector {
       .spawn()
       .unwrap();
 
-    let stderr = child.stderr.as_mut().unwrap();
-    let mut stderr_lines =
-      std::io::BufReader::new(stderr).lines().map(|r| r.unwrap());
-    let ws_url = extract_ws_url_from_stderr(&mut stderr_lines);
-
-    let (socket, response) =
-      tokio_tungstenite::connect_async(ws_url).await.unwrap();
-    assert_eq!(response.status(), 101); // Switching protocols.
-
-    let (mut socket_tx, socket_rx) = socket.split();
-    let mut socket_rx = socket_rx
-      .map(|msg| msg.unwrap().to_string())
-      .filter(|msg| {
-        let pass = !msg.starts_with(r#"{"method":"Debugger.scriptParsed","#);
-        futures::future::ready(pass)
-      })
-      .boxed_local();
-
-    let stdout = child.stdout.as_mut().unwrap();
-    let mut stdout_lines =
-      std::io::BufReader::new(stdout).lines().map(|r| r.unwrap());
+    let mut tester = InspectorTester::create(child, ignore_script_parsed).await;
 
-    assert_stderr_for_inspect_brk(&mut stderr_lines);
+    tester.assert_stderr_for_inspect_brk();
 
-    assert_inspector_messages(
-      &mut socket_tx,
-      &[
-        r#"{"id":1,"method":"Runtime.enable"}"#,
-        r#"{"id":2,"method":"Debugger.enable"}"#,
-      ],
-      &mut socket_rx,
+    tester
+      .send_many(&[
+        json!({"id":1,"method":"Runtime.enable"}),
+        json!({"id":2,"method":"Debugger.enable"}),
+      ])
+      .await;
+    tester.assert_received_messages(
       &[
         r#"{"id":1,"result":{}}"#,
         r#"{"id":2,"result":{"debuggerId":"#,
@@ -1277,31 +1228,30 @@ mod inspector {
     )
     .await;
 
-    assert_inspector_messages(
-      &mut socket_tx,
-      &[r#"{"id":3,"method":"Runtime.runIfWaitingForDebugger"}"#],
-      &mut socket_rx,
-      &[r#"{"id":3,"result":{}}"#],
-      &[r#"{"method":"Debugger.paused","#],
-    )
-    .await;
+    tester
+      .send(json!({"id":3,"method":"Runtime.runIfWaitingForDebugger"}))
+      .await;
+    tester
+      .assert_received_messages(
+        &[r#"{"id":3,"result":{}}"#],
+        &[r#"{"method":"Debugger.paused","#],
+      )
+      .await;
 
-    assert_inspector_messages(
-      &mut socket_tx,
-      &[r#"{"id":4,"method":"Debugger.resume"}"#],
-      &mut socket_rx,
-      &[r#"{"id":4,"result":{}}"#],
-      &[],
-    )
-    .await;
+    tester
+      .send(json!({"id":4,"method":"Debugger.resume"}))
+      .await;
+    tester
+      .assert_received_messages(&[r#"{"id":4,"result":{}}"#], &[])
+      .await;
 
-    assert_eq!(&stdout_lines.next().unwrap(), "this");
-    assert_eq!(&stdout_lines.next().unwrap(), "is");
-    assert_eq!(&stdout_lines.next().unwrap(), "a");
-    assert_eq!(&stdout_lines.next().unwrap(), "test");
+    assert_eq!(&tester.stdout_line(), "this");
+    assert_eq!(&tester.stdout_line(), "is");
+    assert_eq!(&tester.stdout_line(), "a");
+    assert_eq!(&tester.stdout_line(), "test");
 
-    child.kill().unwrap();
-    child.wait().unwrap();
+    tester.child.kill().unwrap();
+    tester.child.wait().unwrap();
   }
 
   #[tokio::test]
@@ -1310,7 +1260,7 @@ mod inspector {
       util::testdata_path().join("inspector/error_with_npm_import.js");
     let _server = http_server();
 
-    let mut child = util::deno_cmd()
+    let child = util::deno_cmd()
       .arg("run")
       .arg("--quiet")
       .arg("-A")
@@ -1322,33 +1272,17 @@ mod inspector {
       .spawn()
       .unwrap();
 
-    let stderr = child.stderr.as_mut().unwrap();
-    let mut stderr_lines =
-      std::io::BufReader::new(stderr).lines().map(|r| r.unwrap());
-    let ws_url = extract_ws_url_from_stderr(&mut stderr_lines);
-
-    let (socket, response) =
-      tokio_tungstenite::connect_async(ws_url).await.unwrap();
-    assert_eq!(response.status(), 101); // Switching protocols.
-
-    let (mut socket_tx, socket_rx) = socket.split();
-    let mut socket_rx = socket_rx
-      .map(|msg| msg.unwrap().to_string())
-      .filter(|msg| {
-        let pass = !msg.starts_with(r#"{"method":"Debugger.scriptParsed","#);
-        futures::future::ready(pass)
-      })
-      .boxed_local();
+    let mut tester = InspectorTester::create(child, ignore_script_parsed).await;
 
-    assert_stderr_for_inspect_brk(&mut stderr_lines);
+    tester.assert_stderr_for_inspect_brk();
 
-    assert_inspector_messages(
-      &mut socket_tx,
-      &[
-        r#"{"id":1,"method":"Runtime.enable"}"#,
-        r#"{"id":2,"method":"Debugger.enable"}"#,
-      ],
-      &mut socket_rx,
+    tester
+      .send_many(&[
+        json!({"id":1,"method":"Runtime.enable"}),
+        json!({"id":2,"method":"Debugger.enable"}),
+      ])
+      .await;
+    tester.assert_received_messages(
       &[
         r#"{"id":1,"result":{}}"#,
         r#"{"id":2,"result":{"debuggerId":"#,
@@ -1359,33 +1293,29 @@ mod inspector {
     )
     .await;
 
-    assert_inspector_messages(
-      &mut socket_tx,
-      &[r#"{"id":3,"method":"Runtime.runIfWaitingForDebugger"}"#],
-      &mut socket_rx,
-      &[r#"{"id":3,"result":{}}"#],
-      &[r#"{"method":"Debugger.paused","#],
-    )
-    .await;
+    tester
+      .send(json!({"id":3,"method":"Runtime.runIfWaitingForDebugger"}))
+      .await;
+    tester
+      .assert_received_messages(
+        &[r#"{"id":3,"result":{}}"#],
+        &[r#"{"method":"Debugger.paused","#],
+      )
+      .await;
 
-    assert_inspector_messages(
-      &mut socket_tx,
-      &[r#"{"id":4,"method":"Debugger.resume"}"#],
-      &mut socket_rx,
-      &[r#"{"id":4,"result":{}}"#],
-      &[],
-    )
-    .await;
+    tester
+      .send(json!({"id":4,"method":"Debugger.resume"}))
+      .await;
+    tester
+      .assert_received_messages(&[r#"{"id":4,"result":{}}"#], &[])
+      .await;
 
     // TODO(bartlomieju): this is a partial fix, we should assert that
     // "Runtime.exceptionThrown" notification was sent, but a bindings for this
     // notification is not yet there
-    assert_eq!(&stderr_lines.next().unwrap(), "Debugger session started.");
-    assert_eq!(
-      &stderr_lines.next().unwrap(),
-      "error: Uncaught Error: boom!"
-    );
+    assert_eq!(&tester.stderr_line(), "Debugger session started.");
+    assert_eq!(&tester.stderr_line(), "error: Uncaught Error: boom!");
 
-    assert_eq!(child.wait().unwrap().code(), Some(1));
+    assert_eq!(tester.child.wait().unwrap().code(), Some(1));
   }
 }