~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2010-09-01 08:02:42 UTC
  • mfrom: (5390.3.3 faster-revert-593560)
  • Revision ID: pqm@pqm.ubuntu.com-20100901080242-esg62ody4frwmy66
(spiv) Avoid repeatedly calling self.target.all_file_ids() in
 InterTree.iter_changes. (Andrew Bennetts)

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005-2013, 2016 Canonical Ltd
 
1
# Copyright (C) 2005-2010 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
 
import gc
21
 
import doctest
 
20
from doctest import ELLIPSIS
22
21
import os
23
22
import signal
24
23
import sys
25
 
import threading
26
24
import time
27
25
import unittest
28
26
import warnings
29
27
 
30
 
from testtools import (
31
 
    ExtendedToOriginalDecorator,
32
 
    MultiTestResult,
33
 
    )
34
 
from testtools.content import Content
 
28
from testtools import MultiTestResult
35
29
from testtools.content_type import ContentType
36
30
from testtools.matchers import (
37
31
    DocTestMatches,
38
32
    Equals,
39
33
    )
40
 
import testtools.testresult.doubles
 
34
import testtools.tests.helpers
41
35
 
42
36
import bzrlib
43
37
from bzrlib import (
44
38
    branchbuilder,
45
39
    bzrdir,
46
 
    controldir,
 
40
    debug,
47
41
    errors,
48
 
    hooks,
49
42
    lockdir,
50
43
    memorytree,
51
44
    osutils,
 
45
    progress,
52
46
    remote,
53
47
    repository,
54
48
    symbol_versioning,
55
49
    tests,
56
50
    transport,
57
51
    workingtree,
58
 
    workingtree_3,
59
 
    workingtree_4,
60
52
    )
61
53
from bzrlib.repofmt import (
62
54
    groupcompress_repo,
 
55
    pack_repo,
 
56
    weaverepo,
63
57
    )
64
58
from bzrlib.symbol_versioning import (
65
59
    deprecated_function,
70
64
    features,
71
65
    test_lsprof,
72
66
    test_server,
 
67
    test_sftp_transport,
73
68
    TestUtil,
74
69
    )
75
 
from bzrlib.trace import note, mutter
 
70
from bzrlib.trace import note
76
71
from bzrlib.transport import memory
 
72
from bzrlib.version import _get_bzr_source_tree
77
73
 
78
74
 
79
75
def _test_ids(test_suite):
81
77
    return [t.id() for t in tests.iter_suite_tests(test_suite)]
82
78
 
83
79
 
 
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
 
84
92
class MetaTestLog(tests.TestCase):
85
93
 
86
94
    def test_logging(self):
92
100
            "text", "plain", {"charset": "utf8"})))
93
101
        self.assertThat(u"".join(log.iter_text()), Equals(self.get_log()))
94
102
        self.assertThat(self.get_log(),
95
 
            DocTestMatches(u"...a test message\n", doctest.ELLIPSIS))
 
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()
96
113
 
97
114
 
98
115
class TestTreeShape(tests.TestCaseInTempDir):
99
116
 
100
117
    def test_unicode_paths(self):
101
 
        self.requireFeature(features.UnicodeFilenameFeature)
 
118
        self.requireFeature(tests.UnicodeFilename)
102
119
 
103
120
        filename = u'hell\u00d8'
104
121
        self.build_tree_contents([(filename, 'contents of hello')])
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
 
122
        self.failUnlessExists(filename)
119
123
 
120
124
 
121
125
class TestTransportScenarios(tests.TestCase):
308
312
        from bzrlib.tests.per_interrepository import make_scenarios
309
313
        server1 = "a"
310
314
        server2 = "b"
311
 
        formats = [("C0", "C1", "C2", "C3"), ("D0", "D1", "D2", "D3")]
 
315
        formats = [("C0", "C1", "C2"), ("D0", "D1", "D2")]
312
316
        scenarios = make_scenarios(server1, server2, formats)
313
317
        self.assertEqual([
314
318
            ('C0,str,str',
315
319
             {'repository_format': 'C1',
316
320
              'repository_format_to': 'C2',
317
321
              'transport_readonly_server': 'b',
318
 
              'transport_server': 'a',
319
 
              'extra_setup': 'C3'}),
 
322
              'transport_server': 'a'}),
320
323
            ('D0,str,str',
321
324
             {'repository_format': 'D1',
322
325
              'repository_format_to': 'D2',
323
326
              'transport_readonly_server': 'b',
324
 
              'transport_server': 'a',
325
 
              'extra_setup': 'D3'})],
 
327
              'transport_server': 'a'})],
326
328
            scenarios)
327
329
 
328
330
 
334
336
        from bzrlib.tests.per_workingtree import make_scenarios
335
337
        server1 = "a"
336
338
        server2 = "b"
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')
 
339
        formats = [workingtree.WorkingTreeFormat2(),
 
340
                   workingtree.WorkingTreeFormat3(),]
 
341
        scenarios = make_scenarios(server1, server2, formats)
343
342
        self.assertEqual([
344
 
            ('WorkingTreeFormat4',
 
343
            ('WorkingTreeFormat2',
345
344
             {'bzrdir_format': formats[0]._matchingbzrdir,
346
345
              'transport_readonly_server': 'b',
347
346
              'transport_server': 'a',
350
349
             {'bzrdir_format': formats[1]._matchingbzrdir,
351
350
              'transport_readonly_server': 'b',
352
351
              'transport_server': 'a',
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)
 
352
              'workingtree_format': formats[1]})],
 
353
            scenarios)
367
354
 
368
355
 
369
356
class TestTreeScenarios(tests.TestCase):
370
357
 
371
358
    def test_scenarios(self):
372
359
        # the tree implementation scenario generator is meant to setup one
373
 
        # instance for each working tree format, one additional instance
 
360
        # instance for each working tree format, and one additional instance
374
361
        # that will use the default wt format, but create a revision tree for
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.
 
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.
380
365
 
381
366
        from bzrlib.tests.per_tree import (
382
367
            _dirstate_tree_from_workingtree,
388
373
            )
389
374
        server1 = "a"
390
375
        server2 = "b"
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(),]
 
376
        formats = [workingtree.WorkingTreeFormat2(),
 
377
                   workingtree.WorkingTreeFormat3(),]
396
378
        scenarios = make_scenarios(server1, server2, formats)
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()
 
379
        self.assertEqual(7, len(scenarios))
 
380
        default_wt_format = workingtree.WorkingTreeFormat4._default_format
 
381
        wt4_format = workingtree.WorkingTreeFormat4()
 
382
        wt5_format = workingtree.WorkingTreeFormat5()
402
383
        expected_scenarios = [
403
 
            ('WorkingTreeFormat4',
 
384
            ('WorkingTreeFormat2',
404
385
             {'bzrdir_format': formats[0]._matchingbzrdir,
405
386
              'transport_readonly_server': 'b',
406
387
              'transport_server': 'a',
414
395
              'workingtree_format': formats[1],
415
396
              '_workingtree_to_test_tree': return_parameter,
416
397
             }),
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
 
             }),
426
398
            ('RevisionTree',
427
399
             {'_workingtree_to_test_tree': revision_tree_from_workingtree,
428
400
              'bzrdir_format': default_wt_format._matchingbzrdir,
475
447
        # ones to add.
476
448
        from bzrlib.tests.per_tree import (
477
449
            return_parameter,
 
450
            revision_tree_from_workingtree
478
451
            )
479
452
        from bzrlib.tests.per_intertree import (
480
453
            make_scenarios,
481
454
            )
482
 
        from bzrlib.workingtree_3 import WorkingTreeFormat3
483
 
        from bzrlib.workingtree_4 import WorkingTreeFormat4
 
455
        from bzrlib.workingtree import WorkingTreeFormat2, WorkingTreeFormat3
484
456
        input_test = TestInterTreeScenarios(
485
457
            "test_scenarios")
486
458
        server1 = "a"
487
459
        server2 = "b"
488
 
        format1 = WorkingTreeFormat4()
 
460
        format1 = WorkingTreeFormat2()
489
461
        format2 = WorkingTreeFormat3()
490
462
        formats = [("1", str, format1, format2, "converter1"),
491
463
            ("2", int, format2, format1, "converter2")]
537
509
        self.assertRaises(AssertionError, self.assertEqualStat,
538
510
            os.lstat("foo"), os.lstat("longname"))
539
511
 
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
 
 
559
512
 
560
513
class TestTestCaseWithMemoryTransport(tests.TestCaseWithMemoryTransport):
561
514
 
595
548
        tree = self.make_branch_and_memory_tree('dir')
596
549
        # Guard against regression into MemoryTransport leaking
597
550
        # files to disk instead of keeping them in memory.
598
 
        self.assertFalse(osutils.lexists('dir'))
 
551
        self.failIf(osutils.lexists('dir'))
599
552
        self.assertIsInstance(tree, memorytree.MemoryTree)
600
553
 
601
554
    def test_make_branch_and_memory_tree_with_format(self):
602
555
        """make_branch_and_memory_tree should accept a format option."""
603
556
        format = bzrdir.BzrDirMetaFormat1()
604
 
        format.repository_format = repository.format_registry.get_default()
 
557
        format.repository_format = weaverepo.RepositoryFormat7()
605
558
        tree = self.make_branch_and_memory_tree('dir', format=format)
606
559
        # Guard against regression into MemoryTransport leaking
607
560
        # files to disk instead of keeping them in memory.
608
 
        self.assertFalse(osutils.lexists('dir'))
 
561
        self.failIf(osutils.lexists('dir'))
609
562
        self.assertIsInstance(tree, memorytree.MemoryTree)
610
563
        self.assertEqual(format.repository_format.__class__,
611
564
            tree.branch.repository._format.__class__)
615
568
        self.assertIsInstance(builder, branchbuilder.BranchBuilder)
616
569
        # Guard against regression into MemoryTransport leaking
617
570
        # files to disk instead of keeping them in memory.
618
 
        self.assertFalse(osutils.lexists('dir'))
 
571
        self.failIf(osutils.lexists('dir'))
619
572
 
620
573
    def test_make_branch_builder_with_format(self):
621
574
        # Use a repo layout that doesn't conform to a 'named' layout, to ensure
622
575
        # that the format objects are used.
623
576
        format = bzrdir.BzrDirMetaFormat1()
624
 
        repo_format = repository.format_registry.get_default()
 
577
        repo_format = weaverepo.RepositoryFormat7()
625
578
        format.repository_format = repo_format
626
579
        builder = self.make_branch_builder('dir', format=format)
627
580
        the_branch = builder.get_branch()
628
581
        # Guard against regression into MemoryTransport leaking
629
582
        # files to disk instead of keeping them in memory.
630
 
        self.assertFalse(osutils.lexists('dir'))
 
583
        self.failIf(osutils.lexists('dir'))
631
584
        self.assertEqual(format.repository_format.__class__,
632
585
                         the_branch.repository._format.__class__)
633
586
        self.assertEqual(repo_format.get_format_string(),
639
592
        the_branch = builder.get_branch()
640
593
        # Guard against regression into MemoryTransport leaking
641
594
        # files to disk instead of keeping them in memory.
642
 
        self.assertFalse(osutils.lexists('dir'))
643
 
        dir_format = controldir.format_registry.make_bzrdir('knit')
 
595
        self.failIf(osutils.lexists('dir'))
 
596
        dir_format = bzrdir.format_registry.make_bzrdir('knit')
644
597
        self.assertEqual(dir_format.repository_format.__class__,
645
598
                         the_branch.repository._format.__class__)
646
599
        self.assertEqual('Bazaar-NG Knit Repository Format 1',
650
603
    def test_dangling_locks_cause_failures(self):
651
604
        class TestDanglingLock(tests.TestCaseWithMemoryTransport):
652
605
            def test_function(self):
653
 
                t = self.get_transport_from_path('.')
 
606
                t = self.get_transport('.')
654
607
                l = lockdir.LockDir(t, 'lock')
655
608
                l.create()
656
609
                l.attempt_lock()
658
611
        result = test.run()
659
612
        total_failures = result.errors + result.failures
660
613
        if self._lock_check_thorough:
661
 
            self.assertEqual(1, len(total_failures))
 
614
            self.assertLength(1, total_failures)
662
615
        else:
663
616
            # When _lock_check_thorough is disabled, then we don't trigger a
664
617
            # failure
665
 
            self.assertEqual(0, len(total_failures))
 
618
            self.assertLength(0, total_failures)
666
619
 
667
620
 
668
621
class TestTestCaseWithTransport(tests.TestCaseWithTransport):
676
629
        # for the server
677
630
        url = self.get_readonly_url()
678
631
        url2 = self.get_readonly_url('foo/bar')
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)
 
632
        t = transport.get_transport(url)
 
633
        t2 = transport.get_transport(url2)
 
634
        self.failUnless(isinstance(t, ReadonlyTransportDecorator))
 
635
        self.failUnless(isinstance(t2, ReadonlyTransportDecorator))
683
636
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
684
637
 
685
638
    def test_get_readonly_url_http(self):
691
644
        url = self.get_readonly_url()
692
645
        url2 = self.get_readonly_url('foo/bar')
693
646
        # the transport returned may be any HttpTransportBase subclass
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)
 
647
        t = transport.get_transport(url)
 
648
        t2 = transport.get_transport(url2)
 
649
        self.failUnless(isinstance(t, HttpTransportBase))
 
650
        self.failUnless(isinstance(t2, HttpTransportBase))
698
651
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
699
652
 
700
653
    def test_is_directory(self):
708
661
    def test_make_branch_builder(self):
709
662
        builder = self.make_branch_builder('dir')
710
663
        rev_id = builder.build_commit()
711
 
        self.assertPathExists('dir')
712
 
        a_dir = controldir.ControlDir.open('dir')
 
664
        self.failUnlessExists('dir')
 
665
        a_dir = bzrdir.BzrDir.open('dir')
713
666
        self.assertRaises(errors.NoWorkingTree, a_dir.open_workingtree)
714
667
        a_branch = a_dir.open_branch()
715
668
        builder_branch = builder.get_branch()
730
683
        self.assertIsInstance(result_bzrdir.transport,
731
684
                              memory.MemoryTransport)
732
685
        # should not be on disk, should only be in memory
733
 
        self.assertPathDoesNotExist('subdir')
 
686
        self.failIfExists('subdir')
734
687
 
735
688
 
736
689
class TestChrootedTest(tests.ChrootedTestCase):
737
690
 
738
691
    def test_root_is_root(self):
739
 
        t = transport.get_transport_from_url(self.get_readonly_url())
 
692
        t = transport.get_transport(self.get_readonly_url())
740
693
        url = t.base
741
694
        self.assertEqual(url, t.clone('..').base)
742
695
 
744
697
class TestProfileResult(tests.TestCase):
745
698
 
746
699
    def test_profiles_tests(self):
747
 
        self.requireFeature(features.lsprof_feature)
748
 
        terminal = testtools.testresult.doubles.ExtendedTestResult()
 
700
        self.requireFeature(test_lsprof.LSProfFeature)
 
701
        terminal = testtools.tests.helpers.ExtendedTestResult()
749
702
        result = tests.ProfileResult(terminal)
750
703
        class Sample(tests.TestCase):
751
704
            def a(self):
768
721
                descriptions=0,
769
722
                verbosity=1,
770
723
                )
771
 
        capture = testtools.testresult.doubles.ExtendedTestResult()
 
724
        capture = testtools.tests.helpers.ExtendedTestResult()
772
725
        test_case.run(MultiTestResult(result, capture))
773
726
        run_case = capture._events[0][1]
774
727
        timed_string = result._testTimeString(run_case)
795
748
        self.check_timing(ShortDelayTestCase('test_short_delay'),
796
749
                          r"^ +[0-9]+ms$")
797
750
 
 
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
 
798
791
    def _time_hello_world_encoding(self):
799
792
        """Profile two sleep calls
800
793
 
805
798
 
806
799
    def test_lsprofiling(self):
807
800
        """Verbose test result prints lsprof statistics from test cases."""
808
 
        self.requireFeature(features.lsprof_feature)
 
801
        self.requireFeature(test_lsprof.LSProfFeature)
809
802
        result_stream = StringIO()
810
803
        result = bzrlib.tests.VerboseTestResult(
811
804
            result_stream,
840
833
        self.assertContainsRe(output,
841
834
            r"LSProf output for <type 'unicode'>\(\('world',\), {'errors': 'replace'}\)\n")
842
835
 
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
 
 
858
836
    def test_known_failure(self):
859
 
        """Using knownFailure should trigger several result actions."""
 
837
        """A KnownFailure being raised should trigger several result actions."""
860
838
        class InstrumentedTestResult(tests.ExtendedTestResult):
861
839
            def stopTestRun(self): pass
862
 
            def report_tests_starting(self): pass
 
840
            def startTests(self): pass
 
841
            def report_test_start(self, test): pass
863
842
            def report_known_failure(self, test, err=None, details=None):
864
843
                self._call = test, 'known failure'
865
844
        result = InstrumentedTestResult(None, None, None, None)
866
845
        class Test(tests.TestCase):
867
846
            def test_function(self):
868
 
                self.knownFailure('failed!')
 
847
                raise tests.KnownFailure('failed!')
869
848
        test = Test("test_function")
870
849
        test.run(result)
871
850
        # it should invoke 'report_known_failure'.
887
866
            descriptions=0,
888
867
            verbosity=2,
889
868
            )
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)")
 
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))
897
886
 
898
887
    def get_passing_test(self):
899
888
        """Return a test object that can't be run usefully."""
905
894
        """Test the behaviour of invoking addNotSupported."""
906
895
        class InstrumentedTestResult(tests.ExtendedTestResult):
907
896
            def stopTestRun(self): pass
908
 
            def report_tests_starting(self): pass
 
897
            def startTests(self): pass
 
898
            def report_test_start(self, test): pass
909
899
            def report_unsupported(self, test, feature):
910
900
                self._call = test, feature
911
901
        result = InstrumentedTestResult(None, None, None, None)
912
902
        test = SampleTestCase('_test_pass')
913
 
        feature = features.Feature()
 
903
        feature = tests.Feature()
914
904
        result.startTest(test)
915
905
        result.addNotSupported(test, feature)
916
906
        # it should invoke 'report_unsupported'.
935
925
            verbosity=2,
936
926
            )
937
927
        test = self.get_passing_test()
938
 
        feature = features.Feature()
 
928
        feature = tests.Feature()
939
929
        result.startTest(test)
940
930
        prefix = len(result_stream.getvalue())
941
931
        result.report_unsupported(test, feature)
950
940
        """An UnavailableFeature being raised should invoke addNotSupported."""
951
941
        class InstrumentedTestResult(tests.ExtendedTestResult):
952
942
            def stopTestRun(self): pass
953
 
            def report_tests_starting(self): pass
 
943
            def startTests(self): pass
 
944
            def report_test_start(self, test): pass
954
945
            def addNotSupported(self, test, feature):
955
946
                self._call = test, feature
956
947
        result = InstrumentedTestResult(None, None, None, None)
957
 
        feature = features.Feature()
 
948
        feature = tests.Feature()
958
949
        class Test(tests.TestCase):
959
950
            def test_function(self):
960
951
                raise tests.UnavailableFeature(feature)
979
970
    def test_strict_with_known_failure(self):
980
971
        result = bzrlib.tests.TextTestResult(self._log_file, descriptions=0,
981
972
                                             verbosity=1)
982
 
        test = _get_test("test_xfail")
983
 
        test.run(result)
 
973
        test = self.get_passing_test()
 
974
        err = (tests.KnownFailure, tests.KnownFailure('foo'), None)
 
975
        result.addExpectedFailure(test, err)
984
976
        self.assertFalse(result.wasStrictlySuccessful())
985
977
        self.assertEqual(None, result._extractBenchmarkTime(test))
986
978
 
997
989
        class InstrumentedTestResult(tests.ExtendedTestResult):
998
990
            calls = 0
999
991
            def startTests(self): self.calls += 1
 
992
            def report_test_start(self, test): pass
1000
993
        result = InstrumentedTestResult(None, None, None, None)
1001
994
        def test_function():
1002
995
            pass
1003
996
        test = unittest.FunctionTestCase(test_function)
1004
997
        test.run(result)
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)
 
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()
1019
1008
 
1020
1009
 
1021
1010
class TestRunner(tests.TestCase):
1034
1023
        because of our use of global state.
1035
1024
        """
1036
1025
        old_root = tests.TestCaseInTempDir.TEST_ROOT
 
1026
        old_leak = tests.TestCase._first_thread_leaker_id
1037
1027
        try:
1038
1028
            tests.TestCaseInTempDir.TEST_ROOT = None
 
1029
            tests.TestCase._first_thread_leaker_id = None
1039
1030
            return testrunner.run(test)
1040
1031
        finally:
1041
1032
            tests.TestCaseInTempDir.TEST_ROOT = old_root
 
1033
            tests.TestCase._first_thread_leaker_id = old_leak
1042
1034
 
1043
1035
    def test_known_failure_failed_run(self):
1044
1036
        # run a test that generates a known failure which should be printed in
1049
1041
        test = unittest.TestSuite()
1050
1042
        test.addTest(Test("known_failure_test"))
1051
1043
        def failing_test():
1052
 
            raise AssertionError('foo')
 
1044
            self.fail('foo')
1053
1045
        test.addTest(unittest.FunctionTestCase(failing_test))
1054
1046
        stream = StringIO()
1055
1047
        runner = tests.TextTestRunner(stream=stream)
1063
1055
            '^----------------------------------------------------------------------\n'
1064
1056
            'Traceback \\(most recent call last\\):\n'
1065
1057
            '  .*' # File .*, line .*, in failing_test' - but maybe not from .pyc
1066
 
            '    raise AssertionError\\(\'foo\'\\)\n'
 
1058
            '    self.fail\\(\'foo\'\\)\n'
1067
1059
            '.*'
1068
1060
            '^----------------------------------------------------------------------\n'
1069
1061
            '.*'
1075
1067
        # the final output.
1076
1068
        class Test(tests.TestCase):
1077
1069
            def known_failure_test(self):
1078
 
                self.knownFailure("Never works...")
 
1070
                self.expectFailure('failed', self.assertTrue, False)
1079
1071
        test = Test("known_failure_test")
1080
1072
        stream = StringIO()
1081
1073
        runner = tests.TextTestRunner(stream=stream)
1087
1079
            '\n'
1088
1080
            'OK \\(known_failures=1\\)\n')
1089
1081
 
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
 
 
1111
1082
    def test_result_decorator(self):
1112
1083
        # decorate results
1113
1084
        calls = []
1114
 
        class LoggingDecorator(ExtendedToOriginalDecorator):
 
1085
        class LoggingDecorator(tests.ForwardingResult):
1115
1086
            def startTest(self, test):
1116
 
                ExtendedToOriginalDecorator.startTest(self, test)
 
1087
                tests.ForwardingResult.startTest(self, test)
1117
1088
                calls.append('start')
1118
1089
        test = unittest.FunctionTestCase(lambda:None)
1119
1090
        stream = StringIO()
1161
1132
        class SkippedTest(tests.TestCase):
1162
1133
 
1163
1134
            def setUp(self):
1164
 
                super(SkippedTest, self).setUp()
 
1135
                tests.TestCase.setUp(self)
1165
1136
                calls.append('setUp')
1166
1137
                self.addCleanup(self.cleanup)
1167
1138
 
1197
1168
 
1198
1169
    def test_unsupported_features_listed(self):
1199
1170
        """When unsupported features are encountered they are detailed."""
1200
 
        class Feature1(features.Feature):
 
1171
        class Feature1(tests.Feature):
1201
1172
            def _probe(self): return False
1202
 
        class Feature2(features.Feature):
 
1173
        class Feature2(tests.Feature):
1203
1174
            def _probe(self): return False
1204
1175
        # create sample tests
1205
1176
        test1 = SampleTestCase('_test_pass')
1220
1191
            ],
1221
1192
            lines[-3:])
1222
1193
 
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")
 
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)
1234
1216
 
1235
1217
    def test_startTestRun(self):
1236
1218
        """run should call result.startTestRun()"""
1237
1219
        calls = []
1238
 
        class LoggingDecorator(ExtendedToOriginalDecorator):
 
1220
        class LoggingDecorator(tests.ForwardingResult):
1239
1221
            def startTestRun(self):
1240
 
                ExtendedToOriginalDecorator.startTestRun(self)
 
1222
                tests.ForwardingResult.startTestRun(self)
1241
1223
                calls.append('startTestRun')
1242
1224
        test = unittest.FunctionTestCase(lambda:None)
1243
1225
        stream = StringIO()
1249
1231
    def test_stopTestRun(self):
1250
1232
        """run should call result.stopTestRun()"""
1251
1233
        calls = []
1252
 
        class LoggingDecorator(ExtendedToOriginalDecorator):
 
1234
        class LoggingDecorator(tests.ForwardingResult):
1253
1235
            def stopTestRun(self):
1254
 
                ExtendedToOriginalDecorator.stopTestRun(self)
 
1236
                tests.ForwardingResult.stopTestRun(self)
1255
1237
                calls.append('stopTestRun')
1256
1238
        test = unittest.FunctionTestCase(lambda:None)
1257
1239
        stream = StringIO()
1260
1242
        result = self.run_test_runner(runner, test)
1261
1243
        self.assertLength(1, calls)
1262
1244
 
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
 
 
1280
1245
 
1281
1246
class SampleTestCase(tests.TestCase):
1282
1247
 
1470
1435
        # Note this test won't fail with hooks that the core library doesn't
1471
1436
        # use - but it trigger with a plugin that adds hooks, so its still a
1472
1437
        # useful warning in that case.
1473
 
        self.assertEqual(bzrlib.branch.BranchHooks(), bzrlib.branch.Branch.hooks)
1474
 
        self.assertEqual(
1475
 
            bzrlib.smart.server.SmartServerHooks(),
 
1438
        self.assertEqual(bzrlib.branch.BranchHooks(),
 
1439
            bzrlib.branch.Branch.hooks)
 
1440
        self.assertEqual(bzrlib.smart.server.SmartServerHooks(),
1476
1441
            bzrlib.smart.server.SmartTCPServer.hooks)
1477
 
        self.assertEqual(
1478
 
            bzrlib.commands.CommandHooks(), bzrlib.commands.Command.hooks)
 
1442
        self.assertEqual(bzrlib.commands.CommandHooks(),
 
1443
            bzrlib.commands.Command.hooks)
1479
1444
 
1480
1445
    def test__gather_lsprof_in_benchmarks(self):
1481
1446
        """When _gather_lsprof_in_benchmarks is on, accumulate profile data.
1482
1447
 
1483
1448
        Each self.time() call is individually and separately profiled.
1484
1449
        """
1485
 
        self.requireFeature(features.lsprof_feature)
 
1450
        self.requireFeature(test_lsprof.LSProfFeature)
1486
1451
        # overrides the class member with an instance member so no cleanup
1487
1452
        # needed.
1488
1453
        self._gather_lsprof_in_benchmarks = True
1507
1472
        transport_server = memory.MemoryServer()
1508
1473
        transport_server.start_server()
1509
1474
        self.addCleanup(transport_server.stop_server)
1510
 
        t = transport.get_transport_from_url(transport_server.get_url())
1511
 
        controldir.ControlDir.create(t.base)
 
1475
        t = transport.get_transport(transport_server.get_url())
 
1476
        bzrdir.BzrDir.create(t.base)
1512
1477
        self.assertRaises(errors.BzrError,
1513
 
            controldir.ControlDir.open_from_transport, t)
 
1478
            bzrdir.BzrDir.open_from_transport, t)
1514
1479
        # But if we declare this as safe, we can open the bzrdir.
1515
1480
        self.permit_url(t.base)
1516
1481
        self._bzr_selftest_roots.append(t.base)
1517
 
        controldir.ControlDir.open_from_transport(t)
 
1482
        bzrdir.BzrDir.open_from_transport(t)
1518
1483
 
1519
1484
    def test_requireFeature_available(self):
1520
1485
        """self.requireFeature(available) is a no-op."""
1521
 
        class Available(features.Feature):
 
1486
        class Available(tests.Feature):
1522
1487
            def _probe(self):return True
1523
1488
        feature = Available()
1524
1489
        self.requireFeature(feature)
1525
1490
 
1526
1491
    def test_requireFeature_unavailable(self):
1527
1492
        """self.requireFeature(unavailable) raises UnavailableFeature."""
1528
 
        class Unavailable(features.Feature):
 
1493
        class Unavailable(tests.Feature):
1529
1494
            def _probe(self):return False
1530
1495
        feature = Unavailable()
1531
1496
        self.assertRaises(tests.UnavailableFeature,
1654
1619
        self.assertRaises(AssertionError,
1655
1620
            self.assertListRaises, _TestException, success_generator)
1656
1621
 
1657
 
    def _run_successful_test(self, test):
1658
 
        result = testtools.TestResult()
1659
 
        test.run(result)
1660
 
        self.assertTrue(result.wasSuccessful())
1661
 
        return result
1662
 
 
1663
1622
    def test_overrideAttr_without_value(self):
1664
1623
        self.test_attr = 'original' # Define a test attribute
1665
1624
        obj = self # Make 'obj' visible to the embedded test
1666
1625
        class Test(tests.TestCase):
1667
1626
 
1668
1627
            def setUp(self):
1669
 
                super(Test, self).setUp()
 
1628
                tests.TestCase.setUp(self)
1670
1629
                self.orig = self.overrideAttr(obj, 'test_attr')
1671
1630
 
1672
1631
            def test_value(self):
1675
1634
                obj.test_attr = 'modified'
1676
1635
                self.assertEqual('modified', obj.test_attr)
1677
1636
 
1678
 
        self._run_successful_test(Test('test_value'))
 
1637
        test = Test('test_value')
 
1638
        test.run(unittest.TestResult())
1679
1639
        self.assertEqual('original', obj.test_attr)
1680
1640
 
1681
1641
    def test_overrideAttr_with_value(self):
1684
1644
        class Test(tests.TestCase):
1685
1645
 
1686
1646
            def setUp(self):
1687
 
                super(Test, self).setUp()
 
1647
                tests.TestCase.setUp(self)
1688
1648
                self.orig = self.overrideAttr(obj, 'test_attr', new='modified')
1689
1649
 
1690
1650
            def test_value(self):
1691
1651
                self.assertEqual('original', self.orig)
1692
1652
                self.assertEqual('modified', obj.test_attr)
1693
1653
 
1694
 
        self._run_successful_test(Test('test_value'))
 
1654
        test = Test('test_value')
 
1655
        test.run(unittest.TestResult())
1695
1656
        self.assertEqual('original', obj.test_attr)
1696
1657
 
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
 
 
1880
1658
 
1881
1659
# NB: Don't delete this; it's not actually from 0.11!
1882
1660
@deprecated_function(deprecated_in((0, 11, 0)))
1912
1690
        self.assertIsInstance(2, int)
1913
1691
        self.assertIsInstance(u'', basestring)
1914
1692
        e = self.assertRaises(AssertionError, self.assertIsInstance, None, int)
1915
 
        self.assertEqual(str(e),
 
1693
        self.assertEquals(str(e),
1916
1694
            "None is an instance of <type 'NoneType'> rather than <type 'int'>")
1917
1695
        self.assertRaises(AssertionError, self.assertIsInstance, 23.3, int)
1918
1696
        e = self.assertRaises(AssertionError,
1919
1697
            self.assertIsInstance, None, int, "it's just not")
1920
 
        self.assertEqual(str(e),
 
1698
        self.assertEquals(str(e),
1921
1699
            "None is an instance of <type 'NoneType'> rather than <type 'int'>"
1922
1700
            ": it's just not")
1923
1701
 
1928
1706
    def test_assertEqualDiff(self):
1929
1707
        e = self.assertRaises(AssertionError,
1930
1708
                              self.assertEqualDiff, '', '\n')
1931
 
        self.assertEqual(str(e),
 
1709
        self.assertEquals(str(e),
1932
1710
                          # Don't blink ! The '+' applies to the second string
1933
1711
                          'first string is missing a final newline.\n+ \n')
1934
1712
        e = self.assertRaises(AssertionError,
1935
1713
                              self.assertEqualDiff, '\n', '')
1936
 
        self.assertEqual(str(e),
 
1714
        self.assertEquals(str(e),
1937
1715
                          # Don't blink ! The '-' applies to the second string
1938
1716
                          'second string is missing a final newline.\n- \n')
1939
1717
 
1996
1774
            warnings.warn("this is your last warning")
1997
1775
            return a + b
1998
1776
        wlist, result = self.callCatchWarnings(meth, 1, 2)
1999
 
        self.assertEqual(3, result)
 
1777
        self.assertEquals(3, result)
2000
1778
        # would like just to compare them, but UserWarning doesn't implement
2001
1779
        # eq well
2002
1780
        w0, = wlist
2003
1781
        self.assertIsInstance(w0, UserWarning)
2004
 
        self.assertEqual("this is your last warning", str(w0))
 
1782
        self.assertEquals("this is your last warning", str(w0))
2005
1783
 
2006
1784
 
2007
1785
class TestConvenienceMakers(tests.TestCaseWithTransport):
2010
1788
    def test_make_branch_and_tree_with_format(self):
2011
1789
        # we should be able to supply a format to make_branch_and_tree
2012
1790
        self.make_branch_and_tree('a', format=bzrlib.bzrdir.BzrDirMetaFormat1())
2013
 
        self.assertIsInstance(bzrlib.controldir.ControlDir.open('a')._format,
 
1791
        self.make_branch_and_tree('b', format=bzrlib.bzrdir.BzrDirFormat6())
 
1792
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('a')._format,
2014
1793
                              bzrlib.bzrdir.BzrDirMetaFormat1)
 
1794
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('b')._format,
 
1795
                              bzrlib.bzrdir.BzrDirFormat6)
2015
1796
 
2016
1797
    def test_make_branch_and_memory_tree(self):
2017
1798
        # we should be able to get a new branch and a mutable tree from
2028
1809
        tree = self.make_branch_and_tree('t1')
2029
1810
        base = tree.bzrdir.root_transport.base
2030
1811
        self.assertStartsWith(base, 'file://')
2031
 
        self.assertEqual(tree.bzrdir.root_transport,
 
1812
        self.assertEquals(tree.bzrdir.root_transport,
2032
1813
                tree.branch.bzrdir.root_transport)
2033
 
        self.assertEqual(tree.bzrdir.root_transport,
 
1814
        self.assertEquals(tree.bzrdir.root_transport,
2034
1815
                tree.branch.repository.bzrdir.root_transport)
2035
1816
 
2036
1817
 
2037
 
class SelfTestHelper(object):
 
1818
class SelfTestHelper:
2038
1819
 
2039
1820
    def run_selftest(self, **kwargs):
2040
1821
        """Run selftest returning its output."""
2094
1875
        self.assertLength(2, output.readlines())
2095
1876
 
2096
1877
    def test_lsprof_tests(self):
2097
 
        self.requireFeature(features.lsprof_feature)
2098
 
        results = []
 
1878
        self.requireFeature(test_lsprof.LSProfFeature)
 
1879
        calls = []
2099
1880
        class Test(object):
2100
1881
            def __call__(test, result):
2101
1882
                test.run(result)
2102
1883
            def run(test, result):
2103
 
                results.append(result)
 
1884
                self.assertIsInstance(result, tests.ForwardingResult)
 
1885
                calls.append("called")
2104
1886
            def countTestCases(self):
2105
1887
                return 1
2106
1888
        self.run_selftest(test_suite_factory=Test, lsprof_tests=True)
2107
 
        self.assertLength(1, results)
2108
 
        self.assertIsInstance(results.pop(), ExtendedToOriginalDecorator)
 
1889
        self.assertLength(1, calls)
2109
1890
 
2110
1891
    def test_random(self):
2111
1892
        # test randomising by listing a number of tests.
2191
1972
            load_list='missing file name', list_only=True)
2192
1973
 
2193
1974
 
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
 
 
2276
1975
class TestRunBzr(tests.TestCase):
2277
1976
 
2278
1977
    out = ''
2300
1999
        self.assertEqual(['rocks'], self.argv)
2301
2000
        self.assertEqual(34, self.retcode)
2302
2001
        self.assertEqual('It sure does!\n', out)
2303
 
        self.assertEqual(out, self.out)
 
2002
        self.assertEquals(out, self.out)
2304
2003
        self.assertEqual('', err)
2305
 
        self.assertEqual(err, self.err)
 
2004
        self.assertEquals(err, self.err)
2306
2005
 
2307
2006
    def test_run_bzr_error_regexes(self):
2308
2007
        self.out = ''
2401
2100
        # stdout and stderr of the invoked run_bzr
2402
2101
        current_factory = bzrlib.ui.ui_factory
2403
2102
        self.run_bzr(['foo'])
2404
 
        self.assertFalse(current_factory is self.factory)
 
2103
        self.failIf(current_factory is self.factory)
2405
2104
        self.assertNotEqual(sys.stdout, self.factory.stdout)
2406
2105
        self.assertNotEqual(sys.stderr, self.factory.stderr)
2407
2106
        self.assertEqual('foo\n', self.factory.stdout.getvalue())
2448
2147
    """Base class for tests testing how we might run bzr."""
2449
2148
 
2450
2149
    def setUp(self):
2451
 
        super(TestWithFakedStartBzrSubprocess, self).setUp()
 
2150
        tests.TestCaseWithTransport.setUp(self)
2452
2151
        self.subprocess_calls = []
2453
2152
 
2454
2153
    def start_bzr_subprocess(self, process_args, env_changes=None,
2564
2263
 
2565
2264
 
2566
2265
class TestStartBzrSubProcess(tests.TestCase):
2567
 
    """Stub test start_bzr_subprocess."""
2568
2266
 
2569
 
    def _subprocess_log_cleanup(self):
2570
 
        """Inhibits the base version as we don't produce a log file."""
 
2267
    def check_popen_state(self):
 
2268
        """Replace to make assertions when popen is called."""
2571
2269
 
2572
2270
    def _popen(self, *args, **kwargs):
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
 
        """
 
2271
        """Record the command that is run, so that we can ensure it is correct"""
2577
2272
        self.check_popen_state()
2578
2273
        self._popen_args = args
2579
2274
        self._popen_kwargs = kwargs
2580
2275
        raise _DontSpawnProcess()
2581
2276
 
2582
 
    def check_popen_state(self):
2583
 
        """Replace to make assertions when popen is called."""
2584
 
 
2585
2277
    def test_run_bzr_subprocess_no_plugins(self):
2586
2278
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [])
2587
2279
        command = self._popen_args[0]
2591
2283
 
2592
2284
    def test_allow_plugins(self):
2593
2285
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2594
 
                          allow_plugins=True)
 
2286
            allow_plugins=True)
2595
2287
        command = self._popen_args[0]
2596
2288
        self.assertEqual([], command[2:])
2597
2289
 
2598
2290
    def test_set_env(self):
2599
 
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
 
2291
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
2600
2292
        # set in the child
2601
2293
        def check_environment():
2602
2294
            self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2603
2295
        self.check_popen_state = check_environment
2604
2296
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2605
 
                          env_changes={'EXISTANT_ENV_VAR':'set variable'})
 
2297
            env_changes={'EXISTANT_ENV_VAR':'set variable'})
2606
2298
        # not set in theparent
2607
2299
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2608
2300
 
2609
2301
    def test_run_bzr_subprocess_env_del(self):
2610
2302
        """run_bzr_subprocess can remove environment variables too."""
2611
 
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
 
2303
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
2612
2304
        def check_environment():
2613
2305
            self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2614
2306
        os.environ['EXISTANT_ENV_VAR'] = 'set variable'
2615
2307
        self.check_popen_state = check_environment
2616
2308
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2617
 
                          env_changes={'EXISTANT_ENV_VAR':None})
 
2309
            env_changes={'EXISTANT_ENV_VAR':None})
2618
2310
        # Still set in parent
2619
2311
        self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2620
2312
        del os.environ['EXISTANT_ENV_VAR']
2621
2313
 
2622
2314
    def test_env_del_missing(self):
2623
 
        self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
 
2315
        self.failIf('NON_EXISTANT_ENV_VAR' in os.environ)
2624
2316
        def check_environment():
2625
2317
            self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2626
2318
        self.check_popen_state = check_environment
2627
2319
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2628
 
                          env_changes={'NON_EXISTANT_ENV_VAR':None})
 
2320
            env_changes={'NON_EXISTANT_ENV_VAR':None})
2629
2321
 
2630
2322
    def test_working_dir(self):
2631
2323
        """Test that we can specify the working dir for the child"""
2634
2326
        chdirs = []
2635
2327
        def chdir(path):
2636
2328
            chdirs.append(path)
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')
 
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
2643
2341
        self.assertEqual(['foo', 'current'], chdirs)
2644
2342
 
2645
2343
    def test_get_bzr_path_with_cwd_bzrlib(self):
2665
2363
        self.assertEqual('bzr: interrupted\n', result[1])
2666
2364
 
2667
2365
 
 
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
 
2668
2440
class TestSelftestFiltering(tests.TestCase):
2669
2441
 
2670
2442
    def setUp(self):
2671
 
        super(TestSelftestFiltering, self).setUp()
 
2443
        tests.TestCase.setUp(self)
2672
2444
        self.suite = TestUtil.TestSuite()
2673
2445
        self.loader = TestUtil.TestLoader()
2674
2446
        self.suite.addTest(self.loader.loadTestsFromModule(
2821
2593
        self.assertEqual(remaining_names, _test_ids(split_suite[1]))
2822
2594
 
2823
2595
 
2824
 
class TestCheckTreeShape(tests.TestCaseWithTransport):
 
2596
class TestCheckInventoryShape(tests.TestCaseWithTransport):
2825
2597
 
2826
 
    def test_check_tree_shape(self):
 
2598
    def test_check_inventory_shape(self):
2827
2599
        files = ['a', 'b/', 'b/c']
2828
2600
        tree = self.make_branch_and_tree('.')
2829
2601
        self.build_tree(files)
2830
2602
        tree.add(files)
2831
2603
        tree.lock_read()
2832
2604
        try:
2833
 
            self.check_tree_shape(tree, files)
 
2605
            self.check_inventory_shape(tree.inventory, files)
2834
2606
        finally:
2835
2607
            tree.unlock()
2836
2608
 
2847
2619
            self.run_bzr, ['assert-fail'])
2848
2620
        # make sure we got the real thing, not an error from somewhere else in
2849
2621
        # the test framework
2850
 
        self.assertEqual('always fails', str(e))
 
2622
        self.assertEquals('always fails', str(e))
2851
2623
        # check that there's no traceback in the test log
2852
2624
        self.assertNotContainsRe(self.get_log(), r'Traceback')
2853
2625
 
2902
2674
    def test_load_tests_from_module_name_smoke_test(self):
2903
2675
        loader = TestUtil.TestLoader()
2904
2676
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2905
 
        self.assertEqual(['bzrlib.tests.test_sampler.DemoTest.test_nothing'],
 
2677
        self.assertEquals(['bzrlib.tests.test_sampler.DemoTest.test_nothing'],
2906
2678
                          _test_ids(suite))
2907
2679
 
2908
2680
    def test_load_tests_from_module_name_with_bogus_module_name(self):
2937
2709
 
2938
2710
    def test_empty_list(self):
2939
2711
        id_list = self._create_id_list([])
2940
 
        self.assertEqual({}, id_list.tests)
2941
 
        self.assertEqual({}, id_list.modules)
 
2712
        self.assertEquals({}, id_list.tests)
 
2713
        self.assertEquals({}, id_list.modules)
2942
2714
 
2943
2715
    def test_valid_list(self):
2944
2716
        id_list = self._create_id_list(
2971
2743
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing',
2972
2744
                     'bogus']
2973
2745
        not_found, duplicates = tests.suite_matches_id_list(suite, test_list)
2974
 
        self.assertEqual(['bogus'], not_found)
2975
 
        self.assertEqual([], duplicates)
 
2746
        self.assertEquals(['bogus'], not_found)
 
2747
        self.assertEquals([], duplicates)
2976
2748
 
2977
2749
    def test_suite_matches_id_list_with_duplicates(self):
2978
2750
        loader = TestUtil.TestLoader()
2985
2757
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing',]
2986
2758
        not_found, duplicates = tests.suite_matches_id_list(
2987
2759
            dupes, test_list)
2988
 
        self.assertEqual([], not_found)
2989
 
        self.assertEqual(['bzrlib.tests.test_sampler.DemoTest.test_nothing'],
 
2760
        self.assertEquals([], not_found)
 
2761
        self.assertEquals(['bzrlib.tests.test_sampler.DemoTest.test_nothing'],
2990
2762
                          duplicates)
2991
2763
 
2992
2764
 
3065
2837
        suite = tests.test_suite(test_list,
3066
2838
                                 ['bzrlib.tests.test_selftest.TestTestSuite'])
3067
2839
        # test_test_suite_list_and_start is not included 
3068
 
        self.assertEqual(test_list, _test_ids(suite))
 
2840
        self.assertEquals(test_list, _test_ids(suite))
3069
2841
 
3070
2842
 
3071
2843
class TestLoadTestIdList(tests.TestCaseInTempDir):
3084
2856
        self._create_test_list_file(test_list_fname,
3085
2857
                                    'mod1.cl1.meth1\nmod2.cl2.meth2\n')
3086
2858
        tlist = tests.load_test_id_list(test_list_fname)
3087
 
        self.assertEqual(2, len(tlist))
3088
 
        self.assertEqual('mod1.cl1.meth1', tlist[0])
3089
 
        self.assertEqual('mod2.cl2.meth2', tlist[1])
 
2859
        self.assertEquals(2, len(tlist))
 
2860
        self.assertEquals('mod1.cl1.meth1', tlist[0])
 
2861
        self.assertEquals('mod2.cl2.meth2', tlist[1])
3090
2862
 
3091
2863
    def test_load_dirty_file(self):
3092
2864
        test_list_fname = 'test.list'
3094
2866
                                    '  mod1.cl1.meth1\n\nmod2.cl2.meth2  \n'
3095
2867
                                    'bar baz\n')
3096
2868
        tlist = tests.load_test_id_list(test_list_fname)
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])
 
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])
3102
2874
 
3103
2875
 
3104
2876
class TestFilteredByModuleTestLoader(tests.TestCase):
3112
2884
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing']
3113
2885
        loader = self._create_loader(test_list)
3114
2886
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
3115
 
        self.assertEqual(test_list, _test_ids(suite))
 
2887
        self.assertEquals(test_list, _test_ids(suite))
3116
2888
 
3117
2889
    def test_exclude_tests(self):
3118
2890
        test_list = ['bogus']
3119
2891
        loader = self._create_loader(test_list)
3120
2892
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
3121
 
        self.assertEqual([], _test_ids(suite))
 
2893
        self.assertEquals([], _test_ids(suite))
3122
2894
 
3123
2895
 
3124
2896
class TestFilteredByNameStartTestLoader(tests.TestCase):
3134
2906
        loader = self._create_loader('bzrlib.tests.test_samp')
3135
2907
 
3136
2908
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
3137
 
        self.assertEqual(test_list, _test_ids(suite))
 
2909
        self.assertEquals(test_list, _test_ids(suite))
3138
2910
 
3139
2911
    def test_load_tests_inside_module(self):
3140
2912
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing']
3141
2913
        loader = self._create_loader('bzrlib.tests.test_sampler.Demo')
3142
2914
 
3143
2915
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
3144
 
        self.assertEqual(test_list, _test_ids(suite))
 
2916
        self.assertEquals(test_list, _test_ids(suite))
3145
2917
 
3146
2918
    def test_exclude_tests(self):
3147
2919
        test_list = ['bogus']
3148
2920
        loader = self._create_loader('bogus')
3149
2921
 
3150
2922
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
3151
 
        self.assertEqual([], _test_ids(suite))
 
2923
        self.assertEquals([], _test_ids(suite))
3152
2924
 
3153
2925
 
3154
2926
class TestTestPrefixRegistry(tests.TestCase):
3160
2932
    def test_register_new_prefix(self):
3161
2933
        tpr = self._get_registry()
3162
2934
        tpr.register('foo', 'fff.ooo.ooo')
3163
 
        self.assertEqual('fff.ooo.ooo', tpr.get('foo'))
 
2935
        self.assertEquals('fff.ooo.ooo', tpr.get('foo'))
3164
2936
 
3165
2937
    def test_register_existing_prefix(self):
3166
2938
        tpr = self._get_registry()
3167
2939
        tpr.register('bar', 'bbb.aaa.rrr')
3168
2940
        tpr.register('bar', 'bBB.aAA.rRR')
3169
 
        self.assertEqual('bbb.aaa.rrr', tpr.get('bar'))
 
2941
        self.assertEquals('bbb.aaa.rrr', tpr.get('bar'))
3170
2942
        self.assertThat(self.get_log(),
3171
 
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR",
3172
 
                           doctest.ELLIPSIS))
 
2943
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR", ELLIPSIS))
3173
2944
 
3174
2945
    def test_get_unknown_prefix(self):
3175
2946
        tpr = self._get_registry()
3178
2949
    def test_resolve_prefix(self):
3179
2950
        tpr = self._get_registry()
3180
2951
        tpr.register('bar', 'bb.aa.rr')
3181
 
        self.assertEqual('bb.aa.rr', tpr.resolve_alias('bar'))
 
2952
        self.assertEquals('bb.aa.rr', tpr.resolve_alias('bar'))
3182
2953
 
3183
2954
    def test_resolve_unknown_alias(self):
3184
2955
        tpr = self._get_registry()
3187
2958
 
3188
2959
    def test_predefined_prefixes(self):
3189
2960
        tpr = tests.test_prefix_alias_registry
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)
 
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'))
3352
2967
 
3353
2968
 
3354
2969
class TestRunSuite(tests.TestCase):
3367
2982
                                                self.verbosity)
3368
2983
        tests.run_suite(suite, runner_class=MyRunner, stream=StringIO())
3369
2984
        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')