1# Copyright (c) 2009-2016 testtools developers. See LICENSE for details.
2
3"""Doubles of test result objects, useful for testing unittest code."""
4
5from collections import namedtuple
6
7from testtools.tags import TagContext
8
9__all__ = [
10    'Python26TestResult',
11    'Python27TestResult',
12    'ExtendedTestResult',
13    'TwistedTestResult',
14    'StreamResult',
15    ]
16
17
18class LoggingBase:
19    """Basic support for logging of results."""
20
21    def __init__(self, event_log=None):
22        if event_log is None:
23            event_log = []
24        self._events = event_log
25
26
27class Python26TestResult(LoggingBase):
28    """A precisely python 2.6 like test result, that logs."""
29
30    def __init__(self, event_log=None):
31        super().__init__(event_log=event_log)
32        self.shouldStop = False
33        self._was_successful = True
34        self.testsRun = 0
35
36    def addError(self, test, err):
37        self._was_successful = False
38        self._events.append(('addError', test, err))
39
40    def addFailure(self, test, err):
41        self._was_successful = False
42        self._events.append(('addFailure', test, err))
43
44    def addSuccess(self, test):
45        self._events.append(('addSuccess', test))
46
47    def startTest(self, test):
48        self._events.append(('startTest', test))
49        self.testsRun += 1
50
51    def stop(self):
52        self.shouldStop = True
53
54    def stopTest(self, test):
55        self._events.append(('stopTest', test))
56
57    def wasSuccessful(self):
58        return self._was_successful
59
60
61class Python27TestResult(Python26TestResult):
62    """A precisely python 2.7 like test result, that logs."""
63
64    def __init__(self, event_log=None):
65        super().__init__(event_log)
66        self.failfast = False
67
68    def addError(self, test, err):
69        super().addError(test, err)
70        if self.failfast:
71            self.stop()
72
73    def addFailure(self, test, err):
74        super().addFailure(test, err)
75        if self.failfast:
76            self.stop()
77
78    def addExpectedFailure(self, test, err):
79        self._events.append(('addExpectedFailure', test, err))
80
81    def addSkip(self, test, reason):
82        self._events.append(('addSkip', test, reason))
83
84    def addUnexpectedSuccess(self, test):
85        self._events.append(('addUnexpectedSuccess', test))
86        if self.failfast:
87            self.stop()
88
89    def startTestRun(self):
90        self._events.append(('startTestRun',))
91
92    def stopTestRun(self):
93        self._events.append(('stopTestRun',))
94
95
96class ExtendedTestResult(Python27TestResult):
97    """A test result like the proposed extended unittest result API."""
98
99    def __init__(self, event_log=None):
100        super().__init__(event_log)
101        self._tags = TagContext()
102
103    def addError(self, test, err=None, details=None):
104        self._was_successful = False
105        self._events.append(('addError', test, err or details))
106
107    def addFailure(self, test, err=None, details=None):
108        self._was_successful = False
109        self._events.append(('addFailure', test, err or details))
110
111    def addExpectedFailure(self, test, err=None, details=None):
112        self._events.append(('addExpectedFailure', test, err or details))
113
114    def addSkip(self, test, reason=None, details=None):
115        self._events.append(('addSkip', test, reason or details))
116
117    def addSuccess(self, test, details=None):
118        if details:
119            self._events.append(('addSuccess', test, details))
120        else:
121            self._events.append(('addSuccess', test))
122
123    def addUnexpectedSuccess(self, test, details=None):
124        self._was_successful = False
125        if details is not None:
126            self._events.append(('addUnexpectedSuccess', test, details))
127        else:
128            self._events.append(('addUnexpectedSuccess', test))
129
130    def progress(self, offset, whence):
131        self._events.append(('progress', offset, whence))
132
133    def startTestRun(self):
134        super().startTestRun()
135        self._was_successful = True
136        self._tags = TagContext()
137
138    def startTest(self, test):
139        super().startTest(test)
140        self._tags = TagContext(self._tags)
141
142    def stopTest(self, test):
143        self._tags = self._tags.parent
144        super().stopTest(test)
145
146    @property
147    def current_tags(self):
148        return self._tags.get_current_tags()
149
150    def tags(self, new_tags, gone_tags):
151        self._tags.change_tags(new_tags, gone_tags)
152        self._events.append(('tags', new_tags, gone_tags))
153
154    def time(self, time):
155        self._events.append(('time', time))
156
157    def wasSuccessful(self):
158        return self._was_successful
159
160
161class TwistedTestResult(LoggingBase):
162    """
163    Emulate the relevant bits of :py:class:`twisted.trial.itrial.IReporter`.
164
165    Used to ensure that we can use ``trial`` as a test runner.
166    """
167
168    def __init__(self, event_log=None):
169        super().__init__(event_log=event_log)
170        self._was_successful = True
171        self.testsRun = 0
172
173    def startTest(self, test):
174        self.testsRun += 1
175        self._events.append(('startTest', test))
176
177    def stopTest(self, test):
178        self._events.append(('stopTest', test))
179
180    def addSuccess(self, test):
181        self._events.append(('addSuccess', test))
182
183    def addError(self, test, error):
184        self._was_successful = False
185        self._events.append(('addError', test, error))
186
187    def addFailure(self, test, error):
188        self._was_successful = False
189        self._events.append(('addFailure', test, error))
190
191    def addExpectedFailure(self, test, failure, todo=None):
192        self._events.append(('addExpectedFailure', test, failure))
193
194    def addUnexpectedSuccess(self, test, todo=None):
195        self._events.append(('addUnexpectedSuccess', test))
196
197    def addSkip(self, test, reason):
198        self._events.append(('addSkip', test, reason))
199
200    def wasSuccessful(self):
201        return self._was_successful
202
203    def done(self):
204        pass
205
206
207class StreamResult(LoggingBase):
208    """A StreamResult implementation for testing.
209
210    All events are logged to _events.
211    """
212
213    def startTestRun(self):
214        self._events.append(('startTestRun',))
215
216    def stopTestRun(self):
217        self._events.append(('stopTestRun',))
218
219    def status(self, test_id=None, test_status=None, test_tags=None,
220               runnable=True, file_name=None, file_bytes=None, eof=False,
221               mime_type=None, route_code=None, timestamp=None):
222        self._events.append(
223            _StatusEvent(
224                'status', test_id, test_status, test_tags, runnable,
225                file_name, file_bytes, eof, mime_type, route_code,
226                timestamp))
227
228
229# Convenience for easier access to status fields
230_StatusEvent = namedtuple(
231    "_Event", [
232        "name", "test_id", "test_status", "test_tags", "runnable", "file_name",
233        "file_bytes", "eof", "mime_type", "route_code", "timestamp"])
234