Merge pull request #4 from strollby/feat/async-fix

Reducing the flakiness of transaction

Author: abhinand-c

mongoengine/signals.py

@@ -38,6 +38,10 @@ def _fail(self, *args, **kwargs):
             )
 
         send = lambda *a, **kw: None  # noqa
+
+        async def send_async(self, *a, **kw):
+            """Fake send_async as no-op fallback when blinker is unavailable."""
+
         connect = disconnect = has_receivers_for = receivers_for = (
             temporarily_connected_to
         ) = _fail

tests/asynchronous/test_context_managers.py

@@ -4,6 +4,7 @@
 
 import pytest
 from pymongo.errors import OperationFailure, InvalidOperation
+from pymongo.read_concern import ReadConcern
 
 from mongoengine import *
 from mongoengine.asynchronous import async_register_connection, async_get_db, async_connect
@@ -367,8 +368,6 @@ async def test_updating_a_document_within_a_transaction_that_fails(self):
         class A(Document):
             name = StringField()
 
-        await A.adrop_collection()
-
         a_doc = await A.aobjects.create(name="a")
 
         with pytest.raises(TestRollbackError):
@@ -385,8 +384,6 @@ async def test_creating_a_document_within_a_transaction(self):
         class A(Document):
             name = StringField()
 
-        await A.adrop_collection()
-
         # ensure the collection is created (needed for transaction with MongoDB <= 4.2)
         await A.aobjects.create(name="test")
         await A.aobjects.delete()
@@ -407,7 +404,6 @@ async def test_creating_a_document_within_a_transaction_that_fails(self):
         class A(Document):
             name = StringField()
 
-        await A.adrop_collection()
         # ensure a collection is created (needed for transaction with MongoDB <= 4.2)
         await A.aobjects.create(name="test")
         await A.aobjects.delete()
@@ -455,19 +451,15 @@ async def test_collection_creation_via_upserts_across_databases_in_transaction(s
         class A(Document):
             name = StringField()
 
-        await A.adrop_collection()
-
         a_doc = await A.aobjects.create(name="a")
 
         class B(Document):
             meta = {"db_alias": "test2"}
             name = StringField()
 
-        await B.adrop_collection()
-
         b_doc = await B.aobjects.create(name="b")
 
-        async with run_in_transaction():
+        async with run_in_transaction(transaction_kwargs={"read_concern": ReadConcern("local")}):
             await a_doc.aupdate(name="a3")
             with switch_db(A, "test2"):
                 await a_doc.aupdate(name="a4", upsert=True)

tests/synchronous/test_context_managers.py

@@ -6,6 +6,7 @@
 
 import pytest
 from pymongo.errors import OperationFailure
+from pymongo.read_concern import ReadConcern
 
 from mongoengine import *
 from mongoengine.session import _get_session
@@ -396,8 +397,6 @@ def test_updating_a_document_within_a_transaction_that_fails(self):
         class A(Document):
             name = StringField()
 
-        A.drop_collection()
-
         a_doc = A.objects.create(name="a")
 
         with pytest.raises(TestRollbackError):
@@ -414,8 +413,6 @@ def test_creating_a_document_within_a_transaction(self):
         class A(Document):
             name = StringField()
 
-        A.drop_collection()
-
         # ensure collection is created (needed for transaction with MongoDB <= 4.2)
         A.objects.create(name="test")
         A.objects.delete()
@@ -436,7 +433,6 @@ def test_creating_a_document_within_a_transaction_that_fails(self):
         class A(Document):
             name = StringField()
 
-        A.drop_collection()
         # ensure a collection is created (needed for transaction with MongoDB <= 4.2)
         A.objects.create(name="test")
         A.objects.delete()
@@ -484,19 +480,15 @@ def test_collection_creation_via_upserts_across_databases_in_transaction(self):
         class A(Document):
             name = StringField()
 
-        A.drop_collection()
-
         a_doc = A.objects.create(name="a")
 
         class B(Document):
             meta = {"db_alias": "test2"}
             name = StringField()
 
-        B.drop_collection()
-
         b_doc = B.objects.create(name="b")
 
-        with run_in_transaction():
+        with run_in_transaction(transaction_kwargs={"read_concern": ReadConcern("local")}):
             a_doc.update(name="a3")
             with switch_db(A, "test2"):
                 a_doc.update(name="a4", upsert=True)