Fix flaky tests with asyncio

flaky/flaky_pytest_plugin.py

@@ -1,3 +1,5 @@
+import asyncio
+
 from _pytest import runner  # pylint:disable=import-error
 
 from flaky._flaky_plugin import _FlakyPlugin
@@ -130,6 +132,7 @@ def call_and_report(self, item, when, log=True, **kwds):
         :type log:
             `bool`
         """
+        self._reset_test_event_loop(item)
         call = runner.call_runtest_hook(item, when, **kwds)
         self._call_infos[item][when] = call
         hook = item.ihook
@@ -399,6 +402,12 @@ def _log_test_failure(self, test_callable_name, err, message):
             '\n',
         ])
 
+    @staticmethod
+    def _reset_test_event_loop(item):
+        if 'asyncio' in item.keywords and 'event_loop' in item.funcargs:
+            # always use new loops for every run
+            item.funcargs['event_loop'] = asyncio.new_event_loop()
+
 
 PLUGIN = FlakyPlugin()
 # pytest only processes hooks defined on the module

requirements-dev.txt

@@ -7,6 +7,7 @@ ordereddict
 pycodestyle
 pylint
 pytest
+pytest-asyncio
 pytest-cov
 pytest-xdist
 tox

test/test_pytest/test_pytest_example.py

@@ -59,6 +59,17 @@ def test_flaky_thing_that_succeeds_then_fails_then_succeeds(self):
         TestExample._threshold += 1
         assert TestExample._threshold != 1
 
+    @pytest.mark.asyncio
+    @flaky(3, 2)
+    async def test_flaky_asyncio_thing_that_fails_then_succeeds(self):
+        """
+        Flaky will run this test 3 times.
+        It will fail once and then succeed twice.
+        """
+        # pylint:disable=no-self-use
+        TestExample._threshold += 1
+        assert TestExample._threshold >= 1
+
     @flaky(2, 2)
     def test_flaky_thing_that_always_passes(self):
         """Flaky will run this test twice.  Both will succeed."""