remove Empty class and cleanup _test_ops funciton

Author: sleepyStick

test/asynchronous/test_comment.py

@@ -31,51 +31,23 @@
 _IS_SYNC = False
 
 
-class Empty:
-    def __getattr__(self, item):
-        try:
-            self.__dict__[item]
-        except KeyError:
-            return self.empty
-
-    def empty(self, *args, **kwargs):
-        return Empty()
-
-
 class AsyncTestComment(AsyncIntegrationTest):
     async def _test_ops(
         self,
         helpers,
         already_supported,
         listener,
-        db=Empty(),  # noqa: B008
-        coll=Empty(),  # noqa: B008
     ):
         for h, args in helpers:
             c = "testing comment with " + h.__name__
             with self.subTest("collection-" + h.__name__ + "-comment"):
                 for cc in [c, {"key": c}, ["any", 1]]:
                     listener.reset()
                     kwargs = {"comment": cc}
-                    if h == coll.rename:
-                        await db.get_collection("temp_temp_temp").drop()
-                        destruct_coll = db.get_collection("test_temp")
-                        await destruct_coll.insert_one({})
-                        maybe_cursor = await destruct_coll.rename(*args, **kwargs)
-                        await destruct_coll.drop()
-                    elif h == db.validate_collection:
-                        coll = db.get_collection("test")
-                        await coll.insert_one({})
-                        maybe_cursor = await db.validate_collection(*args, **kwargs)
-                    else:
-                        if not _IS_SYNC and isinstance(coll, Empty):
-                            coll.create_index("a")
-                        else:
-                            await coll.create_index("a")
-                        if not _IS_SYNC and iscoroutinefunction(h):
-                            maybe_cursor = await h(*args, **kwargs)
-                        else:
-                            maybe_cursor = h(*args, **kwargs)
+                    try:
+                        maybe_cursor = await h(*args, **kwargs)
+                    except Exception:
+                        maybe_cursor = None
                     self.assertIn(
                         "comment",
                         inspect.signature(h).parameters,
@@ -95,7 +67,7 @@ async def _test_ops(
                             self.assertEqual(cc, i.command["comment"])
                             tested = True
                     self.assertTrue(tested)
-                    if h not in [coll.aggregate_raw_batches]:
+                    if h.__name__ != "aggregate_raw_batches":
                         self.assertIn(
                             ":param comment:",
                             h.__doc__,
@@ -128,9 +100,7 @@ async def test_database_helpers(self):
             (db.dereference, [DBRef("collection", 1)]),
         ]
         already_supported = [db.command, db.list_collections, db.list_collection_names]
-        await self._test_ops(
-            helpers, already_supported, listener, db=db, coll=db.get_collection("test")
-        )
+        await self._test_ops(helpers, already_supported, listener)
 
     @async_client_context.require_version_min(4, 7, -1)
     @async_client_context.require_replica_set
@@ -186,7 +156,7 @@ async def test_collection_helpers(self):
             coll.find_one_and_delete,
             coll.find_one_and_update,
         ]
-        await self._test_ops(helpers, already_supported, listener, coll=coll, db=db)
+        await self._test_ops(helpers, already_supported, listener)
 
 
 if __name__ == "__main__":

test/test_comment.py

@@ -31,51 +31,23 @@
 _IS_SYNC = True
 
 
-class Empty:
-    def __getattr__(self, item):
-        try:
-            self.__dict__[item]
-        except KeyError:
-            return self.empty
-
-    def empty(self, *args, **kwargs):
-        return Empty()
-
-
 class TestComment(IntegrationTest):
     def _test_ops(
         self,
         helpers,
         already_supported,
         listener,
-        db=Empty(),  # noqa: B008
-        coll=Empty(),  # noqa: B008
     ):
         for h, args in helpers:
             c = "testing comment with " + h.__name__
             with self.subTest("collection-" + h.__name__ + "-comment"):
                 for cc in [c, {"key": c}, ["any", 1]]:
                     listener.reset()
                     kwargs = {"comment": cc}
-                    if h == coll.rename:
-                        db.get_collection("temp_temp_temp").drop()
-                        destruct_coll = db.get_collection("test_temp")
-                        destruct_coll.insert_one({})
-                        maybe_cursor = destruct_coll.rename(*args, **kwargs)
-                        destruct_coll.drop()
-                    elif h == db.validate_collection:
-                        coll = db.get_collection("test")
-                        coll.insert_one({})
-                        maybe_cursor = db.validate_collection(*args, **kwargs)
-                    else:
-                        if not _IS_SYNC and isinstance(coll, Empty):
-                            coll.create_index("a")
-                        else:
-                            coll.create_index("a")
-                        if not _IS_SYNC and iscoroutinefunction(h):
-                            maybe_cursor = h(*args, **kwargs)
-                        else:
-                            maybe_cursor = h(*args, **kwargs)
+                    try:
+                        maybe_cursor = h(*args, **kwargs)
+                    except Exception:
+                        maybe_cursor = None
                     self.assertIn(
                         "comment",
                         inspect.signature(h).parameters,
@@ -95,7 +67,7 @@ def _test_ops(
                             self.assertEqual(cc, i.command["comment"])
                             tested = True
                     self.assertTrue(tested)
-                    if h not in [coll.aggregate_raw_batches]:
+                    if h.__name__ != "aggregate_raw_batches":
                         self.assertIn(
                             ":param comment:",
                             h.__doc__,
@@ -128,7 +100,7 @@ def test_database_helpers(self):
             (db.dereference, [DBRef("collection", 1)]),
         ]
         already_supported = [db.command, db.list_collections, db.list_collection_names]
-        self._test_ops(helpers, already_supported, listener, db=db, coll=db.get_collection("test"))
+        self._test_ops(helpers, already_supported, listener)
 
     @client_context.require_version_min(4, 7, -1)
     @client_context.require_replica_set
@@ -184,7 +156,7 @@ def test_collection_helpers(self):
             coll.find_one_and_delete,
             coll.find_one_and_update,
         ]
-        self._test_ops(helpers, already_supported, listener, coll=coll, db=db)
+        self._test_ops(helpers, already_supported, listener)
 
 
 if __name__ == "__main__":