RUST-604 Use mongodb::Client for unified runner entities (#314)

patrickfreed · web-flow · commit 8148ee12ad2a · 2021-04-05T16:50:13.000-04:00
diff --git a/src/client/session/test.rs b/src/client/session/test.rs
@@ -242,7 +242,7 @@ async fn cluster_time_in_commands() {
     {
         let mut options = CLIENT_OPTIONS.clone();
         options.heartbeat_freq = Some(Duration::from_secs(1000));
-        let client = EventClient::with_options(options, true).await;
+        let client = EventClient::with_options(options).await;
 
         operation(client.clone())
             .await
diff --git a/src/cmap/test/integration.rs b/src/cmap/test/integration.rs
@@ -79,10 +79,10 @@ async fn concurrent_connections() {
     options.direct_connection = Some(true);
     options.hosts.drain(1..);
 
-    let client = TestClient::with_options(Some(options), true).await;
+    let client = TestClient::with_options(Some(options)).await;
     let version = VersionReq::parse(">= 4.2.9").unwrap();
     // blockConnection failpoint option only supported in 4.2.9+.
-    if !version.matches(&client.server_version.as_ref().unwrap()) {
+    if !version.matches(&client.server_version) {
         println!(
             "skipping concurrent_connections test due to server not supporting failpoint option"
         );
@@ -166,7 +166,7 @@ async fn connection_error_during_establishment() {
     client_options.direct_connection = Some(true);
     client_options.repl_set_name = None;
 
-    let client = TestClient::with_options(Some(client_options.clone()), true).await;
+    let client = TestClient::with_options(Some(client_options.clone())).await;
     if !client.supports_fail_command().await {
         println!(
             "skipping {} due to failCommand not being supported",
@@ -217,7 +217,7 @@ async fn connection_error_during_operation() {
     options.hosts.drain(1..);
     options.max_pool_size = Some(1);
 
-    let client = TestClient::with_options(options.into(), true).await;
+    let client = TestClient::with_options(options.into()).await;
     if !client.supports_fail_command().await {
         println!(
             "skipping {} due to failCommand not being supported",
diff --git a/src/cmap/test/mod.rs b/src/cmap/test/mod.rs
@@ -392,7 +392,7 @@ async fn cmap_spec_tests() {
         let mut options = CLIENT_OPTIONS.clone();
         options.hosts.drain(1..);
         options.direct_connection = Some(true);
-        let client = EventClient::with_options(options, true).await;
+        let client = EventClient::with_options(options).await;
         if let Some(ref run_on) = test_file.run_on {
             let can_run_on = run_on.iter().any(|run_on| run_on.can_run_on(&client));
             if !can_run_on {
diff --git a/src/sdam/description/topology/server_selection/test/in_window.rs b/src/sdam/description/topology/server_selection/test/in_window.rs
@@ -115,11 +115,11 @@ async fn load_balancing_test() {
     let mut setup_client_options = CLIENT_OPTIONS.clone();
     setup_client_options.hosts.drain(1..);
     setup_client_options.direct_connection = Some(true);
-    let setup_client = TestClient::with_options(Some(setup_client_options), true).await;
+    let setup_client = TestClient::with_options(Some(setup_client_options)).await;
 
     let version = VersionReq::parse(">= 4.2.9").unwrap();
     // blockConnection failpoint option only supported in 4.2.9+.
-    if !version.matches(&setup_client.server_version.as_ref().unwrap()) {
+    if !version.matches(&setup_client.server_version) {
         println!(
             "skipping load_balancing_test test due to server not supporting blockConnection option"
         );
diff --git a/src/sdam/test.rs b/src/sdam/test.rs
@@ -34,7 +34,7 @@ async fn min_heartbeat_frequency() {
     setup_client_options.hosts.drain(1..);
     setup_client_options.direct_connection = Some(true);
 
-    let setup_client = TestClient::with_options(Some(setup_client_options.clone()), true).await;
+    let setup_client = TestClient::with_options(Some(setup_client_options.clone())).await;
 
     if !setup_client.supports_fail_command().await {
         println!("skipping min_heartbeat_frequency test due to server not supporting fail points");
@@ -102,13 +102,12 @@ async fn sdam_pool_management() {
         Some(Duration::from_millis(50)),
         None,
         event_handler.clone(),
-        true,
     )
     .await;
 
     if !VersionReq::parse(">= 4.2.9")
         .unwrap()
-        .matches(client.server_version.as_ref().unwrap())
+        .matches(&client.server_version)
     {
         println!(
             "skipping sdam_pool_management test due to server not supporting appName failCommand"
@@ -160,11 +159,11 @@ async fn sdam_min_pool_size_error() {
     setup_client_options.hosts.drain(1..);
     setup_client_options.direct_connection = Some(true);
 
-    let setup_client = TestClient::with_options(Some(setup_client_options.clone()), true).await;
+    let setup_client = TestClient::with_options(Some(setup_client_options.clone())).await;
 
     if !VersionReq::parse(">= 4.9.0")
         .unwrap()
-        .matches(setup_client.server_version.as_ref().unwrap())
+        .matches(&setup_client.server_version)
     {
         println!(
             "skipping sdam_pool_management test due to server not supporting appName failCommand"
@@ -243,10 +242,10 @@ async fn auth_error() {
 
     let mut setup_client_options = CLIENT_OPTIONS.clone();
     setup_client_options.hosts.drain(1..);
-    let setup_client = TestClient::with_options(Some(setup_client_options.clone()), true).await;
+    let setup_client = TestClient::with_options(Some(setup_client_options.clone())).await;
     if !VersionReq::parse(">= 4.4.0")
         .unwrap()
-        .matches(setup_client.server_version.as_ref().unwrap())
+        .matches(&setup_client.server_version)
     {
         println!("skipping auth_error test due to server not supporting appName failCommand");
         return;
diff --git a/src/test/client.rs b/src/test/client.rs
@@ -601,7 +601,7 @@ async fn x509_auth() {
             .build(),
     );
 
-    let client = TestClient::with_options(Some(options), true).await;
+    let client = TestClient::with_options(Some(options)).await;
     client
         .database(function_name!())
         .collection(function_name!())
diff --git a/src/test/coll.rs b/src/test/coll.rs
@@ -654,7 +654,7 @@ async fn find_one_and_delete_hint_test(options: Option<FindOneAndDeleteOptions>,
     let client = EventClient::new().await;
 
     let req = VersionReq::parse(">= 4.2").unwrap();
-    if options.is_some() && !req.matches(&client.server_version.as_ref().unwrap()) {
+    if options.is_some() && !req.matches(&client.server_version) {
         return;
     }
 
@@ -712,10 +712,10 @@ async fn find_one_and_delete_hint_server_version() {
 
     let req1 = VersionReq::parse("< 4.2").unwrap();
     let req2 = VersionReq::parse("4.2.*").unwrap();
-    if req1.matches(&client.server_version.as_ref().unwrap()) {
+    if req1.matches(&client.server_version) {
         let error = res.expect_err("find one and delete should fail");
         assert!(matches!(error.kind, ErrorKind::OperationError { .. }));
-    } else if req2.matches(&client.server_version.as_ref().unwrap()) {
+    } else if req2.matches(&client.server_version) {
         let error = res.expect_err("find one and delete should fail");
         assert!(matches!(error.kind, ErrorKind::CommandError { .. }));
     } else {
@@ -953,7 +953,7 @@ async fn count_documents_with_wc() {
         .build()
         .into();
 
-    let client = TestClient::with_options(Some(options), true).await;
+    let client = TestClient::with_options(Some(options)).await;
     let coll = client
         .database(function_name!())
         .collection(function_name!());
diff --git a/src/test/spec/command_monitoring/mod.rs b/src/test/spec/command_monitoring/mod.rs
@@ -58,15 +58,15 @@ async fn run_command_monitoring_test(test_file: TestFile) {
 
         if let Some(ref max_version) = test_case.max_version {
             let req = VersionReq::parse(&format!("<= {}", &max_version)).unwrap();
-            if !req.matches(&client.server_version.as_ref().unwrap()) {
+            if !req.matches(&client.server_version) {
                 println!("Skipping {}", test_case.description);
                 continue;
             }
         }
 
         if let Some(ref min_version) = test_case.min_version {
             let req = VersionReq::parse(&format!(">= {}", &min_version)).unwrap();
-            if !req.matches(&client.server_version.as_ref().unwrap()) {
+            if !req.matches(&client.server_version) {
                 println!("Skipping {}", test_case.description);
                 continue;
             }
@@ -89,8 +89,7 @@ async fn run_command_monitoring_test(test_file: TestFile) {
             .hosts(CLIENT_OPTIONS.hosts.clone())
             .build();
         let client =
-            EventClient::with_additional_options(Some(options), None, Some(false), None, true)
-                .await;
+            EventClient::with_additional_options(Some(options), None, Some(false), None).await;
 
         let events: Vec<TestEvent> = client
             .run_operation_with_events(
diff --git a/src/test/spec/connection_stepdown.rs b/src/test/spec/connection_stepdown.rs
@@ -15,7 +15,7 @@ use crate::{
         InsertManyOptions,
         WriteConcern,
     },
-    test::{util::EventClient, CLIENT_OPTIONS, LOCK},
+    test::{util::EventClient, LOCK},
     Collection,
     Database,
     RUNTIME,
@@ -24,11 +24,8 @@ use crate::{
 async fn run_test<F: Future>(name: &str, test: impl Fn(EventClient, Database, Collection) -> F) {
     let _guard: RwLockWriteGuard<()> = LOCK.run_exclusively().await;
 
-    let options = ClientOptions::builder()
-        .retry_writes(false)
-        .hosts(CLIENT_OPTIONS.hosts.clone())
-        .build();
-    let client = EventClient::with_additional_options(Some(options), None, None, None, true).await;
+    let options = ClientOptions::builder().retry_writes(false).build();
+    let client = EventClient::with_additional_options(Some(options), None, None, None).await;
 
     if !client.is_replica_set() {
         return;
diff --git a/src/test/spec/retryable_writes/mod.rs b/src/test/spec/retryable_writes/mod.rs
@@ -41,7 +41,6 @@ async fn run_spec_tests() {
                 Some(Duration::from_millis(50)),
                 test_case.use_multiple_mongoses,
                 None,
-                true,
             )
             .await;
 
@@ -200,7 +199,7 @@ async fn transaction_ids_excluded() {
     assert!(excludes_txn_number("aggregate"));
 
     let req = semver::VersionReq::parse(">=4.2").unwrap();
-    if req.matches(&client.server_version.as_ref().unwrap()) {
+    if req.matches(&client.server_version) {
         coll.aggregate(
             vec![
                 doc! { "$match": doc! { "x": 1 } },
@@ -284,7 +283,7 @@ async fn mmapv1_error_raised() {
     let client = TestClient::new().await;
 
     let req = semver::VersionReq::parse("<=4.0").unwrap();
-    if !req.matches(&client.server_version.as_ref().unwrap()) || !client.is_replica_set() {
+    if !req.matches(&client.server_version) || !client.is_replica_set() {
         return;
     }
 
@@ -334,13 +333,13 @@ async fn label_not_added_second_read_error() {
 #[function_name::named]
 async fn label_not_added(retry_reads: bool) {
     let options = ClientOptions::builder().retry_reads(retry_reads).build();
-    let client = TestClient::with_additional_options(Some(options), false, true).await;
+    let client = TestClient::with_additional_options(Some(options), false).await;
 
     // Configuring a failpoint is only supported on 4.0+ replica sets and 4.1.5+ sharded clusters.
     let req = VersionReq::parse(">=4.0").unwrap();
     let sharded_req = VersionReq::parse(">=4.1.5").unwrap();
-    if client.is_sharded() && !sharded_req.matches(&client.server_version.as_ref().unwrap())
-        || !req.matches(&client.server_version.as_ref().unwrap())
+    if client.is_sharded() && !sharded_req.matches(&client.server_version)
+        || !req.matches(&client.server_version)
     {
         return;
     }
diff --git a/src/test/spec/unified_runner/entity.rs b/src/test/spec/unified_runner/entity.rs
@@ -1,6 +1,12 @@
-use std::ops::Deref;
+use std::{ops::Deref, sync::Arc};
 
-use crate::{bson::Bson, test::util::EventClient, Collection, Database};
+use crate::{
+    bson::Bson,
+    test::{CommandEvent, EventHandler},
+    Client,
+    Collection,
+    Database,
+};
 
 #[derive(Clone, Debug)]
 pub enum Entity {
@@ -13,11 +19,63 @@ pub enum Entity {
 
 #[derive(Clone, Debug)]
 pub struct ClientEntity {
-    client: EventClient,
+    client: Client,
+    observer: Arc<EventHandler>,
     pub observe_events: Option<Vec<String>>,
     pub ignore_command_names: Option<Vec<String>>,
 }
 
+impl ClientEntity {
+    pub fn new(
+        client: Client,
+        observer: Arc<EventHandler>,
+        observe_events: Option<Vec<String>>,
+        ignore_command_names: Option<Vec<String>>,
+    ) -> Self {
+        Self {
+            client,
+            observer,
+            observe_events,
+            ignore_command_names,
+        }
+    }
+
+    /// Gets a list of all of the events of the requested event types that occurred on this client.
+    /// Ignores any event with a name in the ignore list. Also ignores all configureFailPoint
+    /// events.
+    pub fn get_filtered_events(
+        &self,
+        observe_events: &Option<Vec<String>>,
+        ignore_command_names: &Option<Vec<String>>,
+    ) -> Vec<CommandEvent> {
+        self.observer.get_filtered_command_events(|event| {
+            if event.command_name() == "configureFailPoint" {
+                return false;
+            }
+            if let Some(observe_events) = observe_events {
+                if !observe_events.iter().any(|name| match event {
+                    CommandEvent::CommandStartedEvent(_) => name.as_str() == "commandStartedEvent",
+                    CommandEvent::CommandSucceededEvent(_) => {
+                        name.as_str() == "commandSucceededEvent"
+                    }
+                    CommandEvent::CommandFailedEvent(_) => name.as_str() == "commandFailedEvent",
+                }) {
+                    return false;
+                }
+            }
+            if let Some(ignore_command_names) = ignore_command_names {
+                if ignore_command_names
+                    .iter()
+                    .any(|name| event.command_name() == name)
+                {
+                    return false;
+                }
+            }
+            true
+        })
+    }
+}
+
 impl From<Database> for Entity {
     fn from(database: Database) -> Self {
         Self::Database(database)
@@ -37,26 +95,14 @@ impl From<Bson> for Entity {
 }
 
 impl Deref for ClientEntity {
-    type Target = EventClient;
+    type Target = Client;
 
     fn deref(&self) -> &Self::Target {
         &self.client
     }
 }
 
 impl Entity {
-    pub fn from_client(
-        client: EventClient,
-        observe_events: Option<Vec<String>>,
-        ignore_command_names: Option<Vec<String>>,
-    ) -> Self {
-        Self::Client(ClientEntity {
-            client,
-            observe_events,
-            ignore_command_names,
-        })
-    }
-
     pub fn as_client(&self) -> &ClientEntity {
         match self {
             Self::Client(client) => client,
diff --git a/src/test/spec/unified_runner/operation.rs b/src/test/spec/unified_runner/operation.rs
@@ -810,11 +810,10 @@ pub(super) struct FailPointCommand {
 impl TestOperation for FailPointCommand {
     async fn execute_test_runner_operation(&self, test_runner: &mut TestRunner) {
         let client = test_runner.get_client(&self.client);
-        let guard = client
-            .enable_failpoint(
-                self.fail_point.clone(),
-                Some(ReadPreference::Primary.into()),
-            )
+        let guard = self
+            .fail_point
+            .clone()
+            .enable(client, Some(ReadPreference::Primary.into()))
             .await
             .unwrap();
         test_runner.fail_point_guards.push(guard);
diff --git a/src/test/spec/unified_runner/test_file.rs b/src/test/spec/unified_runner/test_file.rs
@@ -77,13 +77,13 @@ impl RunOnRequirement {
     pub async fn can_run_on(&self, client: &TestClient) -> bool {
         if let Some(ref min_version) = self.min_server_version {
             let req = VersionReq::parse(&format!(">= {}", &min_version)).unwrap();
-            if !req.matches(&client.server_version.as_ref().unwrap()) {
+            if !req.matches(&client.server_version) {
                 return false;
             }
         }
         if let Some(ref max_version) = self.max_server_version {
             let req = VersionReq::parse(&format!("<= {}", &max_version)).unwrap();
-            if !req.matches(&client.server_version.as_ref().unwrap()) {
+            if !req.matches(&client.server_version) {
                 return false;
             }
         }
@@ -94,7 +94,7 @@ impl RunOnRequirement {
         }
         if let Some(ref actual_server_parameters) = self.server_parameters {
             if !results_match(
-                Some(client.server_parameters.as_ref().unwrap()),
+                Some(&Bson::Document(client.server_parameters.clone())),
                 &Bson::Document(actual_server_parameters.clone()),
                 false,
                 None,
diff --git a/src/test/spec/unified_runner/test_runner.rs b/src/test/spec/unified_runner/test_runner.rs
diff --git a/src/test/spec/v2_runner/mod.rs b/src/test/spec/v2_runner/mod.rs
diff --git a/src/test/spec/v2_runner/test_file.rs b/src/test/spec/v2_runner/test_file.rs
diff --git a/src/test/util/event.rs b/src/test/util/event.rs
diff --git a/src/test/util/failpoint.rs b/src/test/util/failpoint.rs
diff --git a/src/test/util/mod.rs b/src/test/util/mod.rs

Original file line number	Diff line number	Diff line change
`@@ -242,7 +242,7 @@ async fn cluster_time_in_commands() {`
`242`	`242`	`{`
`243`	`243`	`let mut options = CLIENT_OPTIONS.clone();`
`244`	`244`	`options.heartbeat_freq = Some(Duration::from_secs(1000));`
`245`		`- let client = EventClient::with_options(options, true).await;`
	`245`	`+ let client = EventClient::with_options(options).await;`
`246`	`246`
`247`	`247`	`operation(client.clone())`
`248`	`248`	`.await`