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