From ec8592b91254971473b508313fb91b01623f13d7 Mon Sep 17 00:00:00 2001
From: Joongi Kim <me@daybreaker.info>
Date: Thu, 23 Apr 2020 16:10:40 +0900
Subject: [PATCH] Fix up Python 3.8 loop argument warnings (#246)

* Fix up Python 3.8 loop argument warnings

* Remove all "loop=self.loop" expressions.

* Rely on the currently running loop in the constructor of Queue.

* Assuming that janus.Queue objects are created in the functions or
  coroutines called by the event loop, rewrite most test cases to
  be async.

  - No longer manage the event loop lifecycles by ourselves.

  - Adopt pytest-asyncio to seamlessly run test cases in an event loop.

* Add missing .close() / .wait_closed() calls to the end of many test
  cases to ensure proper termination of the queues.

* Insert asyncio.sleep(0) in the wait_closed() method so that all
  task-done callbacks for tasks spawned by _notify_async_not_empty(),
  _notify_async_not_full() internal methods are properly awaited.
  This eliminates hundreds of resource warnings after finishing the test
  suite.

* Ensure dropping of Python 3.3/3.4 in CI configs.

* Add Python 3.7 and 3.8 to CI configs.

* Oops

* Let tox install pytest-asyncio

* Remove PY_33/PY_35 conditional branches as we no longer support Python
  3.4 or older versions.

* Update gitignore

* Let requirements-dev.txt to include pytest-asyncio

* Fix up errors for Python 3.5 and 3.6

* Fix too-long-line error

* Add changelog and update README
---
 .gitignore           |   6 +-
 .travis.yml          |   2 +
 CHANGES.rst          |   5 +
 README.rst           |  55 ++++++-
 janus/__init__.py    |  34 +++--
 requirements-dev.txt |   1 +
 setup.py             |  15 +-
 tests/test_async.py  | 357 +++++++++++++++++++++++--------------------
 tests/test_mixed.py  | 151 ++++++++++--------
 tests/test_sync.py   |  94 +++++++-----
 tox.ini              |   8 +-
 11 files changed, 427 insertions(+), 301 deletions(-)

diff --git a/.gitignore b/.gitignore
index 7375139..efd14fc 100644
--- a/.gitignore
+++ b/.gitignore
@@ -58,4 +58,8 @@ target/
 
 coverage
 
-.pytest_cache
\ No newline at end of file
+.pytest_cache/
+.mypy_cache/
+
+# pyenv
+.python-version
diff --git a/.travis.yml b/.travis.yml
index 39d1a4c..a9eecc5 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -3,6 +3,8 @@ language: python
 python:
   - "3.5"
   - "3.6"
+  - "3.7"
+  - "3.8"
 
 
 install:
diff --git a/CHANGES.rst b/CHANGES.rst
index d85624e..dc1967a 100644
--- a/CHANGES.rst
+++ b/CHANGES.rst
@@ -1,6 +1,11 @@
 Changes
 =======
 
+to be released
+--------------
+
+- Remove explicit loop arguments and forbid creating queues outside event loops #246
+
 0.4.0 (2018-07-28)
 ------------------
 
diff --git a/README.rst b/README.rst
index 97bb295..8527649 100644
--- a/README.rst
+++ b/README.rst
@@ -26,8 +26,43 @@ Synchronous is fully compatible with `standard queue
 follows `asyncio queue design
 <https://docs.python.org/3/library/asyncio-queue.html>`_.
 
-Usage example
-=============
+Usage example (Python 3.7+)
+===========================
+
+.. code:: python
+
+    import asyncio
+    import janus
+
+
+    def threaded(sync_q):
+        for i in range(100):
+            sync_q.put(i)
+        sync_q.join()
+
+
+    async def async_coro(async_q):
+        for i in range(100):
+            val = await async_q.get()
+            assert val == i
+            async_q.task_done()
+
+
+    async def main():
+        queue = janus.Queue()
+        loop = asyncio.get_running_loop()
+        fut = loop.run_in_executor(None, threaded, queue.sync_q)
+        await async_coro(queue.async_q)
+        await fut
+        queue.close()
+        await queue.wait_closed()
+
+
+    asyncio.run(main())
+
+
+Usage example (Python 3.5 and 3.6)
+==================================
 
 .. code:: python
 
@@ -35,7 +70,6 @@ Usage example
     import janus
 
     loop = asyncio.get_event_loop()
-    queue = janus.Queue(loop=loop)
 
 
     def threaded(sync_q):
@@ -51,9 +85,18 @@ Usage example
             async_q.task_done()
 
 
-    fut = loop.run_in_executor(None, threaded, queue.sync_q)
-    loop.run_until_complete(async_coro(queue.async_q))
-    loop.run_until_complete(fut)
+    async def main():
+        queue = janus.Queue()
+        fut = loop.run_in_executor(None, threaded, queue.sync_q)
+        await async_coro(queue.async_q)
+        await fut
+        queue.close()
+        await queue.wait_closed()
+
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()
 
 
 Communication channels
diff --git a/janus/__init__.py b/janus/__init__.py
index e6e4c44..047d2f8 100644
--- a/janus/__init__.py
+++ b/janus/__init__.py
@@ -16,16 +16,17 @@
 
 
 T = TypeVar('T')
-OptLoop = Optional[asyncio.AbstractEventLoop]
 OptInt = Optional[int]
 
 
-class Queue(Generic[T]):
-    def __init__(self, maxsize: int = 0, *, loop: OptLoop = None) -> None:
-        if loop is None:
-            loop = asyncio.get_event_loop()
+current_loop = getattr(asyncio, 'get_running_loop', None)
+if current_loop is None:
+    current_loop = asyncio.get_event_loop
+
 
-        self._loop = loop  # type: asyncio.AbstractEventLoop
+class Queue(Generic[T]):
+    def __init__(self, maxsize: int = 0) -> None:
+        self._loop = current_loop()
         self._maxsize = maxsize
 
         self._init(maxsize)
@@ -37,12 +38,10 @@ def __init__(self, maxsize: int = 0, *, loop: OptLoop = None) -> None:
         self._sync_not_full = threading.Condition(self._sync_mutex)
         self._all_tasks_done = threading.Condition(self._sync_mutex)
 
-        self._async_mutex = asyncio.Lock(loop=self._loop)
-        self._async_not_empty = asyncio.Condition(
-            self._async_mutex, loop=self._loop)
-        self._async_not_full = asyncio.Condition(
-            self._async_mutex, loop=self._loop)
-        self._finished = asyncio.Event(loop=self._loop)
+        self._async_mutex = asyncio.Lock()
+        self._async_not_empty = asyncio.Condition(self._async_mutex)
+        self._async_not_full = asyncio.Condition(self._async_mutex)
+        self._finished = asyncio.Event()
         self._finished.set()
 
         self._closing = False
@@ -78,9 +77,14 @@ async def wait_closed(self) -> None:
         # so lock acquiring is not required
         if not self._closing:
             raise RuntimeError("Waiting for non-closed queue")
+        # give execution chances for the task-done callbacks
+        # of async tasks created inside
+        # _notify_async_not_empty, _notify_async_not_full
+        # methods.
+        await asyncio.sleep(0)
         if not self._pending:
             return
-        await asyncio.wait(self._pending, loop=self._loop)
+        await asyncio.wait(self._pending)
 
     @property
     def closed(self) -> bool:
@@ -143,7 +147,7 @@ async def f() -> None:
                 self._async_not_empty.notify()
 
         def task_maker() -> None:
-            task = asyncio.ensure_future(f(), loop=self._loop)
+            task = self._loop.create_task(f())
             task.add_done_callback(self._pending.discard)
             self._pending.add(task)
 
@@ -158,7 +162,7 @@ async def f() -> None:
                 self._async_not_full.notify()
 
         def task_maker() -> None:
-            task = asyncio.ensure_future(f(), loop=self._loop)
+            task = self._loop.create_task(f())
             task.add_done_callback(self._pending.discard)
             self._pending.add(task)
 
diff --git a/requirements-dev.txt b/requirements-dev.txt
index 03ca1bc..7f87225 100644
--- a/requirements-dev.txt
+++ b/requirements-dev.txt
@@ -7,5 +7,6 @@ mypy==0.770
 pyroma==2.6
 pytest-cov==2.8.1
 pytest==5.4.1
+pytest-asyncio==0.10.0
 tox==3.14.6
 wheel==0.34.2
diff --git a/setup.py b/setup.py
index 1b3f07a..f4a852c 100644
--- a/setup.py
+++ b/setup.py
@@ -6,9 +6,6 @@
 
 from setuptools.command.test import test as TestCommand
 
-PY_33 = sys.version_info < (3, 4)
-PY_35 = sys.version_info >= (3, 5)
-
 
 class PyTest(TestCommand):
     user_options = [('pytest-args=', 'a', "Arguments to pass to py.test")]
@@ -43,13 +40,10 @@ def read(f):
 
 install_requires = []
 
-if PY_33:
-    install_requires.append('asyncio')
-
-# if not PY_35:
-#     install_requires.append('typing')
-
-tests_require = install_requires + ['pytest']
+tests_require = install_requires + [
+    'pytest>=5.4',
+    'pytest-asyncio>=0.10.0',
+]
 extras_require = {}
 
 
@@ -66,6 +60,7 @@ def read(f):
         'Programming Language :: Python :: 3.5',
         'Programming Language :: Python :: 3.6',
         'Programming Language :: Python :: 3.7',
+        'Programming Language :: Python :: 3.8',
         'Topic :: Software Development :: Libraries',
         'Framework :: AsyncIO',
     ],
diff --git a/tests/test_async.py b/tests/test_async.py
index 505b577..9f2a1ae 100644
--- a/tests/test_async.py
+++ b/tests/test_async.py
@@ -1,66 +1,56 @@
 """Tests for queues.py"""
 
 import asyncio
-import concurrent.futures
 import unittest
 
 import janus
+import pytest
 
 
-class _QueueTestBase(unittest.TestCase):
-    def setUp(self):
-        self.loop = asyncio.new_event_loop()
-        self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
-        self.loop.set_default_executor(self.executor)
-        asyncio.set_event_loop(None)
+class QueueBasicTests(unittest.TestCase):
 
-    def tearDown(self):
-        self.executor.shutdown()
-        self.loop.close()
-
-
-class QueueBasicTests(_QueueTestBase):
-    def _test_repr_or_str(self, fn, expect_id):
+    async def _test_repr_or_str(self, fn, expect_id):
         """Test Queue's repr or str.
 
         fn is repr or str. expect_id is True if we expect the Queue's id to
         appear in fn(Queue()).
         """
 
-        _q = janus.Queue(loop=self.loop)
+        _q = janus.Queue()
         q = _q.async_q
         self.assertTrue(fn(q).startswith('<Queue'), fn(q))
         id_is_present = hex(id(q)) in fn(q)
         self.assertEqual(expect_id, id_is_present)
+        loop = janus.current_loop()
 
         async def add_getter():
-            _q = janus.Queue(loop=self.loop)
+            _q = janus.Queue()
             q = _q.async_q
             # Start a task that waits to get.
-            self.loop.create_task(q.get())
+            loop.create_task(q.get())
             # Let it start waiting.
             await asyncio.sleep(0.1)
             self.assertTrue('_getters[1]' in fn(q))
             # resume q.get coroutine to finish generator
             q.put_nowait(0)
 
-        self.loop.run_until_complete(add_getter())
+        await add_getter()
 
         async def add_putter():
-            _q = janus.Queue(maxsize=1, loop=self.loop)
+            _q = janus.Queue(maxsize=1)
             q = _q.async_q
             q.put_nowait(1)
             # Start a task that waits to put.
-            self.loop.create_task(q.put(2))
+            loop.create_task(q.put(2))
             # Let it start waiting.
             await asyncio.sleep(0.1)
             self.assertTrue('_putters[1]' in fn(q))
             # resume q.put coroutine to finish generator
             q.get_nowait()
 
-        self.loop.run_until_complete(add_putter())
+        await add_putter()
 
-        _q = janus.Queue(loop=self.loop)
+        _q = janus.Queue()
         q = _q.async_q
         q.put_nowait(1)
         self.assertTrue('_queue=[1]' in fn(q))
@@ -71,8 +61,9 @@ async def add_putter():
     # def test_str(self):
     #     self._test_repr_or_str(str, False)
 
-    def test_empty(self):
-        _q = janus.Queue(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_empty(self):
+        _q = janus.Queue()
         q = _q.async_q
         self.assertTrue(q.empty())
         q.put_nowait(1)
@@ -82,24 +73,26 @@ def test_empty(self):
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
-    def test_full(self):
-        _q = janus.Queue(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_full(self):
+        _q = janus.Queue()
         q = _q.async_q
         self.assertFalse(q.full())
 
-        _q = janus.Queue(maxsize=1, loop=self.loop)
+        _q = janus.Queue(maxsize=1)
         q = _q.async_q
         q.put_nowait(1)
         self.assertTrue(q.full())
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
-    def test_order(self):
-        _q = janus.Queue(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_order(self):
+        _q = janus.Queue()
         q = _q.async_q
         for i in [1, 3, 2]:
             q.put_nowait(i)
@@ -109,15 +102,17 @@ def test_order(self):
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
-    def test_maxsize(self):
-        _q = janus.Queue(maxsize=2, loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_maxsize(self):
+        loop = janus.current_loop()
+        _q = janus.Queue(maxsize=2)
         q = _q.async_q
         self.assertEqual(2, q.maxsize)
         have_been_put = []
 
-        fut = self.loop.create_future()
+        fut = loop.create_future()
 
         async def putter():
             for i in range(3):
@@ -128,7 +123,7 @@ async def putter():
             return True
 
         async def test():
-            t = self.loop.create_task(putter())
+            t = loop.create_task(putter())
             await fut
 
             # The putter is blocked after putting two items.
@@ -141,58 +136,61 @@ async def test():
             self.assertEqual(1, q.get_nowait())
             self.assertEqual(2, q.get_nowait())
 
-        self.loop.run_until_complete(test())
+        await test()
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
+
 
+class QueueGetTests(unittest.TestCase):
 
-class QueueGetTests(_QueueTestBase):
-    def test_blocking_get(self):
-        _q = janus.Queue(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_blocking_get(self):
+        _q = janus.Queue()
         q = _q.async_q
         q.put_nowait(1)
 
-        async def queue_get():
-            return (await q.get())
-
-        res = self.loop.run_until_complete(queue_get())
+        res = await q.get()
         self.assertEqual(1, res)
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
-    def test_get_with_putters(self):
-        _q = janus.Queue(1, loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_get_with_putters(self):
+        loop = janus.current_loop()
+        _q = janus.Queue(1)
         q = _q.async_q
         q.put_nowait(1)
 
-        fut = self.loop.create_future()
+        fut = loop.create_future()
 
         async def put():
-            t = asyncio.ensure_future(q.put(2), loop=self.loop)
+            t = asyncio.ensure_future(q.put(2))
             await asyncio.sleep(0.01)
             fut.set_result(None)
             return t
 
-        t = self.loop.run_until_complete(put())
+        t = await put()
 
-        res = self.loop.run_until_complete(q.get())
+        res = await q.get()
         self.assertEqual(1, res)
 
-        self.loop.run_until_complete(t)
+        await t
         self.assertEqual(1, q.qsize())
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
-    def test_blocking_get_wait(self):
-        _q = janus.Queue(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_blocking_get_wait(self):
+        loop = janus.current_loop()
+        _q = janus.Queue()
         q = _q.async_q
-        started = asyncio.Event(loop=self.loop)
+        started = asyncio.Event()
         finished = False
 
         async def queue_get():
@@ -203,122 +201,131 @@ async def queue_get():
             return res
 
         async def queue_put():
-            self.loop.call_later(0.01, q.put_nowait, 1)
-            queue_get_task = self.loop.create_task(queue_get())
+            loop.call_later(0.01, q.put_nowait, 1)
+            queue_get_task = loop.create_task(queue_get())
             await started.wait()
             self.assertFalse(finished)
             res = await queue_get_task
             self.assertTrue(finished)
             return res
 
-        res = self.loop.run_until_complete(queue_put())
+        res = await queue_put()
         self.assertEqual(1, res)
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
-    def test_nonblocking_get(self):
-        _q = janus.Queue(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_nonblocking_get(self):
+        _q = janus.Queue()
         q = _q.async_q
         q.put_nowait(1)
         self.assertEqual(1, q.get_nowait())
 
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
-    def test_nonblocking_get_exception(self):
-        _q = janus.Queue(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_nonblocking_get_exception(self):
+        _q = janus.Queue()
         self.assertRaises(asyncio.QueueEmpty, _q.async_q.get_nowait)
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
-    def test_get_cancelled(self):
-        _q = janus.Queue(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_get_cancelled(self):
+        loop = janus.current_loop()
+        _q = janus.Queue()
         q = _q.async_q
 
         async def queue_get():
             return await asyncio.wait_for(q.get(), 0.051)
 
         async def test():
-            get_task = self.loop.create_task(queue_get())
+            get_task = loop.create_task(queue_get())
             await asyncio.sleep(0.01)  # let the task start
             q.put_nowait(1)
             return await get_task
 
-        self.assertEqual(1, self.loop.run_until_complete(test()))
+        self.assertEqual(1, await test())
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
-    def test_get_cancelled_race(self):
-        _q = janus.Queue(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_get_cancelled_race(self):
+        loop = janus.current_loop()
+        _q = janus.Queue()
         q = _q.async_q
 
-        f1 = self.loop.create_future()
+        f1 = loop.create_future()
 
         async def g1():
             f1.set_result(None)
             await q.get()
 
-        t1 = self.loop.create_task(g1())
-        t2 = self.loop.create_task(q.get())
+        t1 = loop.create_task(g1())
+        t2 = loop.create_task(q.get())
 
-        self.loop.run_until_complete(f1)
-        self.loop.run_until_complete(asyncio.sleep(0.01))
+        await f1
+        await asyncio.sleep(0.01)
         t1.cancel()
 
         with self.assertRaises(asyncio.CancelledError):
-            self.loop.run_until_complete(t1)
+            await t1
         self.assertTrue(t1.done())
         q.put_nowait('a')
 
-        self.loop.run_until_complete(t2)
+        await t2
         self.assertEqual(t2.result(), 'a')
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
-    def test_get_with_waiting_putters(self):
-        _q = janus.Queue(loop=self.loop, maxsize=1)
+    @pytest.mark.asyncio
+    async def test_get_with_waiting_putters(self):
+        loop = janus.current_loop()
+        _q = janus.Queue(maxsize=1)
         q = _q.async_q
 
-        self.loop.create_task(q.put('a'))
-        self.loop.create_task(q.put('b'))
+        loop.create_task(q.put('a'))
+        loop.create_task(q.put('b'))
 
-        self.loop.run_until_complete(asyncio.sleep(0.01, loop=self.loop))
+        await asyncio.sleep(0.01)
 
-        self.assertEqual(self.loop.run_until_complete(q.get()), 'a')
-        self.assertEqual(self.loop.run_until_complete(q.get()), 'b')
+        self.assertEqual(await q.get(), 'a')
+        self.assertEqual(await q.get(), 'b')
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
 
-class QueuePutTests(_QueueTestBase):
-    def test_blocking_put(self):
-        _q = janus.Queue(loop=self.loop)
-        q = _q.async_q
+class QueuePutTests(unittest.TestCase):
 
-        async def queue_put():
-            # No maxsize, won't block.
-            await q.put(1)
+    @pytest.mark.asyncio
+    async def test_blocking_put(self):
+        _q = janus.Queue()
+        q = _q.async_q
 
-        self.loop.run_until_complete(queue_put())
+        # No maxsize, won't block.
+        await q.put(1)
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
-    def test_blocking_put_wait(self):
-        _q = janus.Queue(maxsize=1, loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_blocking_put_wait(self):
+        loop = janus.current_loop()
+        _q = janus.Queue(maxsize=1)
         q = _q.async_q
-        started = asyncio.Event(loop=self.loop)
+        started = asyncio.Event()
         finished = False
 
         async def queue_put():
@@ -329,41 +336,44 @@ async def queue_put():
             finished = True
 
         async def queue_get():
-            self.loop.call_later(0.01, q.get_nowait)
-            queue_put_task = self.loop.create_task(queue_put())
+            loop.call_later(0.01, q.get_nowait)
+            queue_put_task = loop.create_task(queue_put())
             await started.wait()
             self.assertFalse(finished)
             await queue_put_task
             self.assertTrue(finished)
 
-        self.loop.run_until_complete(queue_get())
+        await queue_get()
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
-    def test_nonblocking_put(self):
-        _q = janus.Queue(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_nonblocking_put(self):
+        _q = janus.Queue()
         q = _q.async_q
         q.put_nowait(1)
         self.assertEqual(1, q.get_nowait())
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
-    def test_nonblocking_put_exception(self):
-        _q = janus.Queue(maxsize=1, loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_nonblocking_put_exception(self):
+        _q = janus.Queue(maxsize=1)
         q = _q.async_q
         q.put_nowait(1)
         self.assertRaises(asyncio.QueueFull, q.put_nowait, 2)
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
-    def test_float_maxsize(self):
-        _q = janus.Queue(maxsize=1.3, loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_float_maxsize(self):
+        _q = janus.Queue(maxsize=1.3)
         q = _q.async_q
         q.put_nowait(1)
         q.put_nowait(2)
@@ -371,9 +381,9 @@ def test_float_maxsize(self):
         self.assertRaises(asyncio.QueueFull, q.put_nowait, 3)
 
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
-        _q = janus.Queue(maxsize=1.3, loop=self.loop)
+        _q = janus.Queue(maxsize=1.3)
         q = _q.async_q
 
         async def queue_put():
@@ -381,14 +391,16 @@ async def queue_put():
             await q.put(2)
             self.assertTrue(q.full())
 
-        self.loop.run_until_complete(queue_put())
+        await queue_put()
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
-    def test_put_cancelled(self):
-        _q = janus.Queue(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_put_cancelled(self):
+        loop = janus.current_loop()
+        _q = janus.Queue()
         q = _q.async_q
 
         async def queue_put():
@@ -398,30 +410,32 @@ async def queue_put():
         async def test():
             return (await q.get())
 
-        t = self.loop.create_task(queue_put())
-        self.assertEqual(1, self.loop.run_until_complete(test()))
+        t = loop.create_task(queue_put())
+        self.assertEqual(1, await test())
         self.assertTrue(t.done())
         self.assertTrue(t.result())
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
-    def test_put_cancelled_race(self):
-        _q = janus.Queue(loop=self.loop, maxsize=1)
+    @pytest.mark.asyncio
+    async def test_put_cancelled_race(self):
+        loop = janus.current_loop()
+        _q = janus.Queue(maxsize=1)
         q = _q.async_q
 
-        put_a = self.loop.create_task(q.put('a'))
-        put_b = self.loop.create_task(q.put('b'))
-        put_c = self.loop.create_task(q.put('X'))
+        put_a = loop.create_task(q.put('a'))
+        put_b = loop.create_task(q.put('b'))
+        put_c = loop.create_task(q.put('X'))
 
-        self.loop.run_until_complete(put_a)
+        await put_a
         self.assertFalse(put_b.done())
 
         put_c.cancel()
 
         with self.assertRaises(asyncio.CancelledError):
-            self.loop.run_until_complete(put_c)
+            await put_c
 
         async def go():
             a = await q.get()
@@ -430,14 +444,16 @@ async def go():
             self.assertEqual(b, 'b')
             self.assertTrue(put_b.done())
 
-        self.loop.run_until_complete(go())
+        await go()
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
-    def test_put_with_waiting_getters(self):
-        fut = self.loop.create_future()
+    @pytest.mark.asyncio
+    async def test_put_with_waiting_getters(self):
+        loop = janus.current_loop()
+        fut = loop.create_future()
 
         async def go():
             fut.set_result(None)
@@ -447,21 +463,22 @@ async def go():
         async def put():
             await q.put('a')
 
-        _q = janus.Queue(loop=self.loop)
+        _q = janus.Queue()
         q = _q.async_q
-        t = self.loop.create_task(go())
-        self.loop.run_until_complete(fut)
-        self.loop.run_until_complete(put())
-        self.assertEqual(self.loop.run_until_complete(t), 'a')
+        t = loop.create_task(go())
+        await fut
+        await put()
+        self.assertEqual(await t, 'a')
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
 
-class LifoQueueTests(_QueueTestBase):
-    def test_order(self):
-        _q = janus.LifoQueue(loop=self.loop)
+class LifoQueueTests(unittest.TestCase):
+    @pytest.mark.asyncio
+    async def test_order(self):
+        _q = janus.LifoQueue()
         q = _q.async_q
         for i in [1, 3, 2]:
             q.put_nowait(i)
@@ -471,12 +488,13 @@ def test_order(self):
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
 
-class PriorityQueueTests(_QueueTestBase):
-    def test_order(self):
-        _q = janus.PriorityQueue(loop=self.loop)
+class PriorityQueueTests(unittest.TestCase):
+    @pytest.mark.asyncio
+    async def test_order(self):
+        _q = janus.PriorityQueue()
         q = _q.async_q
         for i in [1, 3, 2]:
             q.put_nowait(i)
@@ -486,24 +504,27 @@ def test_order(self):
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
 
 class _QueueJoinTestMixin:
 
     q_class = None
 
-    def test_task_done_underflow(self):
-        _q = self.q_class(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_task_done_underflow(self):
+        _q = self.q_class()
         q = _q.async_q
         self.assertRaises(ValueError, q.task_done)
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
-    def test_task_done(self):
-        _q = self.q_class(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_task_done(self):
+        loop = janus.current_loop()
+        _q = self.q_class()
         q = _q.async_q
         for i in range(100):
             q.put_nowait(i)
@@ -523,27 +544,28 @@ async def worker():
                 q.task_done()
 
         async def test():
-            tasks = [self.loop.create_task(worker())
+            tasks = [loop.create_task(worker())
                      for index in range(2)]
 
             await q.join()
             return tasks
 
-        tasks = self.loop.run_until_complete(test())
+        tasks = await test()
         self.assertEqual(sum(range(100)), accumulator)
 
         # close running generators
         running = False
         for i in range(len(tasks)):
             q.put_nowait(0)
-        self.loop.run_until_complete(asyncio.wait(tasks, loop=self.loop))
+        await asyncio.wait(tasks)
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
-    def test_join_empty_queue(self):
-        _q = self.q_class(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_join_empty_queue(self):
+        _q = self.q_class()
         q = _q.async_q
 
         # Test that a queue join()s successfully, and before anything else
@@ -553,15 +575,16 @@ async def join():
             await q.join()
             await q.join()
 
-        self.loop.run_until_complete(join())
+        await join()
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
     @unittest.expectedFailure
-    def test_format(self):
-        _q = self.q_class(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_format(self):
+        _q = self.q_class()
         q = _q.async_q
         self.assertEqual(q._format(), 'maxsize=0')
 
@@ -570,16 +593,16 @@ def test_format(self):
 
         self.assertFalse(_q._sync_mutex.locked())
         _q.close()
-        self.loop.run_until_complete(_q.wait_closed())
+        await _q.wait_closed()
 
 
-class QueueJoinTests(_QueueJoinTestMixin, _QueueTestBase):
+class QueueJoinTests(_QueueJoinTestMixin, unittest.TestCase):
     q_class = janus.Queue
 
 
-class LifoQueueJoinTests(_QueueJoinTestMixin, _QueueTestBase):
+class LifoQueueJoinTests(_QueueJoinTestMixin, unittest.TestCase):
     q_class = janus.LifoQueue
 
 
-class PriorityQueueJoinTests(_QueueJoinTestMixin, _QueueTestBase):
+class PriorityQueueJoinTests(_QueueJoinTestMixin, unittest.TestCase):
     q_class = janus.PriorityQueue
diff --git a/tests/test_mixed.py b/tests/test_mixed.py
index fbb7bfd..90d6ca9 100644
--- a/tests/test_mixed.py
+++ b/tests/test_mixed.py
@@ -1,50 +1,40 @@
 import asyncio
-import concurrent.futures
+import sys
 import unittest
 
-from unittest import mock
-
 import janus
+import pytest
 
 
 class TestMixedMode(unittest.TestCase):
-    def setUp(self):
-        self.loop = asyncio.new_event_loop()
-        self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
-        self.loop.set_default_executor(self.executor)
-        asyncio.set_event_loop(None)
-
-    def tearDown(self):
-        self.executor.shutdown()
-        self.loop.close()
-
-    def test_ctor_loop(self):
-        loop = mock.Mock()
-        q = janus.Queue(loop=loop)
-        self.assertIs(q._loop, loop)
-
-        q = janus.Queue(loop=self.loop)
-        self.assertIs(q._loop, self.loop)
 
+    @pytest.mark.skipif(
+        sys.version_info < (3, 7),
+        reason="forbidding implicit loop creation works on "
+               "Python 3.7 or higher only",
+    )
     def test_ctor_noloop(self):
-        asyncio.set_event_loop(self.loop)
-        q = janus.Queue()
-        self.assertIs(q._loop, self.loop)
+        with self.assertRaises(RuntimeError):
+            janus.Queue()
 
-    def test_maxsize(self):
-        q = janus.Queue(5, loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_maxsize(self):
+        q = janus.Queue(5)
         self.assertIs(5, q.maxsize)
 
-    def test_maxsize_named_param(self):
-        q = janus.Queue(maxsize=7, loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_maxsize_named_param(self):
+        q = janus.Queue(maxsize=7)
         self.assertIs(7, q.maxsize)
 
-    def test_maxsize_default(self):
-        q = janus.Queue(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_maxsize_default(self):
+        q = janus.Queue()
         self.assertIs(0, q.maxsize)
 
-    def test_unfinished(self):
-        q = janus.Queue(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_unfinished(self):
+        q = janus.Queue()
         self.assertEqual(q.sync_q.unfinished_tasks, 0)
         self.assertEqual(q.async_q.unfinished_tasks, 0)
         q.sync_q.put(1)
@@ -56,16 +46,20 @@ def test_unfinished(self):
         q.sync_q.task_done()
         self.assertEqual(q.sync_q.unfinished_tasks, 0)
         self.assertEqual(q.async_q.unfinished_tasks, 0)
+        q.close()
+        await q.wait_closed()
 
-    def test_sync_put_async_get(self):
-        q = janus.Queue(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_sync_put_async_get(self):
+        loop = janus.current_loop()
+        q = janus.Queue()
 
         def threaded():
             for i in range(5):
                 q.sync_q.put(i)
 
         async def go():
-            f = self.loop.run_in_executor(None, threaded)
+            f = loop.run_in_executor(None, threaded)
             for i in range(5):
                 val = await q.async_q.get()
                 self.assertEqual(val, i)
@@ -75,10 +69,15 @@ async def go():
             await f
 
         for i in range(3):
-            self.loop.run_until_complete(go())
+            await go()
 
-    def test_sync_put_async_join(self):
-        q = janus.Queue(loop=self.loop)
+        q.close()
+        await q.wait_closed()
+
+    @pytest.mark.asyncio
+    async def test_sync_put_async_join(self):
+        loop = janus.current_loop()
+        q = janus.Queue()
 
         for i in range(5):
             q.sync_q.put(i)
@@ -89,16 +88,21 @@ async def do_work():
                 await q.async_q.get()
                 q.async_q.task_done()
 
-        task = self.loop.create_task(do_work())
+        task = loop.create_task(do_work())
 
         async def wait_for_empty_queue():
             await q.async_q.join()
             task.cancel()
 
-        self.loop.run_until_complete(wait_for_empty_queue())
+        await wait_for_empty_queue()
+
+        q.close()
+        await q.wait_closed()
 
-    def test_async_put_sync_get(self):
-        q = janus.Queue(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_async_put_sync_get(self):
+        loop = janus.current_loop()
+        q = janus.Queue()
 
         def threaded():
             for i in range(5):
@@ -106,7 +110,7 @@ def threaded():
                 self.assertEqual(val, i)
 
         async def go():
-            f = self.loop.run_in_executor(None, threaded)
+            f = loop.run_in_executor(None, threaded)
             for i in range(5):
                 await q.async_q.put(i)
 
@@ -114,10 +118,15 @@ async def go():
             self.assertTrue(q.async_q.empty())
 
         for i in range(3):
-            self.loop.run_until_complete(go())
+            await go()
 
-    def test_sync_join_async_done(self):
-        q = janus.Queue(loop=self.loop)
+        q.close()
+        await q.wait_closed()
+
+    @pytest.mark.asyncio
+    async def test_sync_join_async_done(self):
+        loop = janus.current_loop()
+        q = janus.Queue()
 
         def threaded():
             for i in range(5):
@@ -125,7 +134,7 @@ def threaded():
             q.sync_q.join()
 
         async def go():
-            f = self.loop.run_in_executor(None, threaded)
+            f = loop.run_in_executor(None, threaded)
             for i in range(5):
                 val = await q.async_q.get()
                 self.assertEqual(val, i)
@@ -136,10 +145,15 @@ async def go():
             await f
 
         for i in range(3):
-            self.loop.run_until_complete(go())
+            await go()
+
+        q.close()
+        await q.wait_closed()
 
-    def test_async_join_async_done(self):
-        q = janus.Queue(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_async_join_async_done(self):
+        loop = janus.current_loop()
+        q = janus.Queue()
 
         def threaded():
             for i in range(5):
@@ -148,7 +162,7 @@ def threaded():
                 q.sync_q.task_done()
 
         async def go():
-            f = self.loop.run_in_executor(None, threaded)
+            f = loop.run_in_executor(None, threaded)
             for i in range(5):
                 await q.async_q.put(i)
 
@@ -158,19 +172,24 @@ async def go():
             self.assertTrue(q.async_q.empty())
 
         for i in range(3):
-            self.loop.run_until_complete(go())
+            await go()
+
+        q.close()
+        await q.wait_closed()
 
-    def test_wait_without_closing(self):
-        q = janus.Queue(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_wait_without_closing(self):
+        q = janus.Queue()
 
         with self.assertRaises(RuntimeError):
-            self.loop.run_until_complete(q.wait_closed())
+            await q.wait_closed()
 
         q.close()
-        self.loop.run_until_complete(q.wait_closed())
+        await q.wait_closed()
 
-    def test_modifying_forbidden_after_closing(self):
-        q = janus.Queue(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_modifying_forbidden_after_closing(self):
+        q = janus.Queue()
         q.close()
 
         with self.assertRaises(RuntimeError):
@@ -183,7 +202,7 @@ def test_modifying_forbidden_after_closing(self):
             q.sync_q.task_done()
 
         with self.assertRaises(RuntimeError):
-            self.loop.run_until_complete(q.async_q.put(5))
+            await q.async_q.put(5)
 
         with self.assertRaises(RuntimeError):
             q.async_q.put_nowait(5)
@@ -192,18 +211,20 @@ def test_modifying_forbidden_after_closing(self):
             q.async_q.get_nowait()
 
         with self.assertRaises(RuntimeError):
-            self.loop.run_until_complete(q.sync_q.task_done())
+            await q.sync_q.task_done()
 
-        self.loop.run_until_complete(q.wait_closed())
+        await q.wait_closed()
 
-    def test_double_closing(self):
-        q = janus.Queue(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_double_closing(self):
+        q = janus.Queue()
         q.close()
         q.close()
-        self.loop.run_until_complete(q.wait_closed())
+        await q.wait_closed()
 
-    def test_closed(self):
-        q = janus.Queue(loop=self.loop)
+    @pytest.mark.asyncio
+    async def test_closed(self):
+        q = janus.Queue()
         self.assertFalse(q.closed)
         self.assertFalse(q.async_q.closed)
         self.assertFalse(q.sync_q.closed)
diff --git a/tests/test_sync.py b/tests/test_sync.py
index 9911e08..11691da 100644
--- a/tests/test_sync.py
+++ b/tests/test_sync.py
@@ -1,7 +1,5 @@
 # Some simple queue module tests, plus some failure conditions
 # to ensure the Queue locks remain stable.
-import asyncio
-import concurrent.futures
 import queue
 import time
 import unittest
@@ -10,6 +8,7 @@
 import threading
 
 import janus
+import pytest
 
 QUEUE_SIZE = 5
 
@@ -52,16 +51,6 @@ def run(self):
 
 
 class BlockingTestMixin:
-    def setUp(self):
-        asyncio.set_event_loop(None)
-        self.loop = asyncio.new_event_loop()
-        self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
-        self.loop.set_default_executor(self.executor)
-
-    def tearDown(self):
-        self.t = None
-        self.executor.shutdown()
-        self.loop.close()
 
     def do_blocking_test(self, block_func, block_args, trigger_func,
                          trigger_args):
@@ -102,6 +91,7 @@ def do_exceptional_blocking_test(self, block_func, block_args,
 
 
 class BaseQueueTestMixin(BlockingTestMixin):
+
     def setUp(self):
         self.cum = 0
         self.cumlock = threading.Lock()
@@ -187,20 +177,26 @@ def queue_join_test(self, q):
             q.put(-1)  # instruct the threads to close
         q.join()  # verify that you can join twice
 
-    def test_queue_task_done(self):
+    @pytest.mark.asyncio
+    async def test_queue_task_done(self):
         # Test to make sure a queue task completed successfully.
-        q = self.type2test(loop=self.loop).sync_q
+        _q = self.type2test()
+        q = _q.sync_q
         try:
             q.task_done()
         except ValueError:
             pass
         else:
             self.fail("Did not detect task count going negative")
+        _q.close()
+        await _q.wait_closed()
 
-    def test_queue_join(self):
+    @pytest.mark.asyncio
+    async def test_queue_join(self):
         # Test that a queue join()s successfully, and before anything else
         # (done twice for insurance).
-        q = self.type2test(loop=self.loop).sync_q
+        _q = self.type2test()
+        q = _q.sync_q
         self.queue_join_test(q)
         self.queue_join_test(q)
         try:
@@ -209,23 +205,35 @@ def test_queue_join(self):
             pass
         else:
             self.fail("Did not detect task count going negative")
+        finally:
+            _q.close()
+            await _q.wait_closed()
 
-    def test_simple_queue(self):
+    @pytest.mark.asyncio
+    async def test_simple_queue(self):
         # Do it a couple of times on the same queue.
         # Done twice to make sure works with same instance reused.
-        q = self.type2test(QUEUE_SIZE, loop=self.loop)
-        self.simple_queue_test(q)
-        self.simple_queue_test(q)
-
-    def test_negative_timeout_raises_exception(self):
-        q = self.type2test(QUEUE_SIZE, loop=self.loop).sync_q
+        _q = self.type2test(QUEUE_SIZE)
+        self.simple_queue_test(_q)
+        self.simple_queue_test(_q)
+        _q.close()
+        await _q.wait_closed()
+
+    @pytest.mark.asyncio
+    async def test_negative_timeout_raises_exception(self):
+        _q = self.type2test(QUEUE_SIZE)
+        q = _q.sync_q
         with self.assertRaises(ValueError):
             q.put(1, timeout=-1)
         with self.assertRaises(ValueError):
             q.get(1, timeout=-1)
+        _q.close()
+        await _q.wait_closed()
 
-    def test_nowait(self):
-        q = self.type2test(QUEUE_SIZE, loop=self.loop).sync_q
+    @pytest.mark.asyncio
+    async def test_nowait(self):
+        _q = self.type2test(QUEUE_SIZE)
+        q = _q.sync_q
         for i in range(QUEUE_SIZE):
             q.put_nowait(1)
         with self.assertRaises(queue.Full):
@@ -235,10 +243,14 @@ def test_nowait(self):
             q.get_nowait()
         with self.assertRaises(queue.Empty):
             q.get_nowait()
+        _q.close()
+        await _q.wait_closed()
 
-    def test_shrinking_queue(self):
+    @pytest.mark.asyncio
+    async def test_shrinking_queue(self):
         # issue 10110
-        q = self.type2test(3, loop=self.loop).sync_q
+        _q = self.type2test(3)
+        q = _q.sync_q
         q.put(1)
         q.put(2)
         q.put(3)
@@ -248,11 +260,17 @@ def test_shrinking_queue(self):
         q._maxsize = 2  # shrink the queue
         with self.assertRaises(queue.Full):
             q.put_nowait(4)
+        _q.close()
+        await _q.wait_closed()
 
-    def test_maxsize(self):
+    @pytest.mark.asyncio
+    async def test_maxsize(self):
         # Test to make sure a queue task completed successfully.
-        q = self.type2test(maxsize=5, loop=self.loop).sync_q
+        _q = self.type2test(5)
+        q = _q.sync_q
         self.assertEqual(q.maxsize, 5)
+        _q.close()
+        await _q.wait_closed()
 
 
 class QueueTest(BaseQueueTestMixin, unittest.TestCase):
@@ -374,20 +392,28 @@ def failing_queue_test(self, _q):
         q.get()
         self.assertTrue(not q.qsize(), "Queue should be empty")
 
-    def test_failing_queue(self):
+    @pytest.mark.asyncio
+    async def test_failing_queue(self):
         # Test to make sure a queue is functioning correctly.
         # Done twice to the same instance.
-        q = FailingQueue(QUEUE_SIZE, loop=self.loop)
+        q = FailingQueue(QUEUE_SIZE)
         self.failing_queue_test(q)
         self.failing_queue_test(q)
+        q.close()
+        await q.wait_closed()
 
-    def test_closed_loop_non_failing(self):
-        q = janus.Queue(QUEUE_SIZE, loop=self.loop).sync_q
+    @pytest.mark.asyncio
+    async def test_closed_loop_non_failing(self):
+        loop = janus.current_loop()
+        _q = janus.Queue(QUEUE_SIZE)
+        q = _q.sync_q
         # we are pacthing loop to follow setUp/tearDown agreement
-        with patch.object(self.loop, 'call_soon_threadsafe') as func:
+        with patch.object(loop, 'call_soon_threadsafe') as func:
             func.side_effect = RuntimeError()
             q.put_nowait(1)
             self.assertEqual(func.call_count, 1)
+        _q.close()
+        await _q.wait_closed()
 
 
 if __name__ == "__main__":
diff --git a/tox.ini b/tox.ini
index e36af68..1897f57 100644
--- a/tox.ini
+++ b/tox.ini
@@ -1,6 +1,6 @@
 [tox]
 
-envlist = check, {py34,py35}-{debug,release}, report
+envlist = check, {py35,py36,py37,py38}-{debug,release}, report
 
 
 [testenv]
@@ -8,6 +8,7 @@ envlist = check, {py34,py35}-{debug,release}, report
 deps =
     coverage
     pytest
+    pytest-asyncio
 
 commands =
     coverage run -m pytest {posargs}
@@ -17,9 +18,10 @@ setenv =
     debug: PYTHONASYNCIODEBUG = 1
 
 basepython:
-    py33: python3.3
-    py34: python3.4
     py35: python3.5
+    py36: python3.6
+    py37: python3.7
+    py38: python3.8
 
 whitelist_externals =
     coverage