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
197
198
199
200
201
202
203
204
|
#-*- coding: utf-8 -*-
"""
runner
------
A module containing application and daemon process utilities.
@author Isis Agora Lovecruft <isis@leap.se>, 0x2cdb8b35
@version 0.0.1
"""
from os import path as ospath
import re
from twisted.internet import defer
from twisted.trial import runner, reporter, unittest
from twisted.python import usage
from leap.mx.util import log, version
class CheckRequirements(ImportError):
"""
Raised when we're missing something from requirements.pip.
"""
def __init__(self, package_name, pipfile, message=None):
"""
Display an error message with instructions for obtaining missing
dependencies.
@param message: A string describing the error.
@param missing: A string indicating which dependency is missing.
@param pipfile: The path and filename of the pip requirements file,
relative to the top-level repository directory.
"""
if message:
self.message = message
return self
self.package_name = package_name
self.pipfile = pipfile
self.dependencies = self.__read_pip_requirements__()
self.missing = []
for package, version in self.dependencies:
pkg = package.lower() if package == "Twisted" else package
try:
__import__(pkg)
except ImportError:
self.missing.append(package)
if len(self.missing) > 0:
self.message = self.package_name + " requires "
elif len(self.missing) <= 0:
return None
if len(self.missing) >= 1:
for missed in self.missing[:-1]:
self.message += missed + ", "
self.message += "and "
if len(self.missing) == 1:
self.message += self.missing[0] + "."
self.message += "\nPlease see %s for ".format(self.pipfile)
self.message += "instruction on installing dependencies."
raise self(self.message)
def __read_pip_requirements__(self, file=None):
"""
Check the pip requirements file to determine our dependencies.
@param file: The full path of the pip requirements.txt file.
@returns: A list of tuple(package_name, package_version).
"""
if not file:
file = self.pipfile
requirement = re.compile('[^0-9=><]+')
dependencies = []
with open(file) as pipfile:
for line in pipfile.readlines():
shortened = line.strip()
matched = requirement.match(shortened)
package_name = matched.group()
package_version = shortened.split(package_name, 1)[1]
dependencies.append((package_name, package_version))
return dependencies
class TestRunner(runner.TrialRunner):
"""
This class handles loading unittests from a directory, module, file,
class, method, or anything really, and running any unittests found with
twisted.trial.
@param options: A subclass of :class:twisted.python.usage.Options.
@param test: (optional) The thing to load tests from.
"""
def __init__(self, options, test_location=None, *args, **kwargs):
"""
Create a runner for handling unittest runs.
@param options: An the dictionary :ivar:`opts` from the parsed options
of a :class:`twisted.python.usage.Options`.
@param test_location: The path to a directory or filename containing
unittests to run, or the path to a file prefixed with 'test_', or
a subclass of :class:`twisted.trial.unittest.TestCase`, or a
function/method prefixed with 'test_'.
"""
log.debug("Creating TestRunner: %s" % self.__repr__())
if isinstance(options, dict):
log.debug("TestRunner loaded options class")
self.options = options
else:
self.options = None
raise usage.UsageError(
"TestRunner expected t.p.u.Options subclass, got %s"
% options)
self.loader = runner.TestLoader()
self._parse_options()
self.test_location = test_location
self.tests_loaded = self.loadUnittests()
self.test_list = self.runUnittests()
self.results = defer.DeferredList(self.test_list)
def _parse_options(self):
"""
Parse the :class:`twisted.python.usage.Options` for flags pertaining
to running unittests.
"""
if not self.options is None:
if self.options['debug']:
log.debug("Enabled debugging on test runner.")
self.DEBUG = True
if self.options['force-gc']:
## not sure if we need to call it or assign it...
log.debug("Forcing garbage collection between unittest runs.")
self.loader.forceGarbageCollection(True)
if self.options['all-tests']:
repo = version.getRepoDir()
test_directory = ospath.join(repo, 'src/leap/mx/tests')
self.test_location = test_directory
else:
log.warn("TestRunner: got None for t.p.u.Options class!")
def loadUnittests(self):
"""
Load all tests. Tests may be a module or directory, the path to a
filename in the form 'test_*.py", a subclass of
:class:`twisted.trial.unittest.TestCase` or
:class:`twisted.trial.unittest.TestSuite`, or a any function/method
which is prefixed with 'test_'.
@returns: An instance of :class:`twisted.trial.unittest.TestCase` or
:class:`twisted.trial.unittest.TestSuite`.
"""
log.msg("Attempting to load unittests...")
tests_loaded = None
if self.test_location:
log.msg("Loading unittests from %s" % self.test_location)
if ospath.isdir(self.test_location):
log.msg("Found test directory: %s" % test)
tests_loaded = self.loader.loadAnything(self.test_location,
recurse=True)
else:
log.msg("Found test file: %s" % self.test_location)
tests_loaded = self.loader.loadAnything(self.test_location)
else:
log.warn("Test location %s seems to be None!" % self.test_location)
return tests_loaded
def runUnittests(self):
"""xxx fill me in"""
results = []
if not self.tests_loaded is None:
if isinstance(self.tests_loaded, unittest.TestCase):
log.msg("Test case loaded.")
classes = self.loader.findTestClasses(self.tests_loaded)
for cls in classes:
test_instance = cls()
test_instance.setUp() ## xxx does TestRunner handle this?
d = defer.maybeDeferred(test_instance.run())
self.results.append(d)
elif isinstance(self.tests, unittest.TestSuite):
classes = None ## xxx call each TestCase in TestSuite
test_suite = self.tests()
self.results.append(test_suite.visit())
log.msg("Test suite loaded: %d tests to run"
% test_suite.countTestCases)
return results
#return runner.TrialRunner(reporter.TreeReporter, mode=mode,
# profile=profile, logfile=logfile,
# tbformat, rterrors, unclean_warnings,
# temp-directory, force-gc)
|
