RUST-1090 Fix serverless tests (#666)

Author: kmahar

.evergreen/config.yml

@@ -395,10 +395,8 @@ functions:
           export SERVERLESS_ATLAS_USER="${SERVERLESS_ATLAS_USER}"
           export SERVERLESS_ATLAS_PASSWORD="${SERVERLESS_ATLAS_PASSWORD}"
 
-          export MONGODB_URI="${MONGODB_URI}"
           export SSL="${SSL}"
-          export SINGLE_MONGOS_LB_URI="${SINGLE_ATLASPROXY_SERVERLESS_URI}"
-          export MULTI_MONGOS_LB_URI="${MULTI_ATLASPROXY_SERVERLESS_URI}"
+          export SINGLE_MONGOS_LB_URI="${SERVERLESS_URI}"
           . .evergreen/generate-uri.sh
 
           SNAPPY_COMPRESSION_ENABLED="true" \

.evergreen/run-serverless-tests.sh

@@ -45,7 +45,8 @@ cargo_test test::spec::sessions > sessions.xml
 cargo_test test::spec::transactions > transactions.xml
 cargo_test test::spec::load_balancers > load_balancers.xml
 cargo_test test::cursor > cursor.xml
+cargo_test test::spec::collection_management > coll.xml
 
-junit-report-merger results.xml crud.xml retryable_reads.xml retryable_writes.xml versioned_api.xml sessions.xml transactions.xml load_balancers.xml cursor.xml
+junit-report-merger results.xml crud.xml retryable_reads.xml retryable_writes.xml versioned_api.xml sessions.xml transactions.xml load_balancers.xml cursor.xml coll.xml
 
 exit $CARGO_RESULT

Cargo.toml

@@ -137,6 +137,7 @@ features = ["v4"]
 
 [dev-dependencies]
 approx = "0.5.1"
+async_once = "0.2.6"
 derive_more = "0.99.13"
 function_name = "0.2.1"
 futures = "0.3"

src/client/options/mod.rs

@@ -1184,20 +1184,6 @@ impl ClientOptions {
         crate::runtime::block_on(Self::parse_connection_string(conn_str, resolver_config))
     }
 
-    #[cfg(test)]
-    pub(crate) fn parse_without_srv_resolution(s: &str) -> Result<Self> {
-        let mut conn_str = ConnectionString::parse(s)?;
-        let host_info = std::mem::take(&mut conn_str.host_info);
-        let mut options = Self::from_connection_string(conn_str);
-        options.hosts = match host_info {
-            HostInfo::HostIdentifiers(hosts) => hosts,
-            HostInfo::DnsRecord(_) => panic!("Expected non-SRV URI, got {:?}", s),
-        };
-        options.validate()?;
-
-        Ok(options)
-    }
-
     fn from_connection_string(conn_str: ConnectionString) -> Self {
         let mut credential = conn_str.credential;
         // Populate default auth source, if needed.

src/client/session/test/mod.rs

@@ -248,7 +248,7 @@ async fn cluster_time_in_commands() {
         F: Fn(EventClient) -> G,
         G: Future<Output = Result<R>>,
     {
-        let mut options = CLIENT_OPTIONS.clone();
+        let mut options = CLIENT_OPTIONS.get().await.clone();
         options.heartbeat_freq = Some(Duration::from_secs(1000));
         let client = EventClient::with_options(options).await;
 

src/cmap/establish/test.rs

@@ -41,7 +41,7 @@ async fn speculative_auth_test(
             .credential(credential.clone())
             .build(),
     );
-    pool_options.tls_options = CLIENT_OPTIONS.tls_options();
+    pool_options.tls_options = CLIENT_OPTIONS.get().await.tls_options();
 
     let description = client.topology_description();
 

src/cmap/test/integration.rs

@@ -42,7 +42,7 @@ struct DatabaseEntry {
 async fn acquire_connection_and_send_command() {
     let _guard: RwLockReadGuard<()> = LOCK.run_concurrently().await;
 
-    let client_options = CLIENT_OPTIONS.clone();
+    let client_options = CLIENT_OPTIONS.get().await.clone();
     let mut pool_options = ConnectionPoolOptions::from_client_options(&client_options);
     pool_options.ready = Some(true);
 
@@ -86,7 +86,7 @@ async fn acquire_connection_and_send_command() {
 async fn concurrent_connections() {
     let _guard = LOCK.run_exclusively().await;
 
-    let mut options = CLIENT_OPTIONS.clone();
+    let mut options = CLIENT_OPTIONS.get().await.clone();
     if options.load_balanced.unwrap_or(false) {
         log_uncaptured("skipping concurrent_connections test due to load-balanced topology");
         return;
@@ -117,13 +117,13 @@ async fn concurrent_connections() {
         .expect("failpoint should succeed");
 
     let handler = Arc::new(EventHandler::new());
-    let client_options = CLIENT_OPTIONS.clone();
+    let client_options = CLIENT_OPTIONS.get().await.clone();
     let mut options = ConnectionPoolOptions::from_client_options(&client_options);
     options.cmap_event_handler = Some(handler.clone() as Arc<dyn crate::cmap::CmapEventHandler>);
     options.ready = Some(true);
 
     let pool = ConnectionPool::new(
-        CLIENT_OPTIONS.hosts[0].clone(),
+        CLIENT_OPTIONS.get().await.hosts[0].clone(),
         Default::default(),
         TopologyUpdater::channel().0,
         Some(options),
@@ -174,7 +174,7 @@ async fn concurrent_connections() {
 async fn connection_error_during_establishment() {
     let _guard: RwLockWriteGuard<_> = LOCK.run_exclusively().await;
 
-    let mut client_options = CLIENT_OPTIONS.clone();
+    let mut client_options = CLIENT_OPTIONS.get().await.clone();
     if client_options.load_balanced.unwrap_or(false) {
         log_uncaptured(
             "skipping connection_error_during_establishment test due to load-balanced topology",
@@ -235,7 +235,7 @@ async fn connection_error_during_establishment() {
 async fn connection_error_during_operation() {
     let _guard: RwLockWriteGuard<_> = LOCK.run_exclusively().await;
 
-    let mut options = CLIENT_OPTIONS.clone();
+    let mut options = CLIENT_OPTIONS.get().await.clone();
     let handler = Arc::new(EventHandler::new());
     options.cmap_event_handler = Some(handler.clone() as Arc<dyn CmapEventHandler>);
     options.hosts.drain(1..);

src/cmap/test/mod.rs

@@ -122,12 +122,12 @@ impl CmapThread {
 }
 
 impl Executor {
-    fn new(test_file: TestFile) -> Self {
+    async fn new(test_file: TestFile) -> Self {
         let handler = Arc::new(EventHandler::new());
         let error = test_file.error;
 
         let mut pool_options = test_file.pool_options.unwrap_or_default();
-        pool_options.tls_options = CLIENT_OPTIONS.tls_options();
+        pool_options.tls_options = CLIENT_OPTIONS.get().await.tls_options();
         pool_options.cmap_event_handler = Some(handler.clone());
         pool_options.server_api = SERVER_API.clone();
 
@@ -156,7 +156,7 @@ impl Executor {
         let (updater, mut receiver) = TopologyUpdater::channel();
 
         let pool = ConnectionPool::new(
-            CLIENT_OPTIONS.hosts[0].clone(),
+            CLIENT_OPTIONS.get().await.hosts[0].clone(),
             Default::default(),
             updater,
             Some(self.pool_options),
@@ -426,7 +426,7 @@ async fn cmap_spec_tests() {
 
         let _guard: RwLockWriteGuard<()> = LOCK.run_exclusively().await;
 
-        let mut options = CLIENT_OPTIONS.clone();
+        let mut options = CLIENT_OPTIONS.get().await.clone();
         if options.load_balanced.unwrap_or(false) {
             log_uncaptured(format!(
                 "skipping {:?} due to load balanced topology",
@@ -454,7 +454,7 @@ async fn cmap_spec_tests() {
                 .unwrap();
         }
 
-        let executor = Executor::new(test_file);
+        let executor = Executor::new(test_file).await;
         executor.execute_test().await;
 
         if should_disable_fp {

src/compression/test.rs

@@ -69,7 +69,7 @@ fn test_snappy_compressor() {
 #[cfg_attr(feature = "async-std-runtime", async_std::test)]
 #[cfg(feature = "zlib-compression")]
 async fn ping_server_with_zlib_compression() {
-    let mut client_options = CLIENT_OPTIONS.clone();
+    let mut client_options = CLIENT_OPTIONS.get().await.clone();
     client_options.compressors = Some(vec![Compressor::Zlib { level: Some(4) }]);
     send_ping_with_compression(client_options).await;
 }
@@ -78,7 +78,7 @@ async fn ping_server_with_zlib_compression() {
 #[cfg_attr(feature = "async-std-runtime", async_std::test)]
 #[cfg(feature = "zstd-compression")]
 async fn ping_server_with_zstd_compression() {
-    let mut client_options = CLIENT_OPTIONS.clone();
+    let mut client_options = CLIENT_OPTIONS.get().await.clone();
     client_options.compressors = Some(vec![Compressor::Zstd { level: None }]);
     send_ping_with_compression(client_options).await;
 }
@@ -87,7 +87,7 @@ async fn ping_server_with_zstd_compression() {
 #[cfg_attr(feature = "async-std-runtime", async_std::test)]
 #[cfg(feature = "snappy-compression")]
 async fn ping_server_with_snappy_compression() {
-    let mut client_options = CLIENT_OPTIONS.clone();
+    let mut client_options = CLIENT_OPTIONS.get().await.clone();
     client_options.compressors = Some(vec![Compressor::Snappy]);
     send_ping_with_compression(client_options).await;
 }
@@ -100,7 +100,7 @@ async fn ping_server_with_snappy_compression() {
     feature = "snappy-compression"
 ))]
 async fn ping_server_with_all_compressors() {
-    let mut client_options = CLIENT_OPTIONS.clone();
+    let mut client_options = CLIENT_OPTIONS.get().await.clone();
     client_options.compressors = Some(vec![
         Compressor::Zlib { level: None },
         Compressor::Snappy,

src/sdam/description/topology/server_selection/test/in_window.rs

@@ -118,7 +118,7 @@ async fn select_in_window() {
 async fn load_balancing_test() {
     let _guard: RwLockWriteGuard<_> = LOCK.run_exclusively().await;
 
-    let mut setup_client_options = CLIENT_OPTIONS.clone();
+    let mut setup_client_options = CLIENT_OPTIONS.get().await.clone();
 
     if setup_client_options.load_balanced.unwrap_or(false) {
         log_uncaptured("skipping load_balancing_test test due to load-balanced topology");
@@ -148,7 +148,7 @@ async fn load_balancing_test() {
         return;
     }
 
-    if CLIENT_OPTIONS.hosts.len() != 2 {
+    if CLIENT_OPTIONS.get().await.hosts.len() != 2 {
         log_uncaptured("skipping load_balancing_test test due to topology not having 2 mongoses");
         return;
     }
@@ -215,7 +215,7 @@ async fn load_balancing_test() {
 
     let mut handler = EventHandler::new();
     let mut subscriber = handler.subscribe();
-    let mut options = CLIENT_OPTIONS.clone();
+    let mut options = CLIENT_OPTIONS.get().await.clone();
     let max_pool_size = 10;
     let hosts = options.hosts.clone();
     options.local_threshold = Duration::from_secs(30).into();
@@ -260,7 +260,7 @@ async fn load_balancing_test() {
         .build();
     let failpoint = FailPoint::fail_command(&["find"], FailPointMode::AlwaysOn, options);
 
-    let slow_host = CLIENT_OPTIONS.hosts[0].clone();
+    let slow_host = CLIENT_OPTIONS.get().await.hosts[0].clone();
     let criteria = SelectionCriteria::Predicate(Arc::new(move |si| si.address() == &slow_host));
     let fp_guard = setup_client
         .enable_failpoint(failpoint, criteria)