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):
    print 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, loop, *a, **kw):
    f = gen.coroutine(f)
    loop.run_sync(functools.partial(f, loop, *a, **kw))
    loop.stop()


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