~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Tarmac
  • Author(s): Vincent Ladeuil, Patch Queue Manager, Jelmer Vernooij
  • Date: 2017-01-17 16:20:41 UTC
  • mfrom: (6619.1.2 trunk)
  • Revision ID: tarmac-20170117162041-oo62uk1qsmgc9j31
Merge 2.7 into trunk including fixes for bugs #1622039, #1644003, #1579093 and #1645017. [r=vila]

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005-2010 Canonical Ltd
 
1
# Copyright (C) 2005-2013, 2016 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
17
17
"""Tests for the test framework."""
18
18
 
19
19
from cStringIO import StringIO
20
 
from doctest import ELLIPSIS
 
20
import gc
 
21
import doctest
21
22
import os
22
23
import signal
23
24
import sys
 
25
import threading
24
26
import time
25
27
import unittest
26
28
import warnings
27
29
 
28
 
from testtools import MultiTestResult
 
30
from testtools import (
 
31
    ExtendedToOriginalDecorator,
 
32
    MultiTestResult,
 
33
    )
 
34
from testtools.content import Content
29
35
from testtools.content_type import ContentType
30
36
from testtools.matchers import (
31
37
    DocTestMatches,
32
38
    Equals,
33
39
    )
34
 
import testtools.tests.helpers
 
40
import testtools.testresult.doubles
35
41
 
36
42
import bzrlib
37
43
from bzrlib import (
38
44
    branchbuilder,
39
45
    bzrdir,
40
 
    debug,
 
46
    controldir,
41
47
    errors,
 
48
    hooks,
42
49
    lockdir,
43
50
    memorytree,
44
51
    osutils,
45
 
    progress,
46
52
    remote,
47
53
    repository,
48
54
    symbol_versioning,
49
55
    tests,
50
56
    transport,
51
57
    workingtree,
 
58
    workingtree_3,
 
59
    workingtree_4,
52
60
    )
53
61
from bzrlib.repofmt import (
54
62
    groupcompress_repo,
55
 
    pack_repo,
56
 
    weaverepo,
57
63
    )
58
64
from bzrlib.symbol_versioning import (
59
65
    deprecated_function,
64
70
    features,
65
71
    test_lsprof,
66
72
    test_server,
67
 
    test_sftp_transport,
68
73
    TestUtil,
69
74
    )
70
 
from bzrlib.trace import note
 
75
from bzrlib.trace import note, mutter
71
76
from bzrlib.transport import memory
72
 
from bzrlib.version import _get_bzr_source_tree
73
77
 
74
78
 
75
79
def _test_ids(test_suite):
77
81
    return [t.id() for t in tests.iter_suite_tests(test_suite)]
78
82
 
79
83
 
80
 
class SelftestTests(tests.TestCase):
81
 
 
82
 
    def test_import_tests(self):
83
 
        mod = TestUtil._load_module_by_name('bzrlib.tests.test_selftest')
84
 
        self.assertEqual(mod.SelftestTests, SelftestTests)
85
 
 
86
 
    def test_import_test_failure(self):
87
 
        self.assertRaises(ImportError,
88
 
                          TestUtil._load_module_by_name,
89
 
                          'bzrlib.no-name-yet')
90
 
 
91
 
 
92
84
class MetaTestLog(tests.TestCase):
93
85
 
94
86
    def test_logging(self):
100
92
            "text", "plain", {"charset": "utf8"})))
101
93
        self.assertThat(u"".join(log.iter_text()), Equals(self.get_log()))
102
94
        self.assertThat(self.get_log(),
103
 
            DocTestMatches(u"...a test message\n", ELLIPSIS))
104
 
 
105
 
 
106
 
class TestUnicodeFilename(tests.TestCase):
107
 
 
108
 
    def test_probe_passes(self):
109
 
        """UnicodeFilename._probe passes."""
110
 
        # We can't test much more than that because the behaviour depends
111
 
        # on the platform.
112
 
        tests.UnicodeFilename._probe()
 
95
            DocTestMatches(u"...a test message\n", doctest.ELLIPSIS))
113
96
 
114
97
 
115
98
class TestTreeShape(tests.TestCaseInTempDir):
116
99
 
117
100
    def test_unicode_paths(self):
118
 
        self.requireFeature(tests.UnicodeFilename)
 
101
        self.requireFeature(features.UnicodeFilenameFeature)
119
102
 
120
103
        filename = u'hell\u00d8'
121
104
        self.build_tree_contents([(filename, 'contents of hello')])
122
 
        self.failUnlessExists(filename)
 
105
        self.assertPathExists(filename)
 
106
 
 
107
 
 
108
class TestClassesAvailable(tests.TestCase):
 
109
    """As a convenience we expose Test* classes from bzrlib.tests"""
 
110
 
 
111
    def test_test_case(self):
 
112
        from bzrlib.tests import TestCase
 
113
 
 
114
    def test_test_loader(self):
 
115
        from bzrlib.tests import TestLoader
 
116
 
 
117
    def test_test_suite(self):
 
118
        from bzrlib.tests import TestSuite
123
119
 
124
120
 
125
121
class TestTransportScenarios(tests.TestCase):
312
308
        from bzrlib.tests.per_interrepository import make_scenarios
313
309
        server1 = "a"
314
310
        server2 = "b"
315
 
        formats = [("C0", "C1", "C2"), ("D0", "D1", "D2")]
 
311
        formats = [("C0", "C1", "C2", "C3"), ("D0", "D1", "D2", "D3")]
316
312
        scenarios = make_scenarios(server1, server2, formats)
317
313
        self.assertEqual([
318
314
            ('C0,str,str',
319
315
             {'repository_format': 'C1',
320
316
              'repository_format_to': 'C2',
321
317
              'transport_readonly_server': 'b',
322
 
              'transport_server': 'a'}),
 
318
              'transport_server': 'a',
 
319
              'extra_setup': 'C3'}),
323
320
            ('D0,str,str',
324
321
             {'repository_format': 'D1',
325
322
              'repository_format_to': 'D2',
326
323
              'transport_readonly_server': 'b',
327
 
              'transport_server': 'a'})],
 
324
              'transport_server': 'a',
 
325
              'extra_setup': 'D3'})],
328
326
            scenarios)
329
327
 
330
328
 
336
334
        from bzrlib.tests.per_workingtree import make_scenarios
337
335
        server1 = "a"
338
336
        server2 = "b"
339
 
        formats = [workingtree.WorkingTreeFormat2(),
340
 
                   workingtree.WorkingTreeFormat3(),]
341
 
        scenarios = make_scenarios(server1, server2, formats)
 
337
        formats = [workingtree_4.WorkingTreeFormat4(),
 
338
                   workingtree_3.WorkingTreeFormat3(),
 
339
                   workingtree_4.WorkingTreeFormat6()]
 
340
        scenarios = make_scenarios(server1, server2, formats,
 
341
            remote_server='c', remote_readonly_server='d',
 
342
            remote_backing_server='e')
342
343
        self.assertEqual([
343
 
            ('WorkingTreeFormat2',
 
344
            ('WorkingTreeFormat4',
344
345
             {'bzrdir_format': formats[0]._matchingbzrdir,
345
346
              'transport_readonly_server': 'b',
346
347
              'transport_server': 'a',
349
350
             {'bzrdir_format': formats[1]._matchingbzrdir,
350
351
              'transport_readonly_server': 'b',
351
352
              'transport_server': 'a',
352
 
              'workingtree_format': formats[1]})],
353
 
            scenarios)
 
353
              'workingtree_format': formats[1]}),
 
354
            ('WorkingTreeFormat6',
 
355
             {'bzrdir_format': formats[2]._matchingbzrdir,
 
356
              'transport_readonly_server': 'b',
 
357
              'transport_server': 'a',
 
358
              'workingtree_format': formats[2]}),
 
359
            ('WorkingTreeFormat6,remote',
 
360
             {'bzrdir_format': formats[2]._matchingbzrdir,
 
361
              'repo_is_remote': True,
 
362
              'transport_readonly_server': 'd',
 
363
              'transport_server': 'c',
 
364
              'vfs_transport_factory': 'e',
 
365
              'workingtree_format': formats[2]}),
 
366
            ], scenarios)
354
367
 
355
368
 
356
369
class TestTreeScenarios(tests.TestCase):
357
370
 
358
371
    def test_scenarios(self):
359
372
        # the tree implementation scenario generator is meant to setup one
360
 
        # instance for each working tree format, and one additional instance
 
373
        # instance for each working tree format, one additional instance
361
374
        # that will use the default wt format, but create a revision tree for
362
 
        # the tests.  this means that the wt ones should have the
363
 
        # workingtree_to_test_tree attribute set to 'return_parameter' and the
364
 
        # revision one set to revision_tree_from_workingtree.
 
375
        # the tests, and one more that uses the default wt format as a
 
376
        # lightweight checkout of a remote repository.  This means that the wt
 
377
        # ones should have the workingtree_to_test_tree attribute set to
 
378
        # 'return_parameter' and the revision one set to
 
379
        # revision_tree_from_workingtree.
365
380
 
366
381
        from bzrlib.tests.per_tree import (
367
382
            _dirstate_tree_from_workingtree,
373
388
            )
374
389
        server1 = "a"
375
390
        server2 = "b"
376
 
        formats = [workingtree.WorkingTreeFormat2(),
377
 
                   workingtree.WorkingTreeFormat3(),]
 
391
        smart_server = test_server.SmartTCPServer_for_testing
 
392
        smart_readonly_server = test_server.ReadonlySmartTCPServer_for_testing
 
393
        mem_server = memory.MemoryServer
 
394
        formats = [workingtree_4.WorkingTreeFormat4(),
 
395
                   workingtree_3.WorkingTreeFormat3(),]
378
396
        scenarios = make_scenarios(server1, server2, formats)
379
 
        self.assertEqual(7, len(scenarios))
380
 
        default_wt_format = workingtree.WorkingTreeFormat4._default_format
381
 
        wt4_format = workingtree.WorkingTreeFormat4()
382
 
        wt5_format = workingtree.WorkingTreeFormat5()
 
397
        self.assertEqual(8, len(scenarios))
 
398
        default_wt_format = workingtree.format_registry.get_default()
 
399
        wt4_format = workingtree_4.WorkingTreeFormat4()
 
400
        wt5_format = workingtree_4.WorkingTreeFormat5()
 
401
        wt6_format = workingtree_4.WorkingTreeFormat6()
383
402
        expected_scenarios = [
384
 
            ('WorkingTreeFormat2',
 
403
            ('WorkingTreeFormat4',
385
404
             {'bzrdir_format': formats[0]._matchingbzrdir,
386
405
              'transport_readonly_server': 'b',
387
406
              'transport_server': 'a',
395
414
              'workingtree_format': formats[1],
396
415
              '_workingtree_to_test_tree': return_parameter,
397
416
             }),
 
417
            ('WorkingTreeFormat6,remote',
 
418
             {'bzrdir_format': wt6_format._matchingbzrdir,
 
419
              'repo_is_remote': True,
 
420
              'transport_readonly_server': smart_readonly_server,
 
421
              'transport_server': smart_server,
 
422
              'vfs_transport_factory': mem_server,
 
423
              'workingtree_format': wt6_format,
 
424
              '_workingtree_to_test_tree': return_parameter,
 
425
             }),
398
426
            ('RevisionTree',
399
427
             {'_workingtree_to_test_tree': revision_tree_from_workingtree,
400
428
              'bzrdir_format': default_wt_format._matchingbzrdir,
447
475
        # ones to add.
448
476
        from bzrlib.tests.per_tree import (
449
477
            return_parameter,
450
 
            revision_tree_from_workingtree
451
478
            )
452
479
        from bzrlib.tests.per_intertree import (
453
480
            make_scenarios,
454
481
            )
455
 
        from bzrlib.workingtree import WorkingTreeFormat2, WorkingTreeFormat3
 
482
        from bzrlib.workingtree_3 import WorkingTreeFormat3
 
483
        from bzrlib.workingtree_4 import WorkingTreeFormat4
456
484
        input_test = TestInterTreeScenarios(
457
485
            "test_scenarios")
458
486
        server1 = "a"
459
487
        server2 = "b"
460
 
        format1 = WorkingTreeFormat2()
 
488
        format1 = WorkingTreeFormat4()
461
489
        format2 = WorkingTreeFormat3()
462
490
        formats = [("1", str, format1, format2, "converter1"),
463
491
            ("2", int, format2, format1, "converter2")]
509
537
        self.assertRaises(AssertionError, self.assertEqualStat,
510
538
            os.lstat("foo"), os.lstat("longname"))
511
539
 
 
540
    def test_failUnlessExists(self):
 
541
        """Deprecated failUnlessExists and failIfExists"""
 
542
        self.applyDeprecated(
 
543
            deprecated_in((2, 4)),
 
544
            self.failUnlessExists, '.')
 
545
        self.build_tree(['foo/', 'foo/bar'])
 
546
        self.applyDeprecated(
 
547
            deprecated_in((2, 4)),
 
548
            self.failUnlessExists, 'foo/bar')
 
549
        self.applyDeprecated(
 
550
            deprecated_in((2, 4)),
 
551
            self.failIfExists, 'foo/foo')
 
552
 
 
553
    def test_assertPathExists(self):
 
554
        self.assertPathExists('.')
 
555
        self.build_tree(['foo/', 'foo/bar'])
 
556
        self.assertPathExists('foo/bar')
 
557
        self.assertPathDoesNotExist('foo/foo')
 
558
 
512
559
 
513
560
class TestTestCaseWithMemoryTransport(tests.TestCaseWithMemoryTransport):
514
561
 
548
595
        tree = self.make_branch_and_memory_tree('dir')
549
596
        # Guard against regression into MemoryTransport leaking
550
597
        # files to disk instead of keeping them in memory.
551
 
        self.failIf(osutils.lexists('dir'))
 
598
        self.assertFalse(osutils.lexists('dir'))
552
599
        self.assertIsInstance(tree, memorytree.MemoryTree)
553
600
 
554
601
    def test_make_branch_and_memory_tree_with_format(self):
555
602
        """make_branch_and_memory_tree should accept a format option."""
556
603
        format = bzrdir.BzrDirMetaFormat1()
557
 
        format.repository_format = weaverepo.RepositoryFormat7()
 
604
        format.repository_format = repository.format_registry.get_default()
558
605
        tree = self.make_branch_and_memory_tree('dir', format=format)
559
606
        # Guard against regression into MemoryTransport leaking
560
607
        # files to disk instead of keeping them in memory.
561
 
        self.failIf(osutils.lexists('dir'))
 
608
        self.assertFalse(osutils.lexists('dir'))
562
609
        self.assertIsInstance(tree, memorytree.MemoryTree)
563
610
        self.assertEqual(format.repository_format.__class__,
564
611
            tree.branch.repository._format.__class__)
568
615
        self.assertIsInstance(builder, branchbuilder.BranchBuilder)
569
616
        # Guard against regression into MemoryTransport leaking
570
617
        # files to disk instead of keeping them in memory.
571
 
        self.failIf(osutils.lexists('dir'))
 
618
        self.assertFalse(osutils.lexists('dir'))
572
619
 
573
620
    def test_make_branch_builder_with_format(self):
574
621
        # Use a repo layout that doesn't conform to a 'named' layout, to ensure
575
622
        # that the format objects are used.
576
623
        format = bzrdir.BzrDirMetaFormat1()
577
 
        repo_format = weaverepo.RepositoryFormat7()
 
624
        repo_format = repository.format_registry.get_default()
578
625
        format.repository_format = repo_format
579
626
        builder = self.make_branch_builder('dir', format=format)
580
627
        the_branch = builder.get_branch()
581
628
        # Guard against regression into MemoryTransport leaking
582
629
        # files to disk instead of keeping them in memory.
583
 
        self.failIf(osutils.lexists('dir'))
 
630
        self.assertFalse(osutils.lexists('dir'))
584
631
        self.assertEqual(format.repository_format.__class__,
585
632
                         the_branch.repository._format.__class__)
586
633
        self.assertEqual(repo_format.get_format_string(),
592
639
        the_branch = builder.get_branch()
593
640
        # Guard against regression into MemoryTransport leaking
594
641
        # files to disk instead of keeping them in memory.
595
 
        self.failIf(osutils.lexists('dir'))
596
 
        dir_format = bzrdir.format_registry.make_bzrdir('knit')
 
642
        self.assertFalse(osutils.lexists('dir'))
 
643
        dir_format = controldir.format_registry.make_bzrdir('knit')
597
644
        self.assertEqual(dir_format.repository_format.__class__,
598
645
                         the_branch.repository._format.__class__)
599
646
        self.assertEqual('Bazaar-NG Knit Repository Format 1',
603
650
    def test_dangling_locks_cause_failures(self):
604
651
        class TestDanglingLock(tests.TestCaseWithMemoryTransport):
605
652
            def test_function(self):
606
 
                t = self.get_transport('.')
 
653
                t = self.get_transport_from_path('.')
607
654
                l = lockdir.LockDir(t, 'lock')
608
655
                l.create()
609
656
                l.attempt_lock()
611
658
        result = test.run()
612
659
        total_failures = result.errors + result.failures
613
660
        if self._lock_check_thorough:
614
 
            self.assertLength(1, total_failures)
 
661
            self.assertEqual(1, len(total_failures))
615
662
        else:
616
663
            # When _lock_check_thorough is disabled, then we don't trigger a
617
664
            # failure
618
 
            self.assertLength(0, total_failures)
 
665
            self.assertEqual(0, len(total_failures))
619
666
 
620
667
 
621
668
class TestTestCaseWithTransport(tests.TestCaseWithTransport):
629
676
        # for the server
630
677
        url = self.get_readonly_url()
631
678
        url2 = self.get_readonly_url('foo/bar')
632
 
        t = transport.get_transport(url)
633
 
        t2 = transport.get_transport(url2)
634
 
        self.failUnless(isinstance(t, ReadonlyTransportDecorator))
635
 
        self.failUnless(isinstance(t2, ReadonlyTransportDecorator))
 
679
        t = transport.get_transport_from_url(url)
 
680
        t2 = transport.get_transport_from_url(url2)
 
681
        self.assertIsInstance(t, ReadonlyTransportDecorator)
 
682
        self.assertIsInstance(t2, ReadonlyTransportDecorator)
636
683
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
637
684
 
638
685
    def test_get_readonly_url_http(self):
644
691
        url = self.get_readonly_url()
645
692
        url2 = self.get_readonly_url('foo/bar')
646
693
        # the transport returned may be any HttpTransportBase subclass
647
 
        t = transport.get_transport(url)
648
 
        t2 = transport.get_transport(url2)
649
 
        self.failUnless(isinstance(t, HttpTransportBase))
650
 
        self.failUnless(isinstance(t2, HttpTransportBase))
 
694
        t = transport.get_transport_from_url(url)
 
695
        t2 = transport.get_transport_from_url(url2)
 
696
        self.assertIsInstance(t, HttpTransportBase)
 
697
        self.assertIsInstance(t2, HttpTransportBase)
651
698
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
652
699
 
653
700
    def test_is_directory(self):
661
708
    def test_make_branch_builder(self):
662
709
        builder = self.make_branch_builder('dir')
663
710
        rev_id = builder.build_commit()
664
 
        self.failUnlessExists('dir')
665
 
        a_dir = bzrdir.BzrDir.open('dir')
 
711
        self.assertPathExists('dir')
 
712
        a_dir = controldir.ControlDir.open('dir')
666
713
        self.assertRaises(errors.NoWorkingTree, a_dir.open_workingtree)
667
714
        a_branch = a_dir.open_branch()
668
715
        builder_branch = builder.get_branch()
683
730
        self.assertIsInstance(result_bzrdir.transport,
684
731
                              memory.MemoryTransport)
685
732
        # should not be on disk, should only be in memory
686
 
        self.failIfExists('subdir')
 
733
        self.assertPathDoesNotExist('subdir')
687
734
 
688
735
 
689
736
class TestChrootedTest(tests.ChrootedTestCase):
690
737
 
691
738
    def test_root_is_root(self):
692
 
        t = transport.get_transport(self.get_readonly_url())
 
739
        t = transport.get_transport_from_url(self.get_readonly_url())
693
740
        url = t.base
694
741
        self.assertEqual(url, t.clone('..').base)
695
742
 
697
744
class TestProfileResult(tests.TestCase):
698
745
 
699
746
    def test_profiles_tests(self):
700
 
        self.requireFeature(test_lsprof.LSProfFeature)
701
 
        terminal = testtools.tests.helpers.ExtendedTestResult()
 
747
        self.requireFeature(features.lsprof_feature)
 
748
        terminal = testtools.testresult.doubles.ExtendedTestResult()
702
749
        result = tests.ProfileResult(terminal)
703
750
        class Sample(tests.TestCase):
704
751
            def a(self):
721
768
                descriptions=0,
722
769
                verbosity=1,
723
770
                )
724
 
        capture = testtools.tests.helpers.ExtendedTestResult()
 
771
        capture = testtools.testresult.doubles.ExtendedTestResult()
725
772
        test_case.run(MultiTestResult(result, capture))
726
773
        run_case = capture._events[0][1]
727
774
        timed_string = result._testTimeString(run_case)
748
795
        self.check_timing(ShortDelayTestCase('test_short_delay'),
749
796
                          r"^ +[0-9]+ms$")
750
797
 
751
 
    def _patch_get_bzr_source_tree(self):
752
 
        # Reading from the actual source tree breaks isolation, but we don't
753
 
        # want to assume that thats *all* that would happen.
754
 
        self.overrideAttr(bzrlib.version, '_get_bzr_source_tree', lambda: None)
755
 
 
756
 
    def test_assigned_benchmark_file_stores_date(self):
757
 
        self._patch_get_bzr_source_tree()
758
 
        output = StringIO()
759
 
        result = bzrlib.tests.TextTestResult(self._log_file,
760
 
                                        descriptions=0,
761
 
                                        verbosity=1,
762
 
                                        bench_history=output
763
 
                                        )
764
 
        output_string = output.getvalue()
765
 
        # if you are wondering about the regexp please read the comment in
766
 
        # test_bench_history (bzrlib.tests.test_selftest.TestRunner)
767
 
        # XXX: what comment?  -- Andrew Bennetts
768
 
        self.assertContainsRe(output_string, "--date [0-9.]+")
769
 
 
770
 
    def test_benchhistory_records_test_times(self):
771
 
        self._patch_get_bzr_source_tree()
772
 
        result_stream = StringIO()
773
 
        result = bzrlib.tests.TextTestResult(
774
 
            self._log_file,
775
 
            descriptions=0,
776
 
            verbosity=1,
777
 
            bench_history=result_stream
778
 
            )
779
 
 
780
 
        # we want profile a call and check that its test duration is recorded
781
 
        # make a new test instance that when run will generate a benchmark
782
 
        example_test_case = TestTestResult("_time_hello_world_encoding")
783
 
        # execute the test, which should succeed and record times
784
 
        example_test_case.run(result)
785
 
        lines = result_stream.getvalue().splitlines()
786
 
        self.assertEqual(2, len(lines))
787
 
        self.assertContainsRe(lines[1],
788
 
            " *[0-9]+ms bzrlib.tests.test_selftest.TestTestResult"
789
 
            "._time_hello_world_encoding")
790
 
 
791
798
    def _time_hello_world_encoding(self):
792
799
        """Profile two sleep calls
793
800
 
798
805
 
799
806
    def test_lsprofiling(self):
800
807
        """Verbose test result prints lsprof statistics from test cases."""
801
 
        self.requireFeature(test_lsprof.LSProfFeature)
 
808
        self.requireFeature(features.lsprof_feature)
802
809
        result_stream = StringIO()
803
810
        result = bzrlib.tests.VerboseTestResult(
804
811
            result_stream,
833
840
        self.assertContainsRe(output,
834
841
            r"LSProf output for <type 'unicode'>\(\('world',\), {'errors': 'replace'}\)\n")
835
842
 
 
843
    def test_uses_time_from_testtools(self):
 
844
        """Test case timings in verbose results should use testtools times"""
 
845
        import datetime
 
846
        class TimeAddedVerboseTestResult(tests.VerboseTestResult):
 
847
            def startTest(self, test):
 
848
                self.time(datetime.datetime.utcfromtimestamp(1.145))
 
849
                super(TimeAddedVerboseTestResult, self).startTest(test)
 
850
            def addSuccess(self, test):
 
851
                self.time(datetime.datetime.utcfromtimestamp(51.147))
 
852
                super(TimeAddedVerboseTestResult, self).addSuccess(test)
 
853
            def report_tests_starting(self): pass
 
854
        sio = StringIO()
 
855
        self.get_passing_test().run(TimeAddedVerboseTestResult(sio, 0, 2))
 
856
        self.assertEndsWith(sio.getvalue(), "OK    50002ms\n")
 
857
 
836
858
    def test_known_failure(self):
837
 
        """A KnownFailure being raised should trigger several result actions."""
 
859
        """Using knownFailure should trigger several result actions."""
838
860
        class InstrumentedTestResult(tests.ExtendedTestResult):
839
861
            def stopTestRun(self): pass
840
 
            def startTests(self): pass
841
 
            def report_test_start(self, test): pass
 
862
            def report_tests_starting(self): pass
842
863
            def report_known_failure(self, test, err=None, details=None):
843
864
                self._call = test, 'known failure'
844
865
        result = InstrumentedTestResult(None, None, None, None)
845
866
        class Test(tests.TestCase):
846
867
            def test_function(self):
847
 
                raise tests.KnownFailure('failed!')
 
868
                self.knownFailure('failed!')
848
869
        test = Test("test_function")
849
870
        test.run(result)
850
871
        # it should invoke 'report_known_failure'.
866
887
            descriptions=0,
867
888
            verbosity=2,
868
889
            )
869
 
        test = self.get_passing_test()
870
 
        result.startTest(test)
871
 
        prefix = len(result_stream.getvalue())
872
 
        # the err parameter has the shape:
873
 
        # (class, exception object, traceback)
874
 
        # KnownFailures dont get their tracebacks shown though, so we
875
 
        # can skip that.
876
 
        err = (tests.KnownFailure, tests.KnownFailure('foo'), None)
877
 
        result.report_known_failure(test, err)
878
 
        output = result_stream.getvalue()[prefix:]
879
 
        lines = output.splitlines()
880
 
        self.assertContainsRe(lines[0], r'XFAIL *\d+ms$')
881
 
        if sys.version_info > (2, 7):
882
 
            self.expectFailure("_ExpectedFailure on 2.7 loses the message",
883
 
                self.assertNotEqual, lines[1], '    ')
884
 
        self.assertEqual(lines[1], '    foo')
885
 
        self.assertEqual(2, len(lines))
 
890
        _get_test("test_xfail").run(result)
 
891
        self.assertContainsRe(result_stream.getvalue(),
 
892
            "\n\\S+\\.test_xfail\\s+XFAIL\\s+\\d+ms\n"
 
893
            "\\s*(?:Text attachment: )?reason"
 
894
            "(?:\n-+\n|: {{{)"
 
895
            "this_fails"
 
896
            "(?:\n-+\n|}}}\n)")
886
897
 
887
898
    def get_passing_test(self):
888
899
        """Return a test object that can't be run usefully."""
894
905
        """Test the behaviour of invoking addNotSupported."""
895
906
        class InstrumentedTestResult(tests.ExtendedTestResult):
896
907
            def stopTestRun(self): pass
897
 
            def startTests(self): pass
898
 
            def report_test_start(self, test): pass
 
908
            def report_tests_starting(self): pass
899
909
            def report_unsupported(self, test, feature):
900
910
                self._call = test, feature
901
911
        result = InstrumentedTestResult(None, None, None, None)
902
912
        test = SampleTestCase('_test_pass')
903
 
        feature = tests.Feature()
 
913
        feature = features.Feature()
904
914
        result.startTest(test)
905
915
        result.addNotSupported(test, feature)
906
916
        # it should invoke 'report_unsupported'.
925
935
            verbosity=2,
926
936
            )
927
937
        test = self.get_passing_test()
928
 
        feature = tests.Feature()
 
938
        feature = features.Feature()
929
939
        result.startTest(test)
930
940
        prefix = len(result_stream.getvalue())
931
941
        result.report_unsupported(test, feature)
940
950
        """An UnavailableFeature being raised should invoke addNotSupported."""
941
951
        class InstrumentedTestResult(tests.ExtendedTestResult):
942
952
            def stopTestRun(self): pass
943
 
            def startTests(self): pass
944
 
            def report_test_start(self, test): pass
 
953
            def report_tests_starting(self): pass
945
954
            def addNotSupported(self, test, feature):
946
955
                self._call = test, feature
947
956
        result = InstrumentedTestResult(None, None, None, None)
948
 
        feature = tests.Feature()
 
957
        feature = features.Feature()
949
958
        class Test(tests.TestCase):
950
959
            def test_function(self):
951
960
                raise tests.UnavailableFeature(feature)
970
979
    def test_strict_with_known_failure(self):
971
980
        result = bzrlib.tests.TextTestResult(self._log_file, descriptions=0,
972
981
                                             verbosity=1)
973
 
        test = self.get_passing_test()
974
 
        err = (tests.KnownFailure, tests.KnownFailure('foo'), None)
975
 
        result.addExpectedFailure(test, err)
 
982
        test = _get_test("test_xfail")
 
983
        test.run(result)
976
984
        self.assertFalse(result.wasStrictlySuccessful())
977
985
        self.assertEqual(None, result._extractBenchmarkTime(test))
978
986
 
989
997
        class InstrumentedTestResult(tests.ExtendedTestResult):
990
998
            calls = 0
991
999
            def startTests(self): self.calls += 1
992
 
            def report_test_start(self, test): pass
993
1000
        result = InstrumentedTestResult(None, None, None, None)
994
1001
        def test_function():
995
1002
            pass
996
1003
        test = unittest.FunctionTestCase(test_function)
997
1004
        test.run(result)
998
 
        self.assertEquals(1, result.calls)
999
 
 
1000
 
 
1001
 
class TestUnicodeFilenameFeature(tests.TestCase):
1002
 
 
1003
 
    def test_probe_passes(self):
1004
 
        """UnicodeFilenameFeature._probe passes."""
1005
 
        # We can't test much more than that because the behaviour depends
1006
 
        # on the platform.
1007
 
        tests.UnicodeFilenameFeature._probe()
 
1005
        self.assertEqual(1, result.calls)
 
1006
 
 
1007
    def test_startTests_only_once(self):
 
1008
        """With multiple tests startTests should still only be called once"""
 
1009
        class InstrumentedTestResult(tests.ExtendedTestResult):
 
1010
            calls = 0
 
1011
            def startTests(self): self.calls += 1
 
1012
        result = InstrumentedTestResult(None, None, None, None)
 
1013
        suite = unittest.TestSuite([
 
1014
            unittest.FunctionTestCase(lambda: None),
 
1015
            unittest.FunctionTestCase(lambda: None)])
 
1016
        suite.run(result)
 
1017
        self.assertEqual(1, result.calls)
 
1018
        self.assertEqual(2, result.count)
1008
1019
 
1009
1020
 
1010
1021
class TestRunner(tests.TestCase):
1023
1034
        because of our use of global state.
1024
1035
        """
1025
1036
        old_root = tests.TestCaseInTempDir.TEST_ROOT
1026
 
        old_leak = tests.TestCase._first_thread_leaker_id
1027
1037
        try:
1028
1038
            tests.TestCaseInTempDir.TEST_ROOT = None
1029
 
            tests.TestCase._first_thread_leaker_id = None
1030
1039
            return testrunner.run(test)
1031
1040
        finally:
1032
1041
            tests.TestCaseInTempDir.TEST_ROOT = old_root
1033
 
            tests.TestCase._first_thread_leaker_id = old_leak
1034
1042
 
1035
1043
    def test_known_failure_failed_run(self):
1036
1044
        # run a test that generates a known failure which should be printed in
1041
1049
        test = unittest.TestSuite()
1042
1050
        test.addTest(Test("known_failure_test"))
1043
1051
        def failing_test():
1044
 
            self.fail('foo')
 
1052
            raise AssertionError('foo')
1045
1053
        test.addTest(unittest.FunctionTestCase(failing_test))
1046
1054
        stream = StringIO()
1047
1055
        runner = tests.TextTestRunner(stream=stream)
1055
1063
            '^----------------------------------------------------------------------\n'
1056
1064
            'Traceback \\(most recent call last\\):\n'
1057
1065
            '  .*' # File .*, line .*, in failing_test' - but maybe not from .pyc
1058
 
            '    self.fail\\(\'foo\'\\)\n'
 
1066
            '    raise AssertionError\\(\'foo\'\\)\n'
1059
1067
            '.*'
1060
1068
            '^----------------------------------------------------------------------\n'
1061
1069
            '.*'
1067
1075
        # the final output.
1068
1076
        class Test(tests.TestCase):
1069
1077
            def known_failure_test(self):
1070
 
                self.expectFailure('failed', self.assertTrue, False)
 
1078
                self.knownFailure("Never works...")
1071
1079
        test = Test("known_failure_test")
1072
1080
        stream = StringIO()
1073
1081
        runner = tests.TextTestRunner(stream=stream)
1079
1087
            '\n'
1080
1088
            'OK \\(known_failures=1\\)\n')
1081
1089
 
 
1090
    def test_unexpected_success_bad(self):
 
1091
        class Test(tests.TestCase):
 
1092
            def test_truth(self):
 
1093
                self.expectFailure("No absolute truth", self.assertTrue, True)
 
1094
        runner = tests.TextTestRunner(stream=StringIO())
 
1095
        result = self.run_test_runner(runner, Test("test_truth"))
 
1096
        self.assertContainsRe(runner.stream.getvalue(),
 
1097
            "=+\n"
 
1098
            "FAIL: \\S+\.test_truth\n"
 
1099
            "-+\n"
 
1100
            "(?:.*\n)*"
 
1101
            "\\s*(?:Text attachment: )?reason"
 
1102
            "(?:\n-+\n|: {{{)"
 
1103
            "No absolute truth"
 
1104
            "(?:\n-+\n|}}}\n)"
 
1105
            "(?:.*\n)*"
 
1106
            "-+\n"
 
1107
            "Ran 1 test in .*\n"
 
1108
            "\n"
 
1109
            "FAILED \\(failures=1\\)\n\\Z")
 
1110
 
1082
1111
    def test_result_decorator(self):
1083
1112
        # decorate results
1084
1113
        calls = []
1085
 
        class LoggingDecorator(tests.ForwardingResult):
 
1114
        class LoggingDecorator(ExtendedToOriginalDecorator):
1086
1115
            def startTest(self, test):
1087
 
                tests.ForwardingResult.startTest(self, test)
 
1116
                ExtendedToOriginalDecorator.startTest(self, test)
1088
1117
                calls.append('start')
1089
1118
        test = unittest.FunctionTestCase(lambda:None)
1090
1119
        stream = StringIO()
1132
1161
        class SkippedTest(tests.TestCase):
1133
1162
 
1134
1163
            def setUp(self):
1135
 
                tests.TestCase.setUp(self)
 
1164
                super(SkippedTest, self).setUp()
1136
1165
                calls.append('setUp')
1137
1166
                self.addCleanup(self.cleanup)
1138
1167
 
1168
1197
 
1169
1198
    def test_unsupported_features_listed(self):
1170
1199
        """When unsupported features are encountered they are detailed."""
1171
 
        class Feature1(tests.Feature):
 
1200
        class Feature1(features.Feature):
1172
1201
            def _probe(self): return False
1173
 
        class Feature2(tests.Feature):
 
1202
        class Feature2(features.Feature):
1174
1203
            def _probe(self): return False
1175
1204
        # create sample tests
1176
1205
        test1 = SampleTestCase('_test_pass')
1191
1220
            ],
1192
1221
            lines[-3:])
1193
1222
 
1194
 
    def _patch_get_bzr_source_tree(self):
1195
 
        # Reading from the actual source tree breaks isolation, but we don't
1196
 
        # want to assume that thats *all* that would happen.
1197
 
        self._get_source_tree_calls = []
1198
 
        def new_get():
1199
 
            self._get_source_tree_calls.append("called")
1200
 
            return None
1201
 
        self.overrideAttr(bzrlib.version, '_get_bzr_source_tree',  new_get)
1202
 
 
1203
 
    def test_bench_history(self):
1204
 
        # tests that the running the benchmark passes bench_history into
1205
 
        # the test result object. We can tell that happens if
1206
 
        # _get_bzr_source_tree is called.
1207
 
        self._patch_get_bzr_source_tree()
1208
 
        test = TestRunner('dummy_test')
1209
 
        output = StringIO()
1210
 
        runner = tests.TextTestRunner(stream=self._log_file,
1211
 
                                      bench_history=output)
1212
 
        result = self.run_test_runner(runner, test)
1213
 
        output_string = output.getvalue()
1214
 
        self.assertContainsRe(output_string, "--date [0-9.]+")
1215
 
        self.assertLength(1, self._get_source_tree_calls)
 
1223
    def test_verbose_test_count(self):
 
1224
        """A verbose test run reports the right test count at the start"""
 
1225
        suite = TestUtil.TestSuite([
 
1226
            unittest.FunctionTestCase(lambda:None),
 
1227
            unittest.FunctionTestCase(lambda:None)])
 
1228
        self.assertEqual(suite.countTestCases(), 2)
 
1229
        stream = StringIO()
 
1230
        runner = tests.TextTestRunner(stream=stream, verbosity=2)
 
1231
        # Need to use the CountingDecorator as that's what sets num_tests
 
1232
        result = self.run_test_runner(runner, tests.CountingDecorator(suite))
 
1233
        self.assertStartsWith(stream.getvalue(), "running 2 tests")
1216
1234
 
1217
1235
    def test_startTestRun(self):
1218
1236
        """run should call result.startTestRun()"""
1219
1237
        calls = []
1220
 
        class LoggingDecorator(tests.ForwardingResult):
 
1238
        class LoggingDecorator(ExtendedToOriginalDecorator):
1221
1239
            def startTestRun(self):
1222
 
                tests.ForwardingResult.startTestRun(self)
 
1240
                ExtendedToOriginalDecorator.startTestRun(self)
1223
1241
                calls.append('startTestRun')
1224
1242
        test = unittest.FunctionTestCase(lambda:None)
1225
1243
        stream = StringIO()
1231
1249
    def test_stopTestRun(self):
1232
1250
        """run should call result.stopTestRun()"""
1233
1251
        calls = []
1234
 
        class LoggingDecorator(tests.ForwardingResult):
 
1252
        class LoggingDecorator(ExtendedToOriginalDecorator):
1235
1253
            def stopTestRun(self):
1236
 
                tests.ForwardingResult.stopTestRun(self)
 
1254
                ExtendedToOriginalDecorator.stopTestRun(self)
1237
1255
                calls.append('stopTestRun')
1238
1256
        test = unittest.FunctionTestCase(lambda:None)
1239
1257
        stream = StringIO()
1242
1260
        result = self.run_test_runner(runner, test)
1243
1261
        self.assertLength(1, calls)
1244
1262
 
 
1263
    def test_unicode_test_output_on_ascii_stream(self):
 
1264
        """Showing results should always succeed even on an ascii console"""
 
1265
        class FailureWithUnicode(tests.TestCase):
 
1266
            def test_log_unicode(self):
 
1267
                self.log(u"\u2606")
 
1268
                self.fail("Now print that log!")
 
1269
        out = StringIO()
 
1270
        self.overrideAttr(osutils, "get_terminal_encoding",
 
1271
            lambda trace=False: "ascii")
 
1272
        result = self.run_test_runner(tests.TextTestRunner(stream=out),
 
1273
            FailureWithUnicode("test_log_unicode"))
 
1274
        self.assertContainsRe(out.getvalue(),
 
1275
            "(?:Text attachment: )?log"
 
1276
            "(?:\n-+\n|: {{{)"
 
1277
            "\d+\.\d+  \\\\u2606"
 
1278
            "(?:\n-+\n|}}}\n)")
 
1279
 
1245
1280
 
1246
1281
class SampleTestCase(tests.TestCase):
1247
1282
 
1435
1470
        # Note this test won't fail with hooks that the core library doesn't
1436
1471
        # use - but it trigger with a plugin that adds hooks, so its still a
1437
1472
        # useful warning in that case.
1438
 
        self.assertEqual(bzrlib.branch.BranchHooks(),
1439
 
            bzrlib.branch.Branch.hooks)
1440
 
        self.assertEqual(bzrlib.smart.server.SmartServerHooks(),
 
1473
        self.assertEqual(bzrlib.branch.BranchHooks(), bzrlib.branch.Branch.hooks)
 
1474
        self.assertEqual(
 
1475
            bzrlib.smart.server.SmartServerHooks(),
1441
1476
            bzrlib.smart.server.SmartTCPServer.hooks)
1442
 
        self.assertEqual(bzrlib.commands.CommandHooks(),
1443
 
            bzrlib.commands.Command.hooks)
 
1477
        self.assertEqual(
 
1478
            bzrlib.commands.CommandHooks(), bzrlib.commands.Command.hooks)
1444
1479
 
1445
1480
    def test__gather_lsprof_in_benchmarks(self):
1446
1481
        """When _gather_lsprof_in_benchmarks is on, accumulate profile data.
1447
1482
 
1448
1483
        Each self.time() call is individually and separately profiled.
1449
1484
        """
1450
 
        self.requireFeature(test_lsprof.LSProfFeature)
 
1485
        self.requireFeature(features.lsprof_feature)
1451
1486
        # overrides the class member with an instance member so no cleanup
1452
1487
        # needed.
1453
1488
        self._gather_lsprof_in_benchmarks = True
1472
1507
        transport_server = memory.MemoryServer()
1473
1508
        transport_server.start_server()
1474
1509
        self.addCleanup(transport_server.stop_server)
1475
 
        t = transport.get_transport(transport_server.get_url())
1476
 
        bzrdir.BzrDir.create(t.base)
 
1510
        t = transport.get_transport_from_url(transport_server.get_url())
 
1511
        controldir.ControlDir.create(t.base)
1477
1512
        self.assertRaises(errors.BzrError,
1478
 
            bzrdir.BzrDir.open_from_transport, t)
 
1513
            controldir.ControlDir.open_from_transport, t)
1479
1514
        # But if we declare this as safe, we can open the bzrdir.
1480
1515
        self.permit_url(t.base)
1481
1516
        self._bzr_selftest_roots.append(t.base)
1482
 
        bzrdir.BzrDir.open_from_transport(t)
 
1517
        controldir.ControlDir.open_from_transport(t)
1483
1518
 
1484
1519
    def test_requireFeature_available(self):
1485
1520
        """self.requireFeature(available) is a no-op."""
1486
 
        class Available(tests.Feature):
 
1521
        class Available(features.Feature):
1487
1522
            def _probe(self):return True
1488
1523
        feature = Available()
1489
1524
        self.requireFeature(feature)
1490
1525
 
1491
1526
    def test_requireFeature_unavailable(self):
1492
1527
        """self.requireFeature(unavailable) raises UnavailableFeature."""
1493
 
        class Unavailable(tests.Feature):
 
1528
        class Unavailable(features.Feature):
1494
1529
            def _probe(self):return False
1495
1530
        feature = Unavailable()
1496
1531
        self.assertRaises(tests.UnavailableFeature,
1619
1654
        self.assertRaises(AssertionError,
1620
1655
            self.assertListRaises, _TestException, success_generator)
1621
1656
 
 
1657
    def _run_successful_test(self, test):
 
1658
        result = testtools.TestResult()
 
1659
        test.run(result)
 
1660
        self.assertTrue(result.wasSuccessful())
 
1661
        return result
 
1662
 
1622
1663
    def test_overrideAttr_without_value(self):
1623
1664
        self.test_attr = 'original' # Define a test attribute
1624
1665
        obj = self # Make 'obj' visible to the embedded test
1625
1666
        class Test(tests.TestCase):
1626
1667
 
1627
1668
            def setUp(self):
1628
 
                tests.TestCase.setUp(self)
 
1669
                super(Test, self).setUp()
1629
1670
                self.orig = self.overrideAttr(obj, 'test_attr')
1630
1671
 
1631
1672
            def test_value(self):
1634
1675
                obj.test_attr = 'modified'
1635
1676
                self.assertEqual('modified', obj.test_attr)
1636
1677
 
1637
 
        test = Test('test_value')
1638
 
        test.run(unittest.TestResult())
 
1678
        self._run_successful_test(Test('test_value'))
1639
1679
        self.assertEqual('original', obj.test_attr)
1640
1680
 
1641
1681
    def test_overrideAttr_with_value(self):
1644
1684
        class Test(tests.TestCase):
1645
1685
 
1646
1686
            def setUp(self):
1647
 
                tests.TestCase.setUp(self)
 
1687
                super(Test, self).setUp()
1648
1688
                self.orig = self.overrideAttr(obj, 'test_attr', new='modified')
1649
1689
 
1650
1690
            def test_value(self):
1651
1691
                self.assertEqual('original', self.orig)
1652
1692
                self.assertEqual('modified', obj.test_attr)
1653
1693
 
1654
 
        test = Test('test_value')
1655
 
        test.run(unittest.TestResult())
 
1694
        self._run_successful_test(Test('test_value'))
1656
1695
        self.assertEqual('original', obj.test_attr)
1657
1696
 
 
1697
    def test_overrideAttr_with_no_existing_value_and_value(self):
 
1698
        # Do not define the test_attribute
 
1699
        obj = self # Make 'obj' visible to the embedded test
 
1700
        class Test(tests.TestCase):
 
1701
 
 
1702
            def setUp(self):
 
1703
                tests.TestCase.setUp(self)
 
1704
                self.orig = self.overrideAttr(obj, 'test_attr', new='modified')
 
1705
 
 
1706
            def test_value(self):
 
1707
                self.assertEqual(tests._unitialized_attr, self.orig)
 
1708
                self.assertEqual('modified', obj.test_attr)
 
1709
 
 
1710
        self._run_successful_test(Test('test_value'))
 
1711
        self.assertRaises(AttributeError, getattr, obj, 'test_attr')
 
1712
 
 
1713
    def test_overrideAttr_with_no_existing_value_and_no_value(self):
 
1714
        # Do not define the test_attribute
 
1715
        obj = self # Make 'obj' visible to the embedded test
 
1716
        class Test(tests.TestCase):
 
1717
 
 
1718
            def setUp(self):
 
1719
                tests.TestCase.setUp(self)
 
1720
                self.orig = self.overrideAttr(obj, 'test_attr')
 
1721
 
 
1722
            def test_value(self):
 
1723
                self.assertEqual(tests._unitialized_attr, self.orig)
 
1724
                self.assertRaises(AttributeError, getattr, obj, 'test_attr')
 
1725
 
 
1726
        self._run_successful_test(Test('test_value'))
 
1727
        self.assertRaises(AttributeError, getattr, obj, 'test_attr')
 
1728
 
 
1729
    def test_recordCalls(self):
 
1730
        from bzrlib.tests import test_selftest
 
1731
        calls = self.recordCalls(
 
1732
            test_selftest, '_add_numbers')
 
1733
        self.assertEqual(test_selftest._add_numbers(2, 10),
 
1734
            12)
 
1735
        self.assertEqual(calls, [((2, 10), {})])
 
1736
 
 
1737
 
 
1738
def _add_numbers(a, b):
 
1739
    return a + b
 
1740
 
 
1741
 
 
1742
class _MissingFeature(features.Feature):
 
1743
    def _probe(self):
 
1744
        return False
 
1745
missing_feature = _MissingFeature()
 
1746
 
 
1747
 
 
1748
def _get_test(name):
 
1749
    """Get an instance of a specific example test.
 
1750
 
 
1751
    We protect this in a function so that they don't auto-run in the test
 
1752
    suite.
 
1753
    """
 
1754
 
 
1755
    class ExampleTests(tests.TestCase):
 
1756
 
 
1757
        def test_fail(self):
 
1758
            mutter('this was a failing test')
 
1759
            self.fail('this test will fail')
 
1760
 
 
1761
        def test_error(self):
 
1762
            mutter('this test errored')
 
1763
            raise RuntimeError('gotcha')
 
1764
 
 
1765
        def test_missing_feature(self):
 
1766
            mutter('missing the feature')
 
1767
            self.requireFeature(missing_feature)
 
1768
 
 
1769
        def test_skip(self):
 
1770
            mutter('this test will be skipped')
 
1771
            raise tests.TestSkipped('reason')
 
1772
 
 
1773
        def test_success(self):
 
1774
            mutter('this test succeeds')
 
1775
 
 
1776
        def test_xfail(self):
 
1777
            mutter('test with expected failure')
 
1778
            self.knownFailure('this_fails')
 
1779
 
 
1780
        def test_unexpected_success(self):
 
1781
            mutter('test with unexpected success')
 
1782
            self.expectFailure('should_fail', lambda: None)
 
1783
 
 
1784
    return ExampleTests(name)
 
1785
 
 
1786
 
 
1787
class TestTestCaseLogDetails(tests.TestCase):
 
1788
 
 
1789
    def _run_test(self, test_name):
 
1790
        test = _get_test(test_name)
 
1791
        result = testtools.TestResult()
 
1792
        test.run(result)
 
1793
        return result
 
1794
 
 
1795
    def test_fail_has_log(self):
 
1796
        result = self._run_test('test_fail')
 
1797
        self.assertEqual(1, len(result.failures))
 
1798
        result_content = result.failures[0][1]
 
1799
        self.assertContainsRe(result_content,
 
1800
            '(?m)^(?:Text attachment: )?log(?:$|: )')
 
1801
        self.assertContainsRe(result_content, 'this was a failing test')
 
1802
 
 
1803
    def test_error_has_log(self):
 
1804
        result = self._run_test('test_error')
 
1805
        self.assertEqual(1, len(result.errors))
 
1806
        result_content = result.errors[0][1]
 
1807
        self.assertContainsRe(result_content,
 
1808
            '(?m)^(?:Text attachment: )?log(?:$|: )')
 
1809
        self.assertContainsRe(result_content, 'this test errored')
 
1810
 
 
1811
    def test_skip_has_no_log(self):
 
1812
        result = self._run_test('test_skip')
 
1813
        self.assertEqual(['reason'], result.skip_reasons.keys())
 
1814
        skips = result.skip_reasons['reason']
 
1815
        self.assertEqual(1, len(skips))
 
1816
        test = skips[0]
 
1817
        self.assertFalse('log' in test.getDetails())
 
1818
 
 
1819
    def test_missing_feature_has_no_log(self):
 
1820
        # testtools doesn't know about addNotSupported, so it just gets
 
1821
        # considered as a skip
 
1822
        result = self._run_test('test_missing_feature')
 
1823
        self.assertEqual([missing_feature], result.skip_reasons.keys())
 
1824
        skips = result.skip_reasons[missing_feature]
 
1825
        self.assertEqual(1, len(skips))
 
1826
        test = skips[0]
 
1827
        self.assertFalse('log' in test.getDetails())
 
1828
 
 
1829
    def test_xfail_has_no_log(self):
 
1830
        result = self._run_test('test_xfail')
 
1831
        self.assertEqual(1, len(result.expectedFailures))
 
1832
        result_content = result.expectedFailures[0][1]
 
1833
        self.assertNotContainsRe(result_content,
 
1834
            '(?m)^(?:Text attachment: )?log(?:$|: )')
 
1835
        self.assertNotContainsRe(result_content, 'test with expected failure')
 
1836
 
 
1837
    def test_unexpected_success_has_log(self):
 
1838
        result = self._run_test('test_unexpected_success')
 
1839
        self.assertEqual(1, len(result.unexpectedSuccesses))
 
1840
        # Inconsistency, unexpectedSuccesses is a list of tests,
 
1841
        # expectedFailures is a list of reasons?
 
1842
        test = result.unexpectedSuccesses[0]
 
1843
        details = test.getDetails()
 
1844
        self.assertTrue('log' in details)
 
1845
 
 
1846
 
 
1847
class TestTestCloning(tests.TestCase):
 
1848
    """Tests that test cloning of TestCases (as used by multiply_tests)."""
 
1849
 
 
1850
    def test_cloned_testcase_does_not_share_details(self):
 
1851
        """A TestCase cloned with clone_test does not share mutable attributes
 
1852
        such as details or cleanups.
 
1853
        """
 
1854
        class Test(tests.TestCase):
 
1855
            def test_foo(self):
 
1856
                self.addDetail('foo', Content('text/plain', lambda: 'foo'))
 
1857
        orig_test = Test('test_foo')
 
1858
        cloned_test = tests.clone_test(orig_test, orig_test.id() + '(cloned)')
 
1859
        orig_test.run(unittest.TestResult())
 
1860
        self.assertEqual('foo', orig_test.getDetails()['foo'].iter_bytes())
 
1861
        self.assertEqual(None, cloned_test.getDetails().get('foo'))
 
1862
 
 
1863
    def test_double_apply_scenario_preserves_first_scenario(self):
 
1864
        """Applying two levels of scenarios to a test preserves the attributes
 
1865
        added by both scenarios.
 
1866
        """
 
1867
        class Test(tests.TestCase):
 
1868
            def test_foo(self):
 
1869
                pass
 
1870
        test = Test('test_foo')
 
1871
        scenarios_x = [('x=1', {'x': 1}), ('x=2', {'x': 2})]
 
1872
        scenarios_y = [('y=1', {'y': 1}), ('y=2', {'y': 2})]
 
1873
        suite = tests.multiply_tests(test, scenarios_x, unittest.TestSuite())
 
1874
        suite = tests.multiply_tests(suite, scenarios_y, unittest.TestSuite())
 
1875
        all_tests = list(tests.iter_suite_tests(suite))
 
1876
        self.assertLength(4, all_tests)
 
1877
        all_xys = sorted((t.x, t.y) for t in all_tests)
 
1878
        self.assertEqual([(1, 1), (1, 2), (2, 1), (2, 2)], all_xys)
 
1879
 
1658
1880
 
1659
1881
# NB: Don't delete this; it's not actually from 0.11!
1660
1882
@deprecated_function(deprecated_in((0, 11, 0)))
1690
1912
        self.assertIsInstance(2, int)
1691
1913
        self.assertIsInstance(u'', basestring)
1692
1914
        e = self.assertRaises(AssertionError, self.assertIsInstance, None, int)
1693
 
        self.assertEquals(str(e),
 
1915
        self.assertEqual(str(e),
1694
1916
            "None is an instance of <type 'NoneType'> rather than <type 'int'>")
1695
1917
        self.assertRaises(AssertionError, self.assertIsInstance, 23.3, int)
1696
1918
        e = self.assertRaises(AssertionError,
1697
1919
            self.assertIsInstance, None, int, "it's just not")
1698
 
        self.assertEquals(str(e),
 
1920
        self.assertEqual(str(e),
1699
1921
            "None is an instance of <type 'NoneType'> rather than <type 'int'>"
1700
1922
            ": it's just not")
1701
1923
 
1706
1928
    def test_assertEqualDiff(self):
1707
1929
        e = self.assertRaises(AssertionError,
1708
1930
                              self.assertEqualDiff, '', '\n')
1709
 
        self.assertEquals(str(e),
 
1931
        self.assertEqual(str(e),
1710
1932
                          # Don't blink ! The '+' applies to the second string
1711
1933
                          'first string is missing a final newline.\n+ \n')
1712
1934
        e = self.assertRaises(AssertionError,
1713
1935
                              self.assertEqualDiff, '\n', '')
1714
 
        self.assertEquals(str(e),
 
1936
        self.assertEqual(str(e),
1715
1937
                          # Don't blink ! The '-' applies to the second string
1716
1938
                          'second string is missing a final newline.\n- \n')
1717
1939
 
1774
1996
            warnings.warn("this is your last warning")
1775
1997
            return a + b
1776
1998
        wlist, result = self.callCatchWarnings(meth, 1, 2)
1777
 
        self.assertEquals(3, result)
 
1999
        self.assertEqual(3, result)
1778
2000
        # would like just to compare them, but UserWarning doesn't implement
1779
2001
        # eq well
1780
2002
        w0, = wlist
1781
2003
        self.assertIsInstance(w0, UserWarning)
1782
 
        self.assertEquals("this is your last warning", str(w0))
 
2004
        self.assertEqual("this is your last warning", str(w0))
1783
2005
 
1784
2006
 
1785
2007
class TestConvenienceMakers(tests.TestCaseWithTransport):
1788
2010
    def test_make_branch_and_tree_with_format(self):
1789
2011
        # we should be able to supply a format to make_branch_and_tree
1790
2012
        self.make_branch_and_tree('a', format=bzrlib.bzrdir.BzrDirMetaFormat1())
1791
 
        self.make_branch_and_tree('b', format=bzrlib.bzrdir.BzrDirFormat6())
1792
 
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('a')._format,
 
2013
        self.assertIsInstance(bzrlib.controldir.ControlDir.open('a')._format,
1793
2014
                              bzrlib.bzrdir.BzrDirMetaFormat1)
1794
 
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('b')._format,
1795
 
                              bzrlib.bzrdir.BzrDirFormat6)
1796
2015
 
1797
2016
    def test_make_branch_and_memory_tree(self):
1798
2017
        # we should be able to get a new branch and a mutable tree from
1809
2028
        tree = self.make_branch_and_tree('t1')
1810
2029
        base = tree.bzrdir.root_transport.base
1811
2030
        self.assertStartsWith(base, 'file://')
1812
 
        self.assertEquals(tree.bzrdir.root_transport,
 
2031
        self.assertEqual(tree.bzrdir.root_transport,
1813
2032
                tree.branch.bzrdir.root_transport)
1814
 
        self.assertEquals(tree.bzrdir.root_transport,
 
2033
        self.assertEqual(tree.bzrdir.root_transport,
1815
2034
                tree.branch.repository.bzrdir.root_transport)
1816
2035
 
1817
2036
 
1818
 
class SelfTestHelper:
 
2037
class SelfTestHelper(object):
1819
2038
 
1820
2039
    def run_selftest(self, **kwargs):
1821
2040
        """Run selftest returning its output."""
1875
2094
        self.assertLength(2, output.readlines())
1876
2095
 
1877
2096
    def test_lsprof_tests(self):
1878
 
        self.requireFeature(test_lsprof.LSProfFeature)
1879
 
        calls = []
 
2097
        self.requireFeature(features.lsprof_feature)
 
2098
        results = []
1880
2099
        class Test(object):
1881
2100
            def __call__(test, result):
1882
2101
                test.run(result)
1883
2102
            def run(test, result):
1884
 
                self.assertIsInstance(result, tests.ForwardingResult)
1885
 
                calls.append("called")
 
2103
                results.append(result)
1886
2104
            def countTestCases(self):
1887
2105
                return 1
1888
2106
        self.run_selftest(test_suite_factory=Test, lsprof_tests=True)
1889
 
        self.assertLength(1, calls)
 
2107
        self.assertLength(1, results)
 
2108
        self.assertIsInstance(results.pop(), ExtendedToOriginalDecorator)
1890
2109
 
1891
2110
    def test_random(self):
1892
2111
        # test randomising by listing a number of tests.
1972
2191
            load_list='missing file name', list_only=True)
1973
2192
 
1974
2193
 
 
2194
class TestSubunitLogDetails(tests.TestCase, SelfTestHelper):
 
2195
 
 
2196
    _test_needs_features = [features.subunit]
 
2197
 
 
2198
    def run_subunit_stream(self, test_name):
 
2199
        from subunit import ProtocolTestCase
 
2200
        def factory():
 
2201
            return TestUtil.TestSuite([_get_test(test_name)])
 
2202
        stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
 
2203
            test_suite_factory=factory)
 
2204
        test = ProtocolTestCase(stream)
 
2205
        result = testtools.TestResult()
 
2206
        test.run(result)
 
2207
        content = stream.getvalue()
 
2208
        return content, result
 
2209
 
 
2210
    def test_fail_has_log(self):
 
2211
        content, result = self.run_subunit_stream('test_fail')
 
2212
        self.assertEqual(1, len(result.failures))
 
2213
        self.assertContainsRe(content, '(?m)^log$')
 
2214
        self.assertContainsRe(content, 'this test will fail')
 
2215
 
 
2216
    def test_error_has_log(self):
 
2217
        content, result = self.run_subunit_stream('test_error')
 
2218
        self.assertContainsRe(content, '(?m)^log$')
 
2219
        self.assertContainsRe(content, 'this test errored')
 
2220
 
 
2221
    def test_skip_has_no_log(self):
 
2222
        content, result = self.run_subunit_stream('test_skip')
 
2223
        self.assertNotContainsRe(content, '(?m)^log$')
 
2224
        self.assertNotContainsRe(content, 'this test will be skipped')
 
2225
        self.assertEqual(['reason'], result.skip_reasons.keys())
 
2226
        skips = result.skip_reasons['reason']
 
2227
        self.assertEqual(1, len(skips))
 
2228
        test = skips[0]
 
2229
        # RemotedTestCase doesn't preserve the "details"
 
2230
        ## self.assertFalse('log' in test.getDetails())
 
2231
 
 
2232
    def test_missing_feature_has_no_log(self):
 
2233
        content, result = self.run_subunit_stream('test_missing_feature')
 
2234
        self.assertNotContainsRe(content, '(?m)^log$')
 
2235
        self.assertNotContainsRe(content, 'missing the feature')
 
2236
        self.assertEqual(['_MissingFeature\n'], result.skip_reasons.keys())
 
2237
        skips = result.skip_reasons['_MissingFeature\n']
 
2238
        self.assertEqual(1, len(skips))
 
2239
        test = skips[0]
 
2240
        # RemotedTestCase doesn't preserve the "details"
 
2241
        ## self.assertFalse('log' in test.getDetails())
 
2242
 
 
2243
    def test_xfail_has_no_log(self):
 
2244
        content, result = self.run_subunit_stream('test_xfail')
 
2245
        self.assertNotContainsRe(content, '(?m)^log$')
 
2246
        self.assertNotContainsRe(content, 'test with expected failure')
 
2247
        self.assertEqual(1, len(result.expectedFailures))
 
2248
        result_content = result.expectedFailures[0][1]
 
2249
        self.assertNotContainsRe(result_content,
 
2250
            '(?m)^(?:Text attachment: )?log(?:$|: )')
 
2251
        self.assertNotContainsRe(result_content, 'test with expected failure')
 
2252
 
 
2253
    def test_unexpected_success_has_log(self):
 
2254
        content, result = self.run_subunit_stream('test_unexpected_success')
 
2255
        self.assertContainsRe(content, '(?m)^log$')
 
2256
        self.assertContainsRe(content, 'test with unexpected success')
 
2257
        # GZ 2011-05-18: Old versions of subunit treat unexpected success as a
 
2258
        #                success, if a min version check is added remove this
 
2259
        from subunit import TestProtocolClient as _Client
 
2260
        if _Client.addUnexpectedSuccess.im_func is _Client.addSuccess.im_func:
 
2261
            self.expectFailure('subunit treats "unexpectedSuccess"'
 
2262
                               ' as a plain success',
 
2263
                self.assertEqual, 1, len(result.unexpectedSuccesses))
 
2264
        self.assertEqual(1, len(result.unexpectedSuccesses))
 
2265
        test = result.unexpectedSuccesses[0]
 
2266
        # RemotedTestCase doesn't preserve the "details"
 
2267
        ## self.assertTrue('log' in test.getDetails())
 
2268
 
 
2269
    def test_success_has_no_log(self):
 
2270
        content, result = self.run_subunit_stream('test_success')
 
2271
        self.assertEqual(1, result.testsRun)
 
2272
        self.assertNotContainsRe(content, '(?m)^log$')
 
2273
        self.assertNotContainsRe(content, 'this test succeeds')
 
2274
 
 
2275
 
1975
2276
class TestRunBzr(tests.TestCase):
1976
2277
 
1977
2278
    out = ''
1999
2300
        self.assertEqual(['rocks'], self.argv)
2000
2301
        self.assertEqual(34, self.retcode)
2001
2302
        self.assertEqual('It sure does!\n', out)
2002
 
        self.assertEquals(out, self.out)
 
2303
        self.assertEqual(out, self.out)
2003
2304
        self.assertEqual('', err)
2004
 
        self.assertEquals(err, self.err)
 
2305
        self.assertEqual(err, self.err)
2005
2306
 
2006
2307
    def test_run_bzr_error_regexes(self):
2007
2308
        self.out = ''
2100
2401
        # stdout and stderr of the invoked run_bzr
2101
2402
        current_factory = bzrlib.ui.ui_factory
2102
2403
        self.run_bzr(['foo'])
2103
 
        self.failIf(current_factory is self.factory)
 
2404
        self.assertFalse(current_factory is self.factory)
2104
2405
        self.assertNotEqual(sys.stdout, self.factory.stdout)
2105
2406
        self.assertNotEqual(sys.stderr, self.factory.stderr)
2106
2407
        self.assertEqual('foo\n', self.factory.stdout.getvalue())
2147
2448
    """Base class for tests testing how we might run bzr."""
2148
2449
 
2149
2450
    def setUp(self):
2150
 
        tests.TestCaseWithTransport.setUp(self)
 
2451
        super(TestWithFakedStartBzrSubprocess, self).setUp()
2151
2452
        self.subprocess_calls = []
2152
2453
 
2153
2454
    def start_bzr_subprocess(self, process_args, env_changes=None,
2263
2564
 
2264
2565
 
2265
2566
class TestStartBzrSubProcess(tests.TestCase):
 
2567
    """Stub test start_bzr_subprocess."""
2266
2568
 
2267
 
    def check_popen_state(self):
2268
 
        """Replace to make assertions when popen is called."""
 
2569
    def _subprocess_log_cleanup(self):
 
2570
        """Inhibits the base version as we don't produce a log file."""
2269
2571
 
2270
2572
    def _popen(self, *args, **kwargs):
2271
 
        """Record the command that is run, so that we can ensure it is correct"""
 
2573
        """Override the base version to record the command that is run.
 
2574
 
 
2575
        From there we can ensure it is correct without spawning a real process.
 
2576
        """
2272
2577
        self.check_popen_state()
2273
2578
        self._popen_args = args
2274
2579
        self._popen_kwargs = kwargs
2275
2580
        raise _DontSpawnProcess()
2276
2581
 
 
2582
    def check_popen_state(self):
 
2583
        """Replace to make assertions when popen is called."""
 
2584
 
2277
2585
    def test_run_bzr_subprocess_no_plugins(self):
2278
2586
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [])
2279
2587
        command = self._popen_args[0]
2283
2591
 
2284
2592
    def test_allow_plugins(self):
2285
2593
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2286
 
            allow_plugins=True)
 
2594
                          allow_plugins=True)
2287
2595
        command = self._popen_args[0]
2288
2596
        self.assertEqual([], command[2:])
2289
2597
 
2290
2598
    def test_set_env(self):
2291
 
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
 
2599
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2292
2600
        # set in the child
2293
2601
        def check_environment():
2294
2602
            self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2295
2603
        self.check_popen_state = check_environment
2296
2604
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2297
 
            env_changes={'EXISTANT_ENV_VAR':'set variable'})
 
2605
                          env_changes={'EXISTANT_ENV_VAR':'set variable'})
2298
2606
        # not set in theparent
2299
2607
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2300
2608
 
2301
2609
    def test_run_bzr_subprocess_env_del(self):
2302
2610
        """run_bzr_subprocess can remove environment variables too."""
2303
 
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
 
2611
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2304
2612
        def check_environment():
2305
2613
            self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2306
2614
        os.environ['EXISTANT_ENV_VAR'] = 'set variable'
2307
2615
        self.check_popen_state = check_environment
2308
2616
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2309
 
            env_changes={'EXISTANT_ENV_VAR':None})
 
2617
                          env_changes={'EXISTANT_ENV_VAR':None})
2310
2618
        # Still set in parent
2311
2619
        self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2312
2620
        del os.environ['EXISTANT_ENV_VAR']
2313
2621
 
2314
2622
    def test_env_del_missing(self):
2315
 
        self.failIf('NON_EXISTANT_ENV_VAR' in os.environ)
 
2623
        self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2316
2624
        def check_environment():
2317
2625
            self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2318
2626
        self.check_popen_state = check_environment
2319
2627
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2320
 
            env_changes={'NON_EXISTANT_ENV_VAR':None})
 
2628
                          env_changes={'NON_EXISTANT_ENV_VAR':None})
2321
2629
 
2322
2630
    def test_working_dir(self):
2323
2631
        """Test that we can specify the working dir for the child"""
2326
2634
        chdirs = []
2327
2635
        def chdir(path):
2328
2636
            chdirs.append(path)
2329
 
        os.chdir = chdir
2330
 
        try:
2331
 
            def getcwd():
2332
 
                return 'current'
2333
 
            osutils.getcwd = getcwd
2334
 
            try:
2335
 
                self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2336
 
                    working_dir='foo')
2337
 
            finally:
2338
 
                osutils.getcwd = orig_getcwd
2339
 
        finally:
2340
 
            os.chdir = orig_chdir
 
2637
        self.overrideAttr(os, 'chdir', chdir)
 
2638
        def getcwd():
 
2639
            return 'current'
 
2640
        self.overrideAttr(osutils, 'getcwd', getcwd)
 
2641
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
 
2642
                          working_dir='foo')
2341
2643
        self.assertEqual(['foo', 'current'], chdirs)
2342
2644
 
2343
2645
    def test_get_bzr_path_with_cwd_bzrlib(self):
2363
2665
        self.assertEqual('bzr: interrupted\n', result[1])
2364
2666
 
2365
2667
 
2366
 
class TestFeature(tests.TestCase):
2367
 
 
2368
 
    def test_caching(self):
2369
 
        """Feature._probe is called by the feature at most once."""
2370
 
        class InstrumentedFeature(tests.Feature):
2371
 
            def __init__(self):
2372
 
                super(InstrumentedFeature, self).__init__()
2373
 
                self.calls = []
2374
 
            def _probe(self):
2375
 
                self.calls.append('_probe')
2376
 
                return False
2377
 
        feature = InstrumentedFeature()
2378
 
        feature.available()
2379
 
        self.assertEqual(['_probe'], feature.calls)
2380
 
        feature.available()
2381
 
        self.assertEqual(['_probe'], feature.calls)
2382
 
 
2383
 
    def test_named_str(self):
2384
 
        """Feature.__str__ should thunk to feature_name()."""
2385
 
        class NamedFeature(tests.Feature):
2386
 
            def feature_name(self):
2387
 
                return 'symlinks'
2388
 
        feature = NamedFeature()
2389
 
        self.assertEqual('symlinks', str(feature))
2390
 
 
2391
 
    def test_default_str(self):
2392
 
        """Feature.__str__ should default to __class__.__name__."""
2393
 
        class NamedFeature(tests.Feature):
2394
 
            pass
2395
 
        feature = NamedFeature()
2396
 
        self.assertEqual('NamedFeature', str(feature))
2397
 
 
2398
 
 
2399
 
class TestUnavailableFeature(tests.TestCase):
2400
 
 
2401
 
    def test_access_feature(self):
2402
 
        feature = tests.Feature()
2403
 
        exception = tests.UnavailableFeature(feature)
2404
 
        self.assertIs(feature, exception.args[0])
2405
 
 
2406
 
 
2407
 
simple_thunk_feature = tests._CompatabilityThunkFeature(
2408
 
    deprecated_in((2, 1, 0)),
2409
 
    'bzrlib.tests.test_selftest',
2410
 
    'simple_thunk_feature','UnicodeFilename',
2411
 
    replacement_module='bzrlib.tests'
2412
 
    )
2413
 
 
2414
 
class Test_CompatibilityFeature(tests.TestCase):
2415
 
 
2416
 
    def test_does_thunk(self):
2417
 
        res = self.callDeprecated(
2418
 
            ['bzrlib.tests.test_selftest.simple_thunk_feature was deprecated'
2419
 
             ' in version 2.1.0. Use bzrlib.tests.UnicodeFilename instead.'],
2420
 
            simple_thunk_feature.available)
2421
 
        self.assertEqual(tests.UnicodeFilename.available(), res)
2422
 
 
2423
 
 
2424
 
class TestModuleAvailableFeature(tests.TestCase):
2425
 
 
2426
 
    def test_available_module(self):
2427
 
        feature = tests.ModuleAvailableFeature('bzrlib.tests')
2428
 
        self.assertEqual('bzrlib.tests', feature.module_name)
2429
 
        self.assertEqual('bzrlib.tests', str(feature))
2430
 
        self.assertTrue(feature.available())
2431
 
        self.assertIs(tests, feature.module)
2432
 
 
2433
 
    def test_unavailable_module(self):
2434
 
        feature = tests.ModuleAvailableFeature('bzrlib.no_such_module_exists')
2435
 
        self.assertEqual('bzrlib.no_such_module_exists', str(feature))
2436
 
        self.assertFalse(feature.available())
2437
 
        self.assertIs(None, feature.module)
2438
 
 
2439
 
 
2440
2668
class TestSelftestFiltering(tests.TestCase):
2441
2669
 
2442
2670
    def setUp(self):
2443
 
        tests.TestCase.setUp(self)
 
2671
        super(TestSelftestFiltering, self).setUp()
2444
2672
        self.suite = TestUtil.TestSuite()
2445
2673
        self.loader = TestUtil.TestLoader()
2446
2674
        self.suite.addTest(self.loader.loadTestsFromModule(
2593
2821
        self.assertEqual(remaining_names, _test_ids(split_suite[1]))
2594
2822
 
2595
2823
 
2596
 
class TestCheckInventoryShape(tests.TestCaseWithTransport):
 
2824
class TestCheckTreeShape(tests.TestCaseWithTransport):
2597
2825
 
2598
 
    def test_check_inventory_shape(self):
 
2826
    def test_check_tree_shape(self):
2599
2827
        files = ['a', 'b/', 'b/c']
2600
2828
        tree = self.make_branch_and_tree('.')
2601
2829
        self.build_tree(files)
2602
2830
        tree.add(files)
2603
2831
        tree.lock_read()
2604
2832
        try:
2605
 
            self.check_inventory_shape(tree.inventory, files)
 
2833
            self.check_tree_shape(tree, files)
2606
2834
        finally:
2607
2835
            tree.unlock()
2608
2836
 
2619
2847
            self.run_bzr, ['assert-fail'])
2620
2848
        # make sure we got the real thing, not an error from somewhere else in
2621
2849
        # the test framework
2622
 
        self.assertEquals('always fails', str(e))
 
2850
        self.assertEqual('always fails', str(e))
2623
2851
        # check that there's no traceback in the test log
2624
2852
        self.assertNotContainsRe(self.get_log(), r'Traceback')
2625
2853
 
2674
2902
    def test_load_tests_from_module_name_smoke_test(self):
2675
2903
        loader = TestUtil.TestLoader()
2676
2904
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2677
 
        self.assertEquals(['bzrlib.tests.test_sampler.DemoTest.test_nothing'],
 
2905
        self.assertEqual(['bzrlib.tests.test_sampler.DemoTest.test_nothing'],
2678
2906
                          _test_ids(suite))
2679
2907
 
2680
2908
    def test_load_tests_from_module_name_with_bogus_module_name(self):
2709
2937
 
2710
2938
    def test_empty_list(self):
2711
2939
        id_list = self._create_id_list([])
2712
 
        self.assertEquals({}, id_list.tests)
2713
 
        self.assertEquals({}, id_list.modules)
 
2940
        self.assertEqual({}, id_list.tests)
 
2941
        self.assertEqual({}, id_list.modules)
2714
2942
 
2715
2943
    def test_valid_list(self):
2716
2944
        id_list = self._create_id_list(
2743
2971
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing',
2744
2972
                     'bogus']
2745
2973
        not_found, duplicates = tests.suite_matches_id_list(suite, test_list)
2746
 
        self.assertEquals(['bogus'], not_found)
2747
 
        self.assertEquals([], duplicates)
 
2974
        self.assertEqual(['bogus'], not_found)
 
2975
        self.assertEqual([], duplicates)
2748
2976
 
2749
2977
    def test_suite_matches_id_list_with_duplicates(self):
2750
2978
        loader = TestUtil.TestLoader()
2757
2985
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing',]
2758
2986
        not_found, duplicates = tests.suite_matches_id_list(
2759
2987
            dupes, test_list)
2760
 
        self.assertEquals([], not_found)
2761
 
        self.assertEquals(['bzrlib.tests.test_sampler.DemoTest.test_nothing'],
 
2988
        self.assertEqual([], not_found)
 
2989
        self.assertEqual(['bzrlib.tests.test_sampler.DemoTest.test_nothing'],
2762
2990
                          duplicates)
2763
2991
 
2764
2992
 
2837
3065
        suite = tests.test_suite(test_list,
2838
3066
                                 ['bzrlib.tests.test_selftest.TestTestSuite'])
2839
3067
        # test_test_suite_list_and_start is not included 
2840
 
        self.assertEquals(test_list, _test_ids(suite))
 
3068
        self.assertEqual(test_list, _test_ids(suite))
2841
3069
 
2842
3070
 
2843
3071
class TestLoadTestIdList(tests.TestCaseInTempDir):
2856
3084
        self._create_test_list_file(test_list_fname,
2857
3085
                                    'mod1.cl1.meth1\nmod2.cl2.meth2\n')
2858
3086
        tlist = tests.load_test_id_list(test_list_fname)
2859
 
        self.assertEquals(2, len(tlist))
2860
 
        self.assertEquals('mod1.cl1.meth1', tlist[0])
2861
 
        self.assertEquals('mod2.cl2.meth2', tlist[1])
 
3087
        self.assertEqual(2, len(tlist))
 
3088
        self.assertEqual('mod1.cl1.meth1', tlist[0])
 
3089
        self.assertEqual('mod2.cl2.meth2', tlist[1])
2862
3090
 
2863
3091
    def test_load_dirty_file(self):
2864
3092
        test_list_fname = 'test.list'
2866
3094
                                    '  mod1.cl1.meth1\n\nmod2.cl2.meth2  \n'
2867
3095
                                    'bar baz\n')
2868
3096
        tlist = tests.load_test_id_list(test_list_fname)
2869
 
        self.assertEquals(4, len(tlist))
2870
 
        self.assertEquals('mod1.cl1.meth1', tlist[0])
2871
 
        self.assertEquals('', tlist[1])
2872
 
        self.assertEquals('mod2.cl2.meth2', tlist[2])
2873
 
        self.assertEquals('bar baz', tlist[3])
 
3097
        self.assertEqual(4, len(tlist))
 
3098
        self.assertEqual('mod1.cl1.meth1', tlist[0])
 
3099
        self.assertEqual('', tlist[1])
 
3100
        self.assertEqual('mod2.cl2.meth2', tlist[2])
 
3101
        self.assertEqual('bar baz', tlist[3])
2874
3102
 
2875
3103
 
2876
3104
class TestFilteredByModuleTestLoader(tests.TestCase):
2884
3112
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing']
2885
3113
        loader = self._create_loader(test_list)
2886
3114
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2887
 
        self.assertEquals(test_list, _test_ids(suite))
 
3115
        self.assertEqual(test_list, _test_ids(suite))
2888
3116
 
2889
3117
    def test_exclude_tests(self):
2890
3118
        test_list = ['bogus']
2891
3119
        loader = self._create_loader(test_list)
2892
3120
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2893
 
        self.assertEquals([], _test_ids(suite))
 
3121
        self.assertEqual([], _test_ids(suite))
2894
3122
 
2895
3123
 
2896
3124
class TestFilteredByNameStartTestLoader(tests.TestCase):
2906
3134
        loader = self._create_loader('bzrlib.tests.test_samp')
2907
3135
 
2908
3136
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2909
 
        self.assertEquals(test_list, _test_ids(suite))
 
3137
        self.assertEqual(test_list, _test_ids(suite))
2910
3138
 
2911
3139
    def test_load_tests_inside_module(self):
2912
3140
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing']
2913
3141
        loader = self._create_loader('bzrlib.tests.test_sampler.Demo')
2914
3142
 
2915
3143
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2916
 
        self.assertEquals(test_list, _test_ids(suite))
 
3144
        self.assertEqual(test_list, _test_ids(suite))
2917
3145
 
2918
3146
    def test_exclude_tests(self):
2919
3147
        test_list = ['bogus']
2920
3148
        loader = self._create_loader('bogus')
2921
3149
 
2922
3150
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2923
 
        self.assertEquals([], _test_ids(suite))
 
3151
        self.assertEqual([], _test_ids(suite))
2924
3152
 
2925
3153
 
2926
3154
class TestTestPrefixRegistry(tests.TestCase):
2932
3160
    def test_register_new_prefix(self):
2933
3161
        tpr = self._get_registry()
2934
3162
        tpr.register('foo', 'fff.ooo.ooo')
2935
 
        self.assertEquals('fff.ooo.ooo', tpr.get('foo'))
 
3163
        self.assertEqual('fff.ooo.ooo', tpr.get('foo'))
2936
3164
 
2937
3165
    def test_register_existing_prefix(self):
2938
3166
        tpr = self._get_registry()
2939
3167
        tpr.register('bar', 'bbb.aaa.rrr')
2940
3168
        tpr.register('bar', 'bBB.aAA.rRR')
2941
 
        self.assertEquals('bbb.aaa.rrr', tpr.get('bar'))
 
3169
        self.assertEqual('bbb.aaa.rrr', tpr.get('bar'))
2942
3170
        self.assertThat(self.get_log(),
2943
 
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR", ELLIPSIS))
 
3171
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR",
 
3172
                           doctest.ELLIPSIS))
2944
3173
 
2945
3174
    def test_get_unknown_prefix(self):
2946
3175
        tpr = self._get_registry()
2949
3178
    def test_resolve_prefix(self):
2950
3179
        tpr = self._get_registry()
2951
3180
        tpr.register('bar', 'bb.aa.rr')
2952
 
        self.assertEquals('bb.aa.rr', tpr.resolve_alias('bar'))
 
3181
        self.assertEqual('bb.aa.rr', tpr.resolve_alias('bar'))
2953
3182
 
2954
3183
    def test_resolve_unknown_alias(self):
2955
3184
        tpr = self._get_registry()
2958
3187
 
2959
3188
    def test_predefined_prefixes(self):
2960
3189
        tpr = tests.test_prefix_alias_registry
2961
 
        self.assertEquals('bzrlib', tpr.resolve_alias('bzrlib'))
2962
 
        self.assertEquals('bzrlib.doc', tpr.resolve_alias('bd'))
2963
 
        self.assertEquals('bzrlib.utils', tpr.resolve_alias('bu'))
2964
 
        self.assertEquals('bzrlib.tests', tpr.resolve_alias('bt'))
2965
 
        self.assertEquals('bzrlib.tests.blackbox', tpr.resolve_alias('bb'))
2966
 
        self.assertEquals('bzrlib.plugins', tpr.resolve_alias('bp'))
 
3190
        self.assertEqual('bzrlib', tpr.resolve_alias('bzrlib'))
 
3191
        self.assertEqual('bzrlib.doc', tpr.resolve_alias('bd'))
 
3192
        self.assertEqual('bzrlib.utils', tpr.resolve_alias('bu'))
 
3193
        self.assertEqual('bzrlib.tests', tpr.resolve_alias('bt'))
 
3194
        self.assertEqual('bzrlib.tests.blackbox', tpr.resolve_alias('bb'))
 
3195
        self.assertEqual('bzrlib.plugins', tpr.resolve_alias('bp'))
 
3196
 
 
3197
 
 
3198
class TestThreadLeakDetection(tests.TestCase):
 
3199
    """Ensure when tests leak threads we detect and report it"""
 
3200
 
 
3201
    class LeakRecordingResult(tests.ExtendedTestResult):
 
3202
        def __init__(self):
 
3203
            tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
 
3204
            self.leaks = []
 
3205
        def _report_thread_leak(self, test, leaks, alive):
 
3206
            self.leaks.append((test, leaks))
 
3207
 
 
3208
    def test_testcase_without_addCleanups(self):
 
3209
        """Check old TestCase instances don't break with leak detection"""
 
3210
        class Test(unittest.TestCase):
 
3211
            def runTest(self):
 
3212
                pass
 
3213
        result = self.LeakRecordingResult()
 
3214
        test = Test()
 
3215
        result.startTestRun()
 
3216
        test.run(result)
 
3217
        result.stopTestRun()
 
3218
        self.assertEqual(result._tests_leaking_threads_count, 0)
 
3219
        self.assertEqual(result.leaks, [])
 
3220
        
 
3221
    def test_thread_leak(self):
 
3222
        """Ensure a thread that outlives the running of a test is reported
 
3223
 
 
3224
        Uses a thread that blocks on an event, and is started by the inner
 
3225
        test case. As the thread outlives the inner case's run, it should be
 
3226
        detected as a leak, but the event is then set so that the thread can
 
3227
        be safely joined in cleanup so it's not leaked for real.
 
3228
        """
 
3229
        event = threading.Event()
 
3230
        thread = threading.Thread(name="Leaker", target=event.wait)
 
3231
        class Test(tests.TestCase):
 
3232
            def test_leak(self):
 
3233
                thread.start()
 
3234
        result = self.LeakRecordingResult()
 
3235
        test = Test("test_leak")
 
3236
        self.addCleanup(thread.join)
 
3237
        self.addCleanup(event.set)
 
3238
        result.startTestRun()
 
3239
        test.run(result)
 
3240
        result.stopTestRun()
 
3241
        self.assertEqual(result._tests_leaking_threads_count, 1)
 
3242
        self.assertEqual(result._first_thread_leaker_id, test.id())
 
3243
        self.assertEqual(result.leaks, [(test, set([thread]))])
 
3244
        self.assertContainsString(result.stream.getvalue(), "leaking threads")
 
3245
 
 
3246
    def test_multiple_leaks(self):
 
3247
        """Check multiple leaks are blamed on the test cases at fault
 
3248
 
 
3249
        Same concept as the previous test, but has one inner test method that
 
3250
        leaks two threads, and one that doesn't leak at all.
 
3251
        """
 
3252
        event = threading.Event()
 
3253
        thread_a = threading.Thread(name="LeakerA", target=event.wait)
 
3254
        thread_b = threading.Thread(name="LeakerB", target=event.wait)
 
3255
        thread_c = threading.Thread(name="LeakerC", target=event.wait)
 
3256
        class Test(tests.TestCase):
 
3257
            def test_first_leak(self):
 
3258
                thread_b.start()
 
3259
            def test_second_no_leak(self):
 
3260
                pass
 
3261
            def test_third_leak(self):
 
3262
                thread_c.start()
 
3263
                thread_a.start()
 
3264
        result = self.LeakRecordingResult()
 
3265
        first_test = Test("test_first_leak")
 
3266
        third_test = Test("test_third_leak")
 
3267
        self.addCleanup(thread_a.join)
 
3268
        self.addCleanup(thread_b.join)
 
3269
        self.addCleanup(thread_c.join)
 
3270
        self.addCleanup(event.set)
 
3271
        result.startTestRun()
 
3272
        unittest.TestSuite(
 
3273
            [first_test, Test("test_second_no_leak"), third_test]
 
3274
            ).run(result)
 
3275
        result.stopTestRun()
 
3276
        self.assertEqual(result._tests_leaking_threads_count, 2)
 
3277
        self.assertEqual(result._first_thread_leaker_id, first_test.id())
 
3278
        self.assertEqual(result.leaks, [
 
3279
            (first_test, set([thread_b])),
 
3280
            (third_test, set([thread_a, thread_c]))])
 
3281
        self.assertContainsString(result.stream.getvalue(), "leaking threads")
 
3282
 
 
3283
 
 
3284
class TestPostMortemDebugging(tests.TestCase):
 
3285
    """Check post mortem debugging works when tests fail or error"""
 
3286
 
 
3287
    class TracebackRecordingResult(tests.ExtendedTestResult):
 
3288
        def __init__(self):
 
3289
            tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
 
3290
            self.postcode = None
 
3291
        def _post_mortem(self, tb=None):
 
3292
            """Record the code object at the end of the current traceback"""
 
3293
            tb = tb or sys.exc_info()[2]
 
3294
            if tb is not None:
 
3295
                next = tb.tb_next
 
3296
                while next is not None:
 
3297
                    tb = next
 
3298
                    next = next.tb_next
 
3299
                self.postcode = tb.tb_frame.f_code
 
3300
        def report_error(self, test, err):
 
3301
            pass
 
3302
        def report_failure(self, test, err):
 
3303
            pass
 
3304
 
 
3305
    def test_location_unittest_error(self):
 
3306
        """Needs right post mortem traceback with erroring unittest case"""
 
3307
        class Test(unittest.TestCase):
 
3308
            def runTest(self):
 
3309
                raise RuntimeError
 
3310
        result = self.TracebackRecordingResult()
 
3311
        Test().run(result)
 
3312
        self.assertEqual(result.postcode, Test.runTest.func_code)
 
3313
 
 
3314
    def test_location_unittest_failure(self):
 
3315
        """Needs right post mortem traceback with failing unittest case"""
 
3316
        class Test(unittest.TestCase):
 
3317
            def runTest(self):
 
3318
                raise self.failureException
 
3319
        result = self.TracebackRecordingResult()
 
3320
        Test().run(result)
 
3321
        self.assertEqual(result.postcode, Test.runTest.func_code)
 
3322
 
 
3323
    def test_location_bt_error(self):
 
3324
        """Needs right post mortem traceback with erroring bzrlib.tests case"""
 
3325
        class Test(tests.TestCase):
 
3326
            def test_error(self):
 
3327
                raise RuntimeError
 
3328
        result = self.TracebackRecordingResult()
 
3329
        Test("test_error").run(result)
 
3330
        self.assertEqual(result.postcode, Test.test_error.func_code)
 
3331
 
 
3332
    def test_location_bt_failure(self):
 
3333
        """Needs right post mortem traceback with failing bzrlib.tests case"""
 
3334
        class Test(tests.TestCase):
 
3335
            def test_failure(self):
 
3336
                raise self.failureException
 
3337
        result = self.TracebackRecordingResult()
 
3338
        Test("test_failure").run(result)
 
3339
        self.assertEqual(result.postcode, Test.test_failure.func_code)
 
3340
 
 
3341
    def test_env_var_triggers_post_mortem(self):
 
3342
        """Check pdb.post_mortem is called iff BZR_TEST_PDB is set"""
 
3343
        import pdb
 
3344
        result = tests.ExtendedTestResult(StringIO(), 0, 1)
 
3345
        post_mortem_calls = []
 
3346
        self.overrideAttr(pdb, "post_mortem", post_mortem_calls.append)
 
3347
        self.overrideEnv('BZR_TEST_PDB', None)
 
3348
        result._post_mortem(1)
 
3349
        self.overrideEnv('BZR_TEST_PDB', 'on')
 
3350
        result._post_mortem(2)
 
3351
        self.assertEqual([2], post_mortem_calls)
2967
3352
 
2968
3353
 
2969
3354
class TestRunSuite(tests.TestCase):
2982
3367
                                                self.verbosity)
2983
3368
        tests.run_suite(suite, runner_class=MyRunner, stream=StringIO())
2984
3369
        self.assertLength(1, calls)
 
3370
 
 
3371
 
 
3372
class _Selftest(object):
 
3373
    """Mixin for tests needing full selftest output"""
 
3374
 
 
3375
    def _inject_stream_into_subunit(self, stream):
 
3376
        """To be overridden by subclasses that run tests out of process"""
 
3377
 
 
3378
    def _run_selftest(self, **kwargs):
 
3379
        sio = StringIO()
 
3380
        self._inject_stream_into_subunit(sio)
 
3381
        tests.selftest(stream=sio, stop_on_failure=False, **kwargs)
 
3382
        return sio.getvalue()
 
3383
 
 
3384
 
 
3385
class _ForkedSelftest(_Selftest):
 
3386
    """Mixin for tests needing full selftest output with forked children"""
 
3387
 
 
3388
    _test_needs_features = [features.subunit]
 
3389
 
 
3390
    def _inject_stream_into_subunit(self, stream):
 
3391
        """Monkey-patch subunit so the extra output goes to stream not stdout
 
3392
 
 
3393
        Some APIs need rewriting so this kind of bogus hackery can be replaced
 
3394
        by passing the stream param from run_tests down into ProtocolTestCase.
 
3395
        """
 
3396
        from subunit import ProtocolTestCase
 
3397
        _original_init = ProtocolTestCase.__init__
 
3398
        def _init_with_passthrough(self, *args, **kwargs):
 
3399
            _original_init(self, *args, **kwargs)
 
3400
            self._passthrough = stream
 
3401
        self.overrideAttr(ProtocolTestCase, "__init__", _init_with_passthrough)
 
3402
 
 
3403
    def _run_selftest(self, **kwargs):
 
3404
        # GZ 2011-05-26: Add a PosixSystem feature so this check can go away
 
3405
        if getattr(os, "fork", None) is None:
 
3406
            raise tests.TestNotApplicable("Platform doesn't support forking")
 
3407
        # Make sure the fork code is actually invoked by claiming two cores
 
3408
        self.overrideAttr(osutils, "local_concurrency", lambda: 2)
 
3409
        kwargs.setdefault("suite_decorators", []).append(tests.fork_decorator)
 
3410
        return super(_ForkedSelftest, self)._run_selftest(**kwargs)
 
3411
 
 
3412
 
 
3413
class TestParallelFork(_ForkedSelftest, tests.TestCase):
 
3414
    """Check operation of --parallel=fork selftest option"""
 
3415
 
 
3416
    def test_error_in_child_during_fork(self):
 
3417
        """Error in a forked child during test setup should get reported"""
 
3418
        class Test(tests.TestCase):
 
3419
            def testMethod(self):
 
3420
                pass
 
3421
        # We don't care what, just break something that a child will run
 
3422
        self.overrideAttr(tests, "workaround_zealous_crypto_random", None)
 
3423
        out = self._run_selftest(test_suite_factory=Test)
 
3424
        # Lines from the tracebacks of the two child processes may be mixed
 
3425
        # together due to the way subunit parses and forwards the streams,
 
3426
        # so permit extra lines between each part of the error output.
 
3427
        self.assertContainsRe(out,
 
3428
            "Traceback.*:\n"
 
3429
            "(?:.*\n)*"
 
3430
            ".+ in fork_for_tests\n"
 
3431
            "(?:.*\n)*"
 
3432
            "\s*workaround_zealous_crypto_random\(\)\n"
 
3433
            "(?:.*\n)*"
 
3434
            "TypeError:")
 
3435
 
 
3436
 
 
3437
class TestUncollectedWarnings(_Selftest, tests.TestCase):
 
3438
    """Check a test case still alive after being run emits a warning"""
 
3439
 
 
3440
    class Test(tests.TestCase):
 
3441
        def test_pass(self):
 
3442
            pass
 
3443
        def test_self_ref(self):
 
3444
            self.also_self = self.test_self_ref
 
3445
        def test_skip(self):
 
3446
            self.skip("Don't need")
 
3447
 
 
3448
    def _get_suite(self):
 
3449
        return TestUtil.TestSuite([
 
3450
            self.Test("test_pass"),
 
3451
            self.Test("test_self_ref"),
 
3452
            self.Test("test_skip"),
 
3453
            ])
 
3454
 
 
3455
    def _run_selftest_with_suite(self, **kwargs):
 
3456
        old_flags = tests.selftest_debug_flags
 
3457
        tests.selftest_debug_flags = old_flags.union(["uncollected_cases"])
 
3458
        gc_on = gc.isenabled()
 
3459
        if gc_on:
 
3460
            gc.disable()
 
3461
        try:
 
3462
            output = self._run_selftest(test_suite_factory=self._get_suite,
 
3463
                **kwargs)
 
3464
        finally:
 
3465
            if gc_on:
 
3466
                gc.enable()
 
3467
            tests.selftest_debug_flags = old_flags
 
3468
        self.assertNotContainsRe(output, "Uncollected test case.*test_pass")
 
3469
        self.assertContainsRe(output, "Uncollected test case.*test_self_ref")
 
3470
        return output
 
3471
 
 
3472
    def test_testsuite(self):
 
3473
        self._run_selftest_with_suite()
 
3474
 
 
3475
    def test_pattern(self):
 
3476
        out = self._run_selftest_with_suite(pattern="test_(?:pass|self_ref)$")
 
3477
        self.assertNotContainsRe(out, "test_skip")
 
3478
 
 
3479
    def test_exclude_pattern(self):
 
3480
        out = self._run_selftest_with_suite(exclude_pattern="test_skip$")
 
3481
        self.assertNotContainsRe(out, "test_skip")
 
3482
 
 
3483
    def test_random_seed(self):
 
3484
        self._run_selftest_with_suite(random_seed="now")
 
3485
 
 
3486
    def test_matching_tests_first(self):
 
3487
        self._run_selftest_with_suite(matching_tests_first=True,
 
3488
            pattern="test_self_ref$")
 
3489
 
 
3490
    def test_starting_with_and_exclude(self):
 
3491
        out = self._run_selftest_with_suite(starting_with=["bt."],
 
3492
            exclude_pattern="test_skip$")
 
3493
        self.assertNotContainsRe(out, "test_skip")
 
3494
 
 
3495
    def test_additonal_decorator(self):
 
3496
        out = self._run_selftest_with_suite(
 
3497
            suite_decorators=[tests.TestDecorator])
 
3498
 
 
3499
 
 
3500
class TestUncollectedWarningsSubunit(TestUncollectedWarnings):
 
3501
    """Check warnings from tests staying alive are emitted with subunit"""
 
3502
 
 
3503
    _test_needs_features = [features.subunit]
 
3504
 
 
3505
    def _run_selftest_with_suite(self, **kwargs):
 
3506
        return TestUncollectedWarnings._run_selftest_with_suite(self,
 
3507
            runner_class=tests.SubUnitBzrRunner, **kwargs)
 
3508
 
 
3509
 
 
3510
class TestUncollectedWarningsForked(_ForkedSelftest, TestUncollectedWarnings):
 
3511
    """Check warnings from tests staying alive are emitted when forking"""
 
3512
 
 
3513
 
 
3514
class TestEnvironHandling(tests.TestCase):
 
3515
 
 
3516
    def test_overrideEnv_None_called_twice_doesnt_leak(self):
 
3517
        self.assertFalse('MYVAR' in os.environ)
 
3518
        self.overrideEnv('MYVAR', '42')
 
3519
        # We use an embedded test to make sure we fix the _captureVar bug
 
3520
        class Test(tests.TestCase):
 
3521
            def test_me(self):
 
3522
                # The first call save the 42 value
 
3523
                self.overrideEnv('MYVAR', None)
 
3524
                self.assertEqual(None, os.environ.get('MYVAR'))
 
3525
                # Make sure we can call it twice
 
3526
                self.overrideEnv('MYVAR', None)
 
3527
                self.assertEqual(None, os.environ.get('MYVAR'))
 
3528
        output = StringIO()
 
3529
        result = tests.TextTestResult(output, 0, 1)
 
3530
        Test('test_me').run(result)
 
3531
        if not result.wasStrictlySuccessful():
 
3532
            self.fail(output.getvalue())
 
3533
        # We get our value back
 
3534
        self.assertEqual('42', os.environ.get('MYVAR'))
 
3535
 
 
3536
 
 
3537
class TestIsolatedEnv(tests.TestCase):
 
3538
    """Test isolating tests from os.environ.
 
3539
 
 
3540
    Since we use tests that are already isolated from os.environ a bit of care
 
3541
    should be taken when designing the tests to avoid bootstrap side-effects.
 
3542
    The tests start an already clean os.environ which allow doing valid
 
3543
    assertions about which variables are present or not and design tests around
 
3544
    these assertions.
 
3545
    """
 
3546
 
 
3547
    class ScratchMonkey(tests.TestCase):
 
3548
 
 
3549
        def test_me(self):
 
3550
            pass
 
3551
 
 
3552
    def test_basics(self):
 
3553
        # Make sure we know the definition of BZR_HOME: not part of os.environ
 
3554
        # for tests.TestCase.
 
3555
        self.assertTrue('BZR_HOME' in tests.isolated_environ)
 
3556
        self.assertEqual(None, tests.isolated_environ['BZR_HOME'])
 
3557
        # Being part of isolated_environ, BZR_HOME should not appear here
 
3558
        self.assertFalse('BZR_HOME' in os.environ)
 
3559
        # Make sure we know the definition of LINES: part of os.environ for
 
3560
        # tests.TestCase
 
3561
        self.assertTrue('LINES' in tests.isolated_environ)
 
3562
        self.assertEqual('25', tests.isolated_environ['LINES'])
 
3563
        self.assertEqual('25', os.environ['LINES'])
 
3564
 
 
3565
    def test_injecting_unknown_variable(self):
 
3566
        # BZR_HOME is known to be absent from os.environ
 
3567
        test = self.ScratchMonkey('test_me')
 
3568
        tests.override_os_environ(test, {'BZR_HOME': 'foo'})
 
3569
        self.assertEqual('foo', os.environ['BZR_HOME'])
 
3570
        tests.restore_os_environ(test)
 
3571
        self.assertFalse('BZR_HOME' in os.environ)
 
3572
 
 
3573
    def test_injecting_known_variable(self):
 
3574
        test = self.ScratchMonkey('test_me')
 
3575
        # LINES is known to be present in os.environ
 
3576
        tests.override_os_environ(test, {'LINES': '42'})
 
3577
        self.assertEqual('42', os.environ['LINES'])
 
3578
        tests.restore_os_environ(test)
 
3579
        self.assertEqual('25', os.environ['LINES'])
 
3580
 
 
3581
    def test_deleting_variable(self):
 
3582
        test = self.ScratchMonkey('test_me')
 
3583
        # LINES is known to be present in os.environ
 
3584
        tests.override_os_environ(test, {'LINES': None})
 
3585
        self.assertTrue('LINES' not in os.environ)
 
3586
        tests.restore_os_environ(test)
 
3587
        self.assertEqual('25', os.environ['LINES'])
 
3588
 
 
3589
 
 
3590
class TestDocTestSuiteIsolation(tests.TestCase):
 
3591
    """Test that `tests.DocTestSuite` isolates doc tests from os.environ.
 
3592
 
 
3593
    Since tests.TestCase alreay provides an isolation from os.environ, we use
 
3594
    the clean environment as a base for testing. To precisely capture the
 
3595
    isolation provided by tests.DocTestSuite, we use doctest.DocTestSuite to
 
3596
    compare against.
 
3597
 
 
3598
    We want to make sure `tests.DocTestSuite` respect `tests.isolated_environ`,
 
3599
    not `os.environ` so each test overrides it to suit its needs.
 
3600
 
 
3601
    """
 
3602
 
 
3603
    def get_doctest_suite_for_string(self, klass, string):
 
3604
        class Finder(doctest.DocTestFinder):
 
3605
 
 
3606
            def find(*args, **kwargs):
 
3607
                test = doctest.DocTestParser().get_doctest(
 
3608
                    string, {}, 'foo', 'foo.py', 0)
 
3609
                return [test]
 
3610
 
 
3611
        suite = klass(test_finder=Finder())
 
3612
        return suite
 
3613
 
 
3614
    def run_doctest_suite_for_string(self, klass, string):
 
3615
        suite = self.get_doctest_suite_for_string(klass, string)
 
3616
        output = StringIO()
 
3617
        result = tests.TextTestResult(output, 0, 1)
 
3618
        suite.run(result)
 
3619
        return result, output
 
3620
 
 
3621
    def assertDocTestStringSucceds(self, klass, string):
 
3622
        result, output = self.run_doctest_suite_for_string(klass, string)
 
3623
        if not result.wasStrictlySuccessful():
 
3624
            self.fail(output.getvalue())
 
3625
 
 
3626
    def assertDocTestStringFails(self, klass, string):
 
3627
        result, output = self.run_doctest_suite_for_string(klass, string)
 
3628
        if result.wasStrictlySuccessful():
 
3629
            self.fail(output.getvalue())
 
3630
 
 
3631
    def test_injected_variable(self):
 
3632
        self.overrideAttr(tests, 'isolated_environ', {'LINES': '42'})
 
3633
        test = """
 
3634
            >>> import os
 
3635
            >>> os.environ['LINES']
 
3636
            '42'
 
3637
            """
 
3638
        # doctest.DocTestSuite fails as it sees '25'
 
3639
        self.assertDocTestStringFails(doctest.DocTestSuite, test)
 
3640
        # tests.DocTestSuite sees '42'
 
3641
        self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
 
3642
 
 
3643
    def test_deleted_variable(self):
 
3644
        self.overrideAttr(tests, 'isolated_environ', {'LINES': None})
 
3645
        test = """
 
3646
            >>> import os
 
3647
            >>> os.environ.get('LINES')
 
3648
            """
 
3649
        # doctest.DocTestSuite fails as it sees '25'
 
3650
        self.assertDocTestStringFails(doctest.DocTestSuite, test)
 
3651
        # tests.DocTestSuite sees None
 
3652
        self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
 
3653
 
 
3654
 
 
3655
class TestSelftestExcludePatterns(tests.TestCase):
 
3656
 
 
3657
    def setUp(self):
 
3658
        super(TestSelftestExcludePatterns, self).setUp()
 
3659
        self.overrideAttr(tests, 'test_suite', self.suite_factory)
 
3660
 
 
3661
    def suite_factory(self, keep_only=None, starting_with=None):
 
3662
        """A test suite factory with only a few tests."""
 
3663
        class Test(tests.TestCase):
 
3664
            def id(self):
 
3665
                # We don't need the full class path
 
3666
                return self._testMethodName
 
3667
            def a(self):
 
3668
                pass
 
3669
            def b(self):
 
3670
                pass
 
3671
            def c(self):
 
3672
                pass
 
3673
        return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
 
3674
 
 
3675
    def assertTestList(self, expected, *selftest_args):
 
3676
        # We rely on setUp installing the right test suite factory so we can
 
3677
        # test at the command level without loading the whole test suite
 
3678
        out, err = self.run_bzr(('selftest', '--list') + selftest_args)
 
3679
        actual = out.splitlines()
 
3680
        self.assertEqual(expected, actual)
 
3681
 
 
3682
    def test_full_list(self):
 
3683
        self.assertTestList(['a', 'b', 'c'])
 
3684
 
 
3685
    def test_single_exclude(self):
 
3686
        self.assertTestList(['b', 'c'], '-x', 'a')
 
3687
 
 
3688
    def test_mutiple_excludes(self):
 
3689
        self.assertTestList(['c'], '-x', 'a', '-x', 'b')
 
3690
 
 
3691
 
 
3692
class TestCounterHooks(tests.TestCase, SelfTestHelper):
 
3693
 
 
3694
    _test_needs_features = [features.subunit]
 
3695
 
 
3696
    def setUp(self):
 
3697
        super(TestCounterHooks, self).setUp()
 
3698
        class Test(tests.TestCase):
 
3699
 
 
3700
            def setUp(self):
 
3701
                super(Test, self).setUp()
 
3702
                self.hooks = hooks.Hooks()
 
3703
                self.hooks.add_hook('myhook', 'Foo bar blah', (2,4))
 
3704
                self.install_counter_hook(self.hooks, 'myhook')
 
3705
 
 
3706
            def no_hook(self):
 
3707
                pass
 
3708
 
 
3709
            def run_hook_once(self):
 
3710
                for hook in self.hooks['myhook']:
 
3711
                    hook(self)
 
3712
 
 
3713
        self.test_class = Test
 
3714
 
 
3715
    def assertHookCalls(self, expected_calls, test_name):
 
3716
        test = self.test_class(test_name)
 
3717
        result = unittest.TestResult()
 
3718
        test.run(result)
 
3719
        self.assertTrue(hasattr(test, '_counters'))
 
3720
        self.assertTrue(test._counters.has_key('myhook'))
 
3721
        self.assertEqual(expected_calls, test._counters['myhook'])
 
3722
 
 
3723
    def test_no_hook(self):
 
3724
        self.assertHookCalls(0, 'no_hook')
 
3725
 
 
3726
    def test_run_hook_once(self):
 
3727
        tt = features.testtools
 
3728
        if tt.module.__version__ < (0, 9, 8):
 
3729
            raise tests.TestSkipped('testtools-0.9.8 required for addDetail')
 
3730
        self.assertHookCalls(1, 'run_hook_once')