hypothesis_twisted.py

from functools import wraps
from inspect import getargspec

from hypothesis import given
from hypothesis.internal.reflection import copy_argspec
from twisted.internet import reactor
from twisted.internet.threads import deferToThread, blockingCallFromThread


def given_defer(*generator_arguments, **generator_kwargs):
    run_test_with_generator = given(*generator_arguments, **generator_kwargs)

    @wraps(run_test_with_generator)
    def defer_test_with_generator(test):

        @wraps(test)
        @copy_argspec(test.__name__, getargspec(test))
        def run_test_from_thread(*args, **kw):
            return blockingCallFromThread(reactor, test, *args, **kw)

        wrapped_test = run_test_with_generator(run_test_from_thread)

        @wraps(wrapped_test)
        def defer_wrapped_test(*args, **kw):
            return deferToThread(wrapped_test, *args, **kw)

        return defer_wrapped_test

    return defer_test_with_generator