RUST-1588 Testing for RunCursorCommand (#931)

Author: drshika

src/client.rs

@@ -620,7 +620,7 @@ impl Client {
     ///
     /// Note that topology changes require rebuilding the connection pool, so this method cannot
     /// guarantee that the pool will always be filled for the lifetime of the `Client`.
-    /// 
+    ///
     /// Does nothing if `min_pool_size` is unset or zero.
     pub async fn warm_connection_pool(&self) {
         if !self.inner.options.min_pool_size.map_or(false, |s| s > 0) {

src/client/executor.rs

@@ -692,7 +692,6 @@ impl Client {
                             }))
                     }
                 }
-
                 handle_response(self, op, session, is_sharded, response).await
             }
             Err(err) => Err(err),

src/coll/options.rs

@@ -78,7 +78,8 @@ impl Hint {
 }
 
 /// Specifies the type of cursor to return from a find operation.
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug, Clone, Copy, Deserialize)]
+#[serde(rename_all = "camelCase")]
 #[non_exhaustive]
 pub enum CursorType {
     /// Default; close the cursor after the last document is received from the server.

src/db/options.rs

@@ -315,8 +315,10 @@ pub struct ChangeStreamPreAndPostImages {
 
 /// Specifies the options to a
 /// [`Database::RunCursorCommand`](../struct.Database.html#method.run_cursor_command) operation.
-#[derive(Clone, Debug, Default, TypedBuilder)]
+#[derive(Clone, Debug, Default, Deserialize, TypedBuilder)]
 #[builder(field_defaults(default, setter(into)))]
+#[serde(rename_all = "camelCase")]
+#[serde(default)]
 #[non_exhaustive]
 pub struct RunCursorCommandOptions {
     /// The default read preference for operations.
@@ -325,6 +327,8 @@ pub struct RunCursorCommandOptions {
     pub cursor_type: Option<CursorType>,
     /// Number of documents to return per batch.
     pub batch_size: Option<u32>,
+    #[serde(rename = "maxtime", alias = "maxTimeMS")]
+    #[serde(deserialize_with = "serde_util::deserialize_duration_option_from_u64_millis")]
     /// Optional non-negative integer value. Use this value to configure the maxTimeMS option sent
     /// on subsequent getMore commands.
     pub max_time: Option<Duration>,

src/operation/run_cursor_command.rs

@@ -5,7 +5,7 @@ use crate::{
     concern::WriteConcern,
     cursor::CursorSpecification,
     error::{Error, Result},
-    operation::{Operation, RunCommand},
+    operation::{CursorBody, Operation, RunCommand},
     options::RunCursorCommandOptions,
     selection_criteria::SelectionCriteria,
 };
@@ -94,25 +94,18 @@ impl<'conn> Operation for RunCursorCommand<'conn> {
         response: RawCommandResponse,
         description: &StreamDescription,
     ) -> Result<Self::O> {
-        let doc = Operation::handle_response(&self.run_command, response, description)?;
-        let cursor_info = bson::from_document(doc)?;
-        let batch_size = match &self.options {
-            Some(options) => options.batch_size,
-            None => None,
-        };
-        let max_time = match &self.options {
-            Some(options) => options.max_time,
-            None => None,
-        };
+        let cursor_response: CursorBody = response.body()?;
+
         let comment = match &self.options {
             Some(options) => options.comment.clone(),
             None => None,
         };
+
         Ok(CursorSpecification::new(
-            cursor_info,
+            cursor_response.cursor,
             description.server_address.clone(),
-            batch_size,
-            max_time,
+            self.options.as_ref().and_then(|opts| opts.batch_size),
+            self.options.as_ref().and_then(|opts| opts.max_time),
             comment,
         ))
     }

src/test/spec/run_command.rs

@@ -5,11 +5,11 @@ use crate::test::{spec::unified_runner::run_unified_tests, LOCK};
 async fn run_unified() {
     let _guard = LOCK.run_exclusively().await;
     run_unified_tests(&["run-command", "unified"])
-        // TODO RUST-1588: Unskip this file
-        .skip_files(&["runCursorCommand.json"])
         .skip_tests(&[
             // TODO re: RUST-1649: fix withTransaction for new test runner
             "attaches transaction fields to given command",
+            // TODO fix in a follow up PR because need to add IterateOnce
+            "does not close the cursor when receiving an empty batch",
         ])
         .await;
 }

src/test/spec/unified_runner/operation.rs

@@ -45,6 +45,7 @@ use crate::{
     client::session::TransactionState,
     coll::options::Hint,
     collation::Collation,
+    db::options::RunCursorCommandOptions,
     error::{ErrorKind, Result},
     gridfs::options::{GridFsDownloadByNameOptions, GridFsUploadOptions},
     options::{
@@ -288,6 +289,7 @@ impl<'de> Deserialize<'de> for Operation {
             "deleteOne" => deserialize_op::<DeleteOne>(definition.arguments),
             "find" => deserialize_op::<Find>(definition.arguments),
             "createFindCursor" => deserialize_op::<CreateFindCursor>(definition.arguments),
+            "createCommandCursor" => deserialize_op::<CreateCommandCursor>(definition.arguments),
             "aggregate" => deserialize_op::<Aggregate>(definition.arguments),
             "distinct" => deserialize_op::<Distinct>(definition.arguments),
             "countDocuments" => deserialize_op::<CountDocuments>(definition.arguments),
@@ -314,6 +316,7 @@ impl<'de> Deserialize<'de> for Operation {
             "createCollection" => deserialize_op::<CreateCollection>(definition.arguments),
             "dropCollection" => deserialize_op::<DropCollection>(definition.arguments),
             "runCommand" => deserialize_op::<RunCommand>(definition.arguments),
+            "runCursorCommand" => deserialize_op::<RunCursorCommand>(definition.arguments),
             "endSession" => deserialize_op::<EndSession>(definition.arguments),
             "assertSessionTransactionState" => {
                 deserialize_op::<AssertSessionTransactionState>(definition.arguments)
@@ -1561,7 +1564,9 @@ impl TestOperation for CreateCollection {
                     .create_collection(&self.collection, self.options.clone())
                     .await?;
             }
-            Ok(None)
+            Ok(Some(Entity::Collection(
+                database.collection(&self.collection),
+            )))
         }
         .boxed()
     }
@@ -1644,6 +1649,109 @@ impl TestOperation for RunCommand {
     }
 }
 
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub(super) struct RunCursorCommand {
+    command: Document,
+    // We don't need to use this field, but it needs to be included during deserialization so that
+    // we can use the deny_unknown_fields tag.
+    #[serde(rename = "commandName")]
+    _command_name: String,
+
+    #[serde(flatten)]
+    options: RunCursorCommandOptions,
+    session: Option<String>,
+}
+
+impl TestOperation for RunCursorCommand {
+    fn execute_entity_operation<'a>(
+        &'a self,
+        id: &'a str,
+        test_runner: &'a TestRunner,
+    ) -> BoxFuture<'a, Result<Option<Entity>>> {
+        async move {
+            let command = self.command.clone();
+            let db = test_runner.get_database(id).await;
+            let options = self.options.clone();
+
+            let result = match &self.session {
+                Some(session_id) => {
+                    with_mut_session!(test_runner, session_id, |session| async {
+                        let mut cursor = db
+                            .run_cursor_command_with_session(command, options, session)
+                            .await?;
+                        cursor.stream(session).try_collect::<Vec<_>>().await
+                    })
+                    .await?
+                }
+                None => {
+                    let cursor = db.run_cursor_command(command, options).await?;
+                    cursor.try_collect::<Vec<_>>().await?
+                }
+            };
+
+            Ok(Some(bson::to_bson(&result)?.into()))
+        }
+        .boxed()
+    }
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub struct CreateCommandCursor {
+    command: Document,
+    // We don't need to use this field, but it needs to be included during deserialization so that
+    // we can use the deny_unknown_fields tag.
+    #[serde(rename = "commandName")]
+    _command_name: String,
+
+    #[serde(flatten)]
+    options: RunCursorCommandOptions,
+    session: Option<String>,
+}
+
+impl TestOperation for CreateCommandCursor {
+    fn execute_entity_operation<'a>(
+        &'a self,
+        id: &'a str,
+        test_runner: &'a TestRunner,
+    ) -> BoxFuture<'a, Result<Option<Entity>>> {
+        async move {
+            let command = self.command.clone();
+            let db = test_runner.get_database(id).await;
+            let options = self.options.clone();
+
+            match &self.session {
+                Some(session_id) => {
+                    let mut ses_cursor = None;
+                    with_mut_session!(test_runner, session_id, |session| async {
+                        ses_cursor = Some(
+                            db.run_cursor_command_with_session(command, options, session)
+                                .await,
+                        );
+                    })
+                    .await;
+                    let test_cursor = TestCursor::Session {
+                        cursor: ses_cursor.unwrap().unwrap(),
+                        session_id: session_id.clone(),
+                    };
+                    Ok(Some(Entity::Cursor(test_cursor)))
+                }
+                None => {
+                    let doc_cursor = db.run_cursor_command(command, options).await?;
+                    let test_cursor = TestCursor::Normal(Mutex::new(doc_cursor));
+                    Ok(Some(Entity::Cursor(test_cursor)))
+                }
+            }
+        }
+        .boxed()
+    }
+
+    fn returns_root_documents(&self) -> bool {
+        false
+    }
+}
+
 #[derive(Debug, Deserialize)]
 #[serde(rename_all = "camelCase", deny_unknown_fields)]
 pub(super) struct EndSession {}