pygame is sharing code with you
Bitbucket is a code hosting site. Unlimited public and private repositories. Free for small teams.
Don't show this againpygame / unittest_patch.py
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 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 | ################################################################################
import test.unittest as unittest
import re, time, sys, StringIO
from inspect import getdoc
# This is needed for correct tracebacks
__unittest = 1
################################################################################
# Redirect stdout / stderr for the tests
def redirect_output():
yield sys.stderr, sys.stdout
sys.stderr, sys.stdout = StringIO.StringIO(), StringIO.StringIO()
yield sys.stderr, sys.stdout
def restore_output(err, out):
sys.stderr, sys.stdout = err, out
def StringIOContents(io):
io.seek(0)
return io.read()
################################################################################
# TestCase patching
#
def TestCase_run(self, result=None):
if result is None: result = self.defaultTestResult()
result.startTest(self)
testMethod = getattr(self, self._testMethodName)
try:
########################################################################
# Pre run:
result.tests[self.dot_syntax_name()] = {}
tests = result.tests[self.dot_syntax_name()]
(realerr, realout), (stderr, stdout) = redirect_output()
# restore_output(realerr, realout) # DEBUG
t = time.time()
########################################################################
for i in range(self.times_run):
try:
self.setUp()
except KeyboardInterrupt:
raise
except:
result.addError(self, self._exc_info())
return
ok = False
try:
testMethod()
ok = True
except self.failureException:
result.addFailure(self, self._exc_info())
except KeyboardInterrupt:
raise
except:
result.addError(self, self._exc_info())
try:
self.tearDown()
except KeyboardInterrupt:
raise
except:
result.addError(self, self._exc_info())
ok = False
if ok:
if i == 0:
result.addSuccess(self)
else: break
########################################################################
# Post run
t = (time.time() -t) / self.times_run
restore_output(realerr, realout)
tests["time"] = t
tests["stdout"] = StringIOContents(stdout)
tests["stderr"] = StringIOContents(stderr)
########################################################################
finally:
result.stopTest(self)
################################################################################
# TestResult
#
def TestResult___init__(self):
self.failures = []
self.errors = []
self.tests = {}
self.testsRun = 0
self.shouldStop = 0
FILE_LINENUMBER_RE = re.compile(r'File "([^"]+)", line ([0-9]+)')
def errorHandling(key):
def handling(self, test, err):
traceback = self._exc_info_to_string(err, test)
error_file, line_number = FILE_LINENUMBER_RE.search(traceback).groups()
error = (
test.dot_syntax_name(),
traceback,
error_file,
line_number,
)
getattr(self, key).append(error)
# Append it to individual test dict for easy access
self.tests[test.dot_syntax_name()][key[:-1]] = error
return handling
################################################################################
def printErrorList(self, flavour, errors):
for test, err in ((e[0], e[1]) for e in errors):
self.stream.writeln(self.separator1)
self.stream.writeln("%s: %s" % (flavour, test))
self.stream.writeln(self.separator2)
self.stream.writeln("%s" % err)
# DUMP REDIRECTED STDERR / STDOUT ON ERROR / FAILURE
if self.show_redirected_on_errors:
stderr, stdout = map(self.tests[test].get, ('stderr','stdout'))
if stderr or stdout:
if stderr: self.stream.writeln("STDERR:\n%s" % stderr)
if stdout: self.stream.writeln("STDOUT:\n%s" % stdout)
################################################################################
# Exclude by tags
#
TAGS_RE = re.compile(r"\|[tT]ags:([ a-zA-Z,0-9_\n]+)\|", re.M)
def get_tags(obj):
tags = TAGS_RE.search(getdoc(obj) or '')
return tags and [t.strip() for t in tags.group(1).split(',')] or []
def getTestCaseNames(self, testCaseClass):
def test_wanted(attrname, testCaseClass=testCaseClass,
prefix=self.testMethodPrefix):
actual_attr = getattr(testCaseClass, attrname)
filtered = bool([t for t in get_tags(actual_attr) if t in self.exclude])
return ( attrname.startswith(prefix) and callable(actual_attr)
and not filtered )
testFnNames = filter(test_wanted, dir(testCaseClass))
for baseclass in testCaseClass.__bases__:
for testFnName in self.getTestCaseNames(baseclass):
if testFnName not in testFnNames: # handle overridden methods
testFnNames.append(testFnName)
if self.sortTestMethodsUsing:
testFnNames.sort(self.sortTestMethodsUsing)
return testFnNames
################################################################################
def patch(options):
# Tag exclusion
if options.exclude:
unittest.TestLoader.getTestCaseNames = getTestCaseNames
unittest.TestLoader.exclude = (
[e.strip() for e in options.exclude.split(',')] )
# Timing
unittest.TestCase.times_run = options.timings
unittest.TestCase.run = TestCase_run
unittest.TestCase.dot_syntax_name = lambda self: (
"%s.%s"% (self.__class__.__name__, self._testMethodName) )
# Error logging
unittest.TestResult.show_redirected_on_errors = options.show_output
unittest.TestResult.__init__ = TestResult___init__
unittest.TestResult.addError = errorHandling('errors')
unittest.TestResult.addFailure = errorHandling('failures')
unittest._TextTestResult.printErrorList = printErrorList
################################################################################
|