Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
from aiohttp.test_utils import unittest_run_loop
from main import Model, new_manager
import asyncio
import unittest
class ManagerTestCase(unittest.TestCase):
def setUp(self):
self.loop = setup_test_loop()
Model.create_table(fail_silently=True)
@unittest_run_loop
async def test_manager1(self):
manager = new_manager(self.loop)
await manager.create(Model)
@unittest_run_loop
async def test_manager2(self):
manager = new_manager(self.loop)
await manager.create(Model)
def tearDown(self):
teardown_test_loop(self.loop)
Model.drop_table(fail_silently=True)
def setup_test_loop(loop_factory=asyncio.new_event_loop):
"""Create and return an asyncio.BaseEventLoop
instance.
The caller should also call teardown_test_loop,
once they are done with the loop.
"""
loop = loop_factory()
asyncio.set_event_loop(None)
return loop
def teardown_test_loop(loop, fast=False):
"""Teardown and cleanup an event_loop created
by setup_test_loop.
"""
closed = loop.is_closed()
if not closed:
loop.call_soon(loop.stop)
loop.run_forever()
loop.close()
asyncio.set_event_loop(None)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment