PYTHON-5473 - Better test assertions for booleans

Author: NoahStapp

test/asynchronous/test_database.py

@@ -189,7 +189,7 @@ async def test_list_collection_names(self):
             filter={"name": {"$regex": r"^(?!system\.)"}}
         )
         for coll in no_system_collections:
-            self.assertTrue(not coll.startswith("system."))
+            self.assertFalse(coll.startswith("system."))
         self.assertIn("systemcoll.test", no_system_collections)
 
         # Force more than one batch.
@@ -265,19 +265,13 @@ async def test_list_collections(self):
             try:
                 # Found duplicate.
                 coll_cnt[coll] += 1
-                self.assertTrue(False)
+                self.fail("Found duplicate")
             except KeyError:
                 coll_cnt[coll] = 1
         coll_cnt: dict = {}
 
-        # Checking if is there any collection which don't exists.
-        if (
-            len(set(colls) - {"test", "test.mike"}) == 0
-            or len(set(colls) - {"test", "test.mike", "system.indexes"}) == 0
-        ):
-            self.assertTrue(True)
-        else:
-            self.assertTrue(False)
+        # Check if there are any collections which don't exist.
+        self.assertLessEqual(set(colls), {"test", "test.mike", "system.indexes"})
 
         colls = await (await db.list_collections(filter={"name": {"$regex": "^test$"}})).to_list()
         self.assertEqual(1, len(colls))
@@ -307,16 +301,13 @@ async def test_list_collections(self):
             try:
                 # Found duplicate.
                 coll_cnt[coll] += 1
-                self.assertTrue(False)
+                self.fail("Found duplicate")
             except KeyError:
                 coll_cnt[coll] = 1
         coll_cnt = {}
 
-        # Checking if is there any collection which don't exists.
-        if len(set(colls) - {"test"}) == 0 or len(set(colls) - {"test", "system.indexes"}) == 0:
-            self.assertTrue(True)
-        else:
-            self.assertTrue(False)
+        # Check if there are any collections which don't exist.
+        self.assertLessEqual(set(colls), {"test", "system.indexes"})
 
         await self.client.drop_database("pymongo_test")
 

test/asynchronous/test_gridfs_bucket.py

@@ -164,17 +164,16 @@ async def test_upload_ensures_index(self):
         await files.drop()
         await self.fs.upload_from_stream("filename", b"junk")
 
-        self.assertTrue(
-            any(
-                info.get("key") == [("files_id", 1), ("n", 1)]
-                for info in (await chunks.index_information()).values()
-            )
+        self.assertIn(
+            [("files_id", 1), ("n", 1)],
+            [info.get("key") for info in (await chunks.index_information()).values()],
+            "Missing required index on chunks collection: {files_id: 1, n: 1}",
         )
-        self.assertTrue(
-            any(
-                info.get("key") == [("filename", 1), ("uploadDate", 1)]
-                for info in (await files.index_information()).values()
-            )
+
+        self.assertIn(
+            [("filename", 1), ("uploadDate", 1)],
+            [info.get("key") for info in (await files.index_information()).values()],
+            "Missing required index on files collection: {filename: 1, uploadDate: 1}",
         )
 
     async def test_ensure_index_shell_compat(self):
@@ -192,11 +191,10 @@ async def test_ensure_index_shell_compat(self):
             # No error.
             await self.fs.upload_from_stream("filename", b"data")
 
-            self.assertTrue(
-                any(
-                    info.get("key") == [("filename", 1), ("uploadDate", 1)]
-                    for info in (await files.index_information()).values()
-                )
+            self.assertIn(
+                [("filename", 1), ("uploadDate", 1)],
+                [info.get("key") for info in (await files.index_information()).values()],
+                "Missing required index on files collection: {filename: 1, uploadDate: 1}",
             )
             await files.drop()
 

test/asynchronous/test_monitoring.py

@@ -512,9 +512,8 @@ async def test_kill_cursors(self):
             self.assertEqual(cursor.address, succeeded.connection_id)
             # There could be more than one cursor_id here depending on
             # when the thread last ran.
-            self.assertTrue(
-                cursor_id in succeeded.reply["cursorsUnknown"]
-                or cursor_id in succeeded.reply["cursorsKilled"]
+            self.assertIn(
+                cursor_id, succeeded.reply["cursorsUnknown"] + succeeded.reply["cursorsKilled"]
             )
 
     async def test_non_bulk_writes(self):
@@ -1066,7 +1065,7 @@ async def test_write_errors(self):
         self.assertEqual(2, len(errors))
         fields = {"index", "code", "errmsg"}
         for error in errors:
-            self.assertTrue(fields.issubset(set(error)))
+            self.assertLessEqual(fields, set(error))
 
     async def test_first_batch_helper(self):
         # Regardless of server version and use of helpers._first_batch

test/test_database.py

@@ -188,7 +188,7 @@ def test_list_collection_names(self):
             filter={"name": {"$regex": r"^(?!system\.)"}}
         )
         for coll in no_system_collections:
-            self.assertTrue(not coll.startswith("system."))
+            self.assertFalse(coll.startswith("system."))
         self.assertIn("systemcoll.test", no_system_collections)
 
         # Force more than one batch.
@@ -264,19 +264,13 @@ def test_list_collections(self):
             try:
                 # Found duplicate.
                 coll_cnt[coll] += 1
-                self.assertTrue(False)
+                self.fail("Found duplicate")
             except KeyError:
                 coll_cnt[coll] = 1
         coll_cnt: dict = {}
 
-        # Checking if is there any collection which don't exists.
-        if (
-            len(set(colls) - {"test", "test.mike"}) == 0
-            or len(set(colls) - {"test", "test.mike", "system.indexes"}) == 0
-        ):
-            self.assertTrue(True)
-        else:
-            self.assertTrue(False)
+        # 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))
@@ -304,16 +298,13 @@ def test_list_collections(self):
             try:
                 # Found duplicate.
                 coll_cnt[coll] += 1
-                self.assertTrue(False)
+                self.fail("Found duplicate")
             except KeyError:
                 coll_cnt[coll] = 1
         coll_cnt = {}
 
-        # Checking if is there any collection which don't exists.
-        if len(set(colls) - {"test"}) == 0 or len(set(colls) - {"test", "system.indexes"}) == 0:
-            self.assertTrue(True)
-        else:
-            self.assertTrue(False)
+        # Check if there are any collections which don't exist.
+        self.assertLessEqual(set(colls), {"test", "system.indexes"})
 
         self.client.drop_database("pymongo_test")
 

test/test_gridfs_bucket.py

@@ -162,17 +162,16 @@ def test_upload_ensures_index(self):
         files.drop()
         self.fs.upload_from_stream("filename", b"junk")
 
-        self.assertTrue(
-            any(
-                info.get("key") == [("files_id", 1), ("n", 1)]
-                for info in (chunks.index_information()).values()
-            )
+        self.assertIn(
+            [("files_id", 1), ("n", 1)],
+            [info.get("key") for info in (chunks.index_information()).values()],
+            "Missing required index on chunks collection: {files_id: 1, n: 1}",
         )
-        self.assertTrue(
-            any(
-                info.get("key") == [("filename", 1), ("uploadDate", 1)]
-                for info in (files.index_information()).values()
-            )
+
+        self.assertIn(
+            [("filename", 1), ("uploadDate", 1)],
+            [info.get("key") for info in (files.index_information()).values()],
+            "Missing required index on files collection: {filename: 1, uploadDate: 1}",
         )
 
     def test_ensure_index_shell_compat(self):
@@ -190,11 +189,10 @@ def test_ensure_index_shell_compat(self):
             # No error.
             self.fs.upload_from_stream("filename", b"data")
 
-            self.assertTrue(
-                any(
-                    info.get("key") == [("filename", 1), ("uploadDate", 1)]
-                    for info in (files.index_information()).values()
-                )
+            self.assertIn(
+                [("filename", 1), ("uploadDate", 1)],
+                [info.get("key") for info in (files.index_information()).values()],
+                "Missing required index on files collection: {filename: 1, uploadDate: 1}",
             )
             files.drop()
 

test/test_monitoring.py

@@ -510,9 +510,8 @@ def test_kill_cursors(self):
             self.assertEqual(cursor.address, succeeded.connection_id)
             # There could be more than one cursor_id here depending on
             # when the thread last ran.
-            self.assertTrue(
-                cursor_id in succeeded.reply["cursorsUnknown"]
-                or cursor_id in succeeded.reply["cursorsKilled"]
+            self.assertIn(
+                cursor_id, succeeded.reply["cursorsUnknown"] + succeeded.reply["cursorsKilled"]
             )
 
     def test_non_bulk_writes(self):
@@ -1064,7 +1063,7 @@ def test_write_errors(self):
         self.assertEqual(2, len(errors))
         fields = {"index", "code", "errmsg"}
         for error in errors:
-            self.assertTrue(fields.issubset(set(error)))
+            self.assertLessEqual(fields, set(error))
 
     def test_first_batch_helper(self):
         # Regardless of server version and use of helpers._first_batch