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