Implement legacy collection enumeration. (#120)

GODRIVER-492

Change-Id: Iafd2527267b470a582925cc52fdd93bd7f7d7a9e

Author: divjotarora

mongo/client_internal_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/mongodb/mongo-go-driver/mongo/writeconcern"
 	"github.com/mongodb/mongo-go-driver/x/mongo/driver/session"
 	"github.com/mongodb/mongo-go-driver/x/mongo/driver/uuid"
+	"github.com/mongodb/mongo-go-driver/x/network/connstring"
 )
 
 func createTestClient(t *testing.T) *Client {
@@ -41,6 +42,18 @@ func createTestClient(t *testing.T) *Client {
 	}
 }
 
+func createTestClientWithConnstring(t *testing.T, cs connstring.ConnString) *Client {
+	id, _ := uuid.New()
+	return &Client{
+		id:             id,
+		topology:       testutil.TopologyWithConnString(t, cs),
+		connString:     cs,
+		readPreference: readpref.Primary(),
+		clock:          &session.ClusterClock{},
+		registry:       bson.DefaultRegistry,
+	}
+}
+
 func skipIfBelow30(t *testing.T) {
 	serverVersion, err := getServerVersion(createTestDatabase(t, nil))
 	require.NoError(t, err)

mongo/database.go

@@ -232,7 +232,7 @@ func (db *Database) ListCollections(ctx context.Context, filter interface{}, opt
 	cmd := command.ListCollections{
 		DB:       db.name,
 		Filter:   filterDoc,
-		ReadPref: db.readPreference,
+		ReadPref: readpref.Primary(), // list collections must be run on a primary by default
 		Session:  sess,
 		Clock:    db.client.clock,
 	}

mongo/database_internal_test.go

@@ -21,6 +21,8 @@ import (
 	"github.com/mongodb/mongo-go-driver/mongo/readpref"
 	"github.com/mongodb/mongo-go-driver/mongo/writeconcern"
 	"github.com/mongodb/mongo-go-driver/x/bsonx"
+	"github.com/mongodb/mongo-go-driver/x/network/connstring"
+	"github.com/mongodb/mongo-go-driver/x/network/description"
 	"github.com/stretchr/testify/require"
 )
 
@@ -206,7 +208,7 @@ func verifyListCollections(cursor Cursor, uncappedName string, cappedName string
 	var cappedFound bool
 
 	for cursor.Next(context.Background()) {
-		next := bsonx.Doc{}
+		next := &bsonx.Doc{}
 		err = cursor.Decode(next)
 		if err != nil {
 			return err
@@ -223,9 +225,9 @@ func verifyListCollections(cursor Cursor, uncappedName string, cappedName string
 
 		elemName := elem.StringValue()
 
-		if elemName != uncappedName && elemName != cappedName {
-			return fmt.Errorf("incorrect collection name. got: %s. wanted: %s or %s", elemName, uncappedName,
-				cappedName)
+		// legacy servers can return an indexes collection that shouldn't be considered here
+		if elemName != cappedName && elemName != uncappedName {
+			continue
 		}
 
 		if elemName == uncappedName && !uncappedFound {
@@ -257,19 +259,16 @@ func verifyListCollections(cursor Cursor, uncappedName string, cappedName string
 	return nil
 }
 
-func listCollectionsTest(db *Database, cappedOnly bool) error {
-	uncappedName, cappedName, err := setupListCollectionsDb(db)
-	if err != nil {
-		return err
-	}
-
+func listCollectionsTest(db *Database, cappedOnly bool, cappedName, uncappedName string) error {
 	var filter bsonx.Doc
 	if cappedOnly {
 		filter = bsonx.Doc{{"options.capped", bsonx.Boolean(true)}}
 	}
 
+	var cursor Cursor
+	var err error
 	for i := 0; i < 10; i++ {
-		cursor, err := db.ListCollections(context.Background(), filter)
+		cursor, err = db.ListCollections(context.Background(), filter)
 		if err != nil {
 			return err
 		}
@@ -283,43 +282,69 @@ func listCollectionsTest(db *Database, cappedOnly bool) error {
 	return err // all tests failed
 }
 
-func TestDatabase_ListCollections(t *testing.T) {
-	// TODO(GODRIVER-492) - implement legacy list collections
-	skipIfBelow32(t)
+// get the connection string for a direct connection to a secondary in a replica set
+func getSecondaryConnString(t *testing.T) connstring.ConnString {
+	topo := testutil.Topology(t)
+	for _, server := range topo.Description().Servers {
+		if server.Kind != description.RSSecondary {
+			continue
+		}
 
-	rpPrimary := readpref.Primary()
-	rpSecondary := readpref.Secondary()
+		fullAddr := "mongodb://" + server.Addr.String() + "/?connect=direct"
+		cs, err := connstring.Parse(fullAddr)
+		require.NoError(t, err)
+		return cs
+	}
 
+	t.Fatalf("no secondary found for %s", t.Name())
+	return connstring.ConnString{}
+}
+
+func TestDatabase_ListCollections(t *testing.T) {
 	var listCollectionsTable = []struct {
 		name             string
 		expectedTopology string
 		cappedOnly       bool
-		rp               *readpref.ReadPref
+		direct           bool
 	}{
-		{"standalone_nofilter", "server", false, rpPrimary},
-		{"standalone_filter", "server", true, rpPrimary},
-		{"replicaset_nofilter", "replica_set", false, rpPrimary},
-		{"replicaset_filter", "replica_set", true, rpPrimary},
-		{"replicaset_secondary_nofilter", "replica_set", false, rpSecondary},
-		{"replicaset_secondary_filter", "replica_set", true, rpSecondary},
-		{"sharded_nofilter", "sharded_cluster", false, rpPrimary},
-		{"sharded_filter", "sharded_cluster", true, rpPrimary},
+		{"standalone_nofilter", "server", false, false},
+		{"standalone_filter", "server", true, false},
+		{"replicaset_nofilter", "replica_set", false, false},
+		{"replicaset_filter", "replica_set", true, false},
+		{"replicaset_secondary_nofilter", "replica_set", false, true},
+		{"replicaset_secondary_filter", "replica_set", true, true},
+		{"sharded_nofilter", "sharded_cluster", false, false},
+		{"sharded_filter", "sharded_cluster", true, false},
 	}
 
 	for _, tt := range listCollectionsTable {
 		t.Run(tt.name, func(t *testing.T) {
-			if os.Getenv("topology") != tt.expectedTopology {
+			if os.Getenv("TOPOLOGY") != tt.expectedTopology {
 				t.Skip()
 			}
-			dbName := tt.name
-			db := createTestDatabase(t, &dbName, options.Database().SetReadPreference(tt.rp))
 
+			createDb := createTestDatabase(t, &tt.name, options.Database().SetWriteConcern(wcMajority))
 			defer func() {
-				err := db.Drop(context.Background())
+				err := createDb.Drop(context.Background())
 				require.NoError(t, err)
 			}()
 
-			err := listCollectionsTest(db, tt.cappedOnly)
+			uncappedName, cappedName, err := setupListCollectionsDb(createDb)
+			require.NoError(t, err)
+
+			var cs connstring.ConnString
+			if tt.direct {
+				// TODO(GODRIVER-641) - correctly set read preference on direct connections for OP_MSG
+				t.Skip()
+				cs = getSecondaryConnString(t)
+			} else {
+				cs = testutil.ConnString(t)
+			}
+
+			client := createTestClientWithConnstring(t, cs)
+			db := client.Database(tt.name)
+
+			err = listCollectionsTest(db, tt.cappedOnly, cappedName, uncappedName)
 			require.NoError(t, err)
 		})
 	}

x/mongo/driver/list_collections.go

@@ -9,15 +9,20 @@ package driver
 import (
 	"context"
 
+	"errors"
 	"github.com/mongodb/mongo-go-driver/mongo/options"
 	"github.com/mongodb/mongo-go-driver/x/bsonx"
 	"github.com/mongodb/mongo-go-driver/x/mongo/driver/session"
 	"github.com/mongodb/mongo-go-driver/x/mongo/driver/topology"
 	"github.com/mongodb/mongo-go-driver/x/mongo/driver/uuid"
 	"github.com/mongodb/mongo-go-driver/x/network/command"
+	"github.com/mongodb/mongo-go-driver/x/network/connection"
 	"github.com/mongodb/mongo-go-driver/x/network/description"
 )
 
+// ErrFilterType is thrown when a non-string filter is specified.
+var ErrFilterType = errors.New("filter must be a string")
+
 // ListCollections handles the full cycle dispatch and execution of a listCollections command against the provided
 // topology.
 func ListCollections(
@@ -41,6 +46,10 @@ func ListCollections(
 	}
 	defer conn.Close()
 
+	if ss.Description().WireVersion.Max < 3 {
+		return legacyListCollections(ctx, cmd, ss, conn)
+	}
+
 	rp, err := getReadPrefBasedOnTransaction(cmd.ReadPref, cmd.Session)
 	if err != nil {
 		return nil, err
@@ -67,3 +76,49 @@ func ListCollections(
 
 	return c, err
 }
+
+func legacyListCollections(
+	ctx context.Context,
+	cmd command.ListCollections,
+	ss *topology.SelectedServer,
+	conn connection.Connection,
+) (command.Cursor, error) {
+	filter, err := transformFilter(cmd.Filter, cmd.DB)
+	if err != nil {
+		return nil, err
+	}
+
+	findCmd := command.Find{
+		NS:       command.NewNamespace(cmd.DB, "system.namespaces"),
+		ReadPref: cmd.ReadPref,
+		Filter:   filter,
+	}
+
+	// don't need registry because it's used to create BSON docs for find options that don't exist in this case
+	c, err := legacyFind(ctx, findCmd, nil, ss, conn)
+	if err != nil {
+		return nil, err
+	}
+
+	return topology.NewListCollectionsCursor(c), nil
+}
+
+// modify the user-supplied filter to prefix the "name" field with the database name.
+// returns the original filter if the name field is not present or a copy with the modified name field if it is
+func transformFilter(filter bsonx.Doc, dbName string) (bsonx.Doc, error) {
+	if filter == nil {
+		return filter, nil
+	}
+
+	if nameVal, err := filter.LookupErr("name"); err == nil {
+		name, ok := nameVal.StringValueOK()
+		if !ok {
+			return nil, ErrFilterType
+		}
+
+		filterCopy := filter.Copy()
+		filterCopy.Set("name", bsonx.String(dbName+"."+name))
+		return filterCopy, nil
+	}
+	return filter, nil
+}

x/mongo/driver/list_collections_test.go

@@ -0,0 +1,51 @@
+// Copyright (C) MongoDB, Inc. 2017-present.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License. You may obtain
+// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+
+package driver
+
+import (
+	"github.com/mongodb/mongo-go-driver/x/bsonx"
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+func TestListCollections(t *testing.T) {
+	dbName := "db"
+	noNameFilter := bsonx.Doc{
+		{"foo", bsonx.String("bar")},
+	}
+	nonStringFilter := bsonx.Doc{
+		{"name", bsonx.Int32(1)},
+	}
+	nameFilter := bsonx.Doc{
+		{"name", bsonx.String("coll")},
+	}
+	modifiedFilter := bsonx.Doc{
+		{"name", bsonx.String(dbName + ".coll")},
+	}
+
+	t.Run("TestTransformFilter", func(t *testing.T) {
+		testCases := []struct {
+			name           string
+			filter         bsonx.Doc
+			expectedFilter bsonx.Doc
+			err            error
+		}{
+			{"TestNilFilter", nil, nil, nil},
+			{"TestNoName", noNameFilter, noNameFilter, nil},
+			{"TestNonStringName", nonStringFilter, nil, ErrFilterType},
+			{"TestName", nameFilter, modifiedFilter, nil},
+		}
+
+		for _, tc := range testCases {
+			t.Run(tc.name, func(t *testing.T) {
+				newFilter, err := transformFilter(tc.filter, dbName)
+				require.Equal(t, tc.err, err)
+				require.Equal(t, tc.expectedFilter, newFilter)
+			})
+		}
+	})
+}