summaryrefslogtreecommitdiff
path: root/kombu/tests/utils.py
blob: 8ef4cbd4bd00f75d8223450bfa5ba020f90b7d4a (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
import __builtin__
import os
import sys
import types

from StringIO import StringIO

from nose import SkipTest

from kombu.utils.functional import wraps

try:
    import unittest
    unittest.skip
except AttributeError:
    import unittest2 as unittest  # noqa


def redirect_stdouts(fun):

    @wraps(fun)
    def _inner(*args, **kwargs):
        sys.stdout = StringIO()
        sys.stderr = StringIO()
        try:
            return fun(*args, **dict(kwargs, stdout=sys.stdout,
                                             stderr=sys.stderr))
        finally:
            sys.stdout = sys.__stdout__
            sys.stderr = sys.__stderr__

    return _inner


def module_exists(*modules):

    def _inner(fun):

        @wraps(fun)
        def __inner(*args, **kwargs):
            for module in modules:
                if isinstance(module, basestring):
                    module = types.ModuleType(module)
                sys.modules[module.__name__] = module
                try:
                    return fun(*args, **kwargs)
                finally:
                    sys.modules.pop(module.__name__, None)

        return __inner
    return _inner


# Taken from
# http://bitbucket.org/runeh/snippets/src/tip/missing_modules.py
def mask_modules(*modnames):
    """Ban some modules from being importable inside the context

    For example:

        >>> @missing_modules("sys"):
        >>> def foo():
        ...     try:
        ...         import sys
        ...     except ImportError:
        ...         print("sys not found")
        sys not found

        >>> import sys
        >>> sys.version
        (2, 5, 2, 'final', 0)

    """

    def _inner(fun):

        @wraps(fun)
        def __inner(*args, **kwargs):
            realimport = __builtin__.__import__

            def myimp(name, *args, **kwargs):
                if name in modnames:
                    raise ImportError("No module named %s" % name)
                else:
                    return realimport(name, *args, **kwargs)

            __builtin__.__import__ = myimp
            try:
                return fun(*args, **kwargs)
            finally:
                __builtin__.__import__ = realimport

        return __inner
    return _inner


def skip_if_environ(env_var_name):

    def _wrap_test(fun):

        @wraps(fun)
        def _skips_if_environ(*args, **kwargs):
            if os.environ.get(env_var_name):
                raise SkipTest("SKIP %s: %s set\n" % (
                    fun.__name__, env_var_name))
            return fun(*args, **kwargs)

        return _skips_if_environ

    return _wrap_test


def skip_if_module(module):
    def _wrap_test(fun):
        @wraps(fun)
        def _skip_if_module(*args, **kwargs):
            try:
                __import__(module)
                raise SkipTest("SKIP %s: %s available\n" % (
                    fun.__name__, module))
            except ImportError:
                pass
            return fun(*args, **kwargs)
        return _skip_if_module
    return _wrap_test


def skip_if_not_module(module):
    def _wrap_test(fun):
        @wraps(fun)
        def _skip_if_not_module(*args, **kwargs):
            try:
                __import__(module)
            except ImportError:
                raise SkipTest("SKIP %s: %s available\n" % (
                    fun.__name__, module))
            return fun(*args, **kwargs)
        return _skip_if_not_module
    return _wrap_test


def skip_if_quick(fun):
    return skip_if_environ("QUICKTEST")(fun)