pytest.tornado.py

import time
import functools

from decorator import decorator

import pytest

from tornado import ioloop
from tornado import gen


@pytest.fixture
def loop():
    loop = ioloop.IOLoop()
    loop.make_current()
    return loop


@pytest.fixture
def three(loop):
    return 3


def sleep(loop, s):
    result = gen.Future()
    loop.add_timeout(
        time.time()+s, lambda: result.set_result(None))
    return result


@decorator
def async_test(f, *args, **kwargs):
    loop = None
    for arg in args:
        if isinstance(arg, ioloop.IOLoop):
            loop = arg
            break

    if not loop:
        loop = ioloop.IOLoop()

    f = gen.coroutine(f)
    loop.run_sync(functools.partial(f, *args, **kwargs))
    loop.stop()


@async_test
def test_main(loop, three):
    print
    print
    print three
    yield sleep(loop, three)
    print "done"



class TestFoo(object):
    @async_test
    def test_bar(self, loop, three):
        print
        print
        print three
        yield sleep(loop, three)
        print "done"