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