~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Benoît Pierre
  • Date: 2009-11-02 22:24:29 UTC
  • mto: (4634.96.1 integration-2.0)
  • mto: This revision was merged to the branch mainline in revision 4798.
  • Revision ID: benoit.pierre@gmail.com-20091102222429-xqdyo6n8odh3xbbd
Small fix for handling of short option names in shellcomplete_on_options.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005-2011 Canonical Ltd
 
1
# Copyright (C) 2005, 2006, 2007, 2008, 2009 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 doctest
21
20
import os
22
21
import signal
23
22
import sys
24
 
import threading
25
23
import time
26
24
import unittest
27
25
import warnings
28
26
 
29
 
from testtools import (
30
 
    ExtendedToOriginalDecorator,
31
 
    MultiTestResult,
32
 
    __version__ as testtools_version,
33
 
    )
34
 
from testtools.content import Content
35
 
from testtools.content_type import ContentType
36
 
from testtools.matchers import (
37
 
    DocTestMatches,
38
 
    Equals,
39
 
    )
40
 
import testtools.testresult.doubles
41
 
 
42
27
import bzrlib
43
28
from bzrlib import (
44
29
    branchbuilder,
45
30
    bzrdir,
 
31
    debug,
46
32
    errors,
47
 
    hooks,
48
33
    lockdir,
49
34
    memorytree,
50
35
    osutils,
 
36
    progress,
51
37
    remote,
52
38
    repository,
53
39
    symbol_versioning,
54
40
    tests,
55
 
    transport,
56
41
    workingtree,
57
 
    workingtree_3,
58
 
    workingtree_4,
59
42
    )
60
43
from bzrlib.repofmt import (
61
44
    groupcompress_repo,
 
45
    pack_repo,
 
46
    weaverepo,
62
47
    )
63
48
from bzrlib.symbol_versioning import (
64
49
    deprecated_function,
66
51
    deprecated_method,
67
52
    )
68
53
from bzrlib.tests import (
69
 
    features,
 
54
    SubUnitFeature,
70
55
    test_lsprof,
71
 
    test_server,
 
56
    test_sftp_transport,
72
57
    TestUtil,
73
58
    )
74
 
from bzrlib.trace import note, mutter
75
 
from bzrlib.transport import memory
 
59
from bzrlib.trace import note
 
60
from bzrlib.transport.memory import MemoryServer, MemoryTransport
 
61
from bzrlib.version import _get_bzr_source_tree
76
62
 
77
63
 
78
64
def _test_ids(test_suite):
80
66
    return [t.id() for t in tests.iter_suite_tests(test_suite)]
81
67
 
82
68
 
 
69
class SelftestTests(tests.TestCase):
 
70
 
 
71
    def test_import_tests(self):
 
72
        mod = TestUtil._load_module_by_name('bzrlib.tests.test_selftest')
 
73
        self.assertEqual(mod.SelftestTests, SelftestTests)
 
74
 
 
75
    def test_import_test_failure(self):
 
76
        self.assertRaises(ImportError,
 
77
                          TestUtil._load_module_by_name,
 
78
                          'bzrlib.no-name-yet')
 
79
 
83
80
class MetaTestLog(tests.TestCase):
84
81
 
85
82
    def test_logging(self):
86
83
        """Test logs are captured when a test fails."""
87
84
        self.log('a test message')
88
 
        details = self.getDetails()
89
 
        log = details['log']
90
 
        self.assertThat(log.content_type, Equals(ContentType(
91
 
            "text", "plain", {"charset": "utf8"})))
92
 
        self.assertThat(u"".join(log.iter_text()), Equals(self.get_log()))
93
 
        self.assertThat(self.get_log(),
94
 
            DocTestMatches(u"...a test message\n", doctest.ELLIPSIS))
 
85
        self._log_file.flush()
 
86
        self.assertContainsRe(self._get_log(keep_log_file=True),
 
87
                              'a test message\n')
 
88
 
 
89
 
 
90
class TestUnicodeFilename(tests.TestCase):
 
91
 
 
92
    def test_probe_passes(self):
 
93
        """UnicodeFilename._probe passes."""
 
94
        # We can't test much more than that because the behaviour depends
 
95
        # on the platform.
 
96
        tests.UnicodeFilename._probe()
95
97
 
96
98
 
97
99
class TestTreeShape(tests.TestCaseInTempDir):
98
100
 
99
101
    def test_unicode_paths(self):
100
 
        self.requireFeature(features.UnicodeFilenameFeature)
 
102
        self.requireFeature(tests.UnicodeFilename)
101
103
 
102
104
        filename = u'hell\u00d8'
103
105
        self.build_tree_contents([(filename, 'contents of hello')])
104
 
        self.assertPathExists(filename)
105
 
 
106
 
 
107
 
class TestClassesAvailable(tests.TestCase):
108
 
    """As a convenience we expose Test* classes from bzrlib.tests"""
109
 
 
110
 
    def test_test_case(self):
111
 
        from bzrlib.tests import TestCase
112
 
 
113
 
    def test_test_loader(self):
114
 
        from bzrlib.tests import TestLoader
115
 
 
116
 
    def test_test_suite(self):
117
 
        from bzrlib.tests import TestSuite
 
106
        self.failUnlessExists(filename)
118
107
 
119
108
 
120
109
class TestTransportScenarios(tests.TestCase):
203
192
    def test_scenarios(self):
204
193
        # check that constructor parameters are passed through to the adapted
205
194
        # test.
206
 
        from bzrlib.tests.per_controldir import make_scenarios
 
195
        from bzrlib.tests.per_bzrdir import make_scenarios
207
196
        vfs_factory = "v"
208
197
        server1 = "a"
209
198
        server2 = "b"
307
296
        from bzrlib.tests.per_interrepository import make_scenarios
308
297
        server1 = "a"
309
298
        server2 = "b"
310
 
        formats = [("C0", "C1", "C2", "C3"), ("D0", "D1", "D2", "D3")]
 
299
        formats = [("C0", "C1", "C2"), ("D0", "D1", "D2")]
311
300
        scenarios = make_scenarios(server1, server2, formats)
312
301
        self.assertEqual([
313
302
            ('C0,str,str',
314
303
             {'repository_format': 'C1',
315
304
              'repository_format_to': 'C2',
316
305
              'transport_readonly_server': 'b',
317
 
              'transport_server': 'a',
318
 
              'extra_setup': 'C3'}),
 
306
              'transport_server': 'a'}),
319
307
            ('D0,str,str',
320
308
             {'repository_format': 'D1',
321
309
              'repository_format_to': 'D2',
322
310
              'transport_readonly_server': 'b',
323
 
              'transport_server': 'a',
324
 
              'extra_setup': 'D3'})],
 
311
              'transport_server': 'a'})],
325
312
            scenarios)
326
313
 
327
314
 
333
320
        from bzrlib.tests.per_workingtree import make_scenarios
334
321
        server1 = "a"
335
322
        server2 = "b"
336
 
        formats = [workingtree_4.WorkingTreeFormat4(),
337
 
                   workingtree_3.WorkingTreeFormat3(),]
 
323
        formats = [workingtree.WorkingTreeFormat2(),
 
324
                   workingtree.WorkingTreeFormat3(),]
338
325
        scenarios = make_scenarios(server1, server2, formats)
339
326
        self.assertEqual([
340
 
            ('WorkingTreeFormat4',
 
327
            ('WorkingTreeFormat2',
341
328
             {'bzrdir_format': formats[0]._matchingbzrdir,
342
329
              'transport_readonly_server': 'b',
343
330
              'transport_server': 'a',
370
357
            )
371
358
        server1 = "a"
372
359
        server2 = "b"
373
 
        formats = [workingtree_4.WorkingTreeFormat4(),
374
 
                   workingtree_3.WorkingTreeFormat3(),]
 
360
        formats = [workingtree.WorkingTreeFormat2(),
 
361
                   workingtree.WorkingTreeFormat3(),]
375
362
        scenarios = make_scenarios(server1, server2, formats)
376
363
        self.assertEqual(7, len(scenarios))
377
 
        default_wt_format = workingtree.format_registry.get_default()
378
 
        wt4_format = workingtree_4.WorkingTreeFormat4()
379
 
        wt5_format = workingtree_4.WorkingTreeFormat5()
 
364
        default_wt_format = workingtree.WorkingTreeFormat4._default_format
 
365
        wt4_format = workingtree.WorkingTreeFormat4()
 
366
        wt5_format = workingtree.WorkingTreeFormat5()
380
367
        expected_scenarios = [
381
 
            ('WorkingTreeFormat4',
 
368
            ('WorkingTreeFormat2',
382
369
             {'bzrdir_format': formats[0]._matchingbzrdir,
383
370
              'transport_readonly_server': 'b',
384
371
              'transport_server': 'a',
444
431
        # ones to add.
445
432
        from bzrlib.tests.per_tree import (
446
433
            return_parameter,
 
434
            revision_tree_from_workingtree
447
435
            )
448
436
        from bzrlib.tests.per_intertree import (
449
437
            make_scenarios,
450
438
            )
451
 
        from bzrlib.workingtree_3 import WorkingTreeFormat3
452
 
        from bzrlib.workingtree_4 import WorkingTreeFormat4
 
439
        from bzrlib.workingtree import WorkingTreeFormat2, WorkingTreeFormat3
453
440
        input_test = TestInterTreeScenarios(
454
441
            "test_scenarios")
455
442
        server1 = "a"
456
443
        server2 = "b"
457
 
        format1 = WorkingTreeFormat4()
 
444
        format1 = WorkingTreeFormat2()
458
445
        format2 = WorkingTreeFormat3()
459
446
        formats = [("1", str, format1, format2, "converter1"),
460
447
            ("2", int, format2, format1, "converter2")]
502
489
        self.assertEqualStat(real, fake)
503
490
 
504
491
    def test_assertEqualStat_notequal(self):
505
 
        self.build_tree(["foo", "longname"])
 
492
        self.build_tree(["foo", "bar"])
506
493
        self.assertRaises(AssertionError, self.assertEqualStat,
507
 
            os.lstat("foo"), os.lstat("longname"))
508
 
 
509
 
    def test_failUnlessExists(self):
510
 
        """Deprecated failUnlessExists and failIfExists"""
511
 
        self.applyDeprecated(
512
 
            deprecated_in((2, 4)),
513
 
            self.failUnlessExists, '.')
514
 
        self.build_tree(['foo/', 'foo/bar'])
515
 
        self.applyDeprecated(
516
 
            deprecated_in((2, 4)),
517
 
            self.failUnlessExists, 'foo/bar')
518
 
        self.applyDeprecated(
519
 
            deprecated_in((2, 4)),
520
 
            self.failIfExists, 'foo/foo')
521
 
 
522
 
    def test_assertPathExists(self):
523
 
        self.assertPathExists('.')
524
 
        self.build_tree(['foo/', 'foo/bar'])
525
 
        self.assertPathExists('foo/bar')
526
 
        self.assertPathDoesNotExist('foo/foo')
 
494
            os.lstat("foo"), os.lstat("bar"))
527
495
 
528
496
 
529
497
class TestTestCaseWithMemoryTransport(tests.TestCaseWithMemoryTransport):
547
515
        cwd = osutils.getcwd()
548
516
        self.assertIsSameRealPath(self.test_dir, cwd)
549
517
 
550
 
    def test_BZR_HOME_and_HOME_are_bytestrings(self):
551
 
        """The $BZR_HOME and $HOME environment variables should not be unicode.
552
 
 
553
 
        See https://bugs.launchpad.net/bzr/+bug/464174
554
 
        """
555
 
        self.assertIsInstance(os.environ['BZR_HOME'], str)
556
 
        self.assertIsInstance(os.environ['HOME'], str)
557
 
 
558
518
    def test_make_branch_and_memory_tree(self):
559
519
        """In TestCaseWithMemoryTransport we should not make the branch on disk.
560
520
 
564
524
        tree = self.make_branch_and_memory_tree('dir')
565
525
        # Guard against regression into MemoryTransport leaking
566
526
        # files to disk instead of keeping them in memory.
567
 
        self.assertFalse(osutils.lexists('dir'))
 
527
        self.failIf(osutils.lexists('dir'))
568
528
        self.assertIsInstance(tree, memorytree.MemoryTree)
569
529
 
570
530
    def test_make_branch_and_memory_tree_with_format(self):
571
531
        """make_branch_and_memory_tree should accept a format option."""
572
532
        format = bzrdir.BzrDirMetaFormat1()
573
 
        format.repository_format = repository.format_registry.get_default()
 
533
        format.repository_format = weaverepo.RepositoryFormat7()
574
534
        tree = self.make_branch_and_memory_tree('dir', format=format)
575
535
        # Guard against regression into MemoryTransport leaking
576
536
        # files to disk instead of keeping them in memory.
577
 
        self.assertFalse(osutils.lexists('dir'))
 
537
        self.failIf(osutils.lexists('dir'))
578
538
        self.assertIsInstance(tree, memorytree.MemoryTree)
579
539
        self.assertEqual(format.repository_format.__class__,
580
540
            tree.branch.repository._format.__class__)
584
544
        self.assertIsInstance(builder, branchbuilder.BranchBuilder)
585
545
        # Guard against regression into MemoryTransport leaking
586
546
        # files to disk instead of keeping them in memory.
587
 
        self.assertFalse(osutils.lexists('dir'))
 
547
        self.failIf(osutils.lexists('dir'))
588
548
 
589
549
    def test_make_branch_builder_with_format(self):
590
550
        # Use a repo layout that doesn't conform to a 'named' layout, to ensure
591
551
        # that the format objects are used.
592
552
        format = bzrdir.BzrDirMetaFormat1()
593
 
        repo_format = repository.format_registry.get_default()
 
553
        repo_format = weaverepo.RepositoryFormat7()
594
554
        format.repository_format = repo_format
595
555
        builder = self.make_branch_builder('dir', format=format)
596
556
        the_branch = builder.get_branch()
597
557
        # Guard against regression into MemoryTransport leaking
598
558
        # files to disk instead of keeping them in memory.
599
 
        self.assertFalse(osutils.lexists('dir'))
 
559
        self.failIf(osutils.lexists('dir'))
600
560
        self.assertEqual(format.repository_format.__class__,
601
561
                         the_branch.repository._format.__class__)
602
562
        self.assertEqual(repo_format.get_format_string(),
608
568
        the_branch = builder.get_branch()
609
569
        # Guard against regression into MemoryTransport leaking
610
570
        # files to disk instead of keeping them in memory.
611
 
        self.assertFalse(osutils.lexists('dir'))
 
571
        self.failIf(osutils.lexists('dir'))
612
572
        dir_format = bzrdir.format_registry.make_bzrdir('knit')
613
573
        self.assertEqual(dir_format.repository_format.__class__,
614
574
                         the_branch.repository._format.__class__)
616
576
                         self.get_transport().get_bytes(
617
577
                            'dir/.bzr/repository/format'))
618
578
 
 
579
    def test_safety_net(self):
 
580
        """No test should modify the safety .bzr directory.
 
581
 
 
582
        We just test that the _check_safety_net private method raises
 
583
        AssertionError, it's easier than building a test suite with the same
 
584
        test.
 
585
        """
 
586
        # Oops, a commit in the current directory (i.e. without local .bzr
 
587
        # directory) will crawl up the hierarchy to find a .bzr directory.
 
588
        self.run_bzr(['commit', '-mfoo', '--unchanged'])
 
589
        # But we have a safety net in place.
 
590
        self.assertRaises(AssertionError, self._check_safety_net)
 
591
 
619
592
    def test_dangling_locks_cause_failures(self):
620
593
        class TestDanglingLock(tests.TestCaseWithMemoryTransport):
621
594
            def test_function(self):
625
598
                l.attempt_lock()
626
599
        test = TestDanglingLock('test_function')
627
600
        result = test.run()
628
 
        total_failures = result.errors + result.failures
629
601
        if self._lock_check_thorough:
630
 
            self.assertEqual(1, len(total_failures))
 
602
            self.assertEqual(1, len(result.errors))
631
603
        else:
632
604
            # When _lock_check_thorough is disabled, then we don't trigger a
633
605
            # failure
634
 
            self.assertEqual(0, len(total_failures))
 
606
            self.assertEqual(0, len(result.errors))
635
607
 
636
608
 
637
609
class TestTestCaseWithTransport(tests.TestCaseWithTransport):
638
610
    """Tests for the convenience functions TestCaseWithTransport introduces."""
639
611
 
640
612
    def test_get_readonly_url_none(self):
 
613
        from bzrlib.transport import get_transport
 
614
        from bzrlib.transport.memory import MemoryServer
641
615
        from bzrlib.transport.readonly import ReadonlyTransportDecorator
642
 
        self.vfs_transport_factory = memory.MemoryServer
 
616
        self.vfs_transport_factory = MemoryServer
643
617
        self.transport_readonly_server = None
644
618
        # calling get_readonly_transport() constructs a decorator on the url
645
619
        # for the server
646
620
        url = self.get_readonly_url()
647
621
        url2 = self.get_readonly_url('foo/bar')
648
 
        t = transport.get_transport(url)
649
 
        t2 = transport.get_transport(url2)
650
 
        self.assertIsInstance(t, ReadonlyTransportDecorator)
651
 
        self.assertIsInstance(t2, ReadonlyTransportDecorator)
 
622
        t = get_transport(url)
 
623
        t2 = get_transport(url2)
 
624
        self.failUnless(isinstance(t, ReadonlyTransportDecorator))
 
625
        self.failUnless(isinstance(t2, ReadonlyTransportDecorator))
652
626
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
653
627
 
654
628
    def test_get_readonly_url_http(self):
655
629
        from bzrlib.tests.http_server import HttpServer
 
630
        from bzrlib.transport import get_transport
 
631
        from bzrlib.transport.local import LocalURLServer
656
632
        from bzrlib.transport.http import HttpTransportBase
657
 
        self.transport_server = test_server.LocalURLServer
 
633
        self.transport_server = LocalURLServer
658
634
        self.transport_readonly_server = HttpServer
659
635
        # calling get_readonly_transport() gives us a HTTP server instance.
660
636
        url = self.get_readonly_url()
661
637
        url2 = self.get_readonly_url('foo/bar')
662
638
        # the transport returned may be any HttpTransportBase subclass
663
 
        t = transport.get_transport(url)
664
 
        t2 = transport.get_transport(url2)
665
 
        self.assertIsInstance(t, HttpTransportBase)
666
 
        self.assertIsInstance(t2, HttpTransportBase)
 
639
        t = get_transport(url)
 
640
        t2 = get_transport(url2)
 
641
        self.failUnless(isinstance(t, HttpTransportBase))
 
642
        self.failUnless(isinstance(t2, HttpTransportBase))
667
643
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
668
644
 
669
645
    def test_is_directory(self):
677
653
    def test_make_branch_builder(self):
678
654
        builder = self.make_branch_builder('dir')
679
655
        rev_id = builder.build_commit()
680
 
        self.assertPathExists('dir')
 
656
        self.failUnlessExists('dir')
681
657
        a_dir = bzrdir.BzrDir.open('dir')
682
658
        self.assertRaises(errors.NoWorkingTree, a_dir.open_workingtree)
683
659
        a_branch = a_dir.open_branch()
691
667
 
692
668
    def setUp(self):
693
669
        super(TestTestCaseTransports, self).setUp()
694
 
        self.vfs_transport_factory = memory.MemoryServer
 
670
        self.vfs_transport_factory = MemoryServer
695
671
 
696
672
    def test_make_bzrdir_preserves_transport(self):
697
673
        t = self.get_transport()
698
674
        result_bzrdir = self.make_bzrdir('subdir')
699
675
        self.assertIsInstance(result_bzrdir.transport,
700
 
                              memory.MemoryTransport)
 
676
                              MemoryTransport)
701
677
        # should not be on disk, should only be in memory
702
 
        self.assertPathDoesNotExist('subdir')
 
678
        self.failIfExists('subdir')
703
679
 
704
680
 
705
681
class TestChrootedTest(tests.ChrootedTestCase):
706
682
 
707
683
    def test_root_is_root(self):
708
 
        t = transport.get_transport(self.get_readonly_url())
 
684
        from bzrlib.transport import get_transport
 
685
        t = get_transport(self.get_readonly_url())
709
686
        url = t.base
710
687
        self.assertEqual(url, t.clone('..').base)
711
688
 
712
689
 
713
 
class TestProfileResult(tests.TestCase):
714
 
 
715
 
    def test_profiles_tests(self):
716
 
        self.requireFeature(features.lsprof_feature)
717
 
        terminal = testtools.testresult.doubles.ExtendedTestResult()
718
 
        result = tests.ProfileResult(terminal)
719
 
        class Sample(tests.TestCase):
720
 
            def a(self):
721
 
                self.sample_function()
722
 
            def sample_function(self):
723
 
                pass
724
 
        test = Sample("a")
725
 
        test.run(result)
726
 
        case = terminal._events[0][1]
727
 
        self.assertLength(1, case._benchcalls)
728
 
        # We must be able to unpack it as the test reporting code wants
729
 
        (_, _, _), stats = case._benchcalls[0]
730
 
        self.assertTrue(callable(stats.pprint))
731
 
 
732
 
 
733
690
class TestTestResult(tests.TestCase):
734
691
 
735
692
    def check_timing(self, test_case, expected_re):
737
694
                descriptions=0,
738
695
                verbosity=1,
739
696
                )
740
 
        capture = testtools.testresult.doubles.ExtendedTestResult()
741
 
        test_case.run(MultiTestResult(result, capture))
742
 
        run_case = capture._events[0][1]
743
 
        timed_string = result._testTimeString(run_case)
 
697
        test_case.run(result)
 
698
        timed_string = result._testTimeString(test_case)
744
699
        self.assertContainsRe(timed_string, expected_re)
745
700
 
746
701
    def test_test_reporting(self):
764
719
        self.check_timing(ShortDelayTestCase('test_short_delay'),
765
720
                          r"^ +[0-9]+ms$")
766
721
 
 
722
    def test_assigned_benchmark_file_stores_date(self):
 
723
        output = StringIO()
 
724
        result = bzrlib.tests.TextTestResult(self._log_file,
 
725
                                        descriptions=0,
 
726
                                        verbosity=1,
 
727
                                        bench_history=output
 
728
                                        )
 
729
        output_string = output.getvalue()
 
730
        # if you are wondering about the regexp please read the comment in
 
731
        # test_bench_history (bzrlib.tests.test_selftest.TestRunner)
 
732
        # XXX: what comment?  -- Andrew Bennetts
 
733
        self.assertContainsRe(output_string, "--date [0-9.]+")
 
734
 
 
735
    def test_benchhistory_records_test_times(self):
 
736
        result_stream = StringIO()
 
737
        result = bzrlib.tests.TextTestResult(
 
738
            self._log_file,
 
739
            descriptions=0,
 
740
            verbosity=1,
 
741
            bench_history=result_stream
 
742
            )
 
743
 
 
744
        # we want profile a call and check that its test duration is recorded
 
745
        # make a new test instance that when run will generate a benchmark
 
746
        example_test_case = TestTestResult("_time_hello_world_encoding")
 
747
        # execute the test, which should succeed and record times
 
748
        example_test_case.run(result)
 
749
        lines = result_stream.getvalue().splitlines()
 
750
        self.assertEqual(2, len(lines))
 
751
        self.assertContainsRe(lines[1],
 
752
            " *[0-9]+ms bzrlib.tests.test_selftest.TestTestResult"
 
753
            "._time_hello_world_encoding")
 
754
 
767
755
    def _time_hello_world_encoding(self):
768
756
        """Profile two sleep calls
769
757
 
774
762
 
775
763
    def test_lsprofiling(self):
776
764
        """Verbose test result prints lsprof statistics from test cases."""
777
 
        self.requireFeature(features.lsprof_feature)
 
765
        self.requireFeature(test_lsprof.LSProfFeature)
778
766
        result_stream = StringIO()
779
767
        result = bzrlib.tests.VerboseTestResult(
780
 
            result_stream,
 
768
            unittest._WritelnDecorator(result_stream),
781
769
            descriptions=0,
782
770
            verbosity=2,
783
771
            )
809
797
        self.assertContainsRe(output,
810
798
            r"LSProf output for <type 'unicode'>\(\('world',\), {'errors': 'replace'}\)\n")
811
799
 
812
 
    def test_uses_time_from_testtools(self):
813
 
        """Test case timings in verbose results should use testtools times"""
814
 
        import datetime
815
 
        class TimeAddedVerboseTestResult(tests.VerboseTestResult):
816
 
            def startTest(self, test):
817
 
                self.time(datetime.datetime.utcfromtimestamp(1.145))
818
 
                super(TimeAddedVerboseTestResult, self).startTest(test)
819
 
            def addSuccess(self, test):
820
 
                self.time(datetime.datetime.utcfromtimestamp(51.147))
821
 
                super(TimeAddedVerboseTestResult, self).addSuccess(test)
822
 
            def report_tests_starting(self): pass
823
 
        sio = StringIO()
824
 
        self.get_passing_test().run(TimeAddedVerboseTestResult(sio, 0, 2))
825
 
        self.assertEndsWith(sio.getvalue(), "OK    50002ms\n")
826
 
 
827
800
    def test_known_failure(self):
828
 
        """Using knownFailure should trigger several result actions."""
 
801
        """A KnownFailure being raised should trigger several result actions."""
829
802
        class InstrumentedTestResult(tests.ExtendedTestResult):
830
 
            def stopTestRun(self): pass
831
 
            def report_tests_starting(self): pass
832
 
            def report_known_failure(self, test, err=None, details=None):
833
 
                self._call = test, 'known failure'
 
803
            def done(self): pass
 
804
            def startTests(self): pass
 
805
            def report_test_start(self, test): pass
 
806
            def report_known_failure(self, test, err):
 
807
                self._call = test, err
834
808
        result = InstrumentedTestResult(None, None, None, None)
835
 
        class Test(tests.TestCase):
836
 
            def test_function(self):
837
 
                self.knownFailure('failed!')
838
 
        test = Test("test_function")
 
809
        def test_function():
 
810
            raise tests.KnownFailure('failed!')
 
811
        test = unittest.FunctionTestCase(test_function)
839
812
        test.run(result)
840
813
        # it should invoke 'report_known_failure'.
841
814
        self.assertEqual(2, len(result._call))
842
 
        self.assertEqual(test.id(), result._call[0].id())
843
 
        self.assertEqual('known failure', result._call[1])
 
815
        self.assertEqual(test, result._call[0])
 
816
        self.assertEqual(tests.KnownFailure, result._call[1][0])
 
817
        self.assertIsInstance(result._call[1][1], tests.KnownFailure)
844
818
        # we dont introspec the traceback, if the rest is ok, it would be
845
819
        # exceptional for it not to be.
846
820
        # it should update the known_failure_count on the object.
852
826
        # verbose test output formatting
853
827
        result_stream = StringIO()
854
828
        result = bzrlib.tests.VerboseTestResult(
855
 
            result_stream,
 
829
            unittest._WritelnDecorator(result_stream),
856
830
            descriptions=0,
857
831
            verbosity=2,
858
832
            )
859
 
        _get_test("test_xfail").run(result)
860
 
        self.assertContainsRe(result_stream.getvalue(),
861
 
            "\n\\S+\\.test_xfail\\s+XFAIL\\s+\\d+ms\n"
862
 
            "\\s*(?:Text attachment: )?reason"
863
 
            "(?:\n-+\n|: {{{)"
864
 
            "this_fails"
865
 
            "(?:\n-+\n|}}}\n)")
 
833
        test = self.get_passing_test()
 
834
        result.startTest(test)
 
835
        prefix = len(result_stream.getvalue())
 
836
        # the err parameter has the shape:
 
837
        # (class, exception object, traceback)
 
838
        # KnownFailures dont get their tracebacks shown though, so we
 
839
        # can skip that.
 
840
        err = (tests.KnownFailure, tests.KnownFailure('foo'), None)
 
841
        result.report_known_failure(test, err)
 
842
        output = result_stream.getvalue()[prefix:]
 
843
        lines = output.splitlines()
 
844
        self.assertContainsRe(lines[0], r'XFAIL *\d+ms$')
 
845
        self.assertEqual(lines[1], '    foo')
 
846
        self.assertEqual(2, len(lines))
866
847
 
867
848
    def get_passing_test(self):
868
849
        """Return a test object that can't be run usefully."""
873
854
    def test_add_not_supported(self):
874
855
        """Test the behaviour of invoking addNotSupported."""
875
856
        class InstrumentedTestResult(tests.ExtendedTestResult):
876
 
            def stopTestRun(self): pass
877
 
            def report_tests_starting(self): pass
 
857
            def done(self): pass
 
858
            def startTests(self): pass
 
859
            def report_test_start(self, test): pass
878
860
            def report_unsupported(self, test, feature):
879
861
                self._call = test, feature
880
862
        result = InstrumentedTestResult(None, None, None, None)
881
863
        test = SampleTestCase('_test_pass')
882
 
        feature = features.Feature()
 
864
        feature = tests.Feature()
883
865
        result.startTest(test)
884
866
        result.addNotSupported(test, feature)
885
867
        # it should invoke 'report_unsupported'.
899
881
        # verbose test output formatting
900
882
        result_stream = StringIO()
901
883
        result = bzrlib.tests.VerboseTestResult(
902
 
            result_stream,
 
884
            unittest._WritelnDecorator(result_stream),
903
885
            descriptions=0,
904
886
            verbosity=2,
905
887
            )
906
888
        test = self.get_passing_test()
907
 
        feature = features.Feature()
 
889
        feature = tests.Feature()
908
890
        result.startTest(test)
909
891
        prefix = len(result_stream.getvalue())
910
892
        result.report_unsupported(test, feature)
911
893
        output = result_stream.getvalue()[prefix:]
912
894
        lines = output.splitlines()
913
 
        # We don't check for the final '0ms' since it may fail on slow hosts
914
 
        self.assertStartsWith(lines[0], 'NODEP')
915
 
        self.assertEqual(lines[1],
916
 
                         "    The feature 'Feature' is not available.")
 
895
        self.assertEqual(lines, ['NODEP        0ms',
 
896
                                 "    The feature 'Feature' is not available."])
917
897
 
918
898
    def test_unavailable_exception(self):
919
899
        """An UnavailableFeature being raised should invoke addNotSupported."""
920
900
        class InstrumentedTestResult(tests.ExtendedTestResult):
921
 
            def stopTestRun(self): pass
922
 
            def report_tests_starting(self): pass
 
901
            def done(self): pass
 
902
            def startTests(self): pass
 
903
            def report_test_start(self, test): pass
923
904
            def addNotSupported(self, test, feature):
924
905
                self._call = test, feature
925
906
        result = InstrumentedTestResult(None, None, None, None)
926
 
        feature = features.Feature()
927
 
        class Test(tests.TestCase):
928
 
            def test_function(self):
929
 
                raise tests.UnavailableFeature(feature)
930
 
        test = Test("test_function")
 
907
        feature = tests.Feature()
 
908
        def test_function():
 
909
            raise tests.UnavailableFeature(feature)
 
910
        test = unittest.FunctionTestCase(test_function)
931
911
        test.run(result)
932
912
        # it should invoke 'addNotSupported'.
933
913
        self.assertEqual(2, len(result._call))
934
 
        self.assertEqual(test.id(), result._call[0].id())
 
914
        self.assertEqual(test, result._call[0])
935
915
        self.assertEqual(feature, result._call[1])
936
916
        # and not count as an error
937
917
        self.assertEqual(0, result.error_count)
948
928
    def test_strict_with_known_failure(self):
949
929
        result = bzrlib.tests.TextTestResult(self._log_file, descriptions=0,
950
930
                                             verbosity=1)
951
 
        test = _get_test("test_xfail")
952
 
        test.run(result)
 
931
        test = self.get_passing_test()
 
932
        err = (tests.KnownFailure, tests.KnownFailure('foo'), None)
 
933
        result._addKnownFailure(test, err)
953
934
        self.assertFalse(result.wasStrictlySuccessful())
954
935
        self.assertEqual(None, result._extractBenchmarkTime(test))
955
936
 
966
947
        class InstrumentedTestResult(tests.ExtendedTestResult):
967
948
            calls = 0
968
949
            def startTests(self): self.calls += 1
 
950
            def report_test_start(self, test): pass
969
951
        result = InstrumentedTestResult(None, None, None, None)
970
952
        def test_function():
971
953
            pass
973
955
        test.run(result)
974
956
        self.assertEquals(1, result.calls)
975
957
 
976
 
    def test_startTests_only_once(self):
977
 
        """With multiple tests startTests should still only be called once"""
978
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
979
 
            calls = 0
980
 
            def startTests(self): self.calls += 1
981
 
        result = InstrumentedTestResult(None, None, None, None)
982
 
        suite = unittest.TestSuite([
983
 
            unittest.FunctionTestCase(lambda: None),
984
 
            unittest.FunctionTestCase(lambda: None)])
985
 
        suite.run(result)
986
 
        self.assertEquals(1, result.calls)
987
 
        self.assertEquals(2, result.count)
 
958
 
 
959
class TestUnicodeFilenameFeature(tests.TestCase):
 
960
 
 
961
    def test_probe_passes(self):
 
962
        """UnicodeFilenameFeature._probe passes."""
 
963
        # We can't test much more than that because the behaviour depends
 
964
        # on the platform.
 
965
        tests.UnicodeFilenameFeature._probe()
988
966
 
989
967
 
990
968
class TestRunner(tests.TestCase):
1012
990
    def test_known_failure_failed_run(self):
1013
991
        # run a test that generates a known failure which should be printed in
1014
992
        # the final output when real failures occur.
1015
 
        class Test(tests.TestCase):
1016
 
            def known_failure_test(self):
1017
 
                self.expectFailure('failed', self.assertTrue, False)
 
993
        def known_failure_test():
 
994
            raise tests.KnownFailure('failed')
1018
995
        test = unittest.TestSuite()
1019
 
        test.addTest(Test("known_failure_test"))
 
996
        test.addTest(unittest.FunctionTestCase(known_failure_test))
1020
997
        def failing_test():
1021
998
            raise AssertionError('foo')
1022
999
        test.addTest(unittest.FunctionTestCase(failing_test))
1025
1002
        result = self.run_test_runner(runner, test)
1026
1003
        lines = stream.getvalue().splitlines()
1027
1004
        self.assertContainsRe(stream.getvalue(),
1028
 
            '(?sm)^bzr selftest.*$'
 
1005
            '(?sm)^testing.*$'
1029
1006
            '.*'
1030
1007
            '^======================================================================\n'
1031
 
            '^FAIL: failing_test\n'
 
1008
            '^FAIL: unittest.FunctionTestCase \\(failing_test\\)\n'
1032
1009
            '^----------------------------------------------------------------------\n'
1033
1010
            'Traceback \\(most recent call last\\):\n'
1034
1011
            '  .*' # File .*, line .*, in failing_test' - but maybe not from .pyc
1040
1017
            )
1041
1018
 
1042
1019
    def test_known_failure_ok_run(self):
1043
 
        # run a test that generates a known failure which should be printed in
1044
 
        # the final output.
1045
 
        class Test(tests.TestCase):
1046
 
            def known_failure_test(self):
1047
 
                self.knownFailure("Never works...")
1048
 
        test = Test("known_failure_test")
 
1020
        # run a test that generates a known failure which should be printed in the final output.
 
1021
        def known_failure_test():
 
1022
            raise tests.KnownFailure('failed')
 
1023
        test = unittest.FunctionTestCase(known_failure_test)
1049
1024
        stream = StringIO()
1050
1025
        runner = tests.TextTestRunner(stream=stream)
1051
1026
        result = self.run_test_runner(runner, test)
1056
1031
            '\n'
1057
1032
            'OK \\(known_failures=1\\)\n')
1058
1033
 
1059
 
    def test_unexpected_success_bad(self):
1060
 
        class Test(tests.TestCase):
1061
 
            def test_truth(self):
1062
 
                self.expectFailure("No absolute truth", self.assertTrue, True)
1063
 
        runner = tests.TextTestRunner(stream=StringIO())
1064
 
        result = self.run_test_runner(runner, Test("test_truth"))
1065
 
        if testtools_version[:3] <= (0, 9, 11):
1066
 
            self.assertContainsRe(runner.stream.getvalue(),
1067
 
                "=+\n"
1068
 
                "FAIL: \\S+\.test_truth\n"
1069
 
                "-+\n"
1070
 
                "(?:.*\n)*"
1071
 
                "No absolute truth\n"
1072
 
                "(?:.*\n)*"
1073
 
                "-+\n"
1074
 
                "Ran 1 test in .*\n"
1075
 
                "\n"
1076
 
                "FAILED \\(failures=1\\)\n\\Z")
1077
 
        else:
1078
 
            self.assertContainsRe(runner.stream.getvalue(),
1079
 
                "=+\n"
1080
 
                "FAIL: \\S+\.test_truth\n"
1081
 
                "-+\n"
1082
 
                "Empty attachments:\n"
1083
 
                "  log\n"
1084
 
                "\n"
1085
 
                "reason: {{{No absolute truth}}}\n"
1086
 
                "-+\n"
1087
 
                "Ran 1 test in .*\n"
1088
 
                "\n"
1089
 
                "FAILED \\(failures=1\\)\n\\Z")
1090
 
 
1091
 
    def test_result_decorator(self):
1092
 
        # decorate results
1093
 
        calls = []
1094
 
        class LoggingDecorator(ExtendedToOriginalDecorator):
1095
 
            def startTest(self, test):
1096
 
                ExtendedToOriginalDecorator.startTest(self, test)
1097
 
                calls.append('start')
1098
 
        test = unittest.FunctionTestCase(lambda:None)
1099
 
        stream = StringIO()
1100
 
        runner = tests.TextTestRunner(stream=stream,
1101
 
            result_decorators=[LoggingDecorator])
1102
 
        result = self.run_test_runner(runner, test)
1103
 
        self.assertLength(1, calls)
1104
 
 
1105
1034
    def test_skipped_test(self):
1106
1035
        # run a test that is skipped, and check the suite as a whole still
1107
1036
        # succeeds.
1160
1089
 
1161
1090
    def test_not_applicable(self):
1162
1091
        # run a test that is skipped because it's not applicable
1163
 
        class Test(tests.TestCase):
1164
 
            def not_applicable_test(self):
1165
 
                raise tests.TestNotApplicable('this test never runs')
 
1092
        def not_applicable_test():
 
1093
            raise tests.TestNotApplicable('this test never runs')
1166
1094
        out = StringIO()
1167
1095
        runner = tests.TextTestRunner(stream=out, verbosity=2)
1168
 
        test = Test("not_applicable_test")
 
1096
        test = unittest.FunctionTestCase(not_applicable_test)
1169
1097
        result = self.run_test_runner(runner, test)
1170
1098
        self._log_file.write(out.getvalue())
1171
1099
        self.assertTrue(result.wasSuccessful())
1175
1103
        self.assertContainsRe(out.getvalue(),
1176
1104
                r'(?m)^    this test never runs')
1177
1105
 
 
1106
    def test_not_applicable_demo(self):
 
1107
        # just so you can see it in the test output
 
1108
        raise tests.TestNotApplicable('this test is just a demonstation')
 
1109
 
1178
1110
    def test_unsupported_features_listed(self):
1179
1111
        """When unsupported features are encountered they are detailed."""
1180
 
        class Feature1(features.Feature):
 
1112
        class Feature1(tests.Feature):
1181
1113
            def _probe(self): return False
1182
 
        class Feature2(features.Feature):
 
1114
        class Feature2(tests.Feature):
1183
1115
            def _probe(self): return False
1184
1116
        # create sample tests
1185
1117
        test1 = SampleTestCase('_test_pass')
1200
1132
            ],
1201
1133
            lines[-3:])
1202
1134
 
1203
 
    def test_verbose_test_count(self):
1204
 
        """A verbose test run reports the right test count at the start"""
1205
 
        suite = TestUtil.TestSuite([
1206
 
            unittest.FunctionTestCase(lambda:None),
1207
 
            unittest.FunctionTestCase(lambda:None)])
1208
 
        self.assertEqual(suite.countTestCases(), 2)
1209
 
        stream = StringIO()
1210
 
        runner = tests.TextTestRunner(stream=stream, verbosity=2)
1211
 
        # Need to use the CountingDecorator as that's what sets num_tests
1212
 
        result = self.run_test_runner(runner, tests.CountingDecorator(suite))
1213
 
        self.assertStartsWith(stream.getvalue(), "running 2 tests")
1214
 
 
1215
 
    def test_startTestRun(self):
1216
 
        """run should call result.startTestRun()"""
1217
 
        calls = []
1218
 
        class LoggingDecorator(ExtendedToOriginalDecorator):
1219
 
            def startTestRun(self):
1220
 
                ExtendedToOriginalDecorator.startTestRun(self)
1221
 
                calls.append('startTestRun')
1222
 
        test = unittest.FunctionTestCase(lambda:None)
1223
 
        stream = StringIO()
1224
 
        runner = tests.TextTestRunner(stream=stream,
1225
 
            result_decorators=[LoggingDecorator])
1226
 
        result = self.run_test_runner(runner, test)
1227
 
        self.assertLength(1, calls)
1228
 
 
1229
 
    def test_stopTestRun(self):
1230
 
        """run should call result.stopTestRun()"""
1231
 
        calls = []
1232
 
        class LoggingDecorator(ExtendedToOriginalDecorator):
1233
 
            def stopTestRun(self):
1234
 
                ExtendedToOriginalDecorator.stopTestRun(self)
1235
 
                calls.append('stopTestRun')
1236
 
        test = unittest.FunctionTestCase(lambda:None)
1237
 
        stream = StringIO()
1238
 
        runner = tests.TextTestRunner(stream=stream,
1239
 
            result_decorators=[LoggingDecorator])
1240
 
        result = self.run_test_runner(runner, test)
1241
 
        self.assertLength(1, calls)
1242
 
 
1243
 
    def test_unicode_test_output_on_ascii_stream(self):
1244
 
        """Showing results should always succeed even on an ascii console"""
1245
 
        class FailureWithUnicode(tests.TestCase):
1246
 
            def test_log_unicode(self):
1247
 
                self.log(u"\u2606")
1248
 
                self.fail("Now print that log!")
1249
 
        out = StringIO()
1250
 
        self.overrideAttr(osutils, "get_terminal_encoding",
1251
 
            lambda trace=False: "ascii")
1252
 
        result = self.run_test_runner(tests.TextTestRunner(stream=out),
1253
 
            FailureWithUnicode("test_log_unicode"))
1254
 
        if testtools_version[:3] > (0, 9, 11):
1255
 
            self.assertContainsRe(out.getvalue(), "log: {{{\d+\.\d+  \\\\u2606}}}")
1256
 
        else:
1257
 
            self.assertContainsRe(out.getvalue(),
1258
 
                "Text attachment: log\n"
1259
 
                "-+\n"
1260
 
                "\d+\.\d+  \\\\u2606\n"
1261
 
                "-+\n")
 
1135
    def test_bench_history(self):
 
1136
        # tests that the running the benchmark produces a history file
 
1137
        # containing a timestamp and the revision id of the bzrlib source which
 
1138
        # was tested.
 
1139
        workingtree = _get_bzr_source_tree()
 
1140
        test = TestRunner('dummy_test')
 
1141
        output = StringIO()
 
1142
        runner = tests.TextTestRunner(stream=self._log_file,
 
1143
                                      bench_history=output)
 
1144
        result = self.run_test_runner(runner, test)
 
1145
        output_string = output.getvalue()
 
1146
        self.assertContainsRe(output_string, "--date [0-9.]+")
 
1147
        if workingtree is not None:
 
1148
            revision_id = workingtree.get_parent_ids()[0]
 
1149
            self.assertEndsWith(output_string.rstrip(), revision_id)
 
1150
 
 
1151
    def assertLogDeleted(self, test):
 
1152
        log = test._get_log()
 
1153
        self.assertEqual("DELETED log file to reduce memory footprint", log)
 
1154
        self.assertEqual('', test._log_contents)
 
1155
        self.assertIs(None, test._log_file_name)
 
1156
 
 
1157
    def test_success_log_deleted(self):
 
1158
        """Successful tests have their log deleted"""
 
1159
 
 
1160
        class LogTester(tests.TestCase):
 
1161
 
 
1162
            def test_success(self):
 
1163
                self.log('this will be removed\n')
 
1164
 
 
1165
        sio = StringIO()
 
1166
        runner = tests.TextTestRunner(stream=sio)
 
1167
        test = LogTester('test_success')
 
1168
        result = self.run_test_runner(runner, test)
 
1169
 
 
1170
        self.assertLogDeleted(test)
 
1171
 
 
1172
    def test_skipped_log_deleted(self):
 
1173
        """Skipped tests have their log deleted"""
 
1174
 
 
1175
        class LogTester(tests.TestCase):
 
1176
 
 
1177
            def test_skipped(self):
 
1178
                self.log('this will be removed\n')
 
1179
                raise tests.TestSkipped()
 
1180
 
 
1181
        sio = StringIO()
 
1182
        runner = tests.TextTestRunner(stream=sio)
 
1183
        test = LogTester('test_skipped')
 
1184
        result = self.run_test_runner(runner, test)
 
1185
 
 
1186
        self.assertLogDeleted(test)
 
1187
 
 
1188
    def test_not_aplicable_log_deleted(self):
 
1189
        """Not applicable tests have their log deleted"""
 
1190
 
 
1191
        class LogTester(tests.TestCase):
 
1192
 
 
1193
            def test_not_applicable(self):
 
1194
                self.log('this will be removed\n')
 
1195
                raise tests.TestNotApplicable()
 
1196
 
 
1197
        sio = StringIO()
 
1198
        runner = tests.TextTestRunner(stream=sio)
 
1199
        test = LogTester('test_not_applicable')
 
1200
        result = self.run_test_runner(runner, test)
 
1201
 
 
1202
        self.assertLogDeleted(test)
 
1203
 
 
1204
    def test_known_failure_log_deleted(self):
 
1205
        """Know failure tests have their log deleted"""
 
1206
 
 
1207
        class LogTester(tests.TestCase):
 
1208
 
 
1209
            def test_known_failure(self):
 
1210
                self.log('this will be removed\n')
 
1211
                raise tests.KnownFailure()
 
1212
 
 
1213
        sio = StringIO()
 
1214
        runner = tests.TextTestRunner(stream=sio)
 
1215
        test = LogTester('test_known_failure')
 
1216
        result = self.run_test_runner(runner, test)
 
1217
 
 
1218
        self.assertLogDeleted(test)
 
1219
 
 
1220
    def test_fail_log_kept(self):
 
1221
        """Failed tests have their log kept"""
 
1222
 
 
1223
        class LogTester(tests.TestCase):
 
1224
 
 
1225
            def test_fail(self):
 
1226
                self.log('this will be kept\n')
 
1227
                self.fail('this test fails')
 
1228
 
 
1229
        sio = StringIO()
 
1230
        runner = tests.TextTestRunner(stream=sio)
 
1231
        test = LogTester('test_fail')
 
1232
        result = self.run_test_runner(runner, test)
 
1233
 
 
1234
        text = sio.getvalue()
 
1235
        self.assertContainsRe(text, 'this will be kept')
 
1236
        self.assertContainsRe(text, 'this test fails')
 
1237
 
 
1238
        log = test._get_log()
 
1239
        self.assertContainsRe(log, 'this will be kept')
 
1240
        self.assertEqual(log, test._log_contents)
 
1241
 
 
1242
    def test_error_log_kept(self):
 
1243
        """Tests with errors have their log kept"""
 
1244
 
 
1245
        class LogTester(tests.TestCase):
 
1246
 
 
1247
            def test_error(self):
 
1248
                self.log('this will be kept\n')
 
1249
                raise ValueError('random exception raised')
 
1250
 
 
1251
        sio = StringIO()
 
1252
        runner = tests.TextTestRunner(stream=sio)
 
1253
        test = LogTester('test_error')
 
1254
        result = self.run_test_runner(runner, test)
 
1255
 
 
1256
        text = sio.getvalue()
 
1257
        self.assertContainsRe(text, 'this will be kept')
 
1258
        self.assertContainsRe(text, 'random exception raised')
 
1259
 
 
1260
        log = test._get_log()
 
1261
        self.assertContainsRe(log, 'this will be kept')
 
1262
        self.assertEqual(log, test._log_contents)
1262
1263
 
1263
1264
 
1264
1265
class SampleTestCase(tests.TestCase):
1330
1331
        self.assertEqual(flags, bzrlib.debug.debug_flags)
1331
1332
 
1332
1333
    def change_selftest_debug_flags(self, new_flags):
1333
 
        self.overrideAttr(tests, 'selftest_debug_flags', set(new_flags))
 
1334
        orig_selftest_flags = tests.selftest_debug_flags
 
1335
        self.addCleanup(self._restore_selftest_debug_flags, orig_selftest_flags)
 
1336
        tests.selftest_debug_flags = set(new_flags)
 
1337
 
 
1338
    def _restore_selftest_debug_flags(self, flags):
 
1339
        tests.selftest_debug_flags = flags
1334
1340
 
1335
1341
    def test_allow_debug_flag(self):
1336
1342
        """The -Eallow_debug flag prevents bzrlib.debug.debug_flags from being
1396
1402
        self.assertEqual(set(['original-state']), bzrlib.debug.debug_flags)
1397
1403
 
1398
1404
    def make_test_result(self):
1399
 
        """Get a test result that writes to the test log file."""
1400
1405
        return tests.TextTestResult(self._log_file, descriptions=0, verbosity=1)
1401
1406
 
1402
1407
    def inner_test(self):
1410
1415
        result = self.make_test_result()
1411
1416
        self.inner_test.run(result)
1412
1417
        note("outer finish")
1413
 
        self.addCleanup(osutils.delete_any, self._log_file_name)
1414
1418
 
1415
1419
    def test_trace_nesting(self):
1416
1420
        # this tests that each test case nests its trace facility correctly.
1440
1444
        sample_test = TestTestCase("method_that_times_a_bit_twice")
1441
1445
        output_stream = StringIO()
1442
1446
        result = bzrlib.tests.VerboseTestResult(
1443
 
            output_stream,
 
1447
            unittest._WritelnDecorator(output_stream),
1444
1448
            descriptions=0,
1445
1449
            verbosity=2)
1446
1450
        sample_test.run(result)
1453
1457
        # Note this test won't fail with hooks that the core library doesn't
1454
1458
        # use - but it trigger with a plugin that adds hooks, so its still a
1455
1459
        # useful warning in that case.
1456
 
        self.assertEqual(bzrlib.branch.BranchHooks(), bzrlib.branch.Branch.hooks)
1457
 
        self.assertEqual(
1458
 
            bzrlib.smart.server.SmartServerHooks(),
 
1460
        self.assertEqual(bzrlib.branch.BranchHooks(),
 
1461
            bzrlib.branch.Branch.hooks)
 
1462
        self.assertEqual(bzrlib.smart.server.SmartServerHooks(),
1459
1463
            bzrlib.smart.server.SmartTCPServer.hooks)
1460
 
        self.assertEqual(
1461
 
            bzrlib.commands.CommandHooks(), bzrlib.commands.Command.hooks)
 
1464
        self.assertEqual(bzrlib.commands.CommandHooks(),
 
1465
            bzrlib.commands.Command.hooks)
1462
1466
 
1463
1467
    def test__gather_lsprof_in_benchmarks(self):
1464
1468
        """When _gather_lsprof_in_benchmarks is on, accumulate profile data.
1465
1469
 
1466
1470
        Each self.time() call is individually and separately profiled.
1467
1471
        """
1468
 
        self.requireFeature(features.lsprof_feature)
 
1472
        self.requireFeature(test_lsprof.LSProfFeature)
1469
1473
        # overrides the class member with an instance member so no cleanup
1470
1474
        # needed.
1471
1475
        self._gather_lsprof_in_benchmarks = True
1476
1480
        self.assertEqual((time.sleep, (0.003,), {}), self._benchcalls[1][0])
1477
1481
        self.assertIsInstance(self._benchcalls[0][1], bzrlib.lsprof.Stats)
1478
1482
        self.assertIsInstance(self._benchcalls[1][1], bzrlib.lsprof.Stats)
1479
 
        del self._benchcalls[:]
1480
1483
 
1481
1484
    def test_knownFailure(self):
1482
1485
        """Self.knownFailure() should raise a KnownFailure exception."""
1483
1486
        self.assertRaises(tests.KnownFailure, self.knownFailure, "A Failure")
1484
1487
 
1485
 
    def test_open_bzrdir_safe_roots(self):
1486
 
        # even a memory transport should fail to open when its url isn't 
1487
 
        # permitted.
1488
 
        # Manually set one up (TestCase doesn't and shouldn't provide magic
1489
 
        # machinery)
1490
 
        transport_server = memory.MemoryServer()
1491
 
        transport_server.start_server()
1492
 
        self.addCleanup(transport_server.stop_server)
1493
 
        t = transport.get_transport(transport_server.get_url())
1494
 
        bzrdir.BzrDir.create(t.base)
1495
 
        self.assertRaises(errors.BzrError,
1496
 
            bzrdir.BzrDir.open_from_transport, t)
1497
 
        # But if we declare this as safe, we can open the bzrdir.
1498
 
        self.permit_url(t.base)
1499
 
        self._bzr_selftest_roots.append(t.base)
1500
 
        bzrdir.BzrDir.open_from_transport(t)
1501
 
 
1502
1488
    def test_requireFeature_available(self):
1503
1489
        """self.requireFeature(available) is a no-op."""
1504
 
        class Available(features.Feature):
 
1490
        class Available(tests.Feature):
1505
1491
            def _probe(self):return True
1506
1492
        feature = Available()
1507
1493
        self.requireFeature(feature)
1508
1494
 
1509
1495
    def test_requireFeature_unavailable(self):
1510
1496
        """self.requireFeature(unavailable) raises UnavailableFeature."""
1511
 
        class Unavailable(features.Feature):
 
1497
        class Unavailable(tests.Feature):
1512
1498
            def _probe(self):return False
1513
1499
        feature = Unavailable()
1514
1500
        self.assertRaises(tests.UnavailableFeature,
1548
1534
        """Test disabled tests behaviour with support aware results."""
1549
1535
        test = SampleTestCase('_test_pass')
1550
1536
        class DisabledFeature(object):
1551
 
            def __eq__(self, other):
1552
 
                return isinstance(other, DisabledFeature)
1553
1537
            def available(self):
1554
1538
                return False
1555
1539
        the_feature = DisabledFeature()
1566
1550
                self.calls.append(('addNotSupported', test, feature))
1567
1551
        result = InstrumentedTestResult()
1568
1552
        test.run(result)
1569
 
        case = result.calls[0][1]
1570
1553
        self.assertEqual([
1571
 
            ('startTest', case),
1572
 
            ('addNotSupported', case, the_feature),
1573
 
            ('stopTest', case),
 
1554
            ('startTest', test),
 
1555
            ('addNotSupported', test, the_feature),
 
1556
            ('stopTest', test),
1574
1557
            ],
1575
1558
            result.calls)
1576
1559
 
1577
 
    def test_start_server_registers_url(self):
1578
 
        transport_server = memory.MemoryServer()
1579
 
        # A little strict, but unlikely to be changed soon.
1580
 
        self.assertEqual([], self._bzr_selftest_roots)
1581
 
        self.start_server(transport_server)
1582
 
        self.assertSubset([transport_server.get_url()],
1583
 
            self._bzr_selftest_roots)
1584
 
 
1585
1560
    def test_assert_list_raises_on_generator(self):
1586
1561
        def generator_which_will_raise():
1587
1562
            # This will not raise until after the first yield
1637
1612
        self.assertRaises(AssertionError,
1638
1613
            self.assertListRaises, _TestException, success_generator)
1639
1614
 
1640
 
    def test_overrideAttr_without_value(self):
1641
 
        self.test_attr = 'original' # Define a test attribute
1642
 
        obj = self # Make 'obj' visible to the embedded test
1643
 
        class Test(tests.TestCase):
1644
 
 
1645
 
            def setUp(self):
1646
 
                tests.TestCase.setUp(self)
1647
 
                self.orig = self.overrideAttr(obj, 'test_attr')
1648
 
 
1649
 
            def test_value(self):
1650
 
                self.assertEqual('original', self.orig)
1651
 
                self.assertEqual('original', obj.test_attr)
1652
 
                obj.test_attr = 'modified'
1653
 
                self.assertEqual('modified', obj.test_attr)
1654
 
 
1655
 
        test = Test('test_value')
1656
 
        test.run(unittest.TestResult())
1657
 
        self.assertEqual('original', obj.test_attr)
1658
 
 
1659
 
    def test_overrideAttr_with_value(self):
1660
 
        self.test_attr = 'original' # Define a test attribute
1661
 
        obj = self # Make 'obj' visible to the embedded test
1662
 
        class Test(tests.TestCase):
1663
 
 
1664
 
            def setUp(self):
1665
 
                tests.TestCase.setUp(self)
1666
 
                self.orig = self.overrideAttr(obj, 'test_attr', new='modified')
1667
 
 
1668
 
            def test_value(self):
1669
 
                self.assertEqual('original', self.orig)
1670
 
                self.assertEqual('modified', obj.test_attr)
1671
 
 
1672
 
        test = Test('test_value')
1673
 
        test.run(unittest.TestResult())
1674
 
        self.assertEqual('original', obj.test_attr)
1675
 
 
1676
 
    def test_recordCalls(self):
1677
 
        from bzrlib.tests import test_selftest
1678
 
        calls = self.recordCalls(
1679
 
            test_selftest, '_add_numbers')
1680
 
        self.assertEqual(test_selftest._add_numbers(2, 10),
1681
 
            12)
1682
 
        self.assertEquals(calls, [((2, 10), {})])
1683
 
 
1684
 
 
1685
 
def _add_numbers(a, b):
1686
 
    return a + b
1687
 
 
1688
 
 
1689
 
class _MissingFeature(features.Feature):
1690
 
    def _probe(self):
1691
 
        return False
1692
 
missing_feature = _MissingFeature()
1693
 
 
1694
 
 
1695
 
def _get_test(name):
1696
 
    """Get an instance of a specific example test.
1697
 
 
1698
 
    We protect this in a function so that they don't auto-run in the test
1699
 
    suite.
1700
 
    """
1701
 
 
1702
 
    class ExampleTests(tests.TestCase):
1703
 
 
1704
 
        def test_fail(self):
1705
 
            mutter('this was a failing test')
1706
 
            self.fail('this test will fail')
1707
 
 
1708
 
        def test_error(self):
1709
 
            mutter('this test errored')
1710
 
            raise RuntimeError('gotcha')
1711
 
 
1712
 
        def test_missing_feature(self):
1713
 
            mutter('missing the feature')
1714
 
            self.requireFeature(missing_feature)
1715
 
 
1716
 
        def test_skip(self):
1717
 
            mutter('this test will be skipped')
1718
 
            raise tests.TestSkipped('reason')
1719
 
 
1720
 
        def test_success(self):
1721
 
            mutter('this test succeeds')
1722
 
 
1723
 
        def test_xfail(self):
1724
 
            mutter('test with expected failure')
1725
 
            self.knownFailure('this_fails')
1726
 
 
1727
 
        def test_unexpected_success(self):
1728
 
            mutter('test with unexpected success')
1729
 
            self.expectFailure('should_fail', lambda: None)
1730
 
 
1731
 
    return ExampleTests(name)
1732
 
 
1733
 
 
1734
 
class TestTestCaseLogDetails(tests.TestCase):
1735
 
 
1736
 
    def _run_test(self, test_name):
1737
 
        test = _get_test(test_name)
1738
 
        result = testtools.TestResult()
1739
 
        test.run(result)
1740
 
        return result
1741
 
 
1742
 
    def test_fail_has_log(self):
1743
 
        result = self._run_test('test_fail')
1744
 
        self.assertEqual(1, len(result.failures))
1745
 
        result_content = result.failures[0][1]
1746
 
        if testtools_version < (0, 9, 12):
1747
 
            self.assertContainsRe(result_content, 'Text attachment: log')
1748
 
        self.assertContainsRe(result_content, 'this was a failing test')
1749
 
 
1750
 
    def test_error_has_log(self):
1751
 
        result = self._run_test('test_error')
1752
 
        self.assertEqual(1, len(result.errors))
1753
 
        result_content = result.errors[0][1]
1754
 
        if testtools_version < (0, 9, 12):
1755
 
            self.assertContainsRe(result_content, 'Text attachment: log')
1756
 
        self.assertContainsRe(result_content, 'this test errored')
1757
 
 
1758
 
    def test_skip_has_no_log(self):
1759
 
        result = self._run_test('test_skip')
1760
 
        self.assertEqual(['reason'], result.skip_reasons.keys())
1761
 
        skips = result.skip_reasons['reason']
1762
 
        self.assertEqual(1, len(skips))
1763
 
        test = skips[0]
1764
 
        self.assertFalse('log' in test.getDetails())
1765
 
 
1766
 
    def test_missing_feature_has_no_log(self):
1767
 
        # testtools doesn't know about addNotSupported, so it just gets
1768
 
        # considered as a skip
1769
 
        result = self._run_test('test_missing_feature')
1770
 
        self.assertEqual([missing_feature], result.skip_reasons.keys())
1771
 
        skips = result.skip_reasons[missing_feature]
1772
 
        self.assertEqual(1, len(skips))
1773
 
        test = skips[0]
1774
 
        self.assertFalse('log' in test.getDetails())
1775
 
 
1776
 
    def test_xfail_has_no_log(self):
1777
 
        result = self._run_test('test_xfail')
1778
 
        self.assertEqual(1, len(result.expectedFailures))
1779
 
        result_content = result.expectedFailures[0][1]
1780
 
        self.assertNotContainsRe(result_content, 'Text attachment: log')
1781
 
        self.assertNotContainsRe(result_content, 'test with expected failure')
1782
 
 
1783
 
    def test_unexpected_success_has_log(self):
1784
 
        result = self._run_test('test_unexpected_success')
1785
 
        self.assertEqual(1, len(result.unexpectedSuccesses))
1786
 
        # Inconsistency, unexpectedSuccesses is a list of tests,
1787
 
        # expectedFailures is a list of reasons?
1788
 
        test = result.unexpectedSuccesses[0]
1789
 
        details = test.getDetails()
1790
 
        self.assertTrue('log' in details)
1791
 
 
1792
 
 
1793
 
class TestTestCloning(tests.TestCase):
1794
 
    """Tests that test cloning of TestCases (as used by multiply_tests)."""
1795
 
 
1796
 
    def test_cloned_testcase_does_not_share_details(self):
1797
 
        """A TestCase cloned with clone_test does not share mutable attributes
1798
 
        such as details or cleanups.
1799
 
        """
1800
 
        class Test(tests.TestCase):
1801
 
            def test_foo(self):
1802
 
                self.addDetail('foo', Content('text/plain', lambda: 'foo'))
1803
 
        orig_test = Test('test_foo')
1804
 
        cloned_test = tests.clone_test(orig_test, orig_test.id() + '(cloned)')
1805
 
        orig_test.run(unittest.TestResult())
1806
 
        self.assertEqual('foo', orig_test.getDetails()['foo'].iter_bytes())
1807
 
        self.assertEqual(None, cloned_test.getDetails().get('foo'))
1808
 
 
1809
 
    def test_double_apply_scenario_preserves_first_scenario(self):
1810
 
        """Applying two levels of scenarios to a test preserves the attributes
1811
 
        added by both scenarios.
1812
 
        """
1813
 
        class Test(tests.TestCase):
1814
 
            def test_foo(self):
1815
 
                pass
1816
 
        test = Test('test_foo')
1817
 
        scenarios_x = [('x=1', {'x': 1}), ('x=2', {'x': 2})]
1818
 
        scenarios_y = [('y=1', {'y': 1}), ('y=2', {'y': 2})]
1819
 
        suite = tests.multiply_tests(test, scenarios_x, unittest.TestSuite())
1820
 
        suite = tests.multiply_tests(suite, scenarios_y, unittest.TestSuite())
1821
 
        all_tests = list(tests.iter_suite_tests(suite))
1822
 
        self.assertLength(4, all_tests)
1823
 
        all_xys = sorted((t.x, t.y) for t in all_tests)
1824
 
        self.assertEqual([(1, 1), (1, 2), (2, 1), (2, 2)], all_xys)
1825
 
 
1826
1615
 
1827
1616
# NB: Don't delete this; it's not actually from 0.11!
1828
1617
@deprecated_function(deprecated_in((0, 11, 0)))
1871
1660
        self.assertEndsWith('foo', 'oo')
1872
1661
        self.assertRaises(AssertionError, self.assertEndsWith, 'o', 'oo')
1873
1662
 
1874
 
    def test_assertEqualDiff(self):
1875
 
        e = self.assertRaises(AssertionError,
1876
 
                              self.assertEqualDiff, '', '\n')
1877
 
        self.assertEquals(str(e),
1878
 
                          # Don't blink ! The '+' applies to the second string
1879
 
                          'first string is missing a final newline.\n+ \n')
1880
 
        e = self.assertRaises(AssertionError,
1881
 
                              self.assertEqualDiff, '\n', '')
1882
 
        self.assertEquals(str(e),
1883
 
                          # Don't blink ! The '-' applies to the second string
1884
 
                          'second string is missing a final newline.\n- \n')
1885
 
 
1886
 
 
1887
 
class TestDeprecations(tests.TestCase):
1888
 
 
1889
1663
    def test_applyDeprecated_not_deprecated(self):
1890
1664
        sample_object = ApplyDeprecatedHelper()
1891
1665
        # calling an undeprecated callable raises an assertion
1956
1730
    def test_make_branch_and_tree_with_format(self):
1957
1731
        # we should be able to supply a format to make_branch_and_tree
1958
1732
        self.make_branch_and_tree('a', format=bzrlib.bzrdir.BzrDirMetaFormat1())
 
1733
        self.make_branch_and_tree('b', format=bzrlib.bzrdir.BzrDirFormat6())
1959
1734
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('a')._format,
1960
1735
                              bzrlib.bzrdir.BzrDirMetaFormat1)
 
1736
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('b')._format,
 
1737
                              bzrlib.bzrdir.BzrDirFormat6)
1961
1738
 
1962
1739
    def test_make_branch_and_memory_tree(self):
1963
1740
        # we should be able to get a new branch and a mutable tree from
1965
1742
        tree = self.make_branch_and_memory_tree('a')
1966
1743
        self.assertIsInstance(tree, bzrlib.memorytree.MemoryTree)
1967
1744
 
1968
 
    def test_make_tree_for_local_vfs_backed_transport(self):
1969
 
        # make_branch_and_tree has to use local branch and repositories
1970
 
        # when the vfs transport and local disk are colocated, even if
1971
 
        # a different transport is in use for url generation.
1972
 
        self.transport_server = test_server.FakeVFATServer
1973
 
        self.assertFalse(self.get_url('t1').startswith('file://'))
 
1745
 
 
1746
class TestSFTPMakeBranchAndTree(test_sftp_transport.TestCaseWithSFTPServer):
 
1747
 
 
1748
    def test_make_tree_for_sftp_branch(self):
 
1749
        """Transports backed by local directories create local trees."""
 
1750
        # NB: This is arguably a bug in the definition of make_branch_and_tree.
1974
1751
        tree = self.make_branch_and_tree('t1')
1975
1752
        base = tree.bzrdir.root_transport.base
1976
 
        self.assertStartsWith(base, 'file://')
 
1753
        self.failIf(base.startswith('sftp'),
 
1754
                'base %r is on sftp but should be local' % base)
1977
1755
        self.assertEquals(tree.bzrdir.root_transport,
1978
1756
                tree.branch.bzrdir.root_transport)
1979
1757
        self.assertEquals(tree.bzrdir.root_transport,
1980
1758
                tree.branch.repository.bzrdir.root_transport)
1981
1759
 
1982
1760
 
1983
 
class SelfTestHelper(object):
 
1761
class SelfTestHelper:
1984
1762
 
1985
1763
    def run_selftest(self, **kwargs):
1986
1764
        """Run selftest returning its output."""
2039
1817
        self.assertNotContainsRe("Test.b", output.getvalue())
2040
1818
        self.assertLength(2, output.readlines())
2041
1819
 
2042
 
    def test_lsprof_tests(self):
2043
 
        self.requireFeature(features.lsprof_feature)
2044
 
        results = []
2045
 
        class Test(object):
2046
 
            def __call__(test, result):
2047
 
                test.run(result)
2048
 
            def run(test, result):
2049
 
                results.append(result)
2050
 
            def countTestCases(self):
2051
 
                return 1
2052
 
        self.run_selftest(test_suite_factory=Test, lsprof_tests=True)
2053
 
        self.assertLength(1, results)
2054
 
        self.assertIsInstance(results.pop(), ExtendedToOriginalDecorator)
2055
 
 
2056
1820
    def test_random(self):
2057
1821
        # test randomising by listing a number of tests.
2058
1822
        output_123 = self.run_selftest(test_suite_factory=self.factory,
2073
1837
        self.assertEqual(expected.getvalue(), repeated.getvalue())
2074
1838
 
2075
1839
    def test_runner_class(self):
2076
 
        self.requireFeature(features.subunit)
 
1840
        self.requireFeature(SubUnitFeature)
2077
1841
        from subunit import ProtocolTestCase
2078
1842
        stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
2079
1843
            test_suite_factory=self.factory)
2111
1875
        self.assertEqual(transport_server, captured_transport[0])
2112
1876
 
2113
1877
    def test_transport_sftp(self):
2114
 
        self.requireFeature(features.paramiko)
2115
 
        from bzrlib.tests import stub_sftp
2116
 
        self.check_transport_set(stub_sftp.SFTPAbsoluteServer)
 
1878
        try:
 
1879
            import bzrlib.transport.sftp
 
1880
        except errors.ParamikoNotPresent:
 
1881
            raise tests.TestSkipped("Paramiko not present")
 
1882
        self.check_transport_set(bzrlib.transport.sftp.SFTPAbsoluteServer)
2117
1883
 
2118
1884
    def test_transport_memory(self):
2119
 
        self.check_transport_set(memory.MemoryServer)
 
1885
        self.check_transport_set(bzrlib.transport.memory.MemoryServer)
2120
1886
 
2121
1887
 
2122
1888
class TestSelftestWithIdList(tests.TestCaseInTempDir, SelfTestHelper):
2137
1903
            load_list='missing file name', list_only=True)
2138
1904
 
2139
1905
 
2140
 
class TestSubunitLogDetails(tests.TestCase, SelfTestHelper):
2141
 
 
2142
 
    _test_needs_features = [features.subunit]
2143
 
 
2144
 
    def run_subunit_stream(self, test_name):
2145
 
        from subunit import ProtocolTestCase
2146
 
        def factory():
2147
 
            return TestUtil.TestSuite([_get_test(test_name)])
2148
 
        stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
2149
 
            test_suite_factory=factory)
2150
 
        test = ProtocolTestCase(stream)
2151
 
        result = testtools.TestResult()
2152
 
        test.run(result)
2153
 
        content = stream.getvalue()
2154
 
        return content, result
2155
 
 
2156
 
    def test_fail_has_log(self):
2157
 
        content, result = self.run_subunit_stream('test_fail')
2158
 
        self.assertEqual(1, len(result.failures))
2159
 
        self.assertContainsRe(content, '(?m)^log$')
2160
 
        self.assertContainsRe(content, 'this test will fail')
2161
 
 
2162
 
    def test_error_has_log(self):
2163
 
        content, result = self.run_subunit_stream('test_error')
2164
 
        self.assertContainsRe(content, '(?m)^log$')
2165
 
        self.assertContainsRe(content, 'this test errored')
2166
 
 
2167
 
    def test_skip_has_no_log(self):
2168
 
        content, result = self.run_subunit_stream('test_skip')
2169
 
        self.assertNotContainsRe(content, '(?m)^log$')
2170
 
        self.assertNotContainsRe(content, 'this test will be skipped')
2171
 
        self.assertEqual(['reason'], result.skip_reasons.keys())
2172
 
        skips = result.skip_reasons['reason']
2173
 
        self.assertEqual(1, len(skips))
2174
 
        test = skips[0]
2175
 
        # RemotedTestCase doesn't preserve the "details"
2176
 
        ## self.assertFalse('log' in test.getDetails())
2177
 
 
2178
 
    def test_missing_feature_has_no_log(self):
2179
 
        content, result = self.run_subunit_stream('test_missing_feature')
2180
 
        self.assertNotContainsRe(content, '(?m)^log$')
2181
 
        self.assertNotContainsRe(content, 'missing the feature')
2182
 
        self.assertEqual(['_MissingFeature\n'], result.skip_reasons.keys())
2183
 
        skips = result.skip_reasons['_MissingFeature\n']
2184
 
        self.assertEqual(1, len(skips))
2185
 
        test = skips[0]
2186
 
        # RemotedTestCase doesn't preserve the "details"
2187
 
        ## self.assertFalse('log' in test.getDetails())
2188
 
 
2189
 
    def test_xfail_has_no_log(self):
2190
 
        content, result = self.run_subunit_stream('test_xfail')
2191
 
        self.assertNotContainsRe(content, '(?m)^log$')
2192
 
        self.assertNotContainsRe(content, 'test with expected failure')
2193
 
        self.assertEqual(1, len(result.expectedFailures))
2194
 
        result_content = result.expectedFailures[0][1]
2195
 
        self.assertNotContainsRe(result_content, 'Text attachment: log')
2196
 
        self.assertNotContainsRe(result_content, 'test with expected failure')
2197
 
 
2198
 
    def test_unexpected_success_has_log(self):
2199
 
        content, result = self.run_subunit_stream('test_unexpected_success')
2200
 
        self.assertContainsRe(content, '(?m)^log$')
2201
 
        self.assertContainsRe(content, 'test with unexpected success')
2202
 
        # GZ 2011-05-18: Old versions of subunit treat unexpected success as a
2203
 
        #                success, if a min version check is added remove this
2204
 
        from subunit import TestProtocolClient as _Client
2205
 
        if _Client.addUnexpectedSuccess.im_func is _Client.addSuccess.im_func:
2206
 
            self.expectFailure('subunit treats "unexpectedSuccess"'
2207
 
                               ' as a plain success',
2208
 
                self.assertEqual, 1, len(result.unexpectedSuccesses))
2209
 
        self.assertEqual(1, len(result.unexpectedSuccesses))
2210
 
        test = result.unexpectedSuccesses[0]
2211
 
        # RemotedTestCase doesn't preserve the "details"
2212
 
        ## self.assertTrue('log' in test.getDetails())
2213
 
 
2214
 
    def test_success_has_no_log(self):
2215
 
        content, result = self.run_subunit_stream('test_success')
2216
 
        self.assertEqual(1, result.testsRun)
2217
 
        self.assertNotContainsRe(content, '(?m)^log$')
2218
 
        self.assertNotContainsRe(content, 'this test succeeds')
2219
 
 
2220
 
 
2221
1906
class TestRunBzr(tests.TestCase):
2222
1907
 
2223
1908
    out = ''
2229
1914
 
2230
1915
        Attempts to run bzr from inside this class don't actually run it.
2231
1916
 
2232
 
        We test how run_bzr actually invokes bzr in another location.  Here we
2233
 
        only need to test that it passes the right parameters to run_bzr.
 
1917
        We test how run_bzr actually invokes bzr in another location.
 
1918
        Here we only need to test that it is run_bzr passes the right
 
1919
        parameters to run_bzr.
2234
1920
        """
2235
1921
        self.argv = list(argv)
2236
1922
        self.retcode = retcode
2237
1923
        self.encoding = encoding
2238
1924
        self.stdin = stdin
2239
1925
        self.working_dir = working_dir
2240
 
        return self.retcode, self.out, self.err
 
1926
        return self.out, self.err
2241
1927
 
2242
1928
    def test_run_bzr_error(self):
2243
1929
        self.out = "It sure does!\n"
2244
1930
        out, err = self.run_bzr_error(['^$'], ['rocks'], retcode=34)
2245
1931
        self.assertEqual(['rocks'], self.argv)
2246
1932
        self.assertEqual(34, self.retcode)
2247
 
        self.assertEqual('It sure does!\n', out)
2248
 
        self.assertEquals(out, self.out)
2249
 
        self.assertEqual('', err)
2250
 
        self.assertEquals(err, self.err)
 
1933
        self.assertEqual(out, 'It sure does!\n')
2251
1934
 
2252
1935
    def test_run_bzr_error_regexes(self):
2253
1936
        self.out = ''
2254
1937
        self.err = "bzr: ERROR: foobarbaz is not versioned"
2255
1938
        out, err = self.run_bzr_error(
2256
 
            ["bzr: ERROR: foobarbaz is not versioned"],
2257
 
            ['file-id', 'foobarbaz'])
 
1939
                ["bzr: ERROR: foobarbaz is not versioned"],
 
1940
                ['file-id', 'foobarbaz'])
2258
1941
 
2259
1942
    def test_encoding(self):
2260
1943
        """Test that run_bzr passes encoding to _run_bzr_core"""
2346
2029
        # stdout and stderr of the invoked run_bzr
2347
2030
        current_factory = bzrlib.ui.ui_factory
2348
2031
        self.run_bzr(['foo'])
2349
 
        self.assertFalse(current_factory is self.factory)
 
2032
        self.failIf(current_factory is self.factory)
2350
2033
        self.assertNotEqual(sys.stdout, self.factory.stdout)
2351
2034
        self.assertNotEqual(sys.stderr, self.factory.stderr)
2352
2035
        self.assertEqual('foo\n', self.factory.stdout.getvalue())
2389
2072
        return self.out, self.err
2390
2073
 
2391
2074
 
2392
 
class TestWithFakedStartBzrSubprocess(tests.TestCaseWithTransport):
2393
 
    """Base class for tests testing how we might run bzr."""
 
2075
class TestRunBzrSubprocess(tests.TestCaseWithTransport):
2394
2076
 
2395
2077
    def setUp(self):
2396
2078
        tests.TestCaseWithTransport.setUp(self)
2407
2089
            'working_dir':working_dir, 'allow_plugins':allow_plugins})
2408
2090
        return self.next_subprocess
2409
2091
 
2410
 
 
2411
 
class TestRunBzrSubprocess(TestWithFakedStartBzrSubprocess):
2412
 
 
2413
2092
    def assertRunBzrSubprocess(self, expected_args, process, *args, **kwargs):
2414
2093
        """Run run_bzr_subprocess with args and kwargs using a stubbed process.
2415
2094
 
2478
2157
            StubProcess(), '', allow_plugins=True)
2479
2158
 
2480
2159
 
2481
 
class TestFinishBzrSubprocess(TestWithFakedStartBzrSubprocess):
2482
 
 
2483
 
    def test_finish_bzr_subprocess_with_error(self):
2484
 
        """finish_bzr_subprocess allows specification of the desired exit code.
2485
 
        """
2486
 
        process = StubProcess(err="unknown command", retcode=3)
2487
 
        result = self.finish_bzr_subprocess(process, retcode=3)
2488
 
        self.assertEqual('', result[0])
2489
 
        self.assertContainsRe(result[1], 'unknown command')
2490
 
 
2491
 
    def test_finish_bzr_subprocess_ignoring_retcode(self):
2492
 
        """finish_bzr_subprocess allows the exit code to be ignored."""
2493
 
        process = StubProcess(err="unknown command", retcode=3)
2494
 
        result = self.finish_bzr_subprocess(process, retcode=None)
2495
 
        self.assertEqual('', result[0])
2496
 
        self.assertContainsRe(result[1], 'unknown command')
2497
 
 
2498
 
    def test_finish_subprocess_with_unexpected_retcode(self):
2499
 
        """finish_bzr_subprocess raises self.failureException if the retcode is
2500
 
        not the expected one.
2501
 
        """
2502
 
        process = StubProcess(err="unknown command", retcode=3)
2503
 
        self.assertRaises(self.failureException, self.finish_bzr_subprocess,
2504
 
                          process)
2505
 
 
2506
 
 
2507
2160
class _DontSpawnProcess(Exception):
2508
2161
    """A simple exception which just allows us to skip unnecessary steps"""
2509
2162
 
2510
2163
 
2511
2164
class TestStartBzrSubProcess(tests.TestCase):
2512
 
    """Stub test start_bzr_subprocess."""
2513
2165
 
2514
 
    def _subprocess_log_cleanup(self):
2515
 
        """Inhibits the base version as we don't produce a log file."""
 
2166
    def check_popen_state(self):
 
2167
        """Replace to make assertions when popen is called."""
2516
2168
 
2517
2169
    def _popen(self, *args, **kwargs):
2518
 
        """Override the base version to record the command that is run.
2519
 
 
2520
 
        From there we can ensure it is correct without spawning a real process.
2521
 
        """
 
2170
        """Record the command that is run, so that we can ensure it is correct"""
2522
2171
        self.check_popen_state()
2523
2172
        self._popen_args = args
2524
2173
        self._popen_kwargs = kwargs
2525
2174
        raise _DontSpawnProcess()
2526
2175
 
2527
 
    def check_popen_state(self):
2528
 
        """Replace to make assertions when popen is called."""
2529
 
 
2530
2176
    def test_run_bzr_subprocess_no_plugins(self):
2531
2177
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [])
2532
2178
        command = self._popen_args[0]
2536
2182
 
2537
2183
    def test_allow_plugins(self):
2538
2184
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2539
 
                          allow_plugins=True)
 
2185
            allow_plugins=True)
2540
2186
        command = self._popen_args[0]
2541
2187
        self.assertEqual([], command[2:])
2542
2188
 
2543
2189
    def test_set_env(self):
2544
 
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
 
2190
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
2545
2191
        # set in the child
2546
2192
        def check_environment():
2547
2193
            self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2548
2194
        self.check_popen_state = check_environment
2549
2195
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2550
 
                          env_changes={'EXISTANT_ENV_VAR':'set variable'})
 
2196
            env_changes={'EXISTANT_ENV_VAR':'set variable'})
2551
2197
        # not set in theparent
2552
2198
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2553
2199
 
2554
2200
    def test_run_bzr_subprocess_env_del(self):
2555
2201
        """run_bzr_subprocess can remove environment variables too."""
2556
 
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
 
2202
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
2557
2203
        def check_environment():
2558
2204
            self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2559
2205
        os.environ['EXISTANT_ENV_VAR'] = 'set variable'
2560
2206
        self.check_popen_state = check_environment
2561
2207
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2562
 
                          env_changes={'EXISTANT_ENV_VAR':None})
 
2208
            env_changes={'EXISTANT_ENV_VAR':None})
2563
2209
        # Still set in parent
2564
2210
        self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2565
2211
        del os.environ['EXISTANT_ENV_VAR']
2566
2212
 
2567
2213
    def test_env_del_missing(self):
2568
 
        self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
 
2214
        self.failIf('NON_EXISTANT_ENV_VAR' in os.environ)
2569
2215
        def check_environment():
2570
2216
            self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2571
2217
        self.check_popen_state = check_environment
2572
2218
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2573
 
                          env_changes={'NON_EXISTANT_ENV_VAR':None})
 
2219
            env_changes={'NON_EXISTANT_ENV_VAR':None})
2574
2220
 
2575
2221
    def test_working_dir(self):
2576
2222
        """Test that we can specify the working dir for the child"""
2579
2225
        chdirs = []
2580
2226
        def chdir(path):
2581
2227
            chdirs.append(path)
2582
 
        self.overrideAttr(os, 'chdir', chdir)
2583
 
        def getcwd():
2584
 
            return 'current'
2585
 
        self.overrideAttr(osutils, 'getcwd', getcwd)
2586
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2587
 
                          working_dir='foo')
 
2228
        os.chdir = chdir
 
2229
        try:
 
2230
            def getcwd():
 
2231
                return 'current'
 
2232
            osutils.getcwd = getcwd
 
2233
            try:
 
2234
                self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
 
2235
                    working_dir='foo')
 
2236
            finally:
 
2237
                osutils.getcwd = orig_getcwd
 
2238
        finally:
 
2239
            os.chdir = orig_chdir
2588
2240
        self.assertEqual(['foo', 'current'], chdirs)
2589
2241
 
2590
 
    def test_get_bzr_path_with_cwd_bzrlib(self):
2591
 
        self.get_source_path = lambda: ""
2592
 
        self.overrideAttr(os.path, "isfile", lambda path: True)
2593
 
        self.assertEqual(self.get_bzr_path(), "bzr")
2594
 
 
2595
 
 
2596
 
class TestActuallyStartBzrSubprocess(tests.TestCaseWithTransport):
2597
 
    """Tests that really need to do things with an external bzr."""
 
2242
 
 
2243
class TestBzrSubprocess(tests.TestCaseWithTransport):
 
2244
 
 
2245
    def test_start_and_stop_bzr_subprocess(self):
 
2246
        """We can start and perform other test actions while that process is
 
2247
        still alive.
 
2248
        """
 
2249
        process = self.start_bzr_subprocess(['--version'])
 
2250
        result = self.finish_bzr_subprocess(process)
 
2251
        self.assertContainsRe(result[0], 'is free software')
 
2252
        self.assertEqual('', result[1])
 
2253
 
 
2254
    def test_start_and_stop_bzr_subprocess_with_error(self):
 
2255
        """finish_bzr_subprocess allows specification of the desired exit code.
 
2256
        """
 
2257
        process = self.start_bzr_subprocess(['--versionn'])
 
2258
        result = self.finish_bzr_subprocess(process, retcode=3)
 
2259
        self.assertEqual('', result[0])
 
2260
        self.assertContainsRe(result[1], 'unknown command')
 
2261
 
 
2262
    def test_start_and_stop_bzr_subprocess_ignoring_retcode(self):
 
2263
        """finish_bzr_subprocess allows the exit code to be ignored."""
 
2264
        process = self.start_bzr_subprocess(['--versionn'])
 
2265
        result = self.finish_bzr_subprocess(process, retcode=None)
 
2266
        self.assertEqual('', result[0])
 
2267
        self.assertContainsRe(result[1], 'unknown command')
 
2268
 
 
2269
    def test_start_and_stop_bzr_subprocess_with_unexpected_retcode(self):
 
2270
        """finish_bzr_subprocess raises self.failureException if the retcode is
 
2271
        not the expected one.
 
2272
        """
 
2273
        process = self.start_bzr_subprocess(['--versionn'])
 
2274
        self.assertRaises(self.failureException, self.finish_bzr_subprocess,
 
2275
                          process)
2598
2276
 
2599
2277
    def test_start_and_stop_bzr_subprocess_send_signal(self):
2600
2278
        """finish_bzr_subprocess raises self.failureException if the retcode is
2601
2279
        not the expected one.
2602
2280
        """
2603
 
        self.disable_missing_extensions_warning()
2604
2281
        process = self.start_bzr_subprocess(['wait-until-signalled'],
2605
2282
                                            skip_if_plan_to_signal=True)
2606
2283
        self.assertEqual('running\n', process.stdout.readline())
2609
2286
        self.assertEqual('', result[0])
2610
2287
        self.assertEqual('bzr: interrupted\n', result[1])
2611
2288
 
 
2289
    def test_start_and_stop_working_dir(self):
 
2290
        cwd = osutils.getcwd()
 
2291
        self.make_branch_and_tree('one')
 
2292
        process = self.start_bzr_subprocess(['root'], working_dir='one')
 
2293
        result = self.finish_bzr_subprocess(process, universal_newlines=True)
 
2294
        self.assertEndsWith(result[0], 'one\n')
 
2295
        self.assertEqual('', result[1])
 
2296
 
 
2297
 
 
2298
class TestKnownFailure(tests.TestCase):
 
2299
 
 
2300
    def test_known_failure(self):
 
2301
        """Check that KnownFailure is defined appropriately."""
 
2302
        # a KnownFailure is an assertion error for compatability with unaware
 
2303
        # runners.
 
2304
        self.assertIsInstance(tests.KnownFailure(""), AssertionError)
 
2305
 
 
2306
    def test_expect_failure(self):
 
2307
        try:
 
2308
            self.expectFailure("Doomed to failure", self.assertTrue, False)
 
2309
        except tests.KnownFailure, e:
 
2310
            self.assertEqual('Doomed to failure', e.args[0])
 
2311
        try:
 
2312
            self.expectFailure("Doomed to failure", self.assertTrue, True)
 
2313
        except AssertionError, e:
 
2314
            self.assertEqual('Unexpected success.  Should have failed:'
 
2315
                             ' Doomed to failure', e.args[0])
 
2316
        else:
 
2317
            self.fail('Assertion not raised')
 
2318
 
 
2319
 
 
2320
class TestFeature(tests.TestCase):
 
2321
 
 
2322
    def test_caching(self):
 
2323
        """Feature._probe is called by the feature at most once."""
 
2324
        class InstrumentedFeature(tests.Feature):
 
2325
            def __init__(self):
 
2326
                super(InstrumentedFeature, self).__init__()
 
2327
                self.calls = []
 
2328
            def _probe(self):
 
2329
                self.calls.append('_probe')
 
2330
                return False
 
2331
        feature = InstrumentedFeature()
 
2332
        feature.available()
 
2333
        self.assertEqual(['_probe'], feature.calls)
 
2334
        feature.available()
 
2335
        self.assertEqual(['_probe'], feature.calls)
 
2336
 
 
2337
    def test_named_str(self):
 
2338
        """Feature.__str__ should thunk to feature_name()."""
 
2339
        class NamedFeature(tests.Feature):
 
2340
            def feature_name(self):
 
2341
                return 'symlinks'
 
2342
        feature = NamedFeature()
 
2343
        self.assertEqual('symlinks', str(feature))
 
2344
 
 
2345
    def test_default_str(self):
 
2346
        """Feature.__str__ should default to __class__.__name__."""
 
2347
        class NamedFeature(tests.Feature):
 
2348
            pass
 
2349
        feature = NamedFeature()
 
2350
        self.assertEqual('NamedFeature', str(feature))
 
2351
 
 
2352
 
 
2353
class TestUnavailableFeature(tests.TestCase):
 
2354
 
 
2355
    def test_access_feature(self):
 
2356
        feature = tests.Feature()
 
2357
        exception = tests.UnavailableFeature(feature)
 
2358
        self.assertIs(feature, exception.args[0])
 
2359
 
2612
2360
 
2613
2361
class TestSelftestFiltering(tests.TestCase):
2614
2362
 
2766
2514
        self.assertEqual(remaining_names, _test_ids(split_suite[1]))
2767
2515
 
2768
2516
 
2769
 
class TestCheckTreeShape(tests.TestCaseWithTransport):
 
2517
class TestCheckInventoryShape(tests.TestCaseWithTransport):
2770
2518
 
2771
 
    def test_check_tree_shape(self):
 
2519
    def test_check_inventory_shape(self):
2772
2520
        files = ['a', 'b/', 'b/c']
2773
2521
        tree = self.make_branch_and_tree('.')
2774
2522
        self.build_tree(files)
2775
2523
        tree.add(files)
2776
2524
        tree.lock_read()
2777
2525
        try:
2778
 
            self.check_tree_shape(tree, files)
 
2526
            self.check_inventory_shape(tree.inventory, files)
2779
2527
        finally:
2780
2528
            tree.unlock()
2781
2529
 
2794
2542
        # the test framework
2795
2543
        self.assertEquals('always fails', str(e))
2796
2544
        # check that there's no traceback in the test log
2797
 
        self.assertNotContainsRe(self.get_log(), r'Traceback')
 
2545
        self.assertNotContainsRe(self._get_log(keep_log_file=True),
 
2546
            r'Traceback')
2798
2547
 
2799
2548
    def test_run_bzr_user_error_caught(self):
2800
2549
        # Running bzr in blackbox mode, normal/expected/user errors should be
2801
2550
        # caught in the regular way and turned into an error message plus exit
2802
2551
        # code.
2803
 
        transport_server = memory.MemoryServer()
2804
 
        transport_server.start_server()
2805
 
        self.addCleanup(transport_server.stop_server)
2806
 
        url = transport_server.get_url()
2807
 
        self.permit_url(url)
2808
 
        out, err = self.run_bzr(["log", "%s/nonexistantpath" % url], retcode=3)
 
2552
        out, err = self.run_bzr(["log", "/nonexistantpath"], retcode=3)
2809
2553
        self.assertEqual(out, '')
2810
2554
        self.assertContainsRe(err,
2811
2555
            'bzr: ERROR: Not a branch: ".*nonexistantpath/".\n')
2937
2681
 
2938
2682
class TestTestSuite(tests.TestCase):
2939
2683
 
2940
 
    def test__test_suite_testmod_names(self):
2941
 
        # Test that a plausible list of test module names are returned
2942
 
        # by _test_suite_testmod_names.
2943
 
        test_list = tests._test_suite_testmod_names()
2944
 
        self.assertSubset([
2945
 
            'bzrlib.tests.blackbox',
2946
 
            'bzrlib.tests.per_transport',
2947
 
            'bzrlib.tests.test_selftest',
2948
 
            ],
2949
 
            test_list)
2950
 
 
2951
 
    def test__test_suite_modules_to_doctest(self):
2952
 
        # Test that a plausible list of modules to doctest is returned
2953
 
        # by _test_suite_modules_to_doctest.
2954
 
        test_list = tests._test_suite_modules_to_doctest()
2955
 
        if __doc__ is None:
2956
 
            # When docstrings are stripped, there are no modules to doctest
2957
 
            self.assertEqual([], test_list)
2958
 
            return
2959
 
        self.assertSubset([
2960
 
            'bzrlib.timestamp',
2961
 
            ],
2962
 
            test_list)
2963
 
 
2964
2684
    def test_test_suite(self):
2965
 
        # test_suite() loads the entire test suite to operate. To avoid this
2966
 
        # overhead, and yet still be confident that things are happening,
2967
 
        # we temporarily replace two functions used by test_suite with 
2968
 
        # test doubles that supply a few sample tests to load, and check they
2969
 
        # are loaded.
2970
 
        calls = []
2971
 
        def testmod_names():
2972
 
            calls.append("testmod_names")
2973
 
            return [
2974
 
                'bzrlib.tests.blackbox.test_branch',
2975
 
                'bzrlib.tests.per_transport',
2976
 
                'bzrlib.tests.test_selftest',
2977
 
                ]
2978
 
        self.overrideAttr(tests, '_test_suite_testmod_names', testmod_names)
2979
 
        def doctests():
2980
 
            calls.append("modules_to_doctest")
2981
 
            if __doc__ is None:
2982
 
                return []
2983
 
            return ['bzrlib.timestamp']
2984
 
        self.overrideAttr(tests, '_test_suite_modules_to_doctest', doctests)
2985
 
        expected_test_list = [
 
2685
        # This test is slow - it loads the entire test suite to operate, so we
 
2686
        # do a single test with one test in each category
 
2687
        test_list = [
2986
2688
            # testmod_names
2987
2689
            'bzrlib.tests.blackbox.test_branch.TestBranch.test_branch',
2988
2690
            ('bzrlib.tests.per_transport.TransportTests'
2989
 
             '.test_abspath(LocalTransport,LocalURLServer)'),
 
2691
             '.test_abspath(LocalURLServer)'),
2990
2692
            'bzrlib.tests.test_selftest.TestTestSuite.test_test_suite',
 
2693
            # modules_to_doctest
 
2694
            'bzrlib.timestamp.format_highres_date',
2991
2695
            # plugins can't be tested that way since selftest may be run with
2992
2696
            # --no-plugins
2993
2697
            ]
2994
 
        if __doc__ is not None:
2995
 
            expected_test_list.extend([
2996
 
                # modules_to_doctest
2997
 
                'bzrlib.timestamp.format_highres_date',
2998
 
                ])
2999
 
        suite = tests.test_suite()
3000
 
        self.assertEqual(set(["testmod_names", "modules_to_doctest"]),
3001
 
            set(calls))
3002
 
        self.assertSubset(expected_test_list, _test_ids(suite))
 
2698
        suite = tests.test_suite(test_list)
 
2699
        self.assertEquals(test_list, _test_ids(suite))
3003
2700
 
3004
2701
    def test_test_suite_list_and_start(self):
3005
2702
        # We cannot test this at the same time as the main load, because we want
3006
 
        # to know that starting_with == None works. So a second load is
3007
 
        # incurred - note that the starting_with parameter causes a partial load
3008
 
        # rather than a full load so this test should be pretty quick.
 
2703
        # to know that starting_with == None works. So a second full load is
 
2704
        # incurred.
3009
2705
        test_list = ['bzrlib.tests.test_selftest.TestTestSuite.test_test_suite']
3010
2706
        suite = tests.test_suite(test_list,
3011
2707
                                 ['bzrlib.tests.test_selftest.TestTestSuite'])
3056
2752
    def test_load_tests(self):
3057
2753
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing']
3058
2754
        loader = self._create_loader(test_list)
 
2755
 
3059
2756
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
3060
2757
        self.assertEquals(test_list, _test_ids(suite))
3061
2758
 
3062
2759
    def test_exclude_tests(self):
3063
2760
        test_list = ['bogus']
3064
2761
        loader = self._create_loader(test_list)
 
2762
 
3065
2763
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
3066
2764
        self.assertEquals([], _test_ids(suite))
3067
2765
 
3112
2810
        tpr.register('bar', 'bbb.aaa.rrr')
3113
2811
        tpr.register('bar', 'bBB.aAA.rRR')
3114
2812
        self.assertEquals('bbb.aaa.rrr', tpr.get('bar'))
3115
 
        self.assertThat(self.get_log(),
3116
 
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR",
3117
 
                           doctest.ELLIPSIS))
 
2813
        self.assertContainsRe(self._get_log(keep_log_file=True),
 
2814
                              r'.*bar.*bbb.aaa.rrr.*bBB.aAA.rRR')
3118
2815
 
3119
2816
    def test_get_unknown_prefix(self):
3120
2817
        tpr = self._get_registry()
3140
2837
        self.assertEquals('bzrlib.plugins', tpr.resolve_alias('bp'))
3141
2838
 
3142
2839
 
3143
 
class TestThreadLeakDetection(tests.TestCase):
3144
 
    """Ensure when tests leak threads we detect and report it"""
3145
 
 
3146
 
    class LeakRecordingResult(tests.ExtendedTestResult):
3147
 
        def __init__(self):
3148
 
            tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
3149
 
            self.leaks = []
3150
 
        def _report_thread_leak(self, test, leaks, alive):
3151
 
            self.leaks.append((test, leaks))
3152
 
 
3153
 
    def test_testcase_without_addCleanups(self):
3154
 
        """Check old TestCase instances don't break with leak detection"""
3155
 
        class Test(unittest.TestCase):
3156
 
            def runTest(self):
3157
 
                pass
3158
 
        result = self.LeakRecordingResult()
3159
 
        test = Test()
3160
 
        result.startTestRun()
3161
 
        test.run(result)
3162
 
        result.stopTestRun()
3163
 
        self.assertEqual(result._tests_leaking_threads_count, 0)
3164
 
        self.assertEqual(result.leaks, [])
3165
 
        
3166
 
    def test_thread_leak(self):
3167
 
        """Ensure a thread that outlives the running of a test is reported
3168
 
 
3169
 
        Uses a thread that blocks on an event, and is started by the inner
3170
 
        test case. As the thread outlives the inner case's run, it should be
3171
 
        detected as a leak, but the event is then set so that the thread can
3172
 
        be safely joined in cleanup so it's not leaked for real.
3173
 
        """
3174
 
        event = threading.Event()
3175
 
        thread = threading.Thread(name="Leaker", target=event.wait)
3176
 
        class Test(tests.TestCase):
3177
 
            def test_leak(self):
3178
 
                thread.start()
3179
 
        result = self.LeakRecordingResult()
3180
 
        test = Test("test_leak")
3181
 
        self.addCleanup(thread.join)
3182
 
        self.addCleanup(event.set)
3183
 
        result.startTestRun()
3184
 
        test.run(result)
3185
 
        result.stopTestRun()
3186
 
        self.assertEqual(result._tests_leaking_threads_count, 1)
3187
 
        self.assertEqual(result._first_thread_leaker_id, test.id())
3188
 
        self.assertEqual(result.leaks, [(test, set([thread]))])
3189
 
        self.assertContainsString(result.stream.getvalue(), "leaking threads")
3190
 
 
3191
 
    def test_multiple_leaks(self):
3192
 
        """Check multiple leaks are blamed on the test cases at fault
3193
 
 
3194
 
        Same concept as the previous test, but has one inner test method that
3195
 
        leaks two threads, and one that doesn't leak at all.
3196
 
        """
3197
 
        event = threading.Event()
3198
 
        thread_a = threading.Thread(name="LeakerA", target=event.wait)
3199
 
        thread_b = threading.Thread(name="LeakerB", target=event.wait)
3200
 
        thread_c = threading.Thread(name="LeakerC", target=event.wait)
3201
 
        class Test(tests.TestCase):
3202
 
            def test_first_leak(self):
3203
 
                thread_b.start()
3204
 
            def test_second_no_leak(self):
3205
 
                pass
3206
 
            def test_third_leak(self):
3207
 
                thread_c.start()
3208
 
                thread_a.start()
3209
 
        result = self.LeakRecordingResult()
3210
 
        first_test = Test("test_first_leak")
3211
 
        third_test = Test("test_third_leak")
3212
 
        self.addCleanup(thread_a.join)
3213
 
        self.addCleanup(thread_b.join)
3214
 
        self.addCleanup(thread_c.join)
3215
 
        self.addCleanup(event.set)
3216
 
        result.startTestRun()
3217
 
        unittest.TestSuite(
3218
 
            [first_test, Test("test_second_no_leak"), third_test]
3219
 
            ).run(result)
3220
 
        result.stopTestRun()
3221
 
        self.assertEqual(result._tests_leaking_threads_count, 2)
3222
 
        self.assertEqual(result._first_thread_leaker_id, first_test.id())
3223
 
        self.assertEqual(result.leaks, [
3224
 
            (first_test, set([thread_b])),
3225
 
            (third_test, set([thread_a, thread_c]))])
3226
 
        self.assertContainsString(result.stream.getvalue(), "leaking threads")
3227
 
 
3228
 
 
3229
 
class TestPostMortemDebugging(tests.TestCase):
3230
 
    """Check post mortem debugging works when tests fail or error"""
3231
 
 
3232
 
    class TracebackRecordingResult(tests.ExtendedTestResult):
3233
 
        def __init__(self):
3234
 
            tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
3235
 
            self.postcode = None
3236
 
        def _post_mortem(self, tb=None):
3237
 
            """Record the code object at the end of the current traceback"""
3238
 
            tb = tb or sys.exc_info()[2]
3239
 
            if tb is not None:
3240
 
                next = tb.tb_next
3241
 
                while next is not None:
3242
 
                    tb = next
3243
 
                    next = next.tb_next
3244
 
                self.postcode = tb.tb_frame.f_code
3245
 
        def report_error(self, test, err):
3246
 
            pass
3247
 
        def report_failure(self, test, err):
3248
 
            pass
3249
 
 
3250
 
    def test_location_unittest_error(self):
3251
 
        """Needs right post mortem traceback with erroring unittest case"""
3252
 
        class Test(unittest.TestCase):
3253
 
            def runTest(self):
3254
 
                raise RuntimeError
3255
 
        result = self.TracebackRecordingResult()
3256
 
        Test().run(result)
3257
 
        self.assertEqual(result.postcode, Test.runTest.func_code)
3258
 
 
3259
 
    def test_location_unittest_failure(self):
3260
 
        """Needs right post mortem traceback with failing unittest case"""
3261
 
        class Test(unittest.TestCase):
3262
 
            def runTest(self):
3263
 
                raise self.failureException
3264
 
        result = self.TracebackRecordingResult()
3265
 
        Test().run(result)
3266
 
        self.assertEqual(result.postcode, Test.runTest.func_code)
3267
 
 
3268
 
    def test_location_bt_error(self):
3269
 
        """Needs right post mortem traceback with erroring bzrlib.tests case"""
3270
 
        class Test(tests.TestCase):
3271
 
            def test_error(self):
3272
 
                raise RuntimeError
3273
 
        result = self.TracebackRecordingResult()
3274
 
        Test("test_error").run(result)
3275
 
        self.assertEqual(result.postcode, Test.test_error.func_code)
3276
 
 
3277
 
    def test_location_bt_failure(self):
3278
 
        """Needs right post mortem traceback with failing bzrlib.tests case"""
3279
 
        class Test(tests.TestCase):
3280
 
            def test_failure(self):
3281
 
                raise self.failureException
3282
 
        result = self.TracebackRecordingResult()
3283
 
        Test("test_failure").run(result)
3284
 
        self.assertEqual(result.postcode, Test.test_failure.func_code)
3285
 
 
3286
 
    def test_env_var_triggers_post_mortem(self):
3287
 
        """Check pdb.post_mortem is called iff BZR_TEST_PDB is set"""
3288
 
        import pdb
3289
 
        result = tests.ExtendedTestResult(StringIO(), 0, 1)
3290
 
        post_mortem_calls = []
3291
 
        self.overrideAttr(pdb, "post_mortem", post_mortem_calls.append)
3292
 
        self.overrideEnv('BZR_TEST_PDB', None)
3293
 
        result._post_mortem(1)
3294
 
        self.overrideEnv('BZR_TEST_PDB', 'on')
3295
 
        result._post_mortem(2)
3296
 
        self.assertEqual([2], post_mortem_calls)
3297
 
 
3298
 
 
3299
2840
class TestRunSuite(tests.TestCase):
3300
2841
 
3301
2842
    def test_runner_class(self):
3313
2854
        tests.run_suite(suite, runner_class=MyRunner, stream=StringIO())
3314
2855
        self.assertLength(1, calls)
3315
2856
 
3316
 
 
3317
 
class TestEnvironHandling(tests.TestCase):
3318
 
 
3319
 
    def test_overrideEnv_None_called_twice_doesnt_leak(self):
3320
 
        self.assertFalse('MYVAR' in os.environ)
3321
 
        self.overrideEnv('MYVAR', '42')
3322
 
        # We use an embedded test to make sure we fix the _captureVar bug
3323
 
        class Test(tests.TestCase):
3324
 
            def test_me(self):
3325
 
                # The first call save the 42 value
3326
 
                self.overrideEnv('MYVAR', None)
3327
 
                self.assertEquals(None, os.environ.get('MYVAR'))
3328
 
                # Make sure we can call it twice
3329
 
                self.overrideEnv('MYVAR', None)
3330
 
                self.assertEquals(None, os.environ.get('MYVAR'))
3331
 
        output = StringIO()
3332
 
        result = tests.TextTestResult(output, 0, 1)
3333
 
        Test('test_me').run(result)
3334
 
        if not result.wasStrictlySuccessful():
3335
 
            self.fail(output.getvalue())
3336
 
        # We get our value back
3337
 
        self.assertEquals('42', os.environ.get('MYVAR'))
3338
 
 
3339
 
 
3340
 
class TestIsolatedEnv(tests.TestCase):
3341
 
    """Test isolating tests from os.environ.
3342
 
 
3343
 
    Since we use tests that are already isolated from os.environ a bit of care
3344
 
    should be taken when designing the tests to avoid bootstrap side-effects.
3345
 
    The tests start an already clean os.environ which allow doing valid
3346
 
    assertions about which variables are present or not and design tests around
3347
 
    these assertions.
3348
 
    """
3349
 
 
3350
 
    class ScratchMonkey(tests.TestCase):
3351
 
 
3352
 
        def test_me(self):
 
2857
    def test_done(self):
 
2858
        """run_suite should call result.done()"""
 
2859
        self.calls = 0
 
2860
        def one_more_call(): self.calls += 1
 
2861
        def test_function():
3353
2862
            pass
3354
 
 
3355
 
    def test_basics(self):
3356
 
        # Make sure we know the definition of BZR_HOME: not part of os.environ
3357
 
        # for tests.TestCase.
3358
 
        self.assertTrue('BZR_HOME' in tests.isolated_environ)
3359
 
        self.assertEquals(None, tests.isolated_environ['BZR_HOME'])
3360
 
        # Being part of isolated_environ, BZR_HOME should not appear here
3361
 
        self.assertFalse('BZR_HOME' in os.environ)
3362
 
        # Make sure we know the definition of LINES: part of os.environ for
3363
 
        # tests.TestCase
3364
 
        self.assertTrue('LINES' in tests.isolated_environ)
3365
 
        self.assertEquals('25', tests.isolated_environ['LINES'])
3366
 
        self.assertEquals('25', os.environ['LINES'])
3367
 
 
3368
 
    def test_injecting_unknown_variable(self):
3369
 
        # BZR_HOME is known to be absent from os.environ
3370
 
        test = self.ScratchMonkey('test_me')
3371
 
        tests.override_os_environ(test, {'BZR_HOME': 'foo'})
3372
 
        self.assertEquals('foo', os.environ['BZR_HOME'])
3373
 
        tests.restore_os_environ(test)
3374
 
        self.assertFalse('BZR_HOME' in os.environ)
3375
 
 
3376
 
    def test_injecting_known_variable(self):
3377
 
        test = self.ScratchMonkey('test_me')
3378
 
        # LINES is known to be present in os.environ
3379
 
        tests.override_os_environ(test, {'LINES': '42'})
3380
 
        self.assertEquals('42', os.environ['LINES'])
3381
 
        tests.restore_os_environ(test)
3382
 
        self.assertEquals('25', os.environ['LINES'])
3383
 
 
3384
 
    def test_deleting_variable(self):
3385
 
        test = self.ScratchMonkey('test_me')
3386
 
        # LINES is known to be present in os.environ
3387
 
        tests.override_os_environ(test, {'LINES': None})
3388
 
        self.assertTrue('LINES' not in os.environ)
3389
 
        tests.restore_os_environ(test)
3390
 
        self.assertEquals('25', os.environ['LINES'])
3391
 
 
3392
 
 
3393
 
class TestDocTestSuiteIsolation(tests.TestCase):
3394
 
    """Test that `tests.DocTestSuite` isolates doc tests from os.environ.
3395
 
 
3396
 
    Since tests.TestCase alreay provides an isolation from os.environ, we use
3397
 
    the clean environment as a base for testing. To precisely capture the
3398
 
    isolation provided by tests.DocTestSuite, we use doctest.DocTestSuite to
3399
 
    compare against.
3400
 
 
3401
 
    We want to make sure `tests.DocTestSuite` respect `tests.isolated_environ`,
3402
 
    not `os.environ` so each test overrides it to suit its needs.
3403
 
 
3404
 
    """
3405
 
 
3406
 
    def get_doctest_suite_for_string(self, klass, string):
3407
 
        class Finder(doctest.DocTestFinder):
3408
 
 
3409
 
            def find(*args, **kwargs):
3410
 
                test = doctest.DocTestParser().get_doctest(
3411
 
                    string, {}, 'foo', 'foo.py', 0)
3412
 
                return [test]
3413
 
 
3414
 
        suite = klass(test_finder=Finder())
3415
 
        return suite
3416
 
 
3417
 
    def run_doctest_suite_for_string(self, klass, string):
3418
 
        suite = self.get_doctest_suite_for_string(klass, string)
3419
 
        output = StringIO()
3420
 
        result = tests.TextTestResult(output, 0, 1)
3421
 
        suite.run(result)
3422
 
        return result, output
3423
 
 
3424
 
    def assertDocTestStringSucceds(self, klass, string):
3425
 
        result, output = self.run_doctest_suite_for_string(klass, string)
3426
 
        if not result.wasStrictlySuccessful():
3427
 
            self.fail(output.getvalue())
3428
 
 
3429
 
    def assertDocTestStringFails(self, klass, string):
3430
 
        result, output = self.run_doctest_suite_for_string(klass, string)
3431
 
        if result.wasStrictlySuccessful():
3432
 
            self.fail(output.getvalue())
3433
 
 
3434
 
    def test_injected_variable(self):
3435
 
        self.overrideAttr(tests, 'isolated_environ', {'LINES': '42'})
3436
 
        test = """
3437
 
            >>> import os
3438
 
            >>> os.environ['LINES']
3439
 
            '42'
3440
 
            """
3441
 
        # doctest.DocTestSuite fails as it sees '25'
3442
 
        self.assertDocTestStringFails(doctest.DocTestSuite, test)
3443
 
        # tests.DocTestSuite sees '42'
3444
 
        self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
3445
 
 
3446
 
    def test_deleted_variable(self):
3447
 
        self.overrideAttr(tests, 'isolated_environ', {'LINES': None})
3448
 
        test = """
3449
 
            >>> import os
3450
 
            >>> os.environ.get('LINES')
3451
 
            """
3452
 
        # doctest.DocTestSuite fails as it sees '25'
3453
 
        self.assertDocTestStringFails(doctest.DocTestSuite, test)
3454
 
        # tests.DocTestSuite sees None
3455
 
        self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
3456
 
 
3457
 
 
3458
 
class TestSelftestExcludePatterns(tests.TestCase):
3459
 
 
3460
 
    def setUp(self):
3461
 
        super(TestSelftestExcludePatterns, self).setUp()
3462
 
        self.overrideAttr(tests, 'test_suite', self.suite_factory)
3463
 
 
3464
 
    def suite_factory(self, keep_only=None, starting_with=None):
3465
 
        """A test suite factory with only a few tests."""
3466
 
        class Test(tests.TestCase):
3467
 
            def id(self):
3468
 
                # We don't need the full class path
3469
 
                return self._testMethodName
3470
 
            def a(self):
3471
 
                pass
3472
 
            def b(self):
3473
 
                pass
3474
 
            def c(self):
3475
 
                pass
3476
 
        return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
3477
 
 
3478
 
    def assertTestList(self, expected, *selftest_args):
3479
 
        # We rely on setUp installing the right test suite factory so we can
3480
 
        # test at the command level without loading the whole test suite
3481
 
        out, err = self.run_bzr(('selftest', '--list') + selftest_args)
3482
 
        actual = out.splitlines()
3483
 
        self.assertEquals(expected, actual)
3484
 
 
3485
 
    def test_full_list(self):
3486
 
        self.assertTestList(['a', 'b', 'c'])
3487
 
 
3488
 
    def test_single_exclude(self):
3489
 
        self.assertTestList(['b', 'c'], '-x', 'a')
3490
 
 
3491
 
    def test_mutiple_excludes(self):
3492
 
        self.assertTestList(['c'], '-x', 'a', '-x', 'b')
3493
 
 
3494
 
 
3495
 
class TestCounterHooks(tests.TestCase, SelfTestHelper):
3496
 
 
3497
 
    _test_needs_features = [features.subunit]
3498
 
 
3499
 
    def setUp(self):
3500
 
        super(TestCounterHooks, self).setUp()
3501
 
        class Test(tests.TestCase):
3502
 
 
3503
 
            def setUp(self):
3504
 
                super(Test, self).setUp()
3505
 
                self.hooks = hooks.Hooks()
3506
 
                self.hooks.add_hook('myhook', 'Foo bar blah', (2,4))
3507
 
                self.install_counter_hook(self.hooks, 'myhook')
3508
 
 
3509
 
            def no_hook(self):
3510
 
                pass
3511
 
 
3512
 
            def run_hook_once(self):
3513
 
                for hook in self.hooks['myhook']:
3514
 
                    hook(self)
3515
 
 
3516
 
        self.test_class = Test
3517
 
 
3518
 
    def assertHookCalls(self, expected_calls, test_name):
3519
 
        test = self.test_class(test_name)
3520
 
        result = unittest.TestResult()
3521
 
        test.run(result)
3522
 
        self.assertTrue(hasattr(test, '_counters'))
3523
 
        self.assertTrue(test._counters.has_key('myhook'))
3524
 
        self.assertEquals(expected_calls, test._counters['myhook'])
3525
 
 
3526
 
    def test_no_hook(self):
3527
 
        self.assertHookCalls(0, 'no_hook')
3528
 
 
3529
 
    def test_run_hook_once(self):
3530
 
        tt = features.testtools
3531
 
        if tt.module.__version__ < (0, 9, 8):
3532
 
            raise tests.TestSkipped('testtools-0.9.8 required for addDetail')
3533
 
        self.assertHookCalls(1, 'run_hook_once')
 
2863
        test = unittest.FunctionTestCase(test_function)
 
2864
        class InstrumentedTestResult(tests.ExtendedTestResult):
 
2865
            def done(self): one_more_call()
 
2866
        class MyRunner(tests.TextTestRunner):
 
2867
            def run(self, test):
 
2868
                return InstrumentedTestResult(self.stream, self.descriptions,
 
2869
                                              self.verbosity)
 
2870
        tests.run_suite(test, runner_class=MyRunner, stream=StringIO())
 
2871
        self.assertEquals(1, self.calls)