Skip to content

Commit

Permalink
Use asyncio.run instead of get_event_loop()
Browse files Browse the repository at this point in the history
  • Loading branch information
hrnciar committed Feb 22, 2023
1 parent bd49b3d commit da77a56
Showing 1 changed file with 17 additions and 3 deletions.
20 changes: 17 additions & 3 deletions src/tests/test.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,10 @@
import unittest
import decimal
import inspect
from asyncio import get_event_loop
try:
from asyncio import run
except ImportError:
from asyncio import get_event_loop
from collections import defaultdict, ChainMap, abc as c
from decorator import dispatch_on, contextmanager, decorator
try:
Expand All @@ -28,9 +31,20 @@ async def before_after(coro, *args, **kwargs):
return "<before>" + (await coro(*args, **kwargs)) + "<after>"


def asyncio_run(awaitable):
"""asyncio.run for Python 3.5"""
try:
aio_run = run
except AttributeError:
# Python 3.5
return get_event_loop().run_until_complete(awaitable)
else:
return aio_run(awaitable)


@decorator
def coro_to_func(coro, *args, **kw):
return get_event_loop().run_until_complete(coro(*args, **kw))
return asyncio_run(coro(*args, **kw))


class CoroutineTestCase(unittest.TestCase):
Expand All @@ -39,7 +53,7 @@ def test_before_after(self):
async def coro(x):
return x
self.assertTrue(inspect.iscoroutinefunction(coro))
out = get_event_loop().run_until_complete(coro('x'))
out = asyncio_run(coro('x'))
self.assertEqual(out, '<before>x<after>')

def test_coro_to_func(self):
Expand Down

0 comments on commit da77a56

Please sign in to comment.