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