Method test_list_collections

test/test_database.py:244–309

(self)

        self.assertNotIn("listCollections", listener.started_command_names())

    def test_list_collections(self):
        self.client.drop_database("pymongo_test")
        db = Database(self.client, "pymongo_test")
        db.test.insert_one({"dummy": "object"})
        db.test.mike.insert_one({"dummy": "object"})

        results = db.list_collections()
        colls = [result["name"] for result in results]

        # All the collections present.
        self.assertIn("test", colls)
        self.assertIn("test.mike", colls)

        # No collection containing a '$'.
        for coll in colls:
            self.assertNotIn("$", coll)

        # Duplicate check.
        coll_cnt: dict = {}
        for coll in colls:
            try:
                # Found duplicate.
                coll_cnt[coll] += 1
                self.fail("Found duplicate")
            except KeyError:
                coll_cnt[coll] = 1
        coll_cnt: dict = {}

        # Check if there are any collections which don't exist.
        self.assertLessEqual(set(colls), {"test", "test.mike", "system.indexes"})

        colls = (db.list_collections(filter={"name": {"$regex": "^test$"}})).to_list()
        self.assertEqual(1, len(colls))

        colls = (db.list_collections(filter={"name": {"$regex": "^test.mike$"}})).to_list()
        self.assertEqual(1, len(colls))

        db.drop_collection("test")

        db.create_collection("test", capped=True, size=4096)
        results = db.list_collections(filter={"options.capped": True})
        colls = [result["name"] for result in results]

        # Checking only capped collections are present
        self.assertIn("test", colls)
        self.assertNotIn("test.mike", colls)

        # No collection containing a '$'.
        for coll in colls:
            self.assertNotIn("$", coll)

        # Duplicate check.
        coll_cnt = {}
        for coll in colls:
            try:
                # Found duplicate.
                coll_cnt[coll] += 1
                self.fail("Found duplicate")