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
|
"""
Subclass of unittest.TestCase with helpers for spawning a separate
process to perform the actual test case.
"""
from __future__ import absolute_import
import os
import os.path
import unittest
from ... import logging
from ...utils import registry
_TEST_CASES = {}
def make_test_case(test_kind, *args, **kwargs):
"""
Factory function for creating TestCase instances.
"""
if test_kind not in _TEST_CASES:
raise ValueError("Unknown test kind '%s'" % test_kind)
return _TEST_CASES[test_kind](*args, **kwargs)
class TestCase(unittest.TestCase):
"""
A test case to execute.
"""
__metaclass__ = registry.make_registry_metaclass(_TEST_CASES)
REGISTERED_NAME = registry.LEAVE_UNREGISTERED
def __init__(self, logger, test_kind, test_name):
"""
Initializes the TestCase with the name of the test.
"""
unittest.TestCase.__init__(self, methodName="run_test")
if not isinstance(logger, logging.Logger):
raise TypeError("logger must be a Logger instance")
if not isinstance(test_kind, basestring):
raise TypeError("test_kind must be a string")
if not isinstance(test_name, basestring):
raise TypeError("test_name must be a string")
# When the TestCase is created by the TestSuiteExecutor (through a call to make_test_case())
# logger is an instance of TestQueueLogger. When the TestCase is created by a hook
# implementation it is an instance of BaseLogger.
self.logger = logger
# Used to store the logger when overridden by a test logger in Report.startTest().
self._original_logger = None
self.test_kind = test_kind
self.test_name = test_name
self.fixture = None
self.return_code = None
self.is_configured = False
def long_name(self):
"""
Returns the path to the test, relative to the current working directory.
"""
return os.path.relpath(self.test_name)
def basename(self):
"""
Returns the basename of the test.
"""
return os.path.basename(self.test_name)
def short_name(self):
"""
Returns the basename of the test without the file extension.
"""
return os.path.splitext(self.basename())[0]
def id(self):
return self.test_name
def shortDescription(self):
return "%s %s" % (self.test_kind, self.test_name)
def override_logger(self, new_logger):
"""
Overrides this instance's logger with a new logger.
This method is used by the repport to set the test logger.
"""
assert not self._original_logger, "Logger already overridden"
self._original_logger = self.logger
self.logger = new_logger
def reset_logger(self):
"""Resets this instance's logger to its original value."""
assert self._original_logger, "Logger was not overridden"
self.logger = self._original_logger
self._original_logger = None
def configure(self, fixture, *args, **kwargs): # pylint: disable=unused-argument
"""
Stores 'fixture' as an attribute for later use during execution.
"""
if self.is_configured:
raise RuntimeError("configure can only be called once")
self.is_configured = True
self.fixture = fixture
def run_test(self):
"""
Runs the specified test.
"""
raise NotImplementedError("run_test must be implemented by TestCase subclasses")
def as_command(self):
"""
Returns the command invocation used to run the test.
"""
raise NotImplementedError("as_command must be implemented by TestCase subclasses")
class ProcessTestCase(TestCase): # pylint: disable=abstract-method
"""Base class for TestCases that executes an external process."""
def run_test(self):
try:
shell = self._make_process()
self._execute(shell)
except self.failureException:
raise
except:
self.logger.exception("Encountered an error running %s %s", self.test_kind,
self.basename())
raise
def as_command(self):
"""
Returns the command invocation used to run the test.
"""
return self._make_process().as_command()
def _execute(self, process):
"""
Runs the specified process.
"""
self.logger.info("Starting %s...\n%s", self.shortDescription(), process.as_command())
process.start()
self.logger.info("%s started with pid %s.", self.shortDescription(), process.pid)
self.return_code = process.wait()
if self.return_code != 0:
raise self.failureException("%s failed" % (self.shortDescription()))
self.logger.info("%s finished.", self.shortDescription())
def _make_process(self):
"""
Returns a new Process instance that could be used to run the
test or log the command.
"""
raise NotImplementedError("_make_process must be implemented by TestCase subclasses")
|
