CDRIVER-735 show that aggregate ignores wire version

Author: ajdavis

build/generate-future-functions.py

@@ -109,6 +109,14 @@
                      param("bson_ptr", "reply"),
                      param("bson_error_ptr", "error")]),
 
+    future_function("mongoc_cursor_ptr",
+                    "mongoc_collection_aggregate",
+                    [param("mongoc_collection_ptr", "collection"),
+                     param("mongoc_query_flags_t", "flags"),
+                     param("const_bson_ptr", "pipeline"),
+                     param("const_bson_ptr", "options"),
+                     param("const_mongoc_read_prefs_ptr", "read_prefs")]),
+
     future_function("bool",
                     "mongoc_collection_insert_bulk",
                     [param("mongoc_collection_ptr", "collection"),

tests/mock_server/future-functions.c

@@ -82,6 +82,35 @@ background_mongoc_client_command_simple (void *data)
    return NULL;
 }
 
+static void *
+background_mongoc_collection_aggregate (void *data)
+{
+   future_t *future = (future_t *) data;
+
+   /* copy the future so we can unlock it while calling
+    * mongoc_collection_aggregate
+    */
+   future_t *copy = future_new_copy (future);
+   future_value_t return_value;
+
+   return_value.type = future_value_mongoc_cursor_ptr_type;
+
+   future_value_set_mongoc_cursor_ptr (
+      &return_value,
+         mongoc_collection_aggregate (
+         future_value_get_mongoc_collection_ptr (future_get_param(copy, 0)),
+         future_value_get_mongoc_query_flags_t (future_get_param(copy, 1)),
+         future_value_get_const_bson_ptr (future_get_param(copy, 2)),
+         future_value_get_const_bson_ptr (future_get_param(copy, 3)),
+         future_value_get_const_mongoc_read_prefs_ptr (future_get_param(copy, 4))
+      ));
+
+   future_destroy (copy);
+   future_resolve (future, return_value);
+
+   return NULL;
+}
+
 static void *
 background_mongoc_collection_insert_bulk (void *data)
 {
@@ -299,6 +328,36 @@ future_client_command_simple (
    return future;
 }
 
+future_t *
+future_collection_aggregate (
+   mongoc_collection_ptr collection,
+   mongoc_query_flags_t flags,
+   const_bson_ptr pipeline,
+   const_bson_ptr options,
+   const_mongoc_read_prefs_ptr read_prefs)
+{
+   future_t *future = future_new (future_value_mongoc_cursor_ptr_type,
+                                  5);
+   
+   future_value_set_mongoc_collection_ptr (
+      future_get_param (future, 0), collection);
+   
+   future_value_set_mongoc_query_flags_t (
+      future_get_param (future, 1), flags);
+   
+   future_value_set_const_bson_ptr (
+      future_get_param (future, 2), pipeline);
+   
+   future_value_set_const_bson_ptr (
+      future_get_param (future, 3), options);
+   
+   future_value_set_const_mongoc_read_prefs_ptr (
+      future_get_param (future, 4), read_prefs);
+   
+   future_start (future, background_mongoc_collection_aggregate);
+   return future;
+}
+
 future_t *
 future_collection_insert_bulk (
    mongoc_collection_ptr collection,

tests/mock_server/future-functions.h

@@ -35,6 +35,17 @@ future_client_command_simple (
 );
 
 
+future_t *
+future_collection_aggregate (
+
+   mongoc_collection_ptr collection,
+   mongoc_query_flags_t flags,
+   const_bson_ptr pipeline,
+   const_bson_ptr options,
+   const_mongoc_read_prefs_ptr read_prefs
+);
+
+
 future_t *
 future_collection_insert_bulk (
 

tests/mock_server/mock-server.c

@@ -151,6 +151,8 @@ mock_server_t *
 mock_server_with_autoismaster (int32_t max_wire_version)
 {
    mock_server_t *server = mock_server_new ();
+
+   /* TODO: max_wire_version 0 is special */
    char *ismaster = bson_strdup_printf ("{'ok': 1.0,"
                                         " 'ismaster': true,"
                                         " 'minWireVersion': 0,"

tests/test-mongoc-collection.c

@@ -1286,6 +1286,136 @@ test_aggregate (void)
 }
 
 
+typedef struct {
+    bool with_batch_size;
+    bool with_options;
+} test_aggregate_context_t;
+
+
+static const char *
+options_json (test_aggregate_context_t *c)
+{
+   if (c->with_batch_size && c->with_options) {
+      return "{'foo': 1, 'batchSize': 11}";
+   } else if (c->with_batch_size) {
+      return "{'batchSize': 11}";
+   } else if (c->with_options) {
+      return "{'foo': 1}";
+   } else {
+      return "{}";
+   }
+}
+
+
+static void
+test_aggregate_legacy (void *data)
+{
+   test_aggregate_context_t *context = (test_aggregate_context_t *) data;
+   mock_server_t *server;
+   mongoc_client_t *client;
+   mongoc_collection_t *collection;
+   future_t *future;
+   request_t *request;
+   mongoc_cursor_t *cursor;
+   const bson_t *doc;
+
+   /* wire protocol version 0 */
+   server = mock_server_with_autoismaster (0);
+   mock_server_run (server);
+   client = mongoc_client_new_from_uri (mock_server_get_uri (server));
+   collection = mongoc_client_get_collection (client, "db", "collection");
+
+   cursor = mongoc_collection_aggregate (
+      collection,
+      MONGOC_QUERY_NONE,
+      tmp_bson ("[{'a': 1}]"),
+      tmp_bson (options_json (context)),
+      NULL);
+
+   future = future_cursor_next (cursor, &doc);
+
+   /* no "cursor" argument */
+   request = mock_server_receives_command (
+      server, "db", MONGOC_QUERY_NONE,
+      "{'aggregate': 'collection',"
+      " 'pipeline': [{'a': 1}]},"
+      " 'cursor': {'$exists': false} %s",
+      context->with_options ? ", 'foo': 1" : "");
+
+   mock_server_replies_simple (request, "{'ok': 1, 'result': [{'_id': 123}]}");
+   assert (future_get_bool (future));
+   ASSERT_MATCH (doc, "{'_id': 123}");
+
+   /* cursor is completed */
+   assert (!mongoc_cursor_next (cursor, &doc));
+
+   mongoc_cursor_destroy (cursor);
+   request_destroy (request);
+   future_destroy (future);
+   mongoc_collection_destroy (collection);
+   mongoc_client_destroy (client);
+   mock_server_destroy (server);
+}
+
+
+static void
+test_aggregate_modern (void *data)
+{
+   test_aggregate_context_t *context = (test_aggregate_context_t *) data;
+   mock_server_t *server;
+   mongoc_client_t *client;
+   mongoc_collection_t *collection;
+   future_t *future;
+   request_t *request;
+   mongoc_cursor_t *cursor;
+   const bson_t *doc;
+
+   /* wire protocol version 1 */
+   server = mock_server_with_autoismaster (1);
+   mock_server_run (server);
+   client = mongoc_client_new_from_uri (mock_server_get_uri (server));
+   collection = mongoc_client_get_collection (client, "db", "collection");
+
+   future = future_collection_aggregate (
+      collection,
+      MONGOC_QUERY_NONE,
+      tmp_bson ("[{'a': 1}]"),
+      tmp_bson (options_json (context)),
+      NULL);
+
+   /* "cursor" argument always sent if wire version >= 1 */
+   request = mock_server_receives_command (
+      server, "db", MONGOC_QUERY_NONE,
+      "{'aggregate': 'collection',"
+      " 'pipeline': [{'a': 1}],"
+      " 'cursor': %s %s}",
+      context->with_batch_size ? "{'batchSize': 11}" : "{'$empty': true}",
+      context->with_options ? ", 'foo': 1" : "");
+
+   mock_server_replies_simple (request,
+                               "{'ok': 1,"
+                               " 'cursor': {"
+                               "    'id': 0,"
+                               "    'ns': 'db.collection',"
+                               "    'firstBatch': [{'_id': 123}]"
+                               "}}");
+
+   assert ((cursor = future_get_mongoc_cursor_ptr (future)));
+   assert (mongoc_cursor_next (cursor, &doc));
+   ASSERT_MATCH (doc, "{'_id': 123}");
+
+   /* cursor is completed */
+   assert (!mongoc_cursor_next (cursor, &doc));
+
+   mongoc_cursor_destroy (cursor);
+   request_destroy (request);
+   future_destroy (future);
+   mongoc_collection_destroy (collection);
+   mongoc_client_destroy (client);
+   mock_server_destroy (server);
+}
+
+
 static void
 test_validate (void)
 {
@@ -1747,9 +1877,48 @@ test_get_index_info (void)
    mongoc_client_destroy (client);
 }
 
+
+static void
+test_aggregate_install (TestSuite *suite) {
+   static test_aggregate_context_t test_aggregate_contexts[2][2][2];
+
+   int wire_version, with_batch_size, with_options;
+   char *legacy_or_modern;
+   TestFuncWC func;
+   char *name;
+   test_aggregate_context_t *context;
+
+   for (wire_version = 0; wire_version < 2; wire_version++) {
+      for (with_batch_size = 0; with_batch_size < 2; with_batch_size++) {
+         for (with_options = 0; with_options < 2; with_options++) {
+            legacy_or_modern = wire_version ? "legacy" : "modern";
+            func = wire_version ? test_aggregate_legacy : test_aggregate_modern;
+
+            context = &test_aggregate_contexts
+               [wire_version][with_batch_size][with_options];
+
+            context->with_batch_size = (bool) with_batch_size;
+            context->with_options = (bool) with_options;
+
+            name = bson_strdup_printf (
+               "/Collection/aggregate/%s/%s/%s",
+               legacy_or_modern,
+               context->with_batch_size ? "batch_size" : "no_batch_size",
+               context->with_options ? "with_options" : "no_options");
+
+            TestSuite_AddWC (suite, name, func, NULL, (void *) context);
+            bson_free (name);
+         }
+      }
+   }
+}
+
+
 void
 test_collection_install (TestSuite *suite)
 {
+   test_aggregate_install (suite);
+
    TestSuite_Add (suite, "/Collection/insert_bulk", test_insert_bulk);
 
    TestSuite_Add (suite,
@@ -1787,7 +1956,6 @@ test_collection_install (TestSuite *suite)
    TestSuite_Add (suite, "/Collection/count_with_opts", test_count_with_opts);
    TestSuite_Add (suite, "/Collection/drop", test_drop);
    TestSuite_Add (suite, "/Collection/aggregate", test_aggregate);
-   TestSuite_Add (suite, "/Collection/validate", test_validate);
    TestSuite_Add (suite, "/Collection/rename", test_rename);
    TestSuite_Add (suite, "/Collection/stats", test_stats);
    TestSuite_Add (suite, "/Collection/find_and_modify", test_find_and_modify);