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