← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Martin Packman
  • Date: 2012-01-05 12:16:05 UTC
  • mto: This revision was merged to the branch mainline in revision 6432.
  • Revision ID: martin.packman@canonical.com-20120105121605-118byz8r1hol1u27
Clean up bzrlib imports in xml_serializer

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_selftest.py
17
17
"""Tests for the test framework."""
18
18
 
19
19
from cStringIO import StringIO
 
20
import gc
20
21
import doctest
21
22
import os
22
23
import signal
29
30
from testtools import (
30
31
    ExtendedToOriginalDecorator,
31
32
    MultiTestResult,
32
 
    __version__ as testtools_version,
33
33
    )
34
34
from testtools.content import Content
35
35
from testtools.content_type import ContentType
1062
1062
                self.expectFailure("No absolute truth", self.assertTrue, True)
1063
1063
        runner = tests.TextTestRunner(stream=StringIO())
1064
1064
        result = self.run_test_runner(runner, Test("test_truth"))
1065
 
        if testtools_version[:3] <= (0, 9, 11):
1066
 
            self.assertContainsRe(runner.stream.getvalue(),
1067
 
                "=+\n"
1068
 
                "FAIL: \\S+\.test_truth\n"
1069
 
                "-+\n"
1070
 
                "(?:.*\n)*"
1071
 
                "No absolute truth\n"
1072
 
                "(?:.*\n)*"
1073
 
                "-+\n"
1074
 
                "Ran 1 test in .*\n"
1075
 
                "\n"
1076
 
                "FAILED \\(failures=1\\)\n\\Z")
1077
 
        else:
1078
 
            self.assertContainsRe(runner.stream.getvalue(),
1079
 
                "=+\n"
1080
 
                "FAIL: \\S+\.test_truth\n"
1081
 
                "-+\n"
1082
 
                "Empty attachments:\n"
1083
 
                "  log\n"
1084
 
                "\n"
1085
 
                "reason: {{{No absolute truth}}}\n"
1086
 
                "-+\n"
1087
 
                "Ran 1 test in .*\n"
1088
 
                "\n"
1089
 
                "FAILED \\(failures=1\\)\n\\Z")
 
1065
        self.assertContainsRe(runner.stream.getvalue(),
 
1066
            "=+\n"
 
1067
            "FAIL: \\S+\.test_truth\n"
 
1068
            "-+\n"
 
1069
            "(?:.*\n)*"
 
1070
            "\\s*(?:Text attachment: )?reason"
 
1071
            "(?:\n-+\n|: {{{)"
 
1072
            "No absolute truth"
 
1073
            "(?:\n-+\n|}}}\n)"
 
1074
            "(?:.*\n)*"
 
1075
            "-+\n"
 
1076
            "Ran 1 test in .*\n"
 
1077
            "\n"
 
1078
            "FAILED \\(failures=1\\)\n\\Z")
1090
1079
 
1091
1080
    def test_result_decorator(self):
1092
1081
        # decorate results
1251
1240
            lambda trace=False: "ascii")
1252
1241
        result = self.run_test_runner(tests.TextTestRunner(stream=out),
1253
1242
            FailureWithUnicode("test_log_unicode"))
1254
 
        if testtools_version[:3] > (0, 9, 11):
1255
 
            self.assertContainsRe(out.getvalue(), "log: {{{\d+\.\d+  \\\\u2606}}}")
1256
 
        else:
1257
 
            self.assertContainsRe(out.getvalue(),
1258
 
                "Text attachment: log\n"
1259
 
                "-+\n"
1260
 
                "\d+\.\d+  \\\\u2606\n"
1261
 
                "-+\n")
 
1243
        self.assertContainsRe(out.getvalue(),
 
1244
            "(?:Text attachment: )?log"
 
1245
            "(?:\n-+\n|: {{{)"
 
1246
            "\d+\.\d+  \\\\u2606"
 
1247
            "(?:\n-+\n|}}}\n)")
1262
1248
 
1263
1249
 
1264
1250
class SampleTestCase(tests.TestCase):
1743
1729
        result = self._run_test('test_fail')
1744
1730
        self.assertEqual(1, len(result.failures))
1745
1731
        result_content = result.failures[0][1]
1746
 
        if testtools_version < (0, 9, 12):
1747
 
            self.assertContainsRe(result_content, 'Text attachment: log')
 
1732
        self.assertContainsRe(result_content,
 
1733
            '(?m)^(?:Text attachment: )?log(?:$|: )')
1748
1734
        self.assertContainsRe(result_content, 'this was a failing test')
1749
1735
 
1750
1736
    def test_error_has_log(self):
1751
1737
        result = self._run_test('test_error')
1752
1738
        self.assertEqual(1, len(result.errors))
1753
1739
        result_content = result.errors[0][1]
1754
 
        if testtools_version < (0, 9, 12):
1755
 
            self.assertContainsRe(result_content, 'Text attachment: log')
 
1740
        self.assertContainsRe(result_content,
 
1741
            '(?m)^(?:Text attachment: )?log(?:$|: )')
1756
1742
        self.assertContainsRe(result_content, 'this test errored')
1757
1743
 
1758
1744
    def test_skip_has_no_log(self):
1777
1763
        result = self._run_test('test_xfail')
1778
1764
        self.assertEqual(1, len(result.expectedFailures))
1779
1765
        result_content = result.expectedFailures[0][1]
1780
 
        self.assertNotContainsRe(result_content, 'Text attachment: log')
 
1766
        self.assertNotContainsRe(result_content,
 
1767
            '(?m)^(?:Text attachment: )?log(?:$|: )')
1781
1768
        self.assertNotContainsRe(result_content, 'test with expected failure')
1782
1769
 
1783
1770
    def test_unexpected_success_has_log(self):
2192
2179
        self.assertNotContainsRe(content, 'test with expected failure')
2193
2180
        self.assertEqual(1, len(result.expectedFailures))
2194
2181
        result_content = result.expectedFailures[0][1]
2195
 
        self.assertNotContainsRe(result_content, 'Text attachment: log')
 
2182
        self.assertNotContainsRe(result_content,
 
2183
            '(?m)^(?:Text attachment: )?log(?:$|: )')
2196
2184
        self.assertNotContainsRe(result_content, 'test with expected failure')
2197
2185
 
2198
2186
    def test_unexpected_success_has_log(self):
3314
3302
        self.assertLength(1, calls)
3315
3303
 
3316
3304
 
 
3305
class _Selftest(object):
 
3306
    """Mixin for tests needing full selftest output"""
 
3307
 
 
3308
    def _inject_stream_into_subunit(self, stream):
 
3309
        """To be overridden by subclasses that run tests out of process"""
 
3310
 
 
3311
    def _run_selftest(self, **kwargs):
 
3312
        sio = StringIO()
 
3313
        self._inject_stream_into_subunit(sio)
 
3314
        tests.selftest(stream=sio, stop_on_failure=False, **kwargs)
 
3315
        return sio.getvalue()
 
3316
 
 
3317
 
 
3318
class _ForkedSelftest(_Selftest):
 
3319
    """Mixin for tests needing full selftest output with forked children"""
 
3320
 
 
3321
    _test_needs_features = [features.subunit]
 
3322
 
 
3323
    def _inject_stream_into_subunit(self, stream):
 
3324
        """Monkey-patch subunit so the extra output goes to stream not stdout
 
3325
 
 
3326
        Some APIs need rewriting so this kind of bogus hackery can be replaced
 
3327
        by passing the stream param from run_tests down into ProtocolTestCase.
 
3328
        """
 
3329
        from subunit import ProtocolTestCase
 
3330
        _original_init = ProtocolTestCase.__init__
 
3331
        def _init_with_passthrough(self, *args, **kwargs):
 
3332
            _original_init(self, *args, **kwargs)
 
3333
            self._passthrough = stream
 
3334
        self.overrideAttr(ProtocolTestCase, "__init__", _init_with_passthrough)
 
3335
 
 
3336
    def _run_selftest(self, **kwargs):
 
3337
        # GZ 2011-05-26: Add a PosixSystem feature so this check can go away
 
3338
        if getattr(os, "fork", None) is None:
 
3339
            raise tests.TestNotApplicable("Platform doesn't support forking")
 
3340
        # Make sure the fork code is actually invoked by claiming two cores
 
3341
        self.overrideAttr(osutils, "local_concurrency", lambda: 2)
 
3342
        kwargs.setdefault("suite_decorators", []).append(tests.fork_decorator)
 
3343
        return super(_ForkedSelftest, self)._run_selftest(**kwargs)
 
3344
 
 
3345
 
 
3346
class TestParallelFork(_ForkedSelftest, tests.TestCase):
 
3347
    """Check operation of --parallel=fork selftest option"""
 
3348
 
 
3349
    def test_error_in_child_during_fork(self):
 
3350
        """Error in a forked child during test setup should get reported"""
 
3351
        class Test(tests.TestCase):
 
3352
            def testMethod(self):
 
3353
                pass
 
3354
        # We don't care what, just break something that a child will run
 
3355
        self.overrideAttr(tests, "workaround_zealous_crypto_random", None)
 
3356
        out = self._run_selftest(test_suite_factory=Test)
 
3357
        # Lines from the tracebacks of the two child processes may be mixed
 
3358
        # together due to the way subunit parses and forwards the streams,
 
3359
        # so permit extra lines between each part of the error output.
 
3360
        self.assertContainsRe(out,
 
3361
            "Traceback.*:\n"
 
3362
            "(?:.*\n)*"
 
3363
            ".+ in fork_for_tests\n"
 
3364
            "(?:.*\n)*"
 
3365
            "\s*workaround_zealous_crypto_random\(\)\n"
 
3366
            "(?:.*\n)*"
 
3367
            "TypeError:")
 
3368
 
 
3369
 
 
3370
class TestUncollectedWarnings(_Selftest, tests.TestCase):
 
3371
    """Check a test case still alive after being run emits a warning"""
 
3372
 
 
3373
    class Test(tests.TestCase):
 
3374
        def test_pass(self):
 
3375
            pass
 
3376
        def test_self_ref(self):
 
3377
            self.also_self = self.test_self_ref
 
3378
        def test_skip(self):
 
3379
            self.skip("Don't need")
 
3380
 
 
3381
    def _get_suite(self):
 
3382
        return TestUtil.TestSuite([
 
3383
            self.Test("test_pass"),
 
3384
            self.Test("test_self_ref"),
 
3385
            self.Test("test_skip"),
 
3386
            ])
 
3387
 
 
3388
    def _run_selftest_with_suite(self, **kwargs):
 
3389
        old_flags = tests.selftest_debug_flags
 
3390
        tests.selftest_debug_flags = old_flags.union(["uncollected_cases"])
 
3391
        gc_on = gc.isenabled()
 
3392
        if gc_on:
 
3393
            gc.disable()
 
3394
        try:
 
3395
            output = self._run_selftest(test_suite_factory=self._get_suite,
 
3396
                **kwargs)
 
3397
        finally:
 
3398
            if gc_on:
 
3399
                gc.enable()
 
3400
            tests.selftest_debug_flags = old_flags
 
3401
        self.assertNotContainsRe(output, "Uncollected test case.*test_pass")
 
3402
        self.assertContainsRe(output, "Uncollected test case.*test_self_ref")
 
3403
        return output
 
3404
 
 
3405
    def test_testsuite(self):
 
3406
        self._run_selftest_with_suite()
 
3407
 
 
3408
    def test_pattern(self):
 
3409
        out = self._run_selftest_with_suite(pattern="test_(?:pass|self_ref)$")
 
3410
        self.assertNotContainsRe(out, "test_skip")
 
3411
 
 
3412
    def test_exclude_pattern(self):
 
3413
        out = self._run_selftest_with_suite(exclude_pattern="test_skip$")
 
3414
        self.assertNotContainsRe(out, "test_skip")
 
3415
 
 
3416
    def test_random_seed(self):
 
3417
        self._run_selftest_with_suite(random_seed="now")
 
3418
 
 
3419
    def test_matching_tests_first(self):
 
3420
        self._run_selftest_with_suite(matching_tests_first=True,
 
3421
            pattern="test_self_ref$")
 
3422
 
 
3423
    def test_starting_with_and_exclude(self):
 
3424
        out = self._run_selftest_with_suite(starting_with=["bt."],
 
3425
            exclude_pattern="test_skip$")
 
3426
        self.assertNotContainsRe(out, "test_skip")
 
3427
 
 
3428
    def test_additonal_decorator(self):
 
3429
        out = self._run_selftest_with_suite(
 
3430
            suite_decorators=[tests.TestDecorator])
 
3431
 
 
3432
 
 
3433
class TestUncollectedWarningsSubunit(TestUncollectedWarnings):
 
3434
    """Check warnings from tests staying alive are emitted with subunit"""
 
3435
 
 
3436
    _test_needs_features = [features.subunit]
 
3437
 
 
3438
    def _run_selftest_with_suite(self, **kwargs):
 
3439
        return TestUncollectedWarnings._run_selftest_with_suite(self,
 
3440
            runner_class=tests.SubUnitBzrRunner, **kwargs)
 
3441
 
 
3442
 
 
3443
class TestUncollectedWarningsForked(_ForkedSelftest, TestUncollectedWarnings):
 
3444
    """Check warnings from tests staying alive are emitted when forking"""
 
3445
 
 
3446
 
3317
3447
class TestEnvironHandling(tests.TestCase):
3318
3448
 
3319
3449
    def test_overrideEnv_None_called_twice_doesnt_leak(self):