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
|
"""Subclass of unittest.TestCase with helpers for spawning a separate process.
This is used 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 = {} # type: ignore
def make_test_case(test_kind, *args, **kwargs):
"""Provide 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) # type: ignore
REGISTERED_NAME = registry.LEAVE_UNREGISTERED
def __init__(self, logger, test_kind, test_name):
"""Initialize 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.start_test().
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):
"""Return the path to the test, relative to the current working directory."""
return os.path.relpath(self.test_name)
def basename(self):
"""Return the basename of the test."""
return os.path.basename(self.test_name)
def short_name(self):
"""Return the basename of the test without the file extension."""
return os.path.splitext(self.basename())[0]
def id(self):
"""Return the id of the test."""
return self.test_name
def short_description(self):
"""Return the short_description of the test."""
return "%s %s" % (self.test_kind, self.test_name)
def override_logger(self, new_logger):
"""Override 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):
"""Reset 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
"""Store '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):
"""Run the specified test."""
raise NotImplementedError("run_test must be implemented by TestCase subclasses")
def as_command(self):
"""Return 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):
"""Run the test."""
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):
"""Return the command invocation used to run the test."""
return self._make_process().as_command()
def _execute(self, process):
"""Run the specified process."""
self.logger.info("Starting %s...\n%s", self.short_description(), process.as_command())
process.start()
self.logger.info("%s started with pid %s.", self.short_description(), process.pid)
self.return_code = process.wait()
if self.return_code != 0:
raise self.failureException("%s failed" % (self.short_description()))
self.logger.info("%s finished.", self.short_description())
def _make_process(self):
"""Return 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")
|
