← Back to branch summary

~dingqi-lxb/percona-server/5.5_log_queries_in_memory

« back to all changes in this revision

Viewing changes to python-for-subunit2junitxml/subunit/tests/test_test_protocol.py

  • Committer: Stewart Smith
  • Date: 2011-10-06 07:49:44 UTC
  • mfrom: (175.1.4 5.5)
  • Revision ID: stewart@flamingspork.com-20111006074944-xtwnccssnymjqfj0
Merge subunit support for mtr

Show diffs side-by-side

added added

removed removed

Lines of Context:
python-for-subunit2junitxml/subunit/tests/test_test_protocol.py
 
1
#
 
2
#  subunit: extensions to Python unittest to get test results from subprocesses.
 
3
#  Copyright (C) 2005  Robert Collins <robertc@robertcollins.net>
 
4
#
 
5
#  Licensed under either the Apache License, Version 2.0 or the BSD 3-clause
 
6
#  license at the users choice. A copy of both licenses are available in the
 
7
#  project source as Apache-2.0 and BSD. You may not use this file except in
 
8
#  compliance with one of these two licences.
 
9
#
 
10
#  Unless required by applicable law or agreed to in writing, software
 
11
#  distributed under these licenses is distributed on an "AS IS" BASIS, WITHOUT
 
12
#  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
 
13
#  license you chose for the specific language governing permissions and
 
14
#  limitations under that license.
 
15
#
 
16
 
 
17
import datetime
 
18
import unittest
 
19
import os
 
20
 
 
21
from testtools import skipIf, TestCase
 
22
from testtools.compat import _b, _u, BytesIO, StringIO
 
23
from testtools.content import Content, TracebackContent
 
24
from testtools.content_type import ContentType
 
25
from testtools.tests.helpers import (
 
26
    Python26TestResult,
 
27
    Python27TestResult,
 
28
    ExtendedTestResult,
 
29
    )
 
30
 
 
31
import subunit
 
32
from subunit import _remote_exception_str, _remote_exception_str_chunked
 
33
import subunit.iso8601 as iso8601
 
34
 
 
35
 
 
36
class TestTestImports(unittest.TestCase):
 
37
 
 
38
    def test_imports(self):
 
39
        from subunit import DiscardStream
 
40
        from subunit import TestProtocolServer
 
41
        from subunit import RemotedTestCase
 
42
        from subunit import RemoteError
 
43
        from subunit import ExecTestCase
 
44
        from subunit import IsolatedTestCase
 
45
        from subunit import TestProtocolClient
 
46
        from subunit import ProtocolTestCase
 
47
 
 
48
 
 
49
class TestDiscardStream(unittest.TestCase):
 
50
 
 
51
    def test_write(self):
 
52
        subunit.DiscardStream().write("content")
 
53
 
 
54
 
 
55
class TestProtocolServerForward(unittest.TestCase):
 
56
 
 
57
    def test_story(self):
 
58
        client = unittest.TestResult()
 
59
        out = BytesIO()
 
60
        protocol = subunit.TestProtocolServer(client, forward_stream=out)
 
61
        pipe = BytesIO(_b("test old mcdonald\n"
 
62
                        "success old mcdonald\n"))
 
63
        protocol.readFrom(pipe)
 
64
        self.assertEqual(client.testsRun, 1)
 
65
        self.assertEqual(pipe.getvalue(), out.getvalue())
 
66
 
 
67
    def test_not_command(self):
 
68
        client = unittest.TestResult()
 
69
        out = BytesIO()
 
70
        protocol = subunit.TestProtocolServer(client,
 
71
            stream=subunit.DiscardStream(), forward_stream=out)
 
72
        pipe = BytesIO(_b("success old mcdonald\n"))
 
73
        protocol.readFrom(pipe)
 
74
        self.assertEqual(client.testsRun, 0)
 
75
        self.assertEqual(_b(""), out.getvalue())
 
76
 
 
77
 
 
78
class TestTestProtocolServerPipe(unittest.TestCase):
 
79
 
 
80
    def test_story(self):
 
81
        client = unittest.TestResult()
 
82
        protocol = subunit.TestProtocolServer(client)
 
83
        pipe = BytesIO(_b("test old mcdonald\n"
 
84
                        "success old mcdonald\n"
 
85
                        "test bing crosby\n"
 
86
                        "failure bing crosby [\n"
 
87
                        "foo.c:53:ERROR invalid state\n"
 
88
                        "]\n"
 
89
                        "test an error\n"
 
90
                        "error an error\n"))
 
91
        protocol.readFrom(pipe)
 
92
        bing = subunit.RemotedTestCase("bing crosby")
 
93
        an_error = subunit.RemotedTestCase("an error")
 
94
        self.assertEqual(client.errors,
 
95
                         [(an_error, _remote_exception_str + '\n')])
 
96
        self.assertEqual(
 
97
            client.failures,
 
98
            [(bing, _remote_exception_str + ": Text attachment: traceback\n"
 
99
                "------------\nfoo.c:53:ERROR invalid state\n"
 
100
                "------------\n\n")])
 
101
        self.assertEqual(client.testsRun, 3)
 
102
 
 
103
    def test_non_test_characters_forwarded_immediately(self):
 
104
        pass
 
105
 
 
106
 
 
107
class TestTestProtocolServerStartTest(unittest.TestCase):
 
108
 
 
109
    def setUp(self):
 
110
        self.client = Python26TestResult()
 
111
        self.stream = BytesIO()
 
112
        self.protocol = subunit.TestProtocolServer(self.client, self.stream)
 
113
 
 
114
    def test_start_test(self):
 
115
        self.protocol.lineReceived(_b("test old mcdonald\n"))
 
116
        self.assertEqual(self.client._events,
 
117
            [('startTest', subunit.RemotedTestCase("old mcdonald"))])
 
118
 
 
119
    def test_start_testing(self):
 
120
        self.protocol.lineReceived(_b("testing old mcdonald\n"))
 
121
        self.assertEqual(self.client._events,
 
122
            [('startTest', subunit.RemotedTestCase("old mcdonald"))])
 
123
 
 
124
    def test_start_test_colon(self):
 
125
        self.protocol.lineReceived(_b("test: old mcdonald\n"))
 
126
        self.assertEqual(self.client._events,
 
127
            [('startTest', subunit.RemotedTestCase("old mcdonald"))])
 
128
 
 
129
    def test_indented_test_colon_ignored(self):
 
130
        ignored_line = _b(" test: old mcdonald\n")
 
131
        self.protocol.lineReceived(ignored_line)
 
132
        self.assertEqual([], self.client._events)
 
133
        self.assertEqual(self.stream.getvalue(), ignored_line)
 
134
 
 
135
    def test_start_testing_colon(self):
 
136
        self.protocol.lineReceived(_b("testing: old mcdonald\n"))
 
137
        self.assertEqual(self.client._events,
 
138
            [('startTest', subunit.RemotedTestCase("old mcdonald"))])
 
139
 
 
140
 
 
141
class TestTestProtocolServerPassThrough(unittest.TestCase):
 
142
 
 
143
    def setUp(self):
 
144
        self.stdout = BytesIO()
 
145
        self.test = subunit.RemotedTestCase("old mcdonald")
 
146
        self.client = ExtendedTestResult()
 
147
        self.protocol = subunit.TestProtocolServer(self.client, self.stdout)
 
148
 
 
149
    def keywords_before_test(self):
 
150
        self.protocol.lineReceived(_b("failure a\n"))
 
151
        self.protocol.lineReceived(_b("failure: a\n"))
 
152
        self.protocol.lineReceived(_b("error a\n"))
 
153
        self.protocol.lineReceived(_b("error: a\n"))
 
154
        self.protocol.lineReceived(_b("success a\n"))
 
155
        self.protocol.lineReceived(_b("success: a\n"))
 
156
        self.protocol.lineReceived(_b("successful a\n"))
 
157
        self.protocol.lineReceived(_b("successful: a\n"))
 
158
        self.protocol.lineReceived(_b("]\n"))
 
159
        self.assertEqual(self.stdout.getvalue(), _b("failure a\n"
 
160
                                                 "failure: a\n"
 
161
                                                 "error a\n"
 
162
                                                 "error: a\n"
 
163
                                                 "success a\n"
 
164
                                                 "success: a\n"
 
165
                                                 "successful a\n"
 
166
                                                 "successful: a\n"
 
167
                                                 "]\n"))
 
168
 
 
169
    def test_keywords_before_test(self):
 
170
        self.keywords_before_test()
 
171
        self.assertEqual(self.client._events, [])
 
172
 
 
173
    def test_keywords_after_error(self):
 
174
        self.protocol.lineReceived(_b("test old mcdonald\n"))
 
175
        self.protocol.lineReceived(_b("error old mcdonald\n"))
 
176
        self.keywords_before_test()
 
177
        self.assertEqual([
 
178
            ('startTest', self.test),
 
179
            ('addError', self.test, {}),
 
180
            ('stopTest', self.test),
 
181
            ], self.client._events)
 
182
 
 
183
    def test_keywords_after_failure(self):
 
184
        self.protocol.lineReceived(_b("test old mcdonald\n"))
 
185
        self.protocol.lineReceived(_b("failure old mcdonald\n"))
 
186
        self.keywords_before_test()
 
187
        self.assertEqual(self.client._events, [
 
188
            ('startTest', self.test),
 
189
            ('addFailure', self.test, {}),
 
190
            ('stopTest', self.test),
 
191
            ])
 
192
 
 
193
    def test_keywords_after_success(self):
 
194
        self.protocol.lineReceived(_b("test old mcdonald\n"))
 
195
        self.protocol.lineReceived(_b("success old mcdonald\n"))
 
196
        self.keywords_before_test()
 
197
        self.assertEqual([
 
198
            ('startTest', self.test),
 
199
            ('addSuccess', self.test),
 
200
            ('stopTest', self.test),
 
201
            ], self.client._events)
 
202
 
 
203
    def test_keywords_after_test(self):
 
204
        self.protocol.lineReceived(_b("test old mcdonald\n"))
 
205
        self.protocol.lineReceived(_b("test old mcdonald\n"))
 
206
        self.protocol.lineReceived(_b("failure a\n"))
 
207
        self.protocol.lineReceived(_b("failure: a\n"))
 
208
        self.protocol.lineReceived(_b("error a\n"))
 
209
        self.protocol.lineReceived(_b("error: a\n"))
 
210
        self.protocol.lineReceived(_b("success a\n"))
 
211
        self.protocol.lineReceived(_b("success: a\n"))
 
212
        self.protocol.lineReceived(_b("successful a\n"))
 
213
        self.protocol.lineReceived(_b("successful: a\n"))
 
214
        self.protocol.lineReceived(_b("]\n"))
 
215
        self.protocol.lineReceived(_b("failure old mcdonald\n"))
 
216
        self.assertEqual(self.stdout.getvalue(), _b("test old mcdonald\n"
 
217
                                                 "failure a\n"
 
218
                                                 "failure: a\n"
 
219
                                                 "error a\n"
 
220
                                                 "error: a\n"
 
221
                                                 "success a\n"
 
222
                                                 "success: a\n"
 
223
                                                 "successful a\n"
 
224
                                                 "successful: a\n"
 
225
                                                 "]\n"))
 
226
        self.assertEqual(self.client._events, [
 
227
            ('startTest', self.test),
 
228
            ('addFailure', self.test, {}),
 
229
            ('stopTest', self.test),
 
230
            ])
 
231
 
 
232
    def test_keywords_during_failure(self):
 
233
        # A smoke test to make sure that the details parsers have control
 
234
        # appropriately.
 
235
        self.protocol.lineReceived(_b("test old mcdonald\n"))
 
236
        self.protocol.lineReceived(_b("failure: old mcdonald [\n"))
 
237
        self.protocol.lineReceived(_b("test old mcdonald\n"))
 
238
        self.protocol.lineReceived(_b("failure a\n"))
 
239
        self.protocol.lineReceived(_b("failure: a\n"))
 
240
        self.protocol.lineReceived(_b("error a\n"))
 
241
        self.protocol.lineReceived(_b("error: a\n"))
 
242
        self.protocol.lineReceived(_b("success a\n"))
 
243
        self.protocol.lineReceived(_b("success: a\n"))
 
244
        self.protocol.lineReceived(_b("successful a\n"))
 
245
        self.protocol.lineReceived(_b("successful: a\n"))
 
246
        self.protocol.lineReceived(_b(" ]\n"))
 
247
        self.protocol.lineReceived(_b("]\n"))
 
248
        self.assertEqual(self.stdout.getvalue(), _b(""))
 
249
        details = {}
 
250
        details['traceback'] = Content(ContentType("text", "x-traceback",
 
251
            {'charset': 'utf8'}),
 
252
            lambda:[_b(
 
253
            "test old mcdonald\n"
 
254
            "failure a\n"
 
255
            "failure: a\n"
 
256
            "error a\n"
 
257
            "error: a\n"
 
258
            "success a\n"
 
259
            "success: a\n"
 
260
            "successful a\n"
 
261
            "successful: a\n"
 
262
            "]\n")])
 
263
        self.assertEqual(self.client._events, [
 
264
            ('startTest', self.test),
 
265
            ('addFailure', self.test, details),
 
266
            ('stopTest', self.test),
 
267
            ])
 
268
 
 
269
    def test_stdout_passthrough(self):
 
270
        """Lines received which cannot be interpreted as any protocol action
 
271
        should be passed through to sys.stdout.
 
272
        """
 
273
        bytes = _b("randombytes\n")
 
274
        self.protocol.lineReceived(bytes)
 
275
        self.assertEqual(self.stdout.getvalue(), bytes)
 
276
 
 
277
 
 
278
class TestTestProtocolServerLostConnection(unittest.TestCase):
 
279
 
 
280
    def setUp(self):
 
281
        self.client = Python26TestResult()
 
282
        self.protocol = subunit.TestProtocolServer(self.client)
 
283
        self.test = subunit.RemotedTestCase("old mcdonald")
 
284
 
 
285
    def test_lost_connection_no_input(self):
 
286
        self.protocol.lostConnection()
 
287
        self.assertEqual([], self.client._events)
 
288
 
 
289
    def test_lost_connection_after_start(self):
 
290
        self.protocol.lineReceived(_b("test old mcdonald\n"))
 
291
        self.protocol.lostConnection()
 
292
        failure = subunit.RemoteError(
 
293
            _u("lost connection during test 'old mcdonald'"))
 
294
        self.assertEqual([
 
295
            ('startTest', self.test),
 
296
            ('addError', self.test, failure),
 
297
            ('stopTest', self.test),
 
298
            ], self.client._events)
 
299
 
 
300
    def test_lost_connected_after_error(self):
 
301
        self.protocol.lineReceived(_b("test old mcdonald\n"))
 
302
        self.protocol.lineReceived(_b("error old mcdonald\n"))
 
303
        self.protocol.lostConnection()
 
304
        self.assertEqual([
 
305
            ('startTest', self.test),
 
306
            ('addError', self.test, subunit.RemoteError(_u(""))),
 
307
            ('stopTest', self.test),
 
308
            ], self.client._events)
 
309
 
 
310
    def do_connection_lost(self, outcome, opening):
 
311
        self.protocol.lineReceived(_b("test old mcdonald\n"))
 
312
        self.protocol.lineReceived(_b("%s old mcdonald %s" % (outcome, opening)))
 
313
        self.protocol.lostConnection()
 
314
        failure = subunit.RemoteError(
 
315
            _u("lost connection during %s report of test 'old mcdonald'") %
 
316
            outcome)
 
317
        self.assertEqual([
 
318
            ('startTest', self.test),
 
319
            ('addError', self.test, failure),
 
320
            ('stopTest', self.test),
 
321
            ], self.client._events)
 
322
 
 
323
    def test_lost_connection_during_error(self):
 
324
        self.do_connection_lost("error", "[\n")
 
325
 
 
326
    def test_lost_connection_during_error_details(self):
 
327
        self.do_connection_lost("error", "[ multipart\n")
 
328
 
 
329
    def test_lost_connected_after_failure(self):
 
330
        self.protocol.lineReceived(_b("test old mcdonald\n"))
 
331
        self.protocol.lineReceived(_b("failure old mcdonald\n"))
 
332
        self.protocol.lostConnection()
 
333
        self.assertEqual([
 
334
            ('startTest', self.test),
 
335
            ('addFailure', self.test, subunit.RemoteError(_u(""))),
 
336
            ('stopTest', self.test),
 
337
            ], self.client._events)
 
338
 
 
339
    def test_lost_connection_during_failure(self):
 
340
        self.do_connection_lost("failure", "[\n")
 
341
 
 
342
    def test_lost_connection_during_failure_details(self):
 
343
        self.do_connection_lost("failure", "[ multipart\n")
 
344
 
 
345
    def test_lost_connection_after_success(self):
 
346
        self.protocol.lineReceived(_b("test old mcdonald\n"))
 
347
        self.protocol.lineReceived(_b("success old mcdonald\n"))
 
348
        self.protocol.lostConnection()
 
349
        self.assertEqual([
 
350
            ('startTest', self.test),
 
351
            ('addSuccess', self.test),
 
352
            ('stopTest', self.test),
 
353
            ], self.client._events)
 
354
 
 
355
    def test_lost_connection_during_success(self):
 
356
        self.do_connection_lost("success", "[\n")
 
357
 
 
358
    def test_lost_connection_during_success_details(self):
 
359
        self.do_connection_lost("success", "[ multipart\n")
 
360
 
 
361
    def test_lost_connection_during_skip(self):
 
362
        self.do_connection_lost("skip", "[\n")
 
363
 
 
364
    def test_lost_connection_during_skip_details(self):
 
365
        self.do_connection_lost("skip", "[ multipart\n")
 
366
 
 
367
    def test_lost_connection_during_xfail(self):
 
368
        self.do_connection_lost("xfail", "[\n")
 
369
 
 
370
    def test_lost_connection_during_xfail_details(self):
 
371
        self.do_connection_lost("xfail", "[ multipart\n")
 
372
 
 
373
    def test_lost_connection_during_uxsuccess(self):
 
374
        self.do_connection_lost("uxsuccess", "[\n")
 
375
 
 
376
    def test_lost_connection_during_uxsuccess_details(self):
 
377
        self.do_connection_lost("uxsuccess", "[ multipart\n")
 
378
 
 
379
 
 
380
class TestInTestMultipart(unittest.TestCase):
 
381
 
 
382
    def setUp(self):
 
383
        self.client = ExtendedTestResult()
 
384
        self.protocol = subunit.TestProtocolServer(self.client)
 
385
        self.protocol.lineReceived(_b("test mcdonalds farm\n"))
 
386
        self.test = subunit.RemotedTestCase(_u("mcdonalds farm"))
 
387
 
 
388
    def test__outcome_sets_details_parser(self):
 
389
        self.protocol._reading_success_details.details_parser = None
 
390
        self.protocol._state._outcome(0, _b("mcdonalds farm [ multipart\n"),
 
391
            None, self.protocol._reading_success_details)
 
392
        parser = self.protocol._reading_success_details.details_parser
 
393
        self.assertNotEqual(None, parser)
 
394
        self.assertTrue(isinstance(parser,
 
395
            subunit.details.MultipartDetailsParser))
 
396
 
 
397
 
 
398
class TestTestProtocolServerAddError(unittest.TestCase):
 
399
 
 
400
    def setUp(self):
 
401
        self.client = ExtendedTestResult()
 
402
        self.protocol = subunit.TestProtocolServer(self.client)
 
403
        self.protocol.lineReceived(_b("test mcdonalds farm\n"))
 
404
        self.test = subunit.RemotedTestCase("mcdonalds farm")
 
405
 
 
406
    def simple_error_keyword(self, keyword):
 
407
        self.protocol.lineReceived(_b("%s mcdonalds farm\n" % keyword))
 
408
        details = {}
 
409
        self.assertEqual([
 
410
            ('startTest', self.test),
 
411
            ('addError', self.test, details),
 
412
            ('stopTest', self.test),
 
413
            ], self.client._events)
 
414
 
 
415
    def test_simple_error(self):
 
416
        self.simple_error_keyword("error")
 
417
 
 
418
    def test_simple_error_colon(self):
 
419
        self.simple_error_keyword("error:")
 
420
 
 
421
    def test_error_empty_message(self):
 
422
        self.protocol.lineReceived(_b("error mcdonalds farm [\n"))
 
423
        self.protocol.lineReceived(_b("]\n"))
 
424
        details = {}
 
425
        details['traceback'] = Content(ContentType("text", "x-traceback",
 
426
            {'charset': 'utf8'}), lambda:[_b("")])
 
427
        self.assertEqual([
 
428
            ('startTest', self.test),
 
429
            ('addError', self.test, details),
 
430
            ('stopTest', self.test),
 
431
            ], self.client._events)
 
432
 
 
433
    def error_quoted_bracket(self, keyword):
 
434
        self.protocol.lineReceived(_b("%s mcdonalds farm [\n" % keyword))
 
435
        self.protocol.lineReceived(_b(" ]\n"))
 
436
        self.protocol.lineReceived(_b("]\n"))
 
437
        details = {}
 
438
        details['traceback'] = Content(ContentType("text", "x-traceback",
 
439
            {'charset': 'utf8'}), lambda:[_b("]\n")])
 
440
        self.assertEqual([
 
441
            ('startTest', self.test),
 
442
            ('addError', self.test, details),
 
443
            ('stopTest', self.test),
 
444
            ], self.client._events)
 
445
 
 
446
    def test_error_quoted_bracket(self):
 
447
        self.error_quoted_bracket("error")
 
448
 
 
449
    def test_error_colon_quoted_bracket(self):
 
450
        self.error_quoted_bracket("error:")
 
451
 
 
452
 
 
453
class TestTestProtocolServerAddFailure(unittest.TestCase):
 
454
 
 
455
    def setUp(self):
 
456
        self.client = ExtendedTestResult()
 
457
        self.protocol = subunit.TestProtocolServer(self.client)
 
458
        self.protocol.lineReceived(_b("test mcdonalds farm\n"))
 
459
        self.test = subunit.RemotedTestCase("mcdonalds farm")
 
460
 
 
461
    def assertFailure(self, details):
 
462
        self.assertEqual([
 
463
            ('startTest', self.test),
 
464
            ('addFailure', self.test, details),
 
465
            ('stopTest', self.test),
 
466
            ], self.client._events)
 
467
 
 
468
    def simple_failure_keyword(self, keyword):
 
469
        self.protocol.lineReceived(_b("%s mcdonalds farm\n" % keyword))
 
470
        details = {}
 
471
        self.assertFailure(details)
 
472
 
 
473
    def test_simple_failure(self):
 
474
        self.simple_failure_keyword("failure")
 
475
 
 
476
    def test_simple_failure_colon(self):
 
477
        self.simple_failure_keyword("failure:")
 
478
 
 
479
    def test_failure_empty_message(self):
 
480
        self.protocol.lineReceived(_b("failure mcdonalds farm [\n"))
 
481
        self.protocol.lineReceived(_b("]\n"))
 
482
        details = {}
 
483
        details['traceback'] = Content(ContentType("text", "x-traceback",
 
484
            {'charset': 'utf8'}), lambda:[_b("")])
 
485
        self.assertFailure(details)
 
486
 
 
487
    def failure_quoted_bracket(self, keyword):
 
488
        self.protocol.lineReceived(_b("%s mcdonalds farm [\n" % keyword))
 
489
        self.protocol.lineReceived(_b(" ]\n"))
 
490
        self.protocol.lineReceived(_b("]\n"))
 
491
        details = {}
 
492
        details['traceback'] = Content(ContentType("text", "x-traceback",
 
493
            {'charset': 'utf8'}), lambda:[_b("]\n")])
 
494
        self.assertFailure(details)
 
495
 
 
496
    def test_failure_quoted_bracket(self):
 
497
        self.failure_quoted_bracket("failure")
 
498
 
 
499
    def test_failure_colon_quoted_bracket(self):
 
500
        self.failure_quoted_bracket("failure:")
 
501
 
 
502
 
 
503
class TestTestProtocolServerAddxFail(unittest.TestCase):
 
504
    """Tests for the xfail keyword.
 
505
 
 
506
    In Python this can thunk through to Success due to stdlib limitations (see
 
507
    README).
 
508
    """
 
509
 
 
510
    def capture_expected_failure(self, test, err):
 
511
        self._events.append((test, err))
 
512
 
 
513
    def setup_python26(self):
 
514
        """Setup a test object ready to be xfailed and thunk to success."""
 
515
        self.client = Python26TestResult()
 
516
        self.setup_protocol()
 
517
 
 
518
    def setup_python27(self):
 
519
        """Setup a test object ready to be xfailed."""
 
520
        self.client = Python27TestResult()
 
521
        self.setup_protocol()
 
522
 
 
523
    def setup_python_ex(self):
 
524
        """Setup a test object ready to be xfailed with details."""
 
525
        self.client = ExtendedTestResult()
 
526
        self.setup_protocol()
 
527
 
 
528
    def setup_protocol(self):
 
529
        """Setup the protocol based on self.client."""
 
530
        self.protocol = subunit.TestProtocolServer(self.client)
 
531
        self.protocol.lineReceived(_b("test mcdonalds farm\n"))
 
532
        self.test = self.client._events[-1][-1]
 
533
 
 
534
    def simple_xfail_keyword(self, keyword, as_success):
 
535
        self.protocol.lineReceived(_b("%s mcdonalds farm\n" % keyword))
 
536
        self.check_success_or_xfail(as_success)
 
537
 
 
538
    def check_success_or_xfail(self, as_success, error_message=None):
 
539
        if as_success:
 
540
            self.assertEqual([
 
541
                ('startTest', self.test),
 
542
                ('addSuccess', self.test),
 
543
                ('stopTest', self.test),
 
544
                ], self.client._events)
 
545
        else:
 
546
            details = {}
 
547
            if error_message is not None:
 
548
                details['traceback'] = Content(
 
549
                    ContentType("text", "x-traceback", {'charset': 'utf8'}),
 
550
                    lambda:[_b(error_message)])
 
551
            if isinstance(self.client, ExtendedTestResult):
 
552
                value = details
 
553
            else:
 
554
                if error_message is not None:
 
555
                    value = subunit.RemoteError(_u("Text attachment: traceback\n"
 
556
                        "------------\n") + _u(error_message) +
 
557
                        _u("------------\n"))
 
558
                else:
 
559
                    value = subunit.RemoteError()
 
560
            self.assertEqual([
 
561
                ('startTest', self.test),
 
562
                ('addExpectedFailure', self.test, value),
 
563
                ('stopTest', self.test),
 
564
                ], self.client._events)
 
565
 
 
566
    def test_simple_xfail(self):
 
567
        self.setup_python26()
 
568
        self.simple_xfail_keyword("xfail", True)
 
569
        self.setup_python27()
 
570
        self.simple_xfail_keyword("xfail",  False)
 
571
        self.setup_python_ex()
 
572
        self.simple_xfail_keyword("xfail",  False)
 
573
 
 
574
    def test_simple_xfail_colon(self):
 
575
        self.setup_python26()
 
576
        self.simple_xfail_keyword("xfail:", True)
 
577
        self.setup_python27()
 
578
        self.simple_xfail_keyword("xfail:", False)
 
579
        self.setup_python_ex()
 
580
        self.simple_xfail_keyword("xfail:", False)
 
581
 
 
582
    def test_xfail_empty_message(self):
 
583
        self.setup_python26()
 
584
        self.empty_message(True)
 
585
        self.setup_python27()
 
586
        self.empty_message(False)
 
587
        self.setup_python_ex()
 
588
        self.empty_message(False, error_message="")
 
589
 
 
590
    def empty_message(self, as_success, error_message="\n"):
 
591
        self.protocol.lineReceived(_b("xfail mcdonalds farm [\n"))
 
592
        self.protocol.lineReceived(_b("]\n"))
 
593
        self.check_success_or_xfail(as_success, error_message)
 
594
 
 
595
    def xfail_quoted_bracket(self, keyword, as_success):
 
596
        # This tests it is accepted, but cannot test it is used today, because
 
597
        # of not having a way to expose it in Python so far.
 
598
        self.protocol.lineReceived(_b("%s mcdonalds farm [\n" % keyword))
 
599
        self.protocol.lineReceived(_b(" ]\n"))
 
600
        self.protocol.lineReceived(_b("]\n"))
 
601
        self.check_success_or_xfail(as_success, "]\n")
 
602
 
 
603
    def test_xfail_quoted_bracket(self):
 
604
        self.setup_python26()
 
605
        self.xfail_quoted_bracket("xfail", True)
 
606
        self.setup_python27()
 
607
        self.xfail_quoted_bracket("xfail", False)
 
608
        self.setup_python_ex()
 
609
        self.xfail_quoted_bracket("xfail", False)
 
610
 
 
611
    def test_xfail_colon_quoted_bracket(self):
 
612
        self.setup_python26()
 
613
        self.xfail_quoted_bracket("xfail:", True)
 
614
        self.setup_python27()
 
615
        self.xfail_quoted_bracket("xfail:", False)
 
616
        self.setup_python_ex()
 
617
        self.xfail_quoted_bracket("xfail:", False)
 
618
 
 
619
 
 
620
class TestTestProtocolServerAddunexpectedSuccess(TestCase):
 
621
    """Tests for the uxsuccess keyword."""
 
622
 
 
623
    def capture_expected_failure(self, test, err):
 
624
        self._events.append((test, err))
 
625
 
 
626
    def setup_python26(self):
 
627
        """Setup a test object ready to be xfailed and thunk to success."""
 
628
        self.client = Python26TestResult()
 
629
        self.setup_protocol()
 
630
 
 
631
    def setup_python27(self):
 
632
        """Setup a test object ready to be xfailed."""
 
633
        self.client = Python27TestResult()
 
634
        self.setup_protocol()
 
635
 
 
636
    def setup_python_ex(self):
 
637
        """Setup a test object ready to be xfailed with details."""
 
638
        self.client = ExtendedTestResult()
 
639
        self.setup_protocol()
 
640
 
 
641
    def setup_protocol(self):
 
642
        """Setup the protocol based on self.client."""
 
643
        self.protocol = subunit.TestProtocolServer(self.client)
 
644
        self.protocol.lineReceived(_b("test mcdonalds farm\n"))
 
645
        self.test = self.client._events[-1][-1]
 
646
 
 
647
    def simple_uxsuccess_keyword(self, keyword, as_fail):
 
648
        self.protocol.lineReceived(_b("%s mcdonalds farm\n" % keyword))
 
649
        self.check_fail_or_uxsuccess(as_fail)
 
650
 
 
651
    def check_fail_or_uxsuccess(self, as_fail, error_message=None):
 
652
        details = {}
 
653
        if error_message is not None:
 
654
            details['traceback'] = Content(
 
655
                ContentType("text", "x-traceback", {'charset': 'utf8'}),
 
656
                lambda:[_b(error_message)])
 
657
        if isinstance(self.client, ExtendedTestResult):
 
658
            value = details
 
659
        else:
 
660
            value = None
 
661
        if as_fail:
 
662
            self.client._events[1] = self.client._events[1][:2]
 
663
            # The value is generated within the extended to original decorator:
 
664
            # todo use the testtools matcher to check on this.
 
665
            self.assertEqual([
 
666
                ('startTest', self.test),
 
667
                ('addFailure', self.test),
 
668
                ('stopTest', self.test),
 
669
                ], self.client._events)
 
670
        elif value:
 
671
            self.assertEqual([
 
672
                ('startTest', self.test),
 
673
                ('addUnexpectedSuccess', self.test, value),
 
674
                ('stopTest', self.test),
 
675
                ], self.client._events)
 
676
        else:
 
677
            self.assertEqual([
 
678
                ('startTest', self.test),
 
679
                ('addUnexpectedSuccess', self.test),
 
680
                ('stopTest', self.test),
 
681
                ], self.client._events)
 
682
 
 
683
    def test_simple_uxsuccess(self):
 
684
        self.setup_python26()
 
685
        self.simple_uxsuccess_keyword("uxsuccess", True)
 
686
        self.setup_python27()
 
687
        self.simple_uxsuccess_keyword("uxsuccess",  False)
 
688
        self.setup_python_ex()
 
689
        self.simple_uxsuccess_keyword("uxsuccess",  False)
 
690
 
 
691
    def test_simple_uxsuccess_colon(self):
 
692
        self.setup_python26()
 
693
        self.simple_uxsuccess_keyword("uxsuccess:", True)
 
694
        self.setup_python27()
 
695
        self.simple_uxsuccess_keyword("uxsuccess:", False)
 
696
        self.setup_python_ex()
 
697
        self.simple_uxsuccess_keyword("uxsuccess:", False)
 
698
 
 
699
    def test_uxsuccess_empty_message(self):
 
700
        self.setup_python26()
 
701
        self.empty_message(True)
 
702
        self.setup_python27()
 
703
        self.empty_message(False)
 
704
        self.setup_python_ex()
 
705
        self.empty_message(False, error_message="")
 
706
 
 
707
    def empty_message(self, as_fail, error_message="\n"):
 
708
        self.protocol.lineReceived(_b("uxsuccess mcdonalds farm [\n"))
 
709
        self.protocol.lineReceived(_b("]\n"))
 
710
        self.check_fail_or_uxsuccess(as_fail, error_message)
 
711
 
 
712
    def uxsuccess_quoted_bracket(self, keyword, as_fail):
 
713
        self.protocol.lineReceived(_b("%s mcdonalds farm [\n" % keyword))
 
714
        self.protocol.lineReceived(_b(" ]\n"))
 
715
        self.protocol.lineReceived(_b("]\n"))
 
716
        self.check_fail_or_uxsuccess(as_fail, "]\n")
 
717
 
 
718
    def test_uxsuccess_quoted_bracket(self):
 
719
        self.setup_python26()
 
720
        self.uxsuccess_quoted_bracket("uxsuccess", True)
 
721
        self.setup_python27()
 
722
        self.uxsuccess_quoted_bracket("uxsuccess", False)
 
723
        self.setup_python_ex()
 
724
        self.uxsuccess_quoted_bracket("uxsuccess", False)
 
725
 
 
726
    def test_uxsuccess_colon_quoted_bracket(self):
 
727
        self.setup_python26()
 
728
        self.uxsuccess_quoted_bracket("uxsuccess:", True)
 
729
        self.setup_python27()
 
730
        self.uxsuccess_quoted_bracket("uxsuccess:", False)
 
731
        self.setup_python_ex()
 
732
        self.uxsuccess_quoted_bracket("uxsuccess:", False)
 
733
 
 
734
 
 
735
class TestTestProtocolServerAddSkip(unittest.TestCase):
 
736
    """Tests for the skip keyword.
 
737
 
 
738
    In Python this meets the testtools extended TestResult contract.
 
739
    (See https://launchpad.net/testtools).
 
740
    """
 
741
 
 
742
    def setUp(self):
 
743
        """Setup a test object ready to be skipped."""
 
744
        self.client = ExtendedTestResult()
 
745
        self.protocol = subunit.TestProtocolServer(self.client)
 
746
        self.protocol.lineReceived(_b("test mcdonalds farm\n"))
 
747
        self.test = self.client._events[-1][-1]
 
748
 
 
749
    def assertSkip(self, reason):
 
750
        details = {}
 
751
        if reason is not None:
 
752
            details['reason'] = Content(
 
753
                ContentType("text", "plain"), lambda:[reason])
 
754
        self.assertEqual([
 
755
            ('startTest', self.test),
 
756
            ('addSkip', self.test, details),
 
757
            ('stopTest', self.test),
 
758
            ], self.client._events)
 
759
 
 
760
    def simple_skip_keyword(self, keyword):
 
761
        self.protocol.lineReceived(_b("%s mcdonalds farm\n" % keyword))
 
762
        self.assertSkip(None)
 
763
 
 
764
    def test_simple_skip(self):
 
765
        self.simple_skip_keyword("skip")
 
766
 
 
767
    def test_simple_skip_colon(self):
 
768
        self.simple_skip_keyword("skip:")
 
769
 
 
770
    def test_skip_empty_message(self):
 
771
        self.protocol.lineReceived(_b("skip mcdonalds farm [\n"))
 
772
        self.protocol.lineReceived(_b("]\n"))
 
773
        self.assertSkip(_b(""))
 
774
 
 
775
    def skip_quoted_bracket(self, keyword):
 
776
        # This tests it is accepted, but cannot test it is used today, because
 
777
        # of not having a way to expose it in Python so far.
 
778
        self.protocol.lineReceived(_b("%s mcdonalds farm [\n" % keyword))
 
779
        self.protocol.lineReceived(_b(" ]\n"))
 
780
        self.protocol.lineReceived(_b("]\n"))
 
781
        self.assertSkip(_b("]\n"))
 
782
 
 
783
    def test_skip_quoted_bracket(self):
 
784
        self.skip_quoted_bracket("skip")
 
785
 
 
786
    def test_skip_colon_quoted_bracket(self):
 
787
        self.skip_quoted_bracket("skip:")
 
788
 
 
789
 
 
790
class TestTestProtocolServerAddSuccess(unittest.TestCase):
 
791
 
 
792
    def setUp(self):
 
793
        self.client = ExtendedTestResult()
 
794
        self.protocol = subunit.TestProtocolServer(self.client)
 
795
        self.protocol.lineReceived(_b("test mcdonalds farm\n"))
 
796
        self.test = subunit.RemotedTestCase("mcdonalds farm")
 
797
 
 
798
    def simple_success_keyword(self, keyword):
 
799
        self.protocol.lineReceived(_b("%s mcdonalds farm\n" % keyword))
 
800
        self.assertEqual([
 
801
            ('startTest', self.test),
 
802
            ('addSuccess', self.test),
 
803
            ('stopTest', self.test),
 
804
            ], self.client._events)
 
805
 
 
806
    def test_simple_success(self):
 
807
        self.simple_success_keyword("successful")
 
808
 
 
809
    def test_simple_success_colon(self):
 
810
        self.simple_success_keyword("successful:")
 
811
 
 
812
    def assertSuccess(self, details):
 
813
        self.assertEqual([
 
814
            ('startTest', self.test),
 
815
            ('addSuccess', self.test, details),
 
816
            ('stopTest', self.test),
 
817
            ], self.client._events)
 
818
 
 
819
    def test_success_empty_message(self):
 
820
        self.protocol.lineReceived(_b("success mcdonalds farm [\n"))
 
821
        self.protocol.lineReceived(_b("]\n"))
 
822
        details = {}
 
823
        details['message'] = Content(ContentType("text", "plain"),
 
824
            lambda:[_b("")])
 
825
        self.assertSuccess(details)
 
826
 
 
827
    def success_quoted_bracket(self, keyword):
 
828
        # This tests it is accepted, but cannot test it is used today, because
 
829
        # of not having a way to expose it in Python so far.
 
830
        self.protocol.lineReceived(_b("%s mcdonalds farm [\n" % keyword))
 
831
        self.protocol.lineReceived(_b(" ]\n"))
 
832
        self.protocol.lineReceived(_b("]\n"))
 
833
        details = {}
 
834
        details['message'] = Content(ContentType("text", "plain"),
 
835
            lambda:[_b("]\n")])
 
836
        self.assertSuccess(details)
 
837
 
 
838
    def test_success_quoted_bracket(self):
 
839
        self.success_quoted_bracket("success")
 
840
 
 
841
    def test_success_colon_quoted_bracket(self):
 
842
        self.success_quoted_bracket("success:")
 
843
 
 
844
 
 
845
class TestTestProtocolServerProgress(unittest.TestCase):
 
846
    """Test receipt of progress: directives."""
 
847
 
 
848
    def test_progress_accepted_stdlib(self):
 
849
        self.result = Python26TestResult()
 
850
        self.stream = BytesIO()
 
851
        self.protocol = subunit.TestProtocolServer(self.result,
 
852
            stream=self.stream)
 
853
        self.protocol.lineReceived(_b("progress: 23"))
 
854
        self.protocol.lineReceived(_b("progress: -2"))
 
855
        self.protocol.lineReceived(_b("progress: +4"))
 
856
        self.assertEqual(_b(""), self.stream.getvalue())
 
857
 
 
858
    def test_progress_accepted_extended(self):
 
859
        # With a progress capable TestResult, progress events are emitted.
 
860
        self.result = ExtendedTestResult()
 
861
        self.stream = BytesIO()
 
862
        self.protocol = subunit.TestProtocolServer(self.result,
 
863
            stream=self.stream)
 
864
        self.protocol.lineReceived(_b("progress: 23"))
 
865
        self.protocol.lineReceived(_b("progress: push"))
 
866
        self.protocol.lineReceived(_b("progress: -2"))
 
867
        self.protocol.lineReceived(_b("progress: pop"))
 
868
        self.protocol.lineReceived(_b("progress: +4"))
 
869
        self.assertEqual(_b(""), self.stream.getvalue())
 
870
        self.assertEqual([
 
871
            ('progress', 23, subunit.PROGRESS_SET),
 
872
            ('progress', None, subunit.PROGRESS_PUSH),
 
873
            ('progress', -2, subunit.PROGRESS_CUR),
 
874
            ('progress', None, subunit.PROGRESS_POP),
 
875
            ('progress', 4, subunit.PROGRESS_CUR),
 
876
            ], self.result._events)
 
877
 
 
878
 
 
879
class TestTestProtocolServerStreamTags(unittest.TestCase):
 
880
    """Test managing tags on the protocol level."""
 
881
 
 
882
    def setUp(self):
 
883
        self.client = ExtendedTestResult()
 
884
        self.protocol = subunit.TestProtocolServer(self.client)
 
885
 
 
886
    def test_initial_tags(self):
 
887
        self.protocol.lineReceived(_b("tags: foo bar:baz  quux\n"))
 
888
        self.assertEqual([
 
889
            ('tags', set(["foo", "bar:baz", "quux"]), set()),
 
890
            ], self.client._events)
 
891
 
 
892
    def test_minus_removes_tags(self):
 
893
        self.protocol.lineReceived(_b("tags: -bar quux\n"))
 
894
        self.assertEqual([
 
895
            ('tags', set(["quux"]), set(["bar"])),
 
896
            ], self.client._events)
 
897
 
 
898
    def test_tags_do_not_get_set_on_test(self):
 
899
        self.protocol.lineReceived(_b("test mcdonalds farm\n"))
 
900
        test = self.client._events[0][-1]
 
901
        self.assertEqual(None, getattr(test, 'tags', None))
 
902
 
 
903
    def test_tags_do_not_get_set_on_global_tags(self):
 
904
        self.protocol.lineReceived(_b("tags: foo bar\n"))
 
905
        self.protocol.lineReceived(_b("test mcdonalds farm\n"))
 
906
        test = self.client._events[-1][-1]
 
907
        self.assertEqual(None, getattr(test, 'tags', None))
 
908
 
 
909
    def test_tags_get_set_on_test_tags(self):
 
910
        self.protocol.lineReceived(_b("test mcdonalds farm\n"))
 
911
        test = self.client._events[-1][-1]
 
912
        self.protocol.lineReceived(_b("tags: foo bar\n"))
 
913
        self.protocol.lineReceived(_b("success mcdonalds farm\n"))
 
914
        self.assertEqual(None, getattr(test, 'tags', None))
 
915
 
 
916
 
 
917
class TestTestProtocolServerStreamTime(unittest.TestCase):
 
918
    """Test managing time information at the protocol level."""
 
919
 
 
920
    def test_time_accepted_stdlib(self):
 
921
        self.result = Python26TestResult()
 
922
        self.stream = BytesIO()
 
923
        self.protocol = subunit.TestProtocolServer(self.result,
 
924
            stream=self.stream)
 
925
        self.protocol.lineReceived(_b("time: 2001-12-12 12:59:59Z\n"))
 
926
        self.assertEqual(_b(""), self.stream.getvalue())
 
927
 
 
928
    def test_time_accepted_extended(self):
 
929
        self.result = ExtendedTestResult()
 
930
        self.stream = BytesIO()
 
931
        self.protocol = subunit.TestProtocolServer(self.result,
 
932
            stream=self.stream)
 
933
        self.protocol.lineReceived(_b("time: 2001-12-12 12:59:59Z\n"))
 
934
        self.assertEqual(_b(""), self.stream.getvalue())
 
935
        self.assertEqual([
 
936
            ('time', datetime.datetime(2001, 12, 12, 12, 59, 59, 0,
 
937
            iso8601.Utc()))
 
938
            ], self.result._events)
 
939
 
 
940
 
 
941
class TestRemotedTestCase(unittest.TestCase):
 
942
 
 
943
    def test_simple(self):
 
944
        test = subunit.RemotedTestCase("A test description")
 
945
        self.assertRaises(NotImplementedError, test.setUp)
 
946
        self.assertRaises(NotImplementedError, test.tearDown)
 
947
        self.assertEqual("A test description",
 
948
                         test.shortDescription())
 
949
        self.assertEqual("A test description",
 
950
                         test.id())
 
951
        self.assertEqual("A test description (subunit.RemotedTestCase)", "%s" % test)
 
952
        self.assertEqual("<subunit.RemotedTestCase description="
 
953
                         "'A test description'>", "%r" % test)
 
954
        result = unittest.TestResult()
 
955
        test.run(result)
 
956
        self.assertEqual([(test, _remote_exception_str + ": "
 
957
                                 "Cannot run RemotedTestCases.\n\n")],
 
958
                         result.errors)
 
959
        self.assertEqual(1, result.testsRun)
 
960
        another_test = subunit.RemotedTestCase("A test description")
 
961
        self.assertEqual(test, another_test)
 
962
        different_test = subunit.RemotedTestCase("ofo")
 
963
        self.assertNotEqual(test, different_test)
 
964
        self.assertNotEqual(another_test, different_test)
 
965
 
 
966
 
 
967
class TestRemoteError(unittest.TestCase):
 
968
 
 
969
    def test_eq(self):
 
970
        error = subunit.RemoteError(_u("Something went wrong"))
 
971
        another_error = subunit.RemoteError(_u("Something went wrong"))
 
972
        different_error = subunit.RemoteError(_u("boo!"))
 
973
        self.assertEqual(error, another_error)
 
974
        self.assertNotEqual(error, different_error)
 
975
        self.assertNotEqual(different_error, another_error)
 
976
 
 
977
    def test_empty_constructor(self):
 
978
        self.assertEqual(subunit.RemoteError(), subunit.RemoteError(_u("")))
 
979
 
 
980
 
 
981
class TestExecTestCase(unittest.TestCase):
 
982
 
 
983
    class SampleExecTestCase(subunit.ExecTestCase):
 
984
 
 
985
        def test_sample_method(self):
 
986
            """sample-script.py"""
 
987
            # the sample script runs three tests, one each
 
988
            # that fails, errors and succeeds
 
989
 
 
990
        def test_sample_method_args(self):
 
991
            """sample-script.py foo"""
 
992
            # sample that will run just one test.
 
993
 
 
994
    def test_construct(self):
 
995
        test = self.SampleExecTestCase("test_sample_method")
 
996
        self.assertEqual(test.script,
 
997
                         subunit.join_dir(__file__, 'sample-script.py'))
 
998
 
 
999
    def test_args(self):
 
1000
        result = unittest.TestResult()
 
1001
        test = self.SampleExecTestCase("test_sample_method_args")
 
1002
        test.run(result)
 
1003
        self.assertEqual(1, result.testsRun)
 
1004
 
 
1005
    def test_run(self):
 
1006
        result = ExtendedTestResult()
 
1007
        test = self.SampleExecTestCase("test_sample_method")
 
1008
        test.run(result)
 
1009
        mcdonald = subunit.RemotedTestCase("old mcdonald")
 
1010
        bing = subunit.RemotedTestCase("bing crosby")
 
1011
        bing_details = {}
 
1012
        bing_details['traceback'] = Content(ContentType("text", "x-traceback",
 
1013
            {'charset': 'utf8'}), lambda:[_b("foo.c:53:ERROR invalid state\n")])
 
1014
        an_error = subunit.RemotedTestCase("an error")
 
1015
        error_details = {}
 
1016
        self.assertEqual([
 
1017
            ('startTest', mcdonald),
 
1018
            ('addSuccess', mcdonald),
 
1019
            ('stopTest', mcdonald),
 
1020
            ('startTest', bing),
 
1021
            ('addFailure', bing, bing_details),
 
1022
            ('stopTest', bing),
 
1023
            ('startTest', an_error),
 
1024
            ('addError', an_error, error_details),
 
1025
            ('stopTest', an_error),
 
1026
            ], result._events)
 
1027
 
 
1028
    def test_debug(self):
 
1029
        test = self.SampleExecTestCase("test_sample_method")
 
1030
        test.debug()
 
1031
 
 
1032
    def test_count_test_cases(self):
 
1033
        """TODO run the child process and count responses to determine the count."""
 
1034
 
 
1035
    def test_join_dir(self):
 
1036
        sibling = subunit.join_dir(__file__, 'foo')
 
1037
        filedir = os.path.abspath(os.path.dirname(__file__))
 
1038
        expected = os.path.join(filedir, 'foo')
 
1039
        self.assertEqual(sibling, expected)
 
1040
 
 
1041
 
 
1042
class DoExecTestCase(subunit.ExecTestCase):
 
1043
 
 
1044
    def test_working_script(self):
 
1045
        """sample-two-script.py"""
 
1046
 
 
1047
 
 
1048
class TestIsolatedTestCase(TestCase):
 
1049
 
 
1050
    class SampleIsolatedTestCase(subunit.IsolatedTestCase):
 
1051
 
 
1052
        SETUP = False
 
1053
        TEARDOWN = False
 
1054
        TEST = False
 
1055
 
 
1056
        def setUp(self):
 
1057
            TestIsolatedTestCase.SampleIsolatedTestCase.SETUP = True
 
1058
 
 
1059
        def tearDown(self):
 
1060
            TestIsolatedTestCase.SampleIsolatedTestCase.TEARDOWN = True
 
1061
 
 
1062
        def test_sets_global_state(self):
 
1063
            TestIsolatedTestCase.SampleIsolatedTestCase.TEST = True
 
1064
 
 
1065
 
 
1066
    def test_construct(self):
 
1067
        self.SampleIsolatedTestCase("test_sets_global_state")
 
1068
 
 
1069
    @skipIf(os.name != "posix", "Need a posix system for forking tests")
 
1070
    def test_run(self):
 
1071
        result = unittest.TestResult()
 
1072
        test = self.SampleIsolatedTestCase("test_sets_global_state")
 
1073
        test.run(result)
 
1074
        self.assertEqual(result.testsRun, 1)
 
1075
        self.assertEqual(self.SampleIsolatedTestCase.SETUP, False)
 
1076
        self.assertEqual(self.SampleIsolatedTestCase.TEARDOWN, False)
 
1077
        self.assertEqual(self.SampleIsolatedTestCase.TEST, False)
 
1078
 
 
1079
    def test_debug(self):
 
1080
        pass
 
1081
        #test = self.SampleExecTestCase("test_sample_method")
 
1082
        #test.debug()
 
1083
 
 
1084
 
 
1085
class TestIsolatedTestSuite(TestCase):
 
1086
 
 
1087
    class SampleTestToIsolate(unittest.TestCase):
 
1088
 
 
1089
        SETUP = False
 
1090
        TEARDOWN = False
 
1091
        TEST = False
 
1092
 
 
1093
        def setUp(self):
 
1094
            TestIsolatedTestSuite.SampleTestToIsolate.SETUP = True
 
1095
 
 
1096
        def tearDown(self):
 
1097
            TestIsolatedTestSuite.SampleTestToIsolate.TEARDOWN = True
 
1098
 
 
1099
        def test_sets_global_state(self):
 
1100
            TestIsolatedTestSuite.SampleTestToIsolate.TEST = True
 
1101
 
 
1102
 
 
1103
    def test_construct(self):
 
1104
        subunit.IsolatedTestSuite()
 
1105
 
 
1106
    @skipIf(os.name != "posix", "Need a posix system for forking tests")
 
1107
    def test_run(self):
 
1108
        result = unittest.TestResult()
 
1109
        suite = subunit.IsolatedTestSuite()
 
1110
        sub_suite = unittest.TestSuite()
 
1111
        sub_suite.addTest(self.SampleTestToIsolate("test_sets_global_state"))
 
1112
        sub_suite.addTest(self.SampleTestToIsolate("test_sets_global_state"))
 
1113
        suite.addTest(sub_suite)
 
1114
        suite.addTest(self.SampleTestToIsolate("test_sets_global_state"))
 
1115
        suite.run(result)
 
1116
        self.assertEqual(result.testsRun, 3)
 
1117
        self.assertEqual(self.SampleTestToIsolate.SETUP, False)
 
1118
        self.assertEqual(self.SampleTestToIsolate.TEARDOWN, False)
 
1119
        self.assertEqual(self.SampleTestToIsolate.TEST, False)
 
1120
 
 
1121
 
 
1122
class TestTestProtocolClient(unittest.TestCase):
 
1123
 
 
1124
    def setUp(self):
 
1125
        self.io = BytesIO()
 
1126
        self.protocol = subunit.TestProtocolClient(self.io)
 
1127
        self.test = TestTestProtocolClient("test_start_test")
 
1128
        self.sample_details = {'something':Content(
 
1129
            ContentType('text', 'plain'), lambda:[_b('serialised\nform')])}
 
1130
        self.sample_tb_details = dict(self.sample_details)
 
1131
        self.sample_tb_details['traceback'] = TracebackContent(
 
1132
            subunit.RemoteError(_u("boo qux")), self.test)
 
1133
 
 
1134
    def test_start_test(self):
 
1135
        """Test startTest on a TestProtocolClient."""
 
1136
        self.protocol.startTest(self.test)
 
1137
        self.assertEqual(self.io.getvalue(), _b("test: %s\n" % self.test.id()))
 
1138
 
 
1139
    def test_stop_test(self):
 
1140
        # stopTest doesn't output anything.
 
1141
        self.protocol.stopTest(self.test)
 
1142
        self.assertEqual(self.io.getvalue(), _b(""))
 
1143
 
 
1144
    def test_add_success(self):
 
1145
        """Test addSuccess on a TestProtocolClient."""
 
1146
        self.protocol.addSuccess(self.test)
 
1147
        self.assertEqual(
 
1148
            self.io.getvalue(), _b("successful: %s\n" % self.test.id()))
 
1149
 
 
1150
    def test_add_success_details(self):
 
1151
        """Test addSuccess on a TestProtocolClient with details."""
 
1152
        self.protocol.addSuccess(self.test, details=self.sample_details)
 
1153
        self.assertEqual(
 
1154
            self.io.getvalue(), _b("successful: %s [ multipart\n"
 
1155
                "Content-Type: text/plain\n"
 
1156
                "something\n"
 
1157
                "F\r\nserialised\nform0\r\n]\n" % self.test.id()))
 
1158
 
 
1159
    def test_add_failure(self):
 
1160
        """Test addFailure on a TestProtocolClient."""
 
1161
        self.protocol.addFailure(
 
1162
            self.test, subunit.RemoteError(_u("boo qux")))
 
1163
        self.assertEqual(
 
1164
            self.io.getvalue(),
 
1165
            _b(('failure: %s [\n' + _remote_exception_str + ': boo qux\n]\n')
 
1166
            % self.test.id()))
 
1167
 
 
1168
    def test_add_failure_details(self):
 
1169
        """Test addFailure on a TestProtocolClient with details."""
 
1170
        self.protocol.addFailure(
 
1171
            self.test, details=self.sample_tb_details)
 
1172
        self.assertEqual(
 
1173
            self.io.getvalue(),
 
1174
            _b(("failure: %s [ multipart\n"
 
1175
            "Content-Type: text/plain\n"
 
1176
            "something\n"
 
1177
            "F\r\nserialised\nform0\r\n"
 
1178
            "Content-Type: text/x-traceback;charset=utf8,language=python\n"
 
1179
            "traceback\n" + _remote_exception_str_chunked + ": boo qux\n0\r\n"
 
1180
            "]\n") % self.test.id()))
 
1181
 
 
1182
    def test_add_error(self):
 
1183
        """Test stopTest on a TestProtocolClient."""
 
1184
        self.protocol.addError(
 
1185
            self.test, subunit.RemoteError(_u("phwoar crikey")))
 
1186
        self.assertEqual(
 
1187
            self.io.getvalue(),
 
1188
            _b(('error: %s [\n' +
 
1189
            _remote_exception_str + ": phwoar crikey\n"
 
1190
            "]\n") % self.test.id()))
 
1191
 
 
1192
    def test_add_error_details(self):
 
1193
        """Test stopTest on a TestProtocolClient with details."""
 
1194
        self.protocol.addError(
 
1195
            self.test, details=self.sample_tb_details)
 
1196
        self.assertEqual(
 
1197
            self.io.getvalue(),
 
1198
            _b(("error: %s [ multipart\n"
 
1199
            "Content-Type: text/plain\n"
 
1200
            "something\n"
 
1201
            "F\r\nserialised\nform0\r\n"
 
1202
            "Content-Type: text/x-traceback;charset=utf8,language=python\n"
 
1203
            "traceback\n" + _remote_exception_str_chunked + ": boo qux\n0\r\n"
 
1204
            "]\n") % self.test.id()))
 
1205
 
 
1206
    def test_add_expected_failure(self):
 
1207
        """Test addExpectedFailure on a TestProtocolClient."""
 
1208
        self.protocol.addExpectedFailure(
 
1209
            self.test, subunit.RemoteError(_u("phwoar crikey")))
 
1210
        self.assertEqual(
 
1211
            self.io.getvalue(),
 
1212
            _b(('xfail: %s [\n' +
 
1213
            _remote_exception_str + ": phwoar crikey\n"
 
1214
            "]\n") % self.test.id()))
 
1215
 
 
1216
    def test_add_expected_failure_details(self):
 
1217
        """Test addExpectedFailure on a TestProtocolClient with details."""
 
1218
        self.protocol.addExpectedFailure(
 
1219
            self.test, details=self.sample_tb_details)
 
1220
        self.assertEqual(
 
1221
            self.io.getvalue(),
 
1222
            _b(("xfail: %s [ multipart\n"
 
1223
            "Content-Type: text/plain\n"
 
1224
            "something\n"
 
1225
            "F\r\nserialised\nform0\r\n"
 
1226
            "Content-Type: text/x-traceback;charset=utf8,language=python\n"
 
1227
            "traceback\n" + _remote_exception_str_chunked + ": boo qux\n0\r\n"
 
1228
            "]\n") % self.test.id()))
 
1229
 
 
1230
 
 
1231
    def test_add_skip(self):
 
1232
        """Test addSkip on a TestProtocolClient."""
 
1233
        self.protocol.addSkip(
 
1234
            self.test, "Has it really?")
 
1235
        self.assertEqual(
 
1236
            self.io.getvalue(),
 
1237
            _b('skip: %s [\nHas it really?\n]\n' % self.test.id()))
 
1238
 
 
1239
    def test_add_skip_details(self):
 
1240
        """Test addSkip on a TestProtocolClient with details."""
 
1241
        details = {'reason':Content(
 
1242
            ContentType('text', 'plain'), lambda:[_b('Has it really?')])}
 
1243
        self.protocol.addSkip(self.test, details=details)
 
1244
        self.assertEqual(
 
1245
            self.io.getvalue(),
 
1246
            _b("skip: %s [ multipart\n"
 
1247
            "Content-Type: text/plain\n"
 
1248
            "reason\n"
 
1249
            "E\r\nHas it really?0\r\n"
 
1250
            "]\n" % self.test.id()))
 
1251
 
 
1252
    def test_progress_set(self):
 
1253
        self.protocol.progress(23, subunit.PROGRESS_SET)
 
1254
        self.assertEqual(self.io.getvalue(), _b('progress: 23\n'))
 
1255
 
 
1256
    def test_progress_neg_cur(self):
 
1257
        self.protocol.progress(-23, subunit.PROGRESS_CUR)
 
1258
        self.assertEqual(self.io.getvalue(), _b('progress: -23\n'))
 
1259
 
 
1260
    def test_progress_pos_cur(self):
 
1261
        self.protocol.progress(23, subunit.PROGRESS_CUR)
 
1262
        self.assertEqual(self.io.getvalue(), _b('progress: +23\n'))
 
1263
 
 
1264
    def test_progress_pop(self):
 
1265
        self.protocol.progress(1234, subunit.PROGRESS_POP)
 
1266
        self.assertEqual(self.io.getvalue(), _b('progress: pop\n'))
 
1267
 
 
1268
    def test_progress_push(self):
 
1269
        self.protocol.progress(1234, subunit.PROGRESS_PUSH)
 
1270
        self.assertEqual(self.io.getvalue(), _b('progress: push\n'))
 
1271
 
 
1272
    def test_time(self):
 
1273
        # Calling time() outputs a time signal immediately.
 
1274
        self.protocol.time(
 
1275
            datetime.datetime(2009,10,11,12,13,14,15, iso8601.Utc()))
 
1276
        self.assertEqual(
 
1277
            _b("time: 2009-10-11 12:13:14.000015Z\n"),
 
1278
            self.io.getvalue())
 
1279
 
 
1280
    def test_add_unexpected_success(self):
 
1281
        """Test addUnexpectedSuccess on a TestProtocolClient."""
 
1282
        self.protocol.addUnexpectedSuccess(self.test)
 
1283
        self.assertEqual(
 
1284
            self.io.getvalue(), _b("uxsuccess: %s\n" % self.test.id()))
 
1285
 
 
1286
    def test_add_unexpected_success_details(self):
 
1287
        """Test addUnexpectedSuccess on a TestProtocolClient with details."""
 
1288
        self.protocol.addUnexpectedSuccess(self.test, details=self.sample_details)
 
1289
        self.assertEqual(
 
1290
            self.io.getvalue(), _b("uxsuccess: %s [ multipart\n"
 
1291
                "Content-Type: text/plain\n"
 
1292
                "something\n"
 
1293
                "F\r\nserialised\nform0\r\n]\n" % self.test.id()))
 
1294
 
 
1295
 
 
1296
def test_suite():
 
1297
    loader = subunit.tests.TestUtil.TestLoader()
 
1298
    result = loader.loadTestsFromName(__name__)
 
1299
    return result