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
|
# Copyright (C) 2012- Takafumi Arakaki
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import os
import sys
import functools
import io
try:
import unittest
unittest.TestCase.assertIs
except AttributeError:
import unittest2 as unittest
from contextlib import contextmanager
from ..py3compat import Queue, PY3
from ..utils import newthread
@contextmanager
def mockedattr(object, name, replace):
"""
Mock `object.name` attribute using `replace`.
"""
original = getattr(object, name)
try:
setattr(object, name, replace)
yield
finally:
setattr(object, name, original)
def logging_to_stdout(logger):
# it assumes that 0-th hander is the only one stream handler...
return mockedattr(logger.handlers[0], 'stream', sys.stdout)
def streamio():
"""
Return `io.StringIO` for Python 3, otherwise `io.BytesIO`.
"""
if PY3:
return io.StringIO()
else:
return io.BytesIO()
class CaptureStdIO(object):
def __enter__(self):
self._orig_stdin = sys.stdin
self._orig_stdout = sys.stdout
self._orig_stderr = sys.stderr
self.stdin = sys.stdin = streamio()
self.stdout = sys.stdout = streamio()
self.stderr = sys.stderr = streamio()
return self
def __exit__(self, exc_type, exc_value, traceback):
sys.stdin = self._orig_stdin
sys.stdout = self._orig_stdout
sys.stderr = self._orig_stderr
def read_stdout(self):
self.stdout.seek(0)
return self.stdout.read()
def read_stderr(self):
self.stderr.seek(0)
return self.stderr.read()
class BaseTestCase(unittest.TestCase):
TRAVIS = os.getenv('TRAVIS')
if TRAVIS:
timeout = 10
else:
timeout = 1
def skip(reason):
from nose import SkipTest
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwds):
raise SkipTest("Skipping {0} because: {1}"
.format(func.__name__, reason))
return wrapper
return decorator
def post_mortem_in_thread(traceback):
"""
`pdb.post_mortem` that can be used in a daemon thread.
Put the following in the `except`-block::
import sys
from epc.tests.utils import post_mortem_in_thread
exc_info = sys.exc_info()
post_mortem_in_thread(exc_info[2])
"""
import pdb
blocker = Queue.Queue()
thread = newthread(target=blocker.get)
thread.daemon = False
thread.start()
pdb.post_mortem(traceback)
blocker.put(None)
|
