Тестирование асинхронных приложений

@proofit404

email / github / twitter

Зачем вообще заморачиваться?

Who's in da house?

asyncio tornado twisted

asyncio

                        
@asyncio.coroutine
def fetch_page(session, url):
    response = yield from session.get(url)
    assert response.status == 200
    return (yield from response.text())
                        
                    

Event loop

                        
loop = asyncio.get_event_loop()
session = aiohttp.ClientSession()
coroutine = fetch_page(session, 'https://www.python.org')
loop.run_until_complete(coroutine)
                        
                    

Unittest

                        
class MyTest(TestCase):

    def setUp(self):

        self.db = StrictRedis()

    def test_my_test(self):

        self.db.set('foo', 'bar')
        result = self.db.get('foo')
        self.assertEqual('bar', result)
                        
                    

pytest

                        
@pytest.fixture
def redis():
    return StrictRedis()

def test_my_test(redis):
    redis.set('foo', 'bar')
    result = redis.get('foo')
    assert result == 'bar'
                        
                    

В тесте надо:

  • Создать event loop
  • Собрать test case coroutine
  • Запустить тест
  • Дождаться выполнения
  • Всё почистить

test case coroutine

                        
def async_test(f):
    def wrapper(*args, **kwargs):
        loop = asyncio.get_event_loop()
        loop.run_until_complete(f(*args, **kwargs))
    return wrapper
                        
                    

Test definition

                        
class MyTest(TestCase):

    @async_test
    def test_my_test(self):
        session = aiohttp.ClientSession()
        url = 'https://www.python.org'
        result = yield from fetch_page(session, url)
        self.assertIn('</html>', result)
                        
                    

Проблемы

  • @async_test сильно бесит
  • Неудобная подготовка окружения

Async Test Suite

                        
class AsyncTestCase(TestCase):

    def run(self, result=None):
        # ...
        loop = asyncio.get_event_loop()
        loop.run_until_complete(self.setUp())
        loop.run_until_complete(testMethod())
        loop.run_until_complete(self.tearDown())
        # ...
                        
                    

Test rewritten

                        
class MyTest(AsyncTestCase):

    def setUp(self):

        self.db = yield from create_redis(('localhost', 6379))

    def test_my_test(self):

        yield from self.db.set('foo', 'bar')
        result = yield from self.db.get('foo')
        self.assertEqual(b'bar', result)
                        
                    

Pytest suite

                        
@pytest.fixture
def redis():
    pass

def async_test(f):
    def wrapper(**kwargs):
        @asyncio.coroutine
        def coroutine(kwargs):
            if 'redis' in kwargs:
                kwargs['redis'] = yield from create_redis()
            yield from asyncio.coroutine(f)(**kwargs)
        loop = asyncio.new_event_loop()
        loop.run_until_complete(coroutine(kwargs))
    return wrapper
                        
                    

Pytest config

                        
def pytest_pyfunc_call(pyfuncitem):

    path = pyfuncitem.location[0]
    funcargs = pyfuncitem.funcargs
    argnames = pyfuncitem._fixtureinfo.argnames
    testargs = {arg: funcargs[arg] for arg in argnames}
    async_test(pyfuncitem.obj)(**testargs)
    return True
                        
                    

Test rewritten

                        
def test_my_test(redis):
    yield from redis.set('foo', 'bar')
    result = yield from redis.get('foo')
    assert result == b'bar'
                        
                    

Вопросы

Tomato