← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Martin
  • Date: 2010-05-03 20:57:39 UTC
  • mto: This revision was merged to the branch mainline in revision 5204.
  • Revision ID: gzlist@googlemail.com-20100503205739-n326zdvevv0rmruh
Retain original stack and error message when translating to ValueError in bencode

Show diffs side-by-side

added added

removed removed

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