lib/subunit/python/subunit/tests/test_test_protocol.py

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