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
17import datetime
18import unittest
19import os
20
21from testtools import skipIf, TestCase
22from testtools.compat import _b, _u, BytesIO, StringIO
23from testtools.content import Content, TracebackContent
24from testtools.content_type import ContentType
25from testtools.tests.helpers import (
26    Python26TestResult,
27    Python27TestResult,
28    ExtendedTestResult,
29    )
30
31import subunit
32from subunit import _remote_exception_str, _remote_exception_str_chunked
33import subunit.iso8601 as iso8601
34
35
36class 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
49class TestDiscardStream(unittest.TestCase):
50
51    def test_write(self):
52        subunit.DiscardStream().write("content")
53
54
55class 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
78class 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
107class 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
141class 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
278class 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
380class 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
398class 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
453class 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
503class 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
620class 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
735class 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
790class 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
845class 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
879class 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
917class 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
941class 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
967class 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
981class 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
1042class DoExecTestCase(subunit.ExecTestCase):
1043
1044    def test_working_script(self):
1045        """sample-two-script.py"""
1046
1047
1048class 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
1085class 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
1122class 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
1296def test_suite():
1297    loader = subunit.tests.TestUtil.TestLoader()
1298    result = loader.loadTestsFromName(__name__)
1299    return result
1300