~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2006-03-09 08:50:48 UTC
  • mfrom: (1553.5.83 bzr.mbp.locks)
  • Revision ID: pqm@pqm.ubuntu.com-20060309085048-37f21fd146dabe93
[merge] LockDir integration into new formats

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005-2011 Canonical Ltd
 
1
# Copyright (C) 2005 by Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
 
# it under the terms of the GNU General Public License as published by
5
 
# the Free Software Foundation; either version 2 of the License, or
6
 
# (at your option) any later version.
 
4
# it under the terms of the GNU General Public License version 2 as published by
 
5
# the Free Software Foundation.
7
6
#
8
7
# This program is distributed in the hope that it will be useful,
9
8
# but WITHOUT ANY WARRANTY; without even the implied warranty of
12
11
#
13
12
# You should have received a copy of the GNU General Public License
14
13
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
 
 
17
 
"""Tests for the test framework."""
18
 
 
19
 
from cStringIO import StringIO
20
 
import gc
21
 
import doctest
 
14
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
15
 
 
16
"""Tests for the test framework
 
17
"""
 
18
 
22
19
import os
23
 
import signal
24
20
import sys
25
 
import threading
26
 
import time
27
21
import unittest
28
22
import warnings
29
23
 
30
 
from testtools import (
31
 
    ExtendedToOriginalDecorator,
32
 
    MultiTestResult,
33
 
    )
34
 
from testtools.content import Content
35
 
from testtools.content_type import ContentType
36
 
from testtools.matchers import (
37
 
    DocTestMatches,
38
 
    Equals,
39
 
    )
40
 
import testtools.testresult.doubles
41
 
 
42
24
import bzrlib
43
 
from bzrlib import (
44
 
    branchbuilder,
45
 
    bzrdir,
46
 
    controldir,
47
 
    errors,
48
 
    hooks,
49
 
    lockdir,
50
 
    memorytree,
51
 
    osutils,
52
 
    remote,
53
 
    repository,
54
 
    symbol_versioning,
55
 
    tests,
56
 
    transport,
57
 
    workingtree,
58
 
    workingtree_3,
59
 
    workingtree_4,
60
 
    )
61
 
from bzrlib.repofmt import (
62
 
    groupcompress_repo,
63
 
    )
64
 
from bzrlib.symbol_versioning import (
65
 
    deprecated_function,
66
 
    deprecated_in,
67
 
    deprecated_method,
68
 
    )
69
25
from bzrlib.tests import (
70
 
    features,
71
 
    test_lsprof,
72
 
    test_server,
73
 
    TestUtil,
74
 
    )
75
 
from bzrlib.trace import note, mutter
76
 
from bzrlib.transport import memory
77
 
 
78
 
 
79
 
def _test_ids(test_suite):
80
 
    """Get the ids for the tests in a test suite."""
81
 
    return [t.id() for t in tests.iter_suite_tests(test_suite)]
82
 
 
83
 
 
84
 
class MetaTestLog(tests.TestCase):
 
26
                          _load_module_by_name,
 
27
                          ChrootedTestCase,
 
28
                          TestCase,
 
29
                          TestCaseInTempDir,
 
30
                          TestCaseWithTransport,
 
31
                          TestSkipped,
 
32
                          TextTestRunner,
 
33
                          )
 
34
import bzrlib.errors as errors
 
35
 
 
36
 
 
37
class SelftestTests(TestCase):
 
38
 
 
39
    def test_import_tests(self):
 
40
        mod = _load_module_by_name('bzrlib.tests.test_selftest')
 
41
        self.assertEqual(mod.SelftestTests, SelftestTests)
 
42
 
 
43
    def test_import_test_failure(self):
 
44
        self.assertRaises(ImportError,
 
45
                          _load_module_by_name,
 
46
                          'bzrlib.no-name-yet')
 
47
 
 
48
 
 
49
class MetaTestLog(TestCase):
85
50
 
86
51
    def test_logging(self):
87
52
        """Test logs are captured when a test fails."""
88
53
        self.log('a test message')
89
 
        details = self.getDetails()
90
 
        log = details['log']
91
 
        self.assertThat(log.content_type, Equals(ContentType(
92
 
            "text", "plain", {"charset": "utf8"})))
93
 
        self.assertThat(u"".join(log.iter_text()), Equals(self.get_log()))
94
 
        self.assertThat(self.get_log(),
95
 
            DocTestMatches(u"...a test message\n", doctest.ELLIPSIS))
96
 
 
97
 
 
98
 
class TestTreeShape(tests.TestCaseInTempDir):
 
54
        self._log_file.flush()
 
55
        self.assertContainsRe(self._get_log(), 'a test message\n')
 
56
 
 
57
 
 
58
class TestTreeShape(TestCaseInTempDir):
99
59
 
100
60
    def test_unicode_paths(self):
101
 
        self.requireFeature(features.UnicodeFilenameFeature)
102
 
 
103
61
        filename = u'hell\u00d8'
104
 
        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
119
 
 
120
 
 
121
 
class TestTransportScenarios(tests.TestCase):
 
62
        try:
 
63
            self.build_tree_contents([(filename, 'contents of hello')])
 
64
        except UnicodeEncodeError:
 
65
            raise TestSkipped("can't build unicode working tree in "
 
66
                "filesystem encoding %s" % sys.getfilesystemencoding())
 
67
        self.failUnlessExists(filename)
 
68
 
 
69
 
 
70
class TestSkippedTest(TestCase):
 
71
    """Try running a test which is skipped, make sure it's reported properly."""
 
72
 
 
73
    def test_skipped_test(self):
 
74
        # must be hidden in here so it's not run as a real test
 
75
        def skipping_test():
 
76
            raise TestSkipped('test intentionally skipped')
 
77
        runner = TextTestRunner(stream=self._log_file)
 
78
        test = unittest.FunctionTestCase(skipping_test)
 
79
        result = runner.run(test)
 
80
        self.assertTrue(result.wasSuccessful())
 
81
 
 
82
 
 
83
class TestTransportProviderAdapter(TestCase):
122
84
    """A group of tests that test the transport implementation adaption core.
123
85
 
124
 
    This is a meta test that the tests are applied to all available
 
86
    This is a meta test that the tests are applied to all available 
125
87
    transports.
126
88
 
127
 
    This will be generalised in the future which is why it is in this
 
89
    This will be generalised in the future which is why it is in this 
128
90
    test file even though it is specific to transport tests at the moment.
129
91
    """
130
92
 
131
93
    def test_get_transport_permutations(self):
132
 
        # this checks that get_test_permutations defined by the module is
133
 
        # called by the get_transport_test_permutations function.
 
94
        # this checks that we the module get_test_permutations call
 
95
        # is made by the adapter get_transport_test_permitations method.
134
96
        class MockModule(object):
135
97
            def get_test_permutations(self):
136
98
                return sample_permutation
137
99
        sample_permutation = [(1,2), (3,4)]
138
 
        from bzrlib.tests.per_transport import get_transport_test_permutations
 
100
        from bzrlib.transport import TransportTestProviderAdapter
 
101
        adapter = TransportTestProviderAdapter()
139
102
        self.assertEqual(sample_permutation,
140
 
                         get_transport_test_permutations(MockModule()))
 
103
                         adapter.get_transport_test_permutations(MockModule()))
141
104
 
142
 
    def test_scenarios_include_all_modules(self):
143
 
        # this checks that the scenario generator returns as many permutations
144
 
        # as there are in all the registered transport modules - we assume if
145
 
        # this matches its probably doing the right thing especially in
146
 
        # combination with the tests for setting the right classes below.
147
 
        from bzrlib.tests.per_transport import transport_test_permutations
148
 
        from bzrlib.transport import _get_transport_modules
 
105
    def test_adapter_checks_all_modules(self):
 
106
        # this checks that the adapter returns as many permurtations as
 
107
        # there are in all the registered# transport modules for there
 
108
        # - we assume if this matches its probably doing the right thing
 
109
        # especially in combination with the tests for setting the right
 
110
        # classes below.
 
111
        from bzrlib.transport import (TransportTestProviderAdapter,
 
112
                                      _get_transport_modules
 
113
                                      )
149
114
        modules = _get_transport_modules()
150
115
        permutation_count = 0
151
116
        for module in modules:
152
117
            try:
153
 
                permutation_count += len(reduce(getattr,
 
118
                permutation_count += len(reduce(getattr, 
154
119
                    (module + ".get_test_permutations").split('.')[1:],
155
120
                     __import__(module))())
156
121
            except errors.DependencyNotPresent:
157
122
                pass
158
 
        scenarios = transport_test_permutations()
159
 
        self.assertEqual(permutation_count, len(scenarios))
160
 
 
161
 
    def test_scenarios_include_transport_class(self):
162
 
        # This test used to know about all the possible transports and the
163
 
        # order they were returned but that seems overly brittle (mbp
164
 
        # 20060307)
165
 
        from bzrlib.tests.per_transport import transport_test_permutations
166
 
        scenarios = transport_test_permutations()
167
 
        # there are at least that many builtin transports
168
 
        self.assertTrue(len(scenarios) > 6)
169
 
        one_scenario = scenarios[0]
170
 
        self.assertIsInstance(one_scenario[0], str)
171
 
        self.assertTrue(issubclass(one_scenario[1]["transport_class"],
172
 
                                   bzrlib.transport.Transport))
173
 
        self.assertTrue(issubclass(one_scenario[1]["transport_server"],
174
 
                                   bzrlib.transport.Server))
175
 
 
176
 
 
177
 
class TestBranchScenarios(tests.TestCase):
178
 
 
179
 
    def test_scenarios(self):
 
123
        input_test = TestTransportProviderAdapter(
 
124
            "test_adapter_sets_transport_class")
 
125
        adapter = TransportTestProviderAdapter()
 
126
        self.assertEqual(permutation_count,
 
127
                         len(list(iter(adapter.adapt(input_test)))))
 
128
 
 
129
    def test_adapter_sets_transport_class(self):
 
130
        # when the adapter adapts a test it needs to 
 
131
        # place one of the permutations from the transport
 
132
        # providers in each test case copy. This checks
 
133
        # that it does not just use the same one all the time.
 
134
        # and that the id is set correctly so that debugging is
 
135
        # easy.
 
136
        # 
 
137
        # An instance of this test is actually used as the input
 
138
        # for adapting it to all the available transports
 
139
        # (or i think so - ??? mbp)
 
140
        from bzrlib.transport.local import (LocalTransport,
 
141
                                            LocalRelpathServer,
 
142
                                            LocalAbspathServer,
 
143
                                            LocalURLServer
 
144
                                            )
 
145
        try:
 
146
            from bzrlib.transport.sftp import (SFTPTransport,
 
147
                                               SFTPAbsoluteServer,
 
148
                                               SFTPHomeDirServer,
 
149
                                               SFTPSiblingAbsoluteServer,
 
150
                                               )
 
151
        except errors.ParamikoNotPresent, e:
 
152
            warnings.warn(str(e))
 
153
            has_paramiko = False
 
154
        else:
 
155
            has_paramiko = True
 
156
        from bzrlib.transport.http import (HttpTransport,
 
157
                                           HttpServer
 
158
                                           )
 
159
        from bzrlib.transport.ftp import FtpTransport
 
160
        from bzrlib.transport.memory import (MemoryTransport,
 
161
                                             MemoryServer
 
162
                                             )
 
163
        from bzrlib.transport import TransportTestProviderAdapter
 
164
        # FIXME. What we want is a factory for the things
 
165
        # needed to test the implementation. I.e. for transport we want:
 
166
        # the class that connections should get; a local server factory
 
167
        # so we would want the following permutations:
 
168
        # LocalTransport relpath-factory
 
169
        # LocalTransport abspath-factory
 
170
        # LocalTransport file://-factory
 
171
        # SFTPTransport homedir-factory
 
172
        # SFTPTransport abssolute-factory
 
173
        # HTTPTransport http-factory
 
174
        # HTTPTransport https-factory
 
175
        # etc, but we are currently lacking in this, so print out that
 
176
        # this should be fixed.
 
177
        input_test = TestTransportProviderAdapter(
 
178
            "test_adapter_sets_transport_class")
 
179
        suite = TransportTestProviderAdapter().adapt(input_test)
 
180
        # tests are generated in collation order. 
 
181
        # XXX: but i'm not sure the order should really be part of the 
 
182
        # contract of the adapter, should it -- mbp 20060201
 
183
        test_iter = iter(suite)
 
184
        http_test = test_iter.next()
 
185
        local_relpath_test = test_iter.next()
 
186
        local_abspath_test = test_iter.next()
 
187
        local_urlpath_test = test_iter.next()
 
188
        memory_test = test_iter.next()
 
189
        readonly_test = test_iter.next()
 
190
        if has_paramiko:
 
191
            sftp_abs_test = test_iter.next()
 
192
            sftp_homedir_test = test_iter.next()
 
193
            sftp_sibling_abs_test = test_iter.next()
 
194
        # ftp_test = test_iter.next()
 
195
        # should now be at the end of the test
 
196
        self.assertRaises(StopIteration, test_iter.next)
 
197
        self.assertEqual(LocalTransport, local_relpath_test.transport_class)
 
198
        self.assertEqual(LocalRelpathServer, local_relpath_test.transport_server)
 
199
        
 
200
        self.assertEqual(LocalTransport, local_abspath_test.transport_class)
 
201
        self.assertEqual(LocalAbspathServer, local_abspath_test.transport_server)
 
202
 
 
203
        self.assertEqual(LocalTransport, local_urlpath_test.transport_class)
 
204
        self.assertEqual(LocalURLServer, local_urlpath_test.transport_server)
 
205
 
 
206
        if has_paramiko:
 
207
            self.assertEqual(SFTPTransport, sftp_abs_test.transport_class)
 
208
            self.assertEqual(SFTPAbsoluteServer, sftp_abs_test.transport_server)
 
209
            self.assertEqual(SFTPTransport, sftp_homedir_test.transport_class)
 
210
            self.assertEqual(SFTPHomeDirServer, sftp_homedir_test.transport_server)
 
211
            self.assertEqual(SFTPTransport, sftp_sibling_abs_test.transport_class)
 
212
            self.assertEqual(SFTPSiblingAbsoluteServer,
 
213
                             sftp_sibling_abs_test.transport_server)
 
214
 
 
215
        self.assertEqual(HttpTransport, http_test.transport_class)
 
216
        self.assertEqual(HttpServer, http_test.transport_server)
 
217
        # self.assertEqual(FtpTransport, ftp_test.transport_class)
 
218
 
 
219
        self.assertEqual(MemoryTransport, memory_test.transport_class)
 
220
        self.assertEqual(MemoryServer, memory_test.transport_server)
 
221
        
 
222
        # we could test all of them for .id, but two is probably sufficient.
 
223
        self.assertEqual("bzrlib.tests.test_selftest."
 
224
                         "TestTransportProviderAdapter."
 
225
                         "test_adapter_sets_transport_class(MemoryServer)",
 
226
                         memory_test.id())
 
227
        self.assertEqual("bzrlib.tests.test_selftest."
 
228
                         "TestTransportProviderAdapter."
 
229
                         "test_adapter_sets_transport_class(LocalRelpathServer)",
 
230
                         local_relpath_test.id())
 
231
 
 
232
 
 
233
class TestBranchProviderAdapter(TestCase):
 
234
    """A group of tests that test the branch implementation test adapter."""
 
235
 
 
236
    def test_adapted_tests(self):
180
237
        # check that constructor parameters are passed through to the adapted
181
238
        # test.
182
 
        from bzrlib.tests.per_branch import make_scenarios
 
239
        from bzrlib.branch import BranchTestProviderAdapter
 
240
        input_test = TestBranchProviderAdapter(
 
241
            "test_adapted_tests")
183
242
        server1 = "a"
184
243
        server2 = "b"
185
244
        formats = [("c", "C"), ("d", "D")]
186
 
        scenarios = make_scenarios(server1, server2, formats)
187
 
        self.assertEqual(2, len(scenarios))
188
 
        self.assertEqual([
189
 
            ('str',
190
 
             {'branch_format': 'c',
191
 
              'bzrdir_format': 'C',
192
 
              'transport_readonly_server': 'b',
193
 
              'transport_server': 'a'}),
194
 
            ('str',
195
 
             {'branch_format': 'd',
196
 
              'bzrdir_format': 'D',
197
 
              'transport_readonly_server': 'b',
198
 
              'transport_server': 'a'})],
199
 
            scenarios)
200
 
 
201
 
 
202
 
class TestBzrDirScenarios(tests.TestCase):
203
 
 
204
 
    def test_scenarios(self):
 
245
        adapter = BranchTestProviderAdapter(server1, server2, formats)
 
246
        suite = adapter.adapt(input_test)
 
247
        tests = list(iter(suite))
 
248
        self.assertEqual(2, len(tests))
 
249
        self.assertEqual(tests[0].branch_format, formats[0][0])
 
250
        self.assertEqual(tests[0].bzrdir_format, formats[0][1])
 
251
        self.assertEqual(tests[0].transport_server, server1)
 
252
        self.assertEqual(tests[0].transport_readonly_server, server2)
 
253
        self.assertEqual(tests[1].branch_format, formats[1][0])
 
254
        self.assertEqual(tests[1].bzrdir_format, formats[1][1])
 
255
        self.assertEqual(tests[1].transport_server, server1)
 
256
        self.assertEqual(tests[1].transport_readonly_server, server2)
 
257
 
 
258
 
 
259
class TestBzrDirProviderAdapter(TestCase):
 
260
    """A group of tests that test the bzr dir implementation test adapter."""
 
261
 
 
262
    def test_adapted_tests(self):
205
263
        # check that constructor parameters are passed through to the adapted
206
264
        # test.
207
 
        from bzrlib.tests.per_controldir import make_scenarios
208
 
        vfs_factory = "v"
 
265
        from bzrlib.bzrdir import BzrDirTestProviderAdapter
 
266
        input_test = TestBzrDirProviderAdapter(
 
267
            "test_adapted_tests")
209
268
        server1 = "a"
210
269
        server2 = "b"
211
270
        formats = ["c", "d"]
212
 
        scenarios = make_scenarios(vfs_factory, server1, server2, formats)
213
 
        self.assertEqual([
214
 
            ('str',
215
 
             {'bzrdir_format': 'c',
216
 
              'transport_readonly_server': 'b',
217
 
              'transport_server': 'a',
218
 
              'vfs_transport_factory': 'v'}),
219
 
            ('str',
220
 
             {'bzrdir_format': 'd',
221
 
              'transport_readonly_server': 'b',
222
 
              'transport_server': 'a',
223
 
              'vfs_transport_factory': 'v'})],
224
 
            scenarios)
225
 
 
226
 
 
227
 
class TestRepositoryScenarios(tests.TestCase):
228
 
 
229
 
    def test_formats_to_scenarios(self):
230
 
        from bzrlib.tests.per_repository import formats_to_scenarios
231
 
        formats = [("(c)", remote.RemoteRepositoryFormat()),
232
 
                   ("(d)", repository.format_registry.get(
233
 
                    'Bazaar repository format 2a (needs bzr 1.16 or later)\n'))]
234
 
        no_vfs_scenarios = formats_to_scenarios(formats, "server", "readonly",
235
 
            None)
236
 
        vfs_scenarios = formats_to_scenarios(formats, "server", "readonly",
237
 
            vfs_transport_factory="vfs")
238
 
        # no_vfs generate scenarios without vfs_transport_factory
239
 
        expected = [
240
 
            ('RemoteRepositoryFormat(c)',
241
 
             {'bzrdir_format': remote.RemoteBzrDirFormat(),
242
 
              'repository_format': remote.RemoteRepositoryFormat(),
243
 
              'transport_readonly_server': 'readonly',
244
 
              'transport_server': 'server'}),
245
 
            ('RepositoryFormat2a(d)',
246
 
             {'bzrdir_format': bzrdir.BzrDirMetaFormat1(),
247
 
              'repository_format': groupcompress_repo.RepositoryFormat2a(),
248
 
              'transport_readonly_server': 'readonly',
249
 
              'transport_server': 'server'})]
250
 
        self.assertEqual(expected, no_vfs_scenarios)
251
 
        self.assertEqual([
252
 
            ('RemoteRepositoryFormat(c)',
253
 
             {'bzrdir_format': remote.RemoteBzrDirFormat(),
254
 
              'repository_format': remote.RemoteRepositoryFormat(),
255
 
              'transport_readonly_server': 'readonly',
256
 
              'transport_server': 'server',
257
 
              'vfs_transport_factory': 'vfs'}),
258
 
            ('RepositoryFormat2a(d)',
259
 
             {'bzrdir_format': bzrdir.BzrDirMetaFormat1(),
260
 
              'repository_format': groupcompress_repo.RepositoryFormat2a(),
261
 
              'transport_readonly_server': 'readonly',
262
 
              'transport_server': 'server',
263
 
              'vfs_transport_factory': 'vfs'})],
264
 
            vfs_scenarios)
265
 
 
266
 
 
267
 
class TestTestScenarioApplication(tests.TestCase):
268
 
    """Tests for the test adaption facilities."""
269
 
 
270
 
    def test_apply_scenario(self):
271
 
        from bzrlib.tests import apply_scenario
272
 
        input_test = TestTestScenarioApplication("test_apply_scenario")
273
 
        # setup two adapted tests
274
 
        adapted_test1 = apply_scenario(input_test,
275
 
            ("new id",
276
 
            {"bzrdir_format":"bzr_format",
277
 
             "repository_format":"repo_fmt",
278
 
             "transport_server":"transport_server",
279
 
             "transport_readonly_server":"readonly-server"}))
280
 
        adapted_test2 = apply_scenario(input_test,
281
 
            ("new id 2", {"bzrdir_format":None}))
282
 
        # input_test should have been altered.
283
 
        self.assertRaises(AttributeError, getattr, input_test, "bzrdir_format")
284
 
        # the new tests are mutually incompatible, ensuring it has
285
 
        # made new ones, and unspecified elements in the scenario
286
 
        # should not have been altered.
287
 
        self.assertEqual("bzr_format", adapted_test1.bzrdir_format)
288
 
        self.assertEqual("repo_fmt", adapted_test1.repository_format)
289
 
        self.assertEqual("transport_server", adapted_test1.transport_server)
290
 
        self.assertEqual("readonly-server",
291
 
            adapted_test1.transport_readonly_server)
292
 
        self.assertEqual(
293
 
            "bzrlib.tests.test_selftest.TestTestScenarioApplication."
294
 
            "test_apply_scenario(new id)",
295
 
            adapted_test1.id())
296
 
        self.assertEqual(None, adapted_test2.bzrdir_format)
297
 
        self.assertEqual(
298
 
            "bzrlib.tests.test_selftest.TestTestScenarioApplication."
299
 
            "test_apply_scenario(new id 2)",
300
 
            adapted_test2.id())
301
 
 
302
 
 
303
 
class TestInterRepositoryScenarios(tests.TestCase):
304
 
 
305
 
    def test_scenarios(self):
306
 
        # check that constructor parameters are passed through to the adapted
307
 
        # test.
308
 
        from bzrlib.tests.per_interrepository import make_scenarios
309
 
        server1 = "a"
310
 
        server2 = "b"
311
 
        formats = [("C0", "C1", "C2", "C3"), ("D0", "D1", "D2", "D3")]
312
 
        scenarios = make_scenarios(server1, server2, formats)
313
 
        self.assertEqual([
314
 
            ('C0,str,str',
315
 
             {'repository_format': 'C1',
316
 
              'repository_format_to': 'C2',
317
 
              'transport_readonly_server': 'b',
318
 
              'transport_server': 'a',
319
 
              'extra_setup': 'C3'}),
320
 
            ('D0,str,str',
321
 
             {'repository_format': 'D1',
322
 
              'repository_format_to': 'D2',
323
 
              'transport_readonly_server': 'b',
324
 
              'transport_server': 'a',
325
 
              'extra_setup': 'D3'})],
326
 
            scenarios)
327
 
 
328
 
 
329
 
class TestWorkingTreeScenarios(tests.TestCase):
330
 
 
331
 
    def test_scenarios(self):
332
 
        # check that constructor parameters are passed through to the adapted
333
 
        # test.
334
 
        from bzrlib.tests.per_workingtree import make_scenarios
335
 
        server1 = "a"
336
 
        server2 = "b"
337
 
        formats = [workingtree_4.WorkingTreeFormat4(),
338
 
                   workingtree_3.WorkingTreeFormat3(),]
339
 
        scenarios = make_scenarios(server1, server2, formats)
340
 
        self.assertEqual([
341
 
            ('WorkingTreeFormat4',
342
 
             {'bzrdir_format': formats[0]._matchingbzrdir,
343
 
              'transport_readonly_server': 'b',
344
 
              'transport_server': 'a',
345
 
              'workingtree_format': formats[0]}),
346
 
            ('WorkingTreeFormat3',
347
 
             {'bzrdir_format': formats[1]._matchingbzrdir,
348
 
              'transport_readonly_server': 'b',
349
 
              'transport_server': 'a',
350
 
              'workingtree_format': formats[1]})],
351
 
            scenarios)
352
 
 
353
 
 
354
 
class TestTreeScenarios(tests.TestCase):
355
 
 
356
 
    def test_scenarios(self):
357
 
        # the tree implementation scenario generator is meant to setup one
358
 
        # instance for each working tree format, and one additional instance
359
 
        # that will use the default wt format, but create a revision tree for
360
 
        # the tests.  this means that the wt ones should have the
361
 
        # workingtree_to_test_tree attribute set to 'return_parameter' and the
362
 
        # revision one set to revision_tree_from_workingtree.
363
 
 
364
 
        from bzrlib.tests.per_tree import (
365
 
            _dirstate_tree_from_workingtree,
366
 
            make_scenarios,
367
 
            preview_tree_pre,
368
 
            preview_tree_post,
369
 
            return_parameter,
370
 
            revision_tree_from_workingtree
371
 
            )
372
 
        server1 = "a"
373
 
        server2 = "b"
374
 
        formats = [workingtree_4.WorkingTreeFormat4(),
375
 
                   workingtree_3.WorkingTreeFormat3(),]
376
 
        scenarios = make_scenarios(server1, server2, formats)
377
 
        self.assertEqual(7, len(scenarios))
378
 
        default_wt_format = workingtree.format_registry.get_default()
379
 
        wt4_format = workingtree_4.WorkingTreeFormat4()
380
 
        wt5_format = workingtree_4.WorkingTreeFormat5()
381
 
        expected_scenarios = [
382
 
            ('WorkingTreeFormat4',
383
 
             {'bzrdir_format': formats[0]._matchingbzrdir,
384
 
              'transport_readonly_server': 'b',
385
 
              'transport_server': 'a',
386
 
              'workingtree_format': formats[0],
387
 
              '_workingtree_to_test_tree': return_parameter,
388
 
              }),
389
 
            ('WorkingTreeFormat3',
390
 
             {'bzrdir_format': formats[1]._matchingbzrdir,
391
 
              'transport_readonly_server': 'b',
392
 
              'transport_server': 'a',
393
 
              'workingtree_format': formats[1],
394
 
              '_workingtree_to_test_tree': return_parameter,
395
 
             }),
396
 
            ('RevisionTree',
397
 
             {'_workingtree_to_test_tree': revision_tree_from_workingtree,
398
 
              'bzrdir_format': default_wt_format._matchingbzrdir,
399
 
              'transport_readonly_server': 'b',
400
 
              'transport_server': 'a',
401
 
              'workingtree_format': default_wt_format,
402
 
             }),
403
 
            ('DirStateRevisionTree,WT4',
404
 
             {'_workingtree_to_test_tree': _dirstate_tree_from_workingtree,
405
 
              'bzrdir_format': wt4_format._matchingbzrdir,
406
 
              'transport_readonly_server': 'b',
407
 
              'transport_server': 'a',
408
 
              'workingtree_format': wt4_format,
409
 
             }),
410
 
            ('DirStateRevisionTree,WT5',
411
 
             {'_workingtree_to_test_tree': _dirstate_tree_from_workingtree,
412
 
              'bzrdir_format': wt5_format._matchingbzrdir,
413
 
              'transport_readonly_server': 'b',
414
 
              'transport_server': 'a',
415
 
              'workingtree_format': wt5_format,
416
 
             }),
417
 
            ('PreviewTree',
418
 
             {'_workingtree_to_test_tree': preview_tree_pre,
419
 
              'bzrdir_format': default_wt_format._matchingbzrdir,
420
 
              'transport_readonly_server': 'b',
421
 
              'transport_server': 'a',
422
 
              'workingtree_format': default_wt_format}),
423
 
            ('PreviewTreePost',
424
 
             {'_workingtree_to_test_tree': preview_tree_post,
425
 
              'bzrdir_format': default_wt_format._matchingbzrdir,
426
 
              'transport_readonly_server': 'b',
427
 
              'transport_server': 'a',
428
 
              'workingtree_format': default_wt_format}),
429
 
             ]
430
 
        self.assertEqual(expected_scenarios, scenarios)
431
 
 
432
 
 
433
 
class TestInterTreeScenarios(tests.TestCase):
434
 
    """A group of tests that test the InterTreeTestAdapter."""
435
 
 
436
 
    def test_scenarios(self):
437
 
        # check that constructor parameters are passed through to the adapted
438
 
        # test.
439
 
        # for InterTree tests we want the machinery to bring up two trees in
440
 
        # each instance: the base one, and the one we are interacting with.
441
 
        # because each optimiser can be direction specific, we need to test
442
 
        # each optimiser in its chosen direction.
443
 
        # unlike the TestProviderAdapter we dont want to automatically add a
444
 
        # parameterized one for WorkingTree - the optimisers will tell us what
445
 
        # ones to add.
446
 
        from bzrlib.tests.per_tree import (
447
 
            return_parameter,
448
 
            )
449
 
        from bzrlib.tests.per_intertree import (
450
 
            make_scenarios,
451
 
            )
452
 
        from bzrlib.workingtree_3 import WorkingTreeFormat3
453
 
        from bzrlib.workingtree_4 import WorkingTreeFormat4
454
 
        input_test = TestInterTreeScenarios(
455
 
            "test_scenarios")
456
 
        server1 = "a"
457
 
        server2 = "b"
458
 
        format1 = WorkingTreeFormat4()
459
 
        format2 = WorkingTreeFormat3()
460
 
        formats = [("1", str, format1, format2, "converter1"),
461
 
            ("2", int, format2, format1, "converter2")]
462
 
        scenarios = make_scenarios(server1, server2, formats)
463
 
        self.assertEqual(2, len(scenarios))
464
 
        expected_scenarios = [
465
 
            ("1", {
466
 
                "bzrdir_format": format1._matchingbzrdir,
467
 
                "intertree_class": formats[0][1],
468
 
                "workingtree_format": formats[0][2],
469
 
                "workingtree_format_to": formats[0][3],
470
 
                "mutable_trees_to_test_trees": formats[0][4],
471
 
                "_workingtree_to_test_tree": return_parameter,
472
 
                "transport_server": server1,
473
 
                "transport_readonly_server": server2,
474
 
                }),
475
 
            ("2", {
476
 
                "bzrdir_format": format2._matchingbzrdir,
477
 
                "intertree_class": formats[1][1],
478
 
                "workingtree_format": formats[1][2],
479
 
                "workingtree_format_to": formats[1][3],
480
 
                "mutable_trees_to_test_trees": formats[1][4],
481
 
                "_workingtree_to_test_tree": return_parameter,
482
 
                "transport_server": server1,
483
 
                "transport_readonly_server": server2,
484
 
                }),
485
 
            ]
486
 
        self.assertEqual(scenarios, expected_scenarios)
487
 
 
488
 
 
489
 
class TestTestCaseInTempDir(tests.TestCaseInTempDir):
490
 
 
491
 
    def test_home_is_not_working(self):
492
 
        self.assertNotEqual(self.test_dir, self.test_home_dir)
493
 
        cwd = osutils.getcwd()
494
 
        self.assertIsSameRealPath(self.test_dir, cwd)
495
 
        self.assertIsSameRealPath(self.test_home_dir, os.environ['HOME'])
496
 
 
497
 
    def test_assertEqualStat_equal(self):
498
 
        from bzrlib.tests.test_dirstate import _FakeStat
499
 
        self.build_tree(["foo"])
500
 
        real = os.lstat("foo")
501
 
        fake = _FakeStat(real.st_size, real.st_mtime, real.st_ctime,
502
 
            real.st_dev, real.st_ino, real.st_mode)
503
 
        self.assertEqualStat(real, fake)
504
 
 
505
 
    def test_assertEqualStat_notequal(self):
506
 
        self.build_tree(["foo", "longname"])
507
 
        self.assertRaises(AssertionError, self.assertEqualStat,
508
 
            os.lstat("foo"), os.lstat("longname"))
509
 
 
510
 
    def test_failUnlessExists(self):
511
 
        """Deprecated failUnlessExists and failIfExists"""
512
 
        self.applyDeprecated(
513
 
            deprecated_in((2, 4)),
514
 
            self.failUnlessExists, '.')
515
 
        self.build_tree(['foo/', 'foo/bar'])
516
 
        self.applyDeprecated(
517
 
            deprecated_in((2, 4)),
518
 
            self.failUnlessExists, 'foo/bar')
519
 
        self.applyDeprecated(
520
 
            deprecated_in((2, 4)),
521
 
            self.failIfExists, 'foo/foo')
522
 
 
523
 
    def test_assertPathExists(self):
524
 
        self.assertPathExists('.')
525
 
        self.build_tree(['foo/', 'foo/bar'])
526
 
        self.assertPathExists('foo/bar')
527
 
        self.assertPathDoesNotExist('foo/foo')
528
 
 
529
 
 
530
 
class TestTestCaseWithMemoryTransport(tests.TestCaseWithMemoryTransport):
531
 
 
532
 
    def test_home_is_non_existant_dir_under_root(self):
533
 
        """The test_home_dir for TestCaseWithMemoryTransport is missing.
534
 
 
535
 
        This is because TestCaseWithMemoryTransport is for tests that do not
536
 
        need any disk resources: they should be hooked into bzrlib in such a
537
 
        way that no global settings are being changed by the test (only a
538
 
        few tests should need to do that), and having a missing dir as home is
539
 
        an effective way to ensure that this is the case.
540
 
        """
541
 
        self.assertIsSameRealPath(
542
 
            self.TEST_ROOT + "/MemoryTransportMissingHomeDir",
543
 
            self.test_home_dir)
544
 
        self.assertIsSameRealPath(self.test_home_dir, os.environ['HOME'])
545
 
 
546
 
    def test_cwd_is_TEST_ROOT(self):
547
 
        self.assertIsSameRealPath(self.test_dir, self.TEST_ROOT)
548
 
        cwd = osutils.getcwd()
549
 
        self.assertIsSameRealPath(self.test_dir, cwd)
550
 
 
551
 
    def test_BZR_HOME_and_HOME_are_bytestrings(self):
552
 
        """The $BZR_HOME and $HOME environment variables should not be unicode.
553
 
 
554
 
        See https://bugs.launchpad.net/bzr/+bug/464174
555
 
        """
556
 
        self.assertIsInstance(os.environ['BZR_HOME'], str)
557
 
        self.assertIsInstance(os.environ['HOME'], str)
558
 
 
559
 
    def test_make_branch_and_memory_tree(self):
560
 
        """In TestCaseWithMemoryTransport we should not make the branch on disk.
561
 
 
562
 
        This is hard to comprehensively robustly test, so we settle for making
563
 
        a branch and checking no directory was created at its relpath.
564
 
        """
565
 
        tree = self.make_branch_and_memory_tree('dir')
566
 
        # Guard against regression into MemoryTransport leaking
567
 
        # files to disk instead of keeping them in memory.
568
 
        self.assertFalse(osutils.lexists('dir'))
569
 
        self.assertIsInstance(tree, memorytree.MemoryTree)
570
 
 
571
 
    def test_make_branch_and_memory_tree_with_format(self):
572
 
        """make_branch_and_memory_tree should accept a format option."""
573
 
        format = bzrdir.BzrDirMetaFormat1()
574
 
        format.repository_format = repository.format_registry.get_default()
575
 
        tree = self.make_branch_and_memory_tree('dir', format=format)
576
 
        # Guard against regression into MemoryTransport leaking
577
 
        # files to disk instead of keeping them in memory.
578
 
        self.assertFalse(osutils.lexists('dir'))
579
 
        self.assertIsInstance(tree, memorytree.MemoryTree)
580
 
        self.assertEqual(format.repository_format.__class__,
581
 
            tree.branch.repository._format.__class__)
582
 
 
583
 
    def test_make_branch_builder(self):
584
 
        builder = self.make_branch_builder('dir')
585
 
        self.assertIsInstance(builder, branchbuilder.BranchBuilder)
586
 
        # Guard against regression into MemoryTransport leaking
587
 
        # files to disk instead of keeping them in memory.
588
 
        self.assertFalse(osutils.lexists('dir'))
589
 
 
590
 
    def test_make_branch_builder_with_format(self):
591
 
        # Use a repo layout that doesn't conform to a 'named' layout, to ensure
592
 
        # that the format objects are used.
593
 
        format = bzrdir.BzrDirMetaFormat1()
594
 
        repo_format = repository.format_registry.get_default()
595
 
        format.repository_format = repo_format
596
 
        builder = self.make_branch_builder('dir', format=format)
597
 
        the_branch = builder.get_branch()
598
 
        # Guard against regression into MemoryTransport leaking
599
 
        # files to disk instead of keeping them in memory.
600
 
        self.assertFalse(osutils.lexists('dir'))
601
 
        self.assertEqual(format.repository_format.__class__,
602
 
                         the_branch.repository._format.__class__)
603
 
        self.assertEqual(repo_format.get_format_string(),
604
 
                         self.get_transport().get_bytes(
605
 
                            'dir/.bzr/repository/format'))
606
 
 
607
 
    def test_make_branch_builder_with_format_name(self):
608
 
        builder = self.make_branch_builder('dir', format='knit')
609
 
        the_branch = builder.get_branch()
610
 
        # Guard against regression into MemoryTransport leaking
611
 
        # files to disk instead of keeping them in memory.
612
 
        self.assertFalse(osutils.lexists('dir'))
613
 
        dir_format = controldir.format_registry.make_bzrdir('knit')
614
 
        self.assertEqual(dir_format.repository_format.__class__,
615
 
                         the_branch.repository._format.__class__)
616
 
        self.assertEqual('Bazaar-NG Knit Repository Format 1',
617
 
                         self.get_transport().get_bytes(
618
 
                            'dir/.bzr/repository/format'))
619
 
 
620
 
    def test_dangling_locks_cause_failures(self):
621
 
        class TestDanglingLock(tests.TestCaseWithMemoryTransport):
622
 
            def test_function(self):
623
 
                t = self.get_transport_from_path('.')
624
 
                l = lockdir.LockDir(t, 'lock')
625
 
                l.create()
626
 
                l.attempt_lock()
627
 
        test = TestDanglingLock('test_function')
628
 
        result = test.run()
629
 
        total_failures = result.errors + result.failures
630
 
        if self._lock_check_thorough:
631
 
            self.assertEqual(1, len(total_failures))
632
 
        else:
633
 
            # When _lock_check_thorough is disabled, then we don't trigger a
634
 
            # failure
635
 
            self.assertEqual(0, len(total_failures))
636
 
 
637
 
 
638
 
class TestTestCaseWithTransport(tests.TestCaseWithTransport):
 
271
        adapter = BzrDirTestProviderAdapter(server1, server2, formats)
 
272
        suite = adapter.adapt(input_test)
 
273
        tests = list(iter(suite))
 
274
        self.assertEqual(2, len(tests))
 
275
        self.assertEqual(tests[0].bzrdir_format, formats[0])
 
276
        self.assertEqual(tests[0].transport_server, server1)
 
277
        self.assertEqual(tests[0].transport_readonly_server, server2)
 
278
        self.assertEqual(tests[1].bzrdir_format, formats[1])
 
279
        self.assertEqual(tests[1].transport_server, server1)
 
280
        self.assertEqual(tests[1].transport_readonly_server, server2)
 
281
 
 
282
 
 
283
class TestRepositoryProviderAdapter(TestCase):
 
284
    """A group of tests that test the repository implementation test adapter."""
 
285
 
 
286
    def test_adapted_tests(self):
 
287
        # check that constructor parameters are passed through to the adapted
 
288
        # test.
 
289
        from bzrlib.repository import RepositoryTestProviderAdapter
 
290
        input_test = TestRepositoryProviderAdapter(
 
291
            "test_adapted_tests")
 
292
        server1 = "a"
 
293
        server2 = "b"
 
294
        formats = [("c", "C"), ("d", "D")]
 
295
        adapter = RepositoryTestProviderAdapter(server1, server2, formats)
 
296
        suite = adapter.adapt(input_test)
 
297
        tests = list(iter(suite))
 
298
        self.assertEqual(2, len(tests))
 
299
        self.assertEqual(tests[0].bzrdir_format, formats[0][1])
 
300
        self.assertEqual(tests[0].repository_format, formats[0][0])
 
301
        self.assertEqual(tests[0].transport_server, server1)
 
302
        self.assertEqual(tests[0].transport_readonly_server, server2)
 
303
        self.assertEqual(tests[1].bzrdir_format, formats[1][1])
 
304
        self.assertEqual(tests[1].repository_format, formats[1][0])
 
305
        self.assertEqual(tests[1].transport_server, server1)
 
306
        self.assertEqual(tests[1].transport_readonly_server, server2)
 
307
 
 
308
 
 
309
class TestInterRepositoryProviderAdapter(TestCase):
 
310
    """A group of tests that test the InterRepository test adapter."""
 
311
 
 
312
    def test_adapted_tests(self):
 
313
        # check that constructor parameters are passed through to the adapted
 
314
        # test.
 
315
        from bzrlib.repository import InterRepositoryTestProviderAdapter
 
316
        input_test = TestInterRepositoryProviderAdapter(
 
317
            "test_adapted_tests")
 
318
        server1 = "a"
 
319
        server2 = "b"
 
320
        formats = [(str, "C1", "C2"), (int, "D1", "D2")]
 
321
        adapter = InterRepositoryTestProviderAdapter(server1, server2, formats)
 
322
        suite = adapter.adapt(input_test)
 
323
        tests = list(iter(suite))
 
324
        self.assertEqual(2, len(tests))
 
325
        self.assertEqual(tests[0].interrepo_class, formats[0][0])
 
326
        self.assertEqual(tests[0].repository_format, formats[0][1])
 
327
        self.assertEqual(tests[0].repository_format_to, formats[0][2])
 
328
        self.assertEqual(tests[0].transport_server, server1)
 
329
        self.assertEqual(tests[0].transport_readonly_server, server2)
 
330
        self.assertEqual(tests[1].interrepo_class, formats[1][0])
 
331
        self.assertEqual(tests[1].repository_format, formats[1][1])
 
332
        self.assertEqual(tests[1].repository_format_to, formats[1][2])
 
333
        self.assertEqual(tests[1].transport_server, server1)
 
334
        self.assertEqual(tests[1].transport_readonly_server, server2)
 
335
 
 
336
 
 
337
class TestInterVersionedFileProviderAdapter(TestCase):
 
338
    """A group of tests that test the InterVersionedFile test adapter."""
 
339
 
 
340
    def test_adapted_tests(self):
 
341
        # check that constructor parameters are passed through to the adapted
 
342
        # test.
 
343
        from bzrlib.versionedfile import InterVersionedFileTestProviderAdapter
 
344
        input_test = TestInterRepositoryProviderAdapter(
 
345
            "test_adapted_tests")
 
346
        server1 = "a"
 
347
        server2 = "b"
 
348
        formats = [(str, "C1", "C2"), (int, "D1", "D2")]
 
349
        adapter = InterVersionedFileTestProviderAdapter(server1, server2, formats)
 
350
        suite = adapter.adapt(input_test)
 
351
        tests = list(iter(suite))
 
352
        self.assertEqual(2, len(tests))
 
353
        self.assertEqual(tests[0].interversionedfile_class, formats[0][0])
 
354
        self.assertEqual(tests[0].versionedfile_factory, formats[0][1])
 
355
        self.assertEqual(tests[0].versionedfile_factory_to, formats[0][2])
 
356
        self.assertEqual(tests[0].transport_server, server1)
 
357
        self.assertEqual(tests[0].transport_readonly_server, server2)
 
358
        self.assertEqual(tests[1].interversionedfile_class, formats[1][0])
 
359
        self.assertEqual(tests[1].versionedfile_factory, formats[1][1])
 
360
        self.assertEqual(tests[1].versionedfile_factory_to, formats[1][2])
 
361
        self.assertEqual(tests[1].transport_server, server1)
 
362
        self.assertEqual(tests[1].transport_readonly_server, server2)
 
363
 
 
364
 
 
365
class TestRevisionStoreProviderAdapter(TestCase):
 
366
    """A group of tests that test the RevisionStore test adapter."""
 
367
 
 
368
    def test_adapted_tests(self):
 
369
        # check that constructor parameters are passed through to the adapted
 
370
        # test.
 
371
        from bzrlib.store.revision import RevisionStoreTestProviderAdapter
 
372
        input_test = TestRevisionStoreProviderAdapter(
 
373
            "test_adapted_tests")
 
374
        # revision stores need a store factory - i.e. RevisionKnit
 
375
        #, a readonly and rw transport 
 
376
        # transport servers:
 
377
        server1 = "a"
 
378
        server2 = "b"
 
379
        store_factories = ["c", "d"]
 
380
        adapter = RevisionStoreTestProviderAdapter(server1, server2, store_factories)
 
381
        suite = adapter.adapt(input_test)
 
382
        tests = list(iter(suite))
 
383
        self.assertEqual(2, len(tests))
 
384
        self.assertEqual(tests[0].store_factory, store_factories[0][0])
 
385
        self.assertEqual(tests[0].transport_server, server1)
 
386
        self.assertEqual(tests[0].transport_readonly_server, server2)
 
387
        self.assertEqual(tests[1].store_factory, store_factories[1][0])
 
388
        self.assertEqual(tests[1].transport_server, server1)
 
389
        self.assertEqual(tests[1].transport_readonly_server, server2)
 
390
 
 
391
 
 
392
class TestWorkingTreeProviderAdapter(TestCase):
 
393
    """A group of tests that test the workingtree implementation test adapter."""
 
394
 
 
395
    def test_adapted_tests(self):
 
396
        # check that constructor parameters are passed through to the adapted
 
397
        # test.
 
398
        from bzrlib.workingtree import WorkingTreeTestProviderAdapter
 
399
        input_test = TestWorkingTreeProviderAdapter(
 
400
            "test_adapted_tests")
 
401
        server1 = "a"
 
402
        server2 = "b"
 
403
        formats = [("c", "C"), ("d", "D")]
 
404
        adapter = WorkingTreeTestProviderAdapter(server1, server2, formats)
 
405
        suite = adapter.adapt(input_test)
 
406
        tests = list(iter(suite))
 
407
        self.assertEqual(2, len(tests))
 
408
        self.assertEqual(tests[0].workingtree_format, formats[0][0])
 
409
        self.assertEqual(tests[0].bzrdir_format, formats[0][1])
 
410
        self.assertEqual(tests[0].transport_server, server1)
 
411
        self.assertEqual(tests[0].transport_readonly_server, server2)
 
412
        self.assertEqual(tests[1].workingtree_format, formats[1][0])
 
413
        self.assertEqual(tests[1].bzrdir_format, formats[1][1])
 
414
        self.assertEqual(tests[1].transport_server, server1)
 
415
        self.assertEqual(tests[1].transport_readonly_server, server2)
 
416
 
 
417
 
 
418
class TestTestCaseWithTransport(TestCaseWithTransport):
639
419
    """Tests for the convenience functions TestCaseWithTransport introduces."""
640
420
 
641
421
    def test_get_readonly_url_none(self):
 
422
        from bzrlib.transport import get_transport
 
423
        from bzrlib.transport.memory import MemoryServer
642
424
        from bzrlib.transport.readonly import ReadonlyTransportDecorator
643
 
        self.vfs_transport_factory = memory.MemoryServer
 
425
        self.transport_server = MemoryServer
644
426
        self.transport_readonly_server = None
645
427
        # calling get_readonly_transport() constructs a decorator on the url
646
428
        # for the server
647
429
        url = self.get_readonly_url()
648
430
        url2 = self.get_readonly_url('foo/bar')
649
 
        t = transport.get_transport_from_url(url)
650
 
        t2 = transport.get_transport_from_url(url2)
651
 
        self.assertIsInstance(t, ReadonlyTransportDecorator)
652
 
        self.assertIsInstance(t2, ReadonlyTransportDecorator)
 
431
        t = get_transport(url)
 
432
        t2 = get_transport(url2)
 
433
        self.failUnless(isinstance(t, ReadonlyTransportDecorator))
 
434
        self.failUnless(isinstance(t2, ReadonlyTransportDecorator))
653
435
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
654
436
 
655
437
    def test_get_readonly_url_http(self):
656
 
        from bzrlib.tests.http_server import HttpServer
657
 
        from bzrlib.transport.http import HttpTransportBase
658
 
        self.transport_server = test_server.LocalURLServer
 
438
        from bzrlib.transport import get_transport
 
439
        from bzrlib.transport.local import LocalRelpathServer
 
440
        from bzrlib.transport.http import HttpServer, HttpTransport
 
441
        self.transport_server = LocalRelpathServer
659
442
        self.transport_readonly_server = HttpServer
660
443
        # calling get_readonly_transport() gives us a HTTP server instance.
661
444
        url = self.get_readonly_url()
662
445
        url2 = self.get_readonly_url('foo/bar')
663
 
        # the transport returned may be any HttpTransportBase subclass
664
 
        t = transport.get_transport_from_url(url)
665
 
        t2 = transport.get_transport_from_url(url2)
666
 
        self.assertIsInstance(t, HttpTransportBase)
667
 
        self.assertIsInstance(t2, HttpTransportBase)
 
446
        t = get_transport(url)
 
447
        t2 = get_transport(url2)
 
448
        self.failUnless(isinstance(t, HttpTransport))
 
449
        self.failUnless(isinstance(t2, HttpTransport))
668
450
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
669
451
 
670
452
    def test_is_directory(self):
675
457
        self.assertRaises(AssertionError, self.assertIsDirectory, 'a_file', t)
676
458
        self.assertRaises(AssertionError, self.assertIsDirectory, 'not_here', t)
677
459
 
678
 
    def test_make_branch_builder(self):
679
 
        builder = self.make_branch_builder('dir')
680
 
        rev_id = builder.build_commit()
681
 
        self.assertPathExists('dir')
682
 
        a_dir = controldir.ControlDir.open('dir')
683
 
        self.assertRaises(errors.NoWorkingTree, a_dir.open_workingtree)
684
 
        a_branch = a_dir.open_branch()
685
 
        builder_branch = builder.get_branch()
686
 
        self.assertEqual(a_branch.base, builder_branch.base)
687
 
        self.assertEqual((1, rev_id), builder_branch.last_revision_info())
688
 
        self.assertEqual((1, rev_id), a_branch.last_revision_info())
689
 
 
690
 
 
691
 
class TestTestCaseTransports(tests.TestCaseWithTransport):
692
 
 
693
 
    def setUp(self):
694
 
        super(TestTestCaseTransports, self).setUp()
695
 
        self.vfs_transport_factory = memory.MemoryServer
696
 
 
697
 
    def test_make_bzrdir_preserves_transport(self):
698
 
        t = self.get_transport()
699
 
        result_bzrdir = self.make_bzrdir('subdir')
700
 
        self.assertIsInstance(result_bzrdir.transport,
701
 
                              memory.MemoryTransport)
702
 
        # should not be on disk, should only be in memory
703
 
        self.assertPathDoesNotExist('subdir')
704
 
 
705
 
 
706
 
class TestChrootedTest(tests.ChrootedTestCase):
 
460
class TestChrootedTest(ChrootedTestCase):
707
461
 
708
462
    def test_root_is_root(self):
709
 
        t = transport.get_transport_from_url(self.get_readonly_url())
 
463
        from bzrlib.transport import get_transport
 
464
        t = get_transport(self.get_readonly_url())
710
465
        url = t.base
711
466
        self.assertEqual(url, t.clone('..').base)
712
 
 
713
 
 
714
 
class TestProfileResult(tests.TestCase):
715
 
 
716
 
    def test_profiles_tests(self):
717
 
        self.requireFeature(features.lsprof_feature)
718
 
        terminal = testtools.testresult.doubles.ExtendedTestResult()
719
 
        result = tests.ProfileResult(terminal)
720
 
        class Sample(tests.TestCase):
721
 
            def a(self):
722
 
                self.sample_function()
723
 
            def sample_function(self):
724
 
                pass
725
 
        test = Sample("a")
726
 
        test.run(result)
727
 
        case = terminal._events[0][1]
728
 
        self.assertLength(1, case._benchcalls)
729
 
        # We must be able to unpack it as the test reporting code wants
730
 
        (_, _, _), stats = case._benchcalls[0]
731
 
        self.assertTrue(callable(stats.pprint))
732
 
 
733
 
 
734
 
class TestTestResult(tests.TestCase):
735
 
 
736
 
    def check_timing(self, test_case, expected_re):
737
 
        result = bzrlib.tests.TextTestResult(self._log_file,
738
 
                descriptions=0,
739
 
                verbosity=1,
740
 
                )
741
 
        capture = testtools.testresult.doubles.ExtendedTestResult()
742
 
        test_case.run(MultiTestResult(result, capture))
743
 
        run_case = capture._events[0][1]
744
 
        timed_string = result._testTimeString(run_case)
745
 
        self.assertContainsRe(timed_string, expected_re)
746
 
 
747
 
    def test_test_reporting(self):
748
 
        class ShortDelayTestCase(tests.TestCase):
749
 
            def test_short_delay(self):
750
 
                time.sleep(0.003)
751
 
            def test_short_benchmark(self):
752
 
                self.time(time.sleep, 0.003)
753
 
        self.check_timing(ShortDelayTestCase('test_short_delay'),
754
 
                          r"^ +[0-9]+ms$")
755
 
        # if a benchmark time is given, we now show just that time followed by
756
 
        # a star
757
 
        self.check_timing(ShortDelayTestCase('test_short_benchmark'),
758
 
                          r"^ +[0-9]+ms\*$")
759
 
 
760
 
    def test_unittest_reporting_unittest_class(self):
761
 
        # getting the time from a non-bzrlib test works ok
762
 
        class ShortDelayTestCase(unittest.TestCase):
763
 
            def test_short_delay(self):
764
 
                time.sleep(0.003)
765
 
        self.check_timing(ShortDelayTestCase('test_short_delay'),
766
 
                          r"^ +[0-9]+ms$")
767
 
 
768
 
    def _time_hello_world_encoding(self):
769
 
        """Profile two sleep calls
770
 
 
771
 
        This is used to exercise the test framework.
772
 
        """
773
 
        self.time(unicode, 'hello', errors='replace')
774
 
        self.time(unicode, 'world', errors='replace')
775
 
 
776
 
    def test_lsprofiling(self):
777
 
        """Verbose test result prints lsprof statistics from test cases."""
778
 
        self.requireFeature(features.lsprof_feature)
779
 
        result_stream = StringIO()
780
 
        result = bzrlib.tests.VerboseTestResult(
781
 
            result_stream,
782
 
            descriptions=0,
783
 
            verbosity=2,
784
 
            )
785
 
        # we want profile a call of some sort and check it is output by
786
 
        # addSuccess. We dont care about addError or addFailure as they
787
 
        # are not that interesting for performance tuning.
788
 
        # make a new test instance that when run will generate a profile
789
 
        example_test_case = TestTestResult("_time_hello_world_encoding")
790
 
        example_test_case._gather_lsprof_in_benchmarks = True
791
 
        # execute the test, which should succeed and record profiles
792
 
        example_test_case.run(result)
793
 
        # lsprofile_something()
794
 
        # if this worked we want
795
 
        # LSProf output for <built in function unicode> (['hello'], {'errors': 'replace'})
796
 
        #    CallCount    Recursive    Total(ms)   Inline(ms) module:lineno(function)
797
 
        # (the lsprof header)
798
 
        # ... an arbitrary number of lines
799
 
        # and the function call which is time.sleep.
800
 
        #           1        0            ???         ???       ???(sleep)
801
 
        # and then repeated but with 'world', rather than 'hello'.
802
 
        # this should appear in the output stream of our test result.
803
 
        output = result_stream.getvalue()
804
 
        self.assertContainsRe(output,
805
 
            r"LSProf output for <type 'unicode'>\(\('hello',\), {'errors': 'replace'}\)")
806
 
        self.assertContainsRe(output,
807
 
            r" *CallCount *Recursive *Total\(ms\) *Inline\(ms\) *module:lineno\(function\)\n")
808
 
        self.assertContainsRe(output,
809
 
            r"( +1 +0 +0\.\d+ +0\.\d+ +<method 'disable' of '_lsprof\.Profiler' objects>\n)?")
810
 
        self.assertContainsRe(output,
811
 
            r"LSProf output for <type 'unicode'>\(\('world',\), {'errors': 'replace'}\)\n")
812
 
 
813
 
    def test_uses_time_from_testtools(self):
814
 
        """Test case timings in verbose results should use testtools times"""
815
 
        import datetime
816
 
        class TimeAddedVerboseTestResult(tests.VerboseTestResult):
817
 
            def startTest(self, test):
818
 
                self.time(datetime.datetime.utcfromtimestamp(1.145))
819
 
                super(TimeAddedVerboseTestResult, self).startTest(test)
820
 
            def addSuccess(self, test):
821
 
                self.time(datetime.datetime.utcfromtimestamp(51.147))
822
 
                super(TimeAddedVerboseTestResult, self).addSuccess(test)
823
 
            def report_tests_starting(self): pass
824
 
        sio = StringIO()
825
 
        self.get_passing_test().run(TimeAddedVerboseTestResult(sio, 0, 2))
826
 
        self.assertEndsWith(sio.getvalue(), "OK    50002ms\n")
827
 
 
828
 
    def test_known_failure(self):
829
 
        """Using knownFailure should trigger several result actions."""
830
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
831
 
            def stopTestRun(self): pass
832
 
            def report_tests_starting(self): pass
833
 
            def report_known_failure(self, test, err=None, details=None):
834
 
                self._call = test, 'known failure'
835
 
        result = InstrumentedTestResult(None, None, None, None)
836
 
        class Test(tests.TestCase):
837
 
            def test_function(self):
838
 
                self.knownFailure('failed!')
839
 
        test = Test("test_function")
840
 
        test.run(result)
841
 
        # it should invoke 'report_known_failure'.
842
 
        self.assertEqual(2, len(result._call))
843
 
        self.assertEqual(test.id(), result._call[0].id())
844
 
        self.assertEqual('known failure', result._call[1])
845
 
        # we dont introspec the traceback, if the rest is ok, it would be
846
 
        # exceptional for it not to be.
847
 
        # it should update the known_failure_count on the object.
848
 
        self.assertEqual(1, result.known_failure_count)
849
 
        # the result should be successful.
850
 
        self.assertTrue(result.wasSuccessful())
851
 
 
852
 
    def test_verbose_report_known_failure(self):
853
 
        # verbose test output formatting
854
 
        result_stream = StringIO()
855
 
        result = bzrlib.tests.VerboseTestResult(
856
 
            result_stream,
857
 
            descriptions=0,
858
 
            verbosity=2,
859
 
            )
860
 
        _get_test("test_xfail").run(result)
861
 
        self.assertContainsRe(result_stream.getvalue(),
862
 
            "\n\\S+\\.test_xfail\\s+XFAIL\\s+\\d+ms\n"
863
 
            "\\s*(?:Text attachment: )?reason"
864
 
            "(?:\n-+\n|: {{{)"
865
 
            "this_fails"
866
 
            "(?:\n-+\n|}}}\n)")
867
 
 
868
 
    def get_passing_test(self):
869
 
        """Return a test object that can't be run usefully."""
870
 
        def passing_test():
871
 
            pass
872
 
        return unittest.FunctionTestCase(passing_test)
873
 
 
874
 
    def test_add_not_supported(self):
875
 
        """Test the behaviour of invoking addNotSupported."""
876
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
877
 
            def stopTestRun(self): pass
878
 
            def report_tests_starting(self): pass
879
 
            def report_unsupported(self, test, feature):
880
 
                self._call = test, feature
881
 
        result = InstrumentedTestResult(None, None, None, None)
882
 
        test = SampleTestCase('_test_pass')
883
 
        feature = features.Feature()
884
 
        result.startTest(test)
885
 
        result.addNotSupported(test, feature)
886
 
        # it should invoke 'report_unsupported'.
887
 
        self.assertEqual(2, len(result._call))
888
 
        self.assertEqual(test, result._call[0])
889
 
        self.assertEqual(feature, result._call[1])
890
 
        # the result should be successful.
891
 
        self.assertTrue(result.wasSuccessful())
892
 
        # it should record the test against a count of tests not run due to
893
 
        # this feature.
894
 
        self.assertEqual(1, result.unsupported['Feature'])
895
 
        # and invoking it again should increment that counter
896
 
        result.addNotSupported(test, feature)
897
 
        self.assertEqual(2, result.unsupported['Feature'])
898
 
 
899
 
    def test_verbose_report_unsupported(self):
900
 
        # verbose test output formatting
901
 
        result_stream = StringIO()
902
 
        result = bzrlib.tests.VerboseTestResult(
903
 
            result_stream,
904
 
            descriptions=0,
905
 
            verbosity=2,
906
 
            )
907
 
        test = self.get_passing_test()
908
 
        feature = features.Feature()
909
 
        result.startTest(test)
910
 
        prefix = len(result_stream.getvalue())
911
 
        result.report_unsupported(test, feature)
912
 
        output = result_stream.getvalue()[prefix:]
913
 
        lines = output.splitlines()
914
 
        # We don't check for the final '0ms' since it may fail on slow hosts
915
 
        self.assertStartsWith(lines[0], 'NODEP')
916
 
        self.assertEqual(lines[1],
917
 
                         "    The feature 'Feature' is not available.")
918
 
 
919
 
    def test_unavailable_exception(self):
920
 
        """An UnavailableFeature being raised should invoke addNotSupported."""
921
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
922
 
            def stopTestRun(self): pass
923
 
            def report_tests_starting(self): pass
924
 
            def addNotSupported(self, test, feature):
925
 
                self._call = test, feature
926
 
        result = InstrumentedTestResult(None, None, None, None)
927
 
        feature = features.Feature()
928
 
        class Test(tests.TestCase):
929
 
            def test_function(self):
930
 
                raise tests.UnavailableFeature(feature)
931
 
        test = Test("test_function")
932
 
        test.run(result)
933
 
        # it should invoke 'addNotSupported'.
934
 
        self.assertEqual(2, len(result._call))
935
 
        self.assertEqual(test.id(), result._call[0].id())
936
 
        self.assertEqual(feature, result._call[1])
937
 
        # and not count as an error
938
 
        self.assertEqual(0, result.error_count)
939
 
 
940
 
    def test_strict_with_unsupported_feature(self):
941
 
        result = bzrlib.tests.TextTestResult(self._log_file, descriptions=0,
942
 
                                             verbosity=1)
943
 
        test = self.get_passing_test()
944
 
        feature = "Unsupported Feature"
945
 
        result.addNotSupported(test, feature)
946
 
        self.assertFalse(result.wasStrictlySuccessful())
947
 
        self.assertEqual(None, result._extractBenchmarkTime(test))
948
 
 
949
 
    def test_strict_with_known_failure(self):
950
 
        result = bzrlib.tests.TextTestResult(self._log_file, descriptions=0,
951
 
                                             verbosity=1)
952
 
        test = _get_test("test_xfail")
953
 
        test.run(result)
954
 
        self.assertFalse(result.wasStrictlySuccessful())
955
 
        self.assertEqual(None, result._extractBenchmarkTime(test))
956
 
 
957
 
    def test_strict_with_success(self):
958
 
        result = bzrlib.tests.TextTestResult(self._log_file, descriptions=0,
959
 
                                             verbosity=1)
960
 
        test = self.get_passing_test()
961
 
        result.addSuccess(test)
962
 
        self.assertTrue(result.wasStrictlySuccessful())
963
 
        self.assertEqual(None, result._extractBenchmarkTime(test))
964
 
 
965
 
    def test_startTests(self):
966
 
        """Starting the first test should trigger startTests."""
967
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
968
 
            calls = 0
969
 
            def startTests(self): self.calls += 1
970
 
        result = InstrumentedTestResult(None, None, None, None)
971
 
        def test_function():
972
 
            pass
973
 
        test = unittest.FunctionTestCase(test_function)
974
 
        test.run(result)
975
 
        self.assertEquals(1, result.calls)
976
 
 
977
 
    def test_startTests_only_once(self):
978
 
        """With multiple tests startTests should still only be called once"""
979
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
980
 
            calls = 0
981
 
            def startTests(self): self.calls += 1
982
 
        result = InstrumentedTestResult(None, None, None, None)
983
 
        suite = unittest.TestSuite([
984
 
            unittest.FunctionTestCase(lambda: None),
985
 
            unittest.FunctionTestCase(lambda: None)])
986
 
        suite.run(result)
987
 
        self.assertEquals(1, result.calls)
988
 
        self.assertEquals(2, result.count)
989
 
 
990
 
 
991
 
class TestRunner(tests.TestCase):
992
 
 
993
 
    def dummy_test(self):
994
 
        pass
995
 
 
996
 
    def run_test_runner(self, testrunner, test):
997
 
        """Run suite in testrunner, saving global state and restoring it.
998
 
 
999
 
        This current saves and restores:
1000
 
        TestCaseInTempDir.TEST_ROOT
1001
 
 
1002
 
        There should be no tests in this file that use
1003
 
        bzrlib.tests.TextTestRunner without using this convenience method,
1004
 
        because of our use of global state.
1005
 
        """
1006
 
        old_root = tests.TestCaseInTempDir.TEST_ROOT
1007
 
        try:
1008
 
            tests.TestCaseInTempDir.TEST_ROOT = None
1009
 
            return testrunner.run(test)
1010
 
        finally:
1011
 
            tests.TestCaseInTempDir.TEST_ROOT = old_root
1012
 
 
1013
 
    def test_known_failure_failed_run(self):
1014
 
        # run a test that generates a known failure which should be printed in
1015
 
        # the final output when real failures occur.
1016
 
        class Test(tests.TestCase):
1017
 
            def known_failure_test(self):
1018
 
                self.expectFailure('failed', self.assertTrue, False)
1019
 
        test = unittest.TestSuite()
1020
 
        test.addTest(Test("known_failure_test"))
1021
 
        def failing_test():
1022
 
            raise AssertionError('foo')
1023
 
        test.addTest(unittest.FunctionTestCase(failing_test))
1024
 
        stream = StringIO()
1025
 
        runner = tests.TextTestRunner(stream=stream)
1026
 
        result = self.run_test_runner(runner, test)
1027
 
        lines = stream.getvalue().splitlines()
1028
 
        self.assertContainsRe(stream.getvalue(),
1029
 
            '(?sm)^bzr selftest.*$'
1030
 
            '.*'
1031
 
            '^======================================================================\n'
1032
 
            '^FAIL: failing_test\n'
1033
 
            '^----------------------------------------------------------------------\n'
1034
 
            'Traceback \\(most recent call last\\):\n'
1035
 
            '  .*' # File .*, line .*, in failing_test' - but maybe not from .pyc
1036
 
            '    raise AssertionError\\(\'foo\'\\)\n'
1037
 
            '.*'
1038
 
            '^----------------------------------------------------------------------\n'
1039
 
            '.*'
1040
 
            'FAILED \\(failures=1, known_failure_count=1\\)'
1041
 
            )
1042
 
 
1043
 
    def test_known_failure_ok_run(self):
1044
 
        # run a test that generates a known failure which should be printed in
1045
 
        # the final output.
1046
 
        class Test(tests.TestCase):
1047
 
            def known_failure_test(self):
1048
 
                self.knownFailure("Never works...")
1049
 
        test = Test("known_failure_test")
1050
 
        stream = StringIO()
1051
 
        runner = tests.TextTestRunner(stream=stream)
1052
 
        result = self.run_test_runner(runner, test)
1053
 
        self.assertContainsRe(stream.getvalue(),
1054
 
            '\n'
1055
 
            '-*\n'
1056
 
            'Ran 1 test in .*\n'
1057
 
            '\n'
1058
 
            'OK \\(known_failures=1\\)\n')
1059
 
 
1060
 
    def test_unexpected_success_bad(self):
1061
 
        class Test(tests.TestCase):
1062
 
            def test_truth(self):
1063
 
                self.expectFailure("No absolute truth", self.assertTrue, True)
1064
 
        runner = tests.TextTestRunner(stream=StringIO())
1065
 
        result = self.run_test_runner(runner, Test("test_truth"))
1066
 
        self.assertContainsRe(runner.stream.getvalue(),
1067
 
            "=+\n"
1068
 
            "FAIL: \\S+\.test_truth\n"
1069
 
            "-+\n"
1070
 
            "(?:.*\n)*"
1071
 
            "\\s*(?:Text attachment: )?reason"
1072
 
            "(?:\n-+\n|: {{{)"
1073
 
            "No absolute truth"
1074
 
            "(?:\n-+\n|}}}\n)"
1075
 
            "(?:.*\n)*"
1076
 
            "-+\n"
1077
 
            "Ran 1 test in .*\n"
1078
 
            "\n"
1079
 
            "FAILED \\(failures=1\\)\n\\Z")
1080
 
 
1081
 
    def test_result_decorator(self):
1082
 
        # decorate results
1083
 
        calls = []
1084
 
        class LoggingDecorator(ExtendedToOriginalDecorator):
1085
 
            def startTest(self, test):
1086
 
                ExtendedToOriginalDecorator.startTest(self, test)
1087
 
                calls.append('start')
1088
 
        test = unittest.FunctionTestCase(lambda:None)
1089
 
        stream = StringIO()
1090
 
        runner = tests.TextTestRunner(stream=stream,
1091
 
            result_decorators=[LoggingDecorator])
1092
 
        result = self.run_test_runner(runner, test)
1093
 
        self.assertLength(1, calls)
1094
 
 
1095
 
    def test_skipped_test(self):
1096
 
        # run a test that is skipped, and check the suite as a whole still
1097
 
        # succeeds.
1098
 
        # skipping_test must be hidden in here so it's not run as a real test
1099
 
        class SkippingTest(tests.TestCase):
1100
 
            def skipping_test(self):
1101
 
                raise tests.TestSkipped('test intentionally skipped')
1102
 
        runner = tests.TextTestRunner(stream=self._log_file)
1103
 
        test = SkippingTest("skipping_test")
1104
 
        result = self.run_test_runner(runner, test)
1105
 
        self.assertTrue(result.wasSuccessful())
1106
 
 
1107
 
    def test_skipped_from_setup(self):
1108
 
        calls = []
1109
 
        class SkippedSetupTest(tests.TestCase):
1110
 
 
1111
 
            def setUp(self):
1112
 
                calls.append('setUp')
1113
 
                self.addCleanup(self.cleanup)
1114
 
                raise tests.TestSkipped('skipped setup')
1115
 
 
1116
 
            def test_skip(self):
1117
 
                self.fail('test reached')
1118
 
 
1119
 
            def cleanup(self):
1120
 
                calls.append('cleanup')
1121
 
 
1122
 
        runner = tests.TextTestRunner(stream=self._log_file)
1123
 
        test = SkippedSetupTest('test_skip')
1124
 
        result = self.run_test_runner(runner, test)
1125
 
        self.assertTrue(result.wasSuccessful())
1126
 
        # Check if cleanup was called the right number of times.
1127
 
        self.assertEqual(['setUp', 'cleanup'], calls)
1128
 
 
1129
 
    def test_skipped_from_test(self):
1130
 
        calls = []
1131
 
        class SkippedTest(tests.TestCase):
1132
 
 
1133
 
            def setUp(self):
1134
 
                tests.TestCase.setUp(self)
1135
 
                calls.append('setUp')
1136
 
                self.addCleanup(self.cleanup)
1137
 
 
1138
 
            def test_skip(self):
1139
 
                raise tests.TestSkipped('skipped test')
1140
 
 
1141
 
            def cleanup(self):
1142
 
                calls.append('cleanup')
1143
 
 
1144
 
        runner = tests.TextTestRunner(stream=self._log_file)
1145
 
        test = SkippedTest('test_skip')
1146
 
        result = self.run_test_runner(runner, test)
1147
 
        self.assertTrue(result.wasSuccessful())
1148
 
        # Check if cleanup was called the right number of times.
1149
 
        self.assertEqual(['setUp', 'cleanup'], calls)
1150
 
 
1151
 
    def test_not_applicable(self):
1152
 
        # run a test that is skipped because it's not applicable
1153
 
        class Test(tests.TestCase):
1154
 
            def not_applicable_test(self):
1155
 
                raise tests.TestNotApplicable('this test never runs')
1156
 
        out = StringIO()
1157
 
        runner = tests.TextTestRunner(stream=out, verbosity=2)
1158
 
        test = Test("not_applicable_test")
1159
 
        result = self.run_test_runner(runner, test)
1160
 
        self._log_file.write(out.getvalue())
1161
 
        self.assertTrue(result.wasSuccessful())
1162
 
        self.assertTrue(result.wasStrictlySuccessful())
1163
 
        self.assertContainsRe(out.getvalue(),
1164
 
                r'(?m)not_applicable_test   * N/A')
1165
 
        self.assertContainsRe(out.getvalue(),
1166
 
                r'(?m)^    this test never runs')
1167
 
 
1168
 
    def test_unsupported_features_listed(self):
1169
 
        """When unsupported features are encountered they are detailed."""
1170
 
        class Feature1(features.Feature):
1171
 
            def _probe(self): return False
1172
 
        class Feature2(features.Feature):
1173
 
            def _probe(self): return False
1174
 
        # create sample tests
1175
 
        test1 = SampleTestCase('_test_pass')
1176
 
        test1._test_needs_features = [Feature1()]
1177
 
        test2 = SampleTestCase('_test_pass')
1178
 
        test2._test_needs_features = [Feature2()]
1179
 
        test = unittest.TestSuite()
1180
 
        test.addTest(test1)
1181
 
        test.addTest(test2)
1182
 
        stream = StringIO()
1183
 
        runner = tests.TextTestRunner(stream=stream)
1184
 
        result = self.run_test_runner(runner, test)
1185
 
        lines = stream.getvalue().splitlines()
1186
 
        self.assertEqual([
1187
 
            'OK',
1188
 
            "Missing feature 'Feature1' skipped 1 tests.",
1189
 
            "Missing feature 'Feature2' skipped 1 tests.",
1190
 
            ],
1191
 
            lines[-3:])
1192
 
 
1193
 
    def test_verbose_test_count(self):
1194
 
        """A verbose test run reports the right test count at the start"""
1195
 
        suite = TestUtil.TestSuite([
1196
 
            unittest.FunctionTestCase(lambda:None),
1197
 
            unittest.FunctionTestCase(lambda:None)])
1198
 
        self.assertEqual(suite.countTestCases(), 2)
1199
 
        stream = StringIO()
1200
 
        runner = tests.TextTestRunner(stream=stream, verbosity=2)
1201
 
        # Need to use the CountingDecorator as that's what sets num_tests
1202
 
        result = self.run_test_runner(runner, tests.CountingDecorator(suite))
1203
 
        self.assertStartsWith(stream.getvalue(), "running 2 tests")
1204
 
 
1205
 
    def test_startTestRun(self):
1206
 
        """run should call result.startTestRun()"""
1207
 
        calls = []
1208
 
        class LoggingDecorator(ExtendedToOriginalDecorator):
1209
 
            def startTestRun(self):
1210
 
                ExtendedToOriginalDecorator.startTestRun(self)
1211
 
                calls.append('startTestRun')
1212
 
        test = unittest.FunctionTestCase(lambda:None)
1213
 
        stream = StringIO()
1214
 
        runner = tests.TextTestRunner(stream=stream,
1215
 
            result_decorators=[LoggingDecorator])
1216
 
        result = self.run_test_runner(runner, test)
1217
 
        self.assertLength(1, calls)
1218
 
 
1219
 
    def test_stopTestRun(self):
1220
 
        """run should call result.stopTestRun()"""
1221
 
        calls = []
1222
 
        class LoggingDecorator(ExtendedToOriginalDecorator):
1223
 
            def stopTestRun(self):
1224
 
                ExtendedToOriginalDecorator.stopTestRun(self)
1225
 
                calls.append('stopTestRun')
1226
 
        test = unittest.FunctionTestCase(lambda:None)
1227
 
        stream = StringIO()
1228
 
        runner = tests.TextTestRunner(stream=stream,
1229
 
            result_decorators=[LoggingDecorator])
1230
 
        result = self.run_test_runner(runner, test)
1231
 
        self.assertLength(1, calls)
1232
 
 
1233
 
    def test_unicode_test_output_on_ascii_stream(self):
1234
 
        """Showing results should always succeed even on an ascii console"""
1235
 
        class FailureWithUnicode(tests.TestCase):
1236
 
            def test_log_unicode(self):
1237
 
                self.log(u"\u2606")
1238
 
                self.fail("Now print that log!")
1239
 
        out = StringIO()
1240
 
        self.overrideAttr(osutils, "get_terminal_encoding",
1241
 
            lambda trace=False: "ascii")
1242
 
        result = self.run_test_runner(tests.TextTestRunner(stream=out),
1243
 
            FailureWithUnicode("test_log_unicode"))
1244
 
        self.assertContainsRe(out.getvalue(),
1245
 
            "(?:Text attachment: )?log"
1246
 
            "(?:\n-+\n|: {{{)"
1247
 
            "\d+\.\d+  \\\\u2606"
1248
 
            "(?:\n-+\n|}}}\n)")
1249
 
 
1250
 
 
1251
 
class SampleTestCase(tests.TestCase):
1252
 
 
1253
 
    def _test_pass(self):
1254
 
        pass
1255
 
 
1256
 
class _TestException(Exception):
1257
 
    pass
1258
 
 
1259
 
 
1260
 
class TestTestCase(tests.TestCase):
1261
 
    """Tests that test the core bzrlib TestCase."""
1262
 
 
1263
 
    def test_assertLength_matches_empty(self):
1264
 
        a_list = []
1265
 
        self.assertLength(0, a_list)
1266
 
 
1267
 
    def test_assertLength_matches_nonempty(self):
1268
 
        a_list = [1, 2, 3]
1269
 
        self.assertLength(3, a_list)
1270
 
 
1271
 
    def test_assertLength_fails_different(self):
1272
 
        a_list = []
1273
 
        self.assertRaises(AssertionError, self.assertLength, 1, a_list)
1274
 
 
1275
 
    def test_assertLength_shows_sequence_in_failure(self):
1276
 
        a_list = [1, 2, 3]
1277
 
        exception = self.assertRaises(AssertionError, self.assertLength, 2,
1278
 
            a_list)
1279
 
        self.assertEqual('Incorrect length: wanted 2, got 3 for [1, 2, 3]',
1280
 
            exception.args[0])
1281
 
 
1282
 
    def test_base_setUp_not_called_causes_failure(self):
1283
 
        class TestCaseWithBrokenSetUp(tests.TestCase):
1284
 
            def setUp(self):
1285
 
                pass # does not call TestCase.setUp
1286
 
            def test_foo(self):
1287
 
                pass
1288
 
        test = TestCaseWithBrokenSetUp('test_foo')
1289
 
        result = unittest.TestResult()
1290
 
        test.run(result)
1291
 
        self.assertFalse(result.wasSuccessful())
1292
 
        self.assertEqual(1, result.testsRun)
1293
 
 
1294
 
    def test_base_tearDown_not_called_causes_failure(self):
1295
 
        class TestCaseWithBrokenTearDown(tests.TestCase):
1296
 
            def tearDown(self):
1297
 
                pass # does not call TestCase.tearDown
1298
 
            def test_foo(self):
1299
 
                pass
1300
 
        test = TestCaseWithBrokenTearDown('test_foo')
1301
 
        result = unittest.TestResult()
1302
 
        test.run(result)
1303
 
        self.assertFalse(result.wasSuccessful())
1304
 
        self.assertEqual(1, result.testsRun)
1305
 
 
1306
 
    def test_debug_flags_sanitised(self):
1307
 
        """The bzrlib debug flags should be sanitised by setUp."""
1308
 
        if 'allow_debug' in tests.selftest_debug_flags:
1309
 
            raise tests.TestNotApplicable(
1310
 
                '-Eallow_debug option prevents debug flag sanitisation')
1311
 
        # we could set something and run a test that will check
1312
 
        # it gets santised, but this is probably sufficient for now:
1313
 
        # if someone runs the test with -Dsomething it will error.
1314
 
        flags = set()
1315
 
        if self._lock_check_thorough:
1316
 
            flags.add('strict_locks')
1317
 
        self.assertEqual(flags, bzrlib.debug.debug_flags)
1318
 
 
1319
 
    def change_selftest_debug_flags(self, new_flags):
1320
 
        self.overrideAttr(tests, 'selftest_debug_flags', set(new_flags))
1321
 
 
1322
 
    def test_allow_debug_flag(self):
1323
 
        """The -Eallow_debug flag prevents bzrlib.debug.debug_flags from being
1324
 
        sanitised (i.e. cleared) before running a test.
1325
 
        """
1326
 
        self.change_selftest_debug_flags(set(['allow_debug']))
1327
 
        bzrlib.debug.debug_flags = set(['a-flag'])
1328
 
        class TestThatRecordsFlags(tests.TestCase):
1329
 
            def test_foo(nested_self):
1330
 
                self.flags = set(bzrlib.debug.debug_flags)
1331
 
        test = TestThatRecordsFlags('test_foo')
1332
 
        test.run(self.make_test_result())
1333
 
        flags = set(['a-flag'])
1334
 
        if 'disable_lock_checks' not in tests.selftest_debug_flags:
1335
 
            flags.add('strict_locks')
1336
 
        self.assertEqual(flags, self.flags)
1337
 
 
1338
 
    def test_disable_lock_checks(self):
1339
 
        """The -Edisable_lock_checks flag disables thorough checks."""
1340
 
        class TestThatRecordsFlags(tests.TestCase):
1341
 
            def test_foo(nested_self):
1342
 
                self.flags = set(bzrlib.debug.debug_flags)
1343
 
                self.test_lock_check_thorough = nested_self._lock_check_thorough
1344
 
        self.change_selftest_debug_flags(set())
1345
 
        test = TestThatRecordsFlags('test_foo')
1346
 
        test.run(self.make_test_result())
1347
 
        # By default we do strict lock checking and thorough lock/unlock
1348
 
        # tracking.
1349
 
        self.assertTrue(self.test_lock_check_thorough)
1350
 
        self.assertEqual(set(['strict_locks']), self.flags)
1351
 
        # Now set the disable_lock_checks flag, and show that this changed.
1352
 
        self.change_selftest_debug_flags(set(['disable_lock_checks']))
1353
 
        test = TestThatRecordsFlags('test_foo')
1354
 
        test.run(self.make_test_result())
1355
 
        self.assertFalse(self.test_lock_check_thorough)
1356
 
        self.assertEqual(set(), self.flags)
1357
 
 
1358
 
    def test_this_fails_strict_lock_check(self):
1359
 
        class TestThatRecordsFlags(tests.TestCase):
1360
 
            def test_foo(nested_self):
1361
 
                self.flags1 = set(bzrlib.debug.debug_flags)
1362
 
                self.thisFailsStrictLockCheck()
1363
 
                self.flags2 = set(bzrlib.debug.debug_flags)
1364
 
        # Make sure lock checking is active
1365
 
        self.change_selftest_debug_flags(set())
1366
 
        test = TestThatRecordsFlags('test_foo')
1367
 
        test.run(self.make_test_result())
1368
 
        self.assertEqual(set(['strict_locks']), self.flags1)
1369
 
        self.assertEqual(set(), self.flags2)
1370
 
 
1371
 
    def test_debug_flags_restored(self):
1372
 
        """The bzrlib debug flags should be restored to their original state
1373
 
        after the test was run, even if allow_debug is set.
1374
 
        """
1375
 
        self.change_selftest_debug_flags(set(['allow_debug']))
1376
 
        # Now run a test that modifies debug.debug_flags.
1377
 
        bzrlib.debug.debug_flags = set(['original-state'])
1378
 
        class TestThatModifiesFlags(tests.TestCase):
1379
 
            def test_foo(self):
1380
 
                bzrlib.debug.debug_flags = set(['modified'])
1381
 
        test = TestThatModifiesFlags('test_foo')
1382
 
        test.run(self.make_test_result())
1383
 
        self.assertEqual(set(['original-state']), bzrlib.debug.debug_flags)
1384
 
 
1385
 
    def make_test_result(self):
1386
 
        """Get a test result that writes to the test log file."""
1387
 
        return tests.TextTestResult(self._log_file, descriptions=0, verbosity=1)
1388
 
 
1389
 
    def inner_test(self):
1390
 
        # the inner child test
1391
 
        note("inner_test")
1392
 
 
1393
 
    def outer_child(self):
1394
 
        # the outer child test
1395
 
        note("outer_start")
1396
 
        self.inner_test = TestTestCase("inner_child")
1397
 
        result = self.make_test_result()
1398
 
        self.inner_test.run(result)
1399
 
        note("outer finish")
1400
 
        self.addCleanup(osutils.delete_any, self._log_file_name)
1401
 
 
1402
 
    def test_trace_nesting(self):
1403
 
        # this tests that each test case nests its trace facility correctly.
1404
 
        # we do this by running a test case manually. That test case (A)
1405
 
        # should setup a new log, log content to it, setup a child case (B),
1406
 
        # which should log independently, then case (A) should log a trailer
1407
 
        # and return.
1408
 
        # we do two nested children so that we can verify the state of the
1409
 
        # logs after the outer child finishes is correct, which a bad clean
1410
 
        # up routine in tearDown might trigger a fault in our test with only
1411
 
        # one child, we should instead see the bad result inside our test with
1412
 
        # the two children.
1413
 
        # the outer child test
1414
 
        original_trace = bzrlib.trace._trace_file
1415
 
        outer_test = TestTestCase("outer_child")
1416
 
        result = self.make_test_result()
1417
 
        outer_test.run(result)
1418
 
        self.assertEqual(original_trace, bzrlib.trace._trace_file)
1419
 
 
1420
 
    def method_that_times_a_bit_twice(self):
1421
 
        # call self.time twice to ensure it aggregates
1422
 
        self.time(time.sleep, 0.007)
1423
 
        self.time(time.sleep, 0.007)
1424
 
 
1425
 
    def test_time_creates_benchmark_in_result(self):
1426
 
        """Test that the TestCase.time() method accumulates a benchmark time."""
1427
 
        sample_test = TestTestCase("method_that_times_a_bit_twice")
1428
 
        output_stream = StringIO()
1429
 
        result = bzrlib.tests.VerboseTestResult(
1430
 
            output_stream,
1431
 
            descriptions=0,
1432
 
            verbosity=2)
1433
 
        sample_test.run(result)
1434
 
        self.assertContainsRe(
1435
 
            output_stream.getvalue(),
1436
 
            r"\d+ms\*\n$")
1437
 
 
1438
 
    def test_hooks_sanitised(self):
1439
 
        """The bzrlib hooks should be sanitised by setUp."""
1440
 
        # Note this test won't fail with hooks that the core library doesn't
1441
 
        # use - but it trigger with a plugin that adds hooks, so its still a
1442
 
        # useful warning in that case.
1443
 
        self.assertEqual(bzrlib.branch.BranchHooks(), bzrlib.branch.Branch.hooks)
1444
 
        self.assertEqual(
1445
 
            bzrlib.smart.server.SmartServerHooks(),
1446
 
            bzrlib.smart.server.SmartTCPServer.hooks)
1447
 
        self.assertEqual(
1448
 
            bzrlib.commands.CommandHooks(), bzrlib.commands.Command.hooks)
1449
 
 
1450
 
    def test__gather_lsprof_in_benchmarks(self):
1451
 
        """When _gather_lsprof_in_benchmarks is on, accumulate profile data.
1452
 
 
1453
 
        Each self.time() call is individually and separately profiled.
1454
 
        """
1455
 
        self.requireFeature(features.lsprof_feature)
1456
 
        # overrides the class member with an instance member so no cleanup
1457
 
        # needed.
1458
 
        self._gather_lsprof_in_benchmarks = True
1459
 
        self.time(time.sleep, 0.000)
1460
 
        self.time(time.sleep, 0.003)
1461
 
        self.assertEqual(2, len(self._benchcalls))
1462
 
        self.assertEqual((time.sleep, (0.000,), {}), self._benchcalls[0][0])
1463
 
        self.assertEqual((time.sleep, (0.003,), {}), self._benchcalls[1][0])
1464
 
        self.assertIsInstance(self._benchcalls[0][1], bzrlib.lsprof.Stats)
1465
 
        self.assertIsInstance(self._benchcalls[1][1], bzrlib.lsprof.Stats)
1466
 
        del self._benchcalls[:]
1467
 
 
1468
 
    def test_knownFailure(self):
1469
 
        """Self.knownFailure() should raise a KnownFailure exception."""
1470
 
        self.assertRaises(tests.KnownFailure, self.knownFailure, "A Failure")
1471
 
 
1472
 
    def test_open_bzrdir_safe_roots(self):
1473
 
        # even a memory transport should fail to open when its url isn't 
1474
 
        # permitted.
1475
 
        # Manually set one up (TestCase doesn't and shouldn't provide magic
1476
 
        # machinery)
1477
 
        transport_server = memory.MemoryServer()
1478
 
        transport_server.start_server()
1479
 
        self.addCleanup(transport_server.stop_server)
1480
 
        t = transport.get_transport_from_url(transport_server.get_url())
1481
 
        controldir.ControlDir.create(t.base)
1482
 
        self.assertRaises(errors.BzrError,
1483
 
            controldir.ControlDir.open_from_transport, t)
1484
 
        # But if we declare this as safe, we can open the bzrdir.
1485
 
        self.permit_url(t.base)
1486
 
        self._bzr_selftest_roots.append(t.base)
1487
 
        controldir.ControlDir.open_from_transport(t)
1488
 
 
1489
 
    def test_requireFeature_available(self):
1490
 
        """self.requireFeature(available) is a no-op."""
1491
 
        class Available(features.Feature):
1492
 
            def _probe(self):return True
1493
 
        feature = Available()
1494
 
        self.requireFeature(feature)
1495
 
 
1496
 
    def test_requireFeature_unavailable(self):
1497
 
        """self.requireFeature(unavailable) raises UnavailableFeature."""
1498
 
        class Unavailable(features.Feature):
1499
 
            def _probe(self):return False
1500
 
        feature = Unavailable()
1501
 
        self.assertRaises(tests.UnavailableFeature,
1502
 
                          self.requireFeature, feature)
1503
 
 
1504
 
    def test_run_no_parameters(self):
1505
 
        test = SampleTestCase('_test_pass')
1506
 
        test.run()
1507
 
 
1508
 
    def test_run_enabled_unittest_result(self):
1509
 
        """Test we revert to regular behaviour when the test is enabled."""
1510
 
        test = SampleTestCase('_test_pass')
1511
 
        class EnabledFeature(object):
1512
 
            def available(self):
1513
 
                return True
1514
 
        test._test_needs_features = [EnabledFeature()]
1515
 
        result = unittest.TestResult()
1516
 
        test.run(result)
1517
 
        self.assertEqual(1, result.testsRun)
1518
 
        self.assertEqual([], result.errors)
1519
 
        self.assertEqual([], result.failures)
1520
 
 
1521
 
    def test_run_disabled_unittest_result(self):
1522
 
        """Test our compatability for disabled tests with unittest results."""
1523
 
        test = SampleTestCase('_test_pass')
1524
 
        class DisabledFeature(object):
1525
 
            def available(self):
1526
 
                return False
1527
 
        test._test_needs_features = [DisabledFeature()]
1528
 
        result = unittest.TestResult()
1529
 
        test.run(result)
1530
 
        self.assertEqual(1, result.testsRun)
1531
 
        self.assertEqual([], result.errors)
1532
 
        self.assertEqual([], result.failures)
1533
 
 
1534
 
    def test_run_disabled_supporting_result(self):
1535
 
        """Test disabled tests behaviour with support aware results."""
1536
 
        test = SampleTestCase('_test_pass')
1537
 
        class DisabledFeature(object):
1538
 
            def __eq__(self, other):
1539
 
                return isinstance(other, DisabledFeature)
1540
 
            def available(self):
1541
 
                return False
1542
 
        the_feature = DisabledFeature()
1543
 
        test._test_needs_features = [the_feature]
1544
 
        class InstrumentedTestResult(unittest.TestResult):
1545
 
            def __init__(self):
1546
 
                unittest.TestResult.__init__(self)
1547
 
                self.calls = []
1548
 
            def startTest(self, test):
1549
 
                self.calls.append(('startTest', test))
1550
 
            def stopTest(self, test):
1551
 
                self.calls.append(('stopTest', test))
1552
 
            def addNotSupported(self, test, feature):
1553
 
                self.calls.append(('addNotSupported', test, feature))
1554
 
        result = InstrumentedTestResult()
1555
 
        test.run(result)
1556
 
        case = result.calls[0][1]
1557
 
        self.assertEqual([
1558
 
            ('startTest', case),
1559
 
            ('addNotSupported', case, the_feature),
1560
 
            ('stopTest', case),
1561
 
            ],
1562
 
            result.calls)
1563
 
 
1564
 
    def test_start_server_registers_url(self):
1565
 
        transport_server = memory.MemoryServer()
1566
 
        # A little strict, but unlikely to be changed soon.
1567
 
        self.assertEqual([], self._bzr_selftest_roots)
1568
 
        self.start_server(transport_server)
1569
 
        self.assertSubset([transport_server.get_url()],
1570
 
            self._bzr_selftest_roots)
1571
 
 
1572
 
    def test_assert_list_raises_on_generator(self):
1573
 
        def generator_which_will_raise():
1574
 
            # This will not raise until after the first yield
1575
 
            yield 1
1576
 
            raise _TestException()
1577
 
 
1578
 
        e = self.assertListRaises(_TestException, generator_which_will_raise)
1579
 
        self.assertIsInstance(e, _TestException)
1580
 
 
1581
 
        e = self.assertListRaises(Exception, generator_which_will_raise)
1582
 
        self.assertIsInstance(e, _TestException)
1583
 
 
1584
 
    def test_assert_list_raises_on_plain(self):
1585
 
        def plain_exception():
1586
 
            raise _TestException()
1587
 
            return []
1588
 
 
1589
 
        e = self.assertListRaises(_TestException, plain_exception)
1590
 
        self.assertIsInstance(e, _TestException)
1591
 
 
1592
 
        e = self.assertListRaises(Exception, plain_exception)
1593
 
        self.assertIsInstance(e, _TestException)
1594
 
 
1595
 
    def test_assert_list_raises_assert_wrong_exception(self):
1596
 
        class _NotTestException(Exception):
1597
 
            pass
1598
 
 
1599
 
        def wrong_exception():
1600
 
            raise _NotTestException()
1601
 
 
1602
 
        def wrong_exception_generator():
1603
 
            yield 1
1604
 
            yield 2
1605
 
            raise _NotTestException()
1606
 
 
1607
 
        # Wrong exceptions are not intercepted
1608
 
        self.assertRaises(_NotTestException,
1609
 
            self.assertListRaises, _TestException, wrong_exception)
1610
 
        self.assertRaises(_NotTestException,
1611
 
            self.assertListRaises, _TestException, wrong_exception_generator)
1612
 
 
1613
 
    def test_assert_list_raises_no_exception(self):
1614
 
        def success():
1615
 
            return []
1616
 
 
1617
 
        def success_generator():
1618
 
            yield 1
1619
 
            yield 2
1620
 
 
1621
 
        self.assertRaises(AssertionError,
1622
 
            self.assertListRaises, _TestException, success)
1623
 
 
1624
 
        self.assertRaises(AssertionError,
1625
 
            self.assertListRaises, _TestException, success_generator)
1626
 
 
1627
 
    def test_overrideAttr_without_value(self):
1628
 
        self.test_attr = 'original' # Define a test attribute
1629
 
        obj = self # Make 'obj' visible to the embedded test
1630
 
        class Test(tests.TestCase):
1631
 
 
1632
 
            def setUp(self):
1633
 
                tests.TestCase.setUp(self)
1634
 
                self.orig = self.overrideAttr(obj, 'test_attr')
1635
 
 
1636
 
            def test_value(self):
1637
 
                self.assertEqual('original', self.orig)
1638
 
                self.assertEqual('original', obj.test_attr)
1639
 
                obj.test_attr = 'modified'
1640
 
                self.assertEqual('modified', obj.test_attr)
1641
 
 
1642
 
        test = Test('test_value')
1643
 
        test.run(unittest.TestResult())
1644
 
        self.assertEqual('original', obj.test_attr)
1645
 
 
1646
 
    def test_overrideAttr_with_value(self):
1647
 
        self.test_attr = 'original' # Define a test attribute
1648
 
        obj = self # Make 'obj' visible to the embedded test
1649
 
        class Test(tests.TestCase):
1650
 
 
1651
 
            def setUp(self):
1652
 
                tests.TestCase.setUp(self)
1653
 
                self.orig = self.overrideAttr(obj, 'test_attr', new='modified')
1654
 
 
1655
 
            def test_value(self):
1656
 
                self.assertEqual('original', self.orig)
1657
 
                self.assertEqual('modified', obj.test_attr)
1658
 
 
1659
 
        test = Test('test_value')
1660
 
        test.run(unittest.TestResult())
1661
 
        self.assertEqual('original', obj.test_attr)
1662
 
 
1663
 
    def test_recordCalls(self):
1664
 
        from bzrlib.tests import test_selftest
1665
 
        calls = self.recordCalls(
1666
 
            test_selftest, '_add_numbers')
1667
 
        self.assertEqual(test_selftest._add_numbers(2, 10),
1668
 
            12)
1669
 
        self.assertEquals(calls, [((2, 10), {})])
1670
 
 
1671
 
 
1672
 
def _add_numbers(a, b):
1673
 
    return a + b
1674
 
 
1675
 
 
1676
 
class _MissingFeature(features.Feature):
1677
 
    def _probe(self):
1678
 
        return False
1679
 
missing_feature = _MissingFeature()
1680
 
 
1681
 
 
1682
 
def _get_test(name):
1683
 
    """Get an instance of a specific example test.
1684
 
 
1685
 
    We protect this in a function so that they don't auto-run in the test
1686
 
    suite.
1687
 
    """
1688
 
 
1689
 
    class ExampleTests(tests.TestCase):
1690
 
 
1691
 
        def test_fail(self):
1692
 
            mutter('this was a failing test')
1693
 
            self.fail('this test will fail')
1694
 
 
1695
 
        def test_error(self):
1696
 
            mutter('this test errored')
1697
 
            raise RuntimeError('gotcha')
1698
 
 
1699
 
        def test_missing_feature(self):
1700
 
            mutter('missing the feature')
1701
 
            self.requireFeature(missing_feature)
1702
 
 
1703
 
        def test_skip(self):
1704
 
            mutter('this test will be skipped')
1705
 
            raise tests.TestSkipped('reason')
1706
 
 
1707
 
        def test_success(self):
1708
 
            mutter('this test succeeds')
1709
 
 
1710
 
        def test_xfail(self):
1711
 
            mutter('test with expected failure')
1712
 
            self.knownFailure('this_fails')
1713
 
 
1714
 
        def test_unexpected_success(self):
1715
 
            mutter('test with unexpected success')
1716
 
            self.expectFailure('should_fail', lambda: None)
1717
 
 
1718
 
    return ExampleTests(name)
1719
 
 
1720
 
 
1721
 
class TestTestCaseLogDetails(tests.TestCase):
1722
 
 
1723
 
    def _run_test(self, test_name):
1724
 
        test = _get_test(test_name)
1725
 
        result = testtools.TestResult()
1726
 
        test.run(result)
1727
 
        return result
1728
 
 
1729
 
    def test_fail_has_log(self):
1730
 
        result = self._run_test('test_fail')
1731
 
        self.assertEqual(1, len(result.failures))
1732
 
        result_content = result.failures[0][1]
1733
 
        self.assertContainsRe(result_content,
1734
 
            '(?m)^(?:Text attachment: )?log(?:$|: )')
1735
 
        self.assertContainsRe(result_content, 'this was a failing test')
1736
 
 
1737
 
    def test_error_has_log(self):
1738
 
        result = self._run_test('test_error')
1739
 
        self.assertEqual(1, len(result.errors))
1740
 
        result_content = result.errors[0][1]
1741
 
        self.assertContainsRe(result_content,
1742
 
            '(?m)^(?:Text attachment: )?log(?:$|: )')
1743
 
        self.assertContainsRe(result_content, 'this test errored')
1744
 
 
1745
 
    def test_skip_has_no_log(self):
1746
 
        result = self._run_test('test_skip')
1747
 
        self.assertEqual(['reason'], result.skip_reasons.keys())
1748
 
        skips = result.skip_reasons['reason']
1749
 
        self.assertEqual(1, len(skips))
1750
 
        test = skips[0]
1751
 
        self.assertFalse('log' in test.getDetails())
1752
 
 
1753
 
    def test_missing_feature_has_no_log(self):
1754
 
        # testtools doesn't know about addNotSupported, so it just gets
1755
 
        # considered as a skip
1756
 
        result = self._run_test('test_missing_feature')
1757
 
        self.assertEqual([missing_feature], result.skip_reasons.keys())
1758
 
        skips = result.skip_reasons[missing_feature]
1759
 
        self.assertEqual(1, len(skips))
1760
 
        test = skips[0]
1761
 
        self.assertFalse('log' in test.getDetails())
1762
 
 
1763
 
    def test_xfail_has_no_log(self):
1764
 
        result = self._run_test('test_xfail')
1765
 
        self.assertEqual(1, len(result.expectedFailures))
1766
 
        result_content = result.expectedFailures[0][1]
1767
 
        self.assertNotContainsRe(result_content,
1768
 
            '(?m)^(?:Text attachment: )?log(?:$|: )')
1769
 
        self.assertNotContainsRe(result_content, 'test with expected failure')
1770
 
 
1771
 
    def test_unexpected_success_has_log(self):
1772
 
        result = self._run_test('test_unexpected_success')
1773
 
        self.assertEqual(1, len(result.unexpectedSuccesses))
1774
 
        # Inconsistency, unexpectedSuccesses is a list of tests,
1775
 
        # expectedFailures is a list of reasons?
1776
 
        test = result.unexpectedSuccesses[0]
1777
 
        details = test.getDetails()
1778
 
        self.assertTrue('log' in details)
1779
 
 
1780
 
 
1781
 
class TestTestCloning(tests.TestCase):
1782
 
    """Tests that test cloning of TestCases (as used by multiply_tests)."""
1783
 
 
1784
 
    def test_cloned_testcase_does_not_share_details(self):
1785
 
        """A TestCase cloned with clone_test does not share mutable attributes
1786
 
        such as details or cleanups.
1787
 
        """
1788
 
        class Test(tests.TestCase):
1789
 
            def test_foo(self):
1790
 
                self.addDetail('foo', Content('text/plain', lambda: 'foo'))
1791
 
        orig_test = Test('test_foo')
1792
 
        cloned_test = tests.clone_test(orig_test, orig_test.id() + '(cloned)')
1793
 
        orig_test.run(unittest.TestResult())
1794
 
        self.assertEqual('foo', orig_test.getDetails()['foo'].iter_bytes())
1795
 
        self.assertEqual(None, cloned_test.getDetails().get('foo'))
1796
 
 
1797
 
    def test_double_apply_scenario_preserves_first_scenario(self):
1798
 
        """Applying two levels of scenarios to a test preserves the attributes
1799
 
        added by both scenarios.
1800
 
        """
1801
 
        class Test(tests.TestCase):
1802
 
            def test_foo(self):
1803
 
                pass
1804
 
        test = Test('test_foo')
1805
 
        scenarios_x = [('x=1', {'x': 1}), ('x=2', {'x': 2})]
1806
 
        scenarios_y = [('y=1', {'y': 1}), ('y=2', {'y': 2})]
1807
 
        suite = tests.multiply_tests(test, scenarios_x, unittest.TestSuite())
1808
 
        suite = tests.multiply_tests(suite, scenarios_y, unittest.TestSuite())
1809
 
        all_tests = list(tests.iter_suite_tests(suite))
1810
 
        self.assertLength(4, all_tests)
1811
 
        all_xys = sorted((t.x, t.y) for t in all_tests)
1812
 
        self.assertEqual([(1, 1), (1, 2), (2, 1), (2, 2)], all_xys)
1813
 
 
1814
 
 
1815
 
# NB: Don't delete this; it's not actually from 0.11!
1816
 
@deprecated_function(deprecated_in((0, 11, 0)))
1817
 
def sample_deprecated_function():
1818
 
    """A deprecated function to test applyDeprecated with."""
1819
 
    return 2
1820
 
 
1821
 
 
1822
 
def sample_undeprecated_function(a_param):
1823
 
    """A undeprecated function to test applyDeprecated with."""
1824
 
 
1825
 
 
1826
 
class ApplyDeprecatedHelper(object):
1827
 
    """A helper class for ApplyDeprecated tests."""
1828
 
 
1829
 
    @deprecated_method(deprecated_in((0, 11, 0)))
1830
 
    def sample_deprecated_method(self, param_one):
1831
 
        """A deprecated method for testing with."""
1832
 
        return param_one
1833
 
 
1834
 
    def sample_normal_method(self):
1835
 
        """A undeprecated method."""
1836
 
 
1837
 
    @deprecated_method(deprecated_in((0, 10, 0)))
1838
 
    def sample_nested_deprecation(self):
1839
 
        return sample_deprecated_function()
1840
 
 
1841
 
 
1842
 
class TestExtraAssertions(tests.TestCase):
1843
 
    """Tests for new test assertions in bzrlib test suite"""
1844
 
 
1845
 
    def test_assert_isinstance(self):
1846
 
        self.assertIsInstance(2, int)
1847
 
        self.assertIsInstance(u'', basestring)
1848
 
        e = self.assertRaises(AssertionError, self.assertIsInstance, None, int)
1849
 
        self.assertEquals(str(e),
1850
 
            "None is an instance of <type 'NoneType'> rather than <type 'int'>")
1851
 
        self.assertRaises(AssertionError, self.assertIsInstance, 23.3, int)
1852
 
        e = self.assertRaises(AssertionError,
1853
 
            self.assertIsInstance, None, int, "it's just not")
1854
 
        self.assertEquals(str(e),
1855
 
            "None is an instance of <type 'NoneType'> rather than <type 'int'>"
1856
 
            ": it's just not")
1857
 
 
1858
 
    def test_assertEndsWith(self):
1859
 
        self.assertEndsWith('foo', 'oo')
1860
 
        self.assertRaises(AssertionError, self.assertEndsWith, 'o', 'oo')
1861
 
 
1862
 
    def test_assertEqualDiff(self):
1863
 
        e = self.assertRaises(AssertionError,
1864
 
                              self.assertEqualDiff, '', '\n')
1865
 
        self.assertEquals(str(e),
1866
 
                          # Don't blink ! The '+' applies to the second string
1867
 
                          'first string is missing a final newline.\n+ \n')
1868
 
        e = self.assertRaises(AssertionError,
1869
 
                              self.assertEqualDiff, '\n', '')
1870
 
        self.assertEquals(str(e),
1871
 
                          # Don't blink ! The '-' applies to the second string
1872
 
                          'second string is missing a final newline.\n- \n')
1873
 
 
1874
 
 
1875
 
class TestDeprecations(tests.TestCase):
1876
 
 
1877
 
    def test_applyDeprecated_not_deprecated(self):
1878
 
        sample_object = ApplyDeprecatedHelper()
1879
 
        # calling an undeprecated callable raises an assertion
1880
 
        self.assertRaises(AssertionError, self.applyDeprecated,
1881
 
            deprecated_in((0, 11, 0)),
1882
 
            sample_object.sample_normal_method)
1883
 
        self.assertRaises(AssertionError, self.applyDeprecated,
1884
 
            deprecated_in((0, 11, 0)),
1885
 
            sample_undeprecated_function, "a param value")
1886
 
        # calling a deprecated callable (function or method) with the wrong
1887
 
        # expected deprecation fails.
1888
 
        self.assertRaises(AssertionError, self.applyDeprecated,
1889
 
            deprecated_in((0, 10, 0)),
1890
 
            sample_object.sample_deprecated_method, "a param value")
1891
 
        self.assertRaises(AssertionError, self.applyDeprecated,
1892
 
            deprecated_in((0, 10, 0)),
1893
 
            sample_deprecated_function)
1894
 
        # calling a deprecated callable (function or method) with the right
1895
 
        # expected deprecation returns the functions result.
1896
 
        self.assertEqual("a param value",
1897
 
            self.applyDeprecated(deprecated_in((0, 11, 0)),
1898
 
            sample_object.sample_deprecated_method, "a param value"))
1899
 
        self.assertEqual(2, self.applyDeprecated(deprecated_in((0, 11, 0)),
1900
 
            sample_deprecated_function))
1901
 
        # calling a nested deprecation with the wrong deprecation version
1902
 
        # fails even if a deeper nested function was deprecated with the
1903
 
        # supplied version.
1904
 
        self.assertRaises(AssertionError, self.applyDeprecated,
1905
 
            deprecated_in((0, 11, 0)), sample_object.sample_nested_deprecation)
1906
 
        # calling a nested deprecation with the right deprecation value
1907
 
        # returns the calls result.
1908
 
        self.assertEqual(2, self.applyDeprecated(deprecated_in((0, 10, 0)),
1909
 
            sample_object.sample_nested_deprecation))
1910
 
 
1911
 
    def test_callDeprecated(self):
1912
 
        def testfunc(be_deprecated, result=None):
1913
 
            if be_deprecated is True:
1914
 
                symbol_versioning.warn('i am deprecated', DeprecationWarning,
1915
 
                                       stacklevel=1)
1916
 
            return result
1917
 
        result = self.callDeprecated(['i am deprecated'], testfunc, True)
1918
 
        self.assertIs(None, result)
1919
 
        result = self.callDeprecated([], testfunc, False, 'result')
1920
 
        self.assertEqual('result', result)
1921
 
        self.callDeprecated(['i am deprecated'], testfunc, be_deprecated=True)
1922
 
        self.callDeprecated([], testfunc, be_deprecated=False)
1923
 
 
1924
 
 
1925
 
class TestWarningTests(tests.TestCase):
1926
 
    """Tests for calling methods that raise warnings."""
1927
 
 
1928
 
    def test_callCatchWarnings(self):
1929
 
        def meth(a, b):
1930
 
            warnings.warn("this is your last warning")
1931
 
            return a + b
1932
 
        wlist, result = self.callCatchWarnings(meth, 1, 2)
1933
 
        self.assertEquals(3, result)
1934
 
        # would like just to compare them, but UserWarning doesn't implement
1935
 
        # eq well
1936
 
        w0, = wlist
1937
 
        self.assertIsInstance(w0, UserWarning)
1938
 
        self.assertEquals("this is your last warning", str(w0))
1939
 
 
1940
 
 
1941
 
class TestConvenienceMakers(tests.TestCaseWithTransport):
1942
 
    """Test for the make_* convenience functions."""
1943
 
 
1944
 
    def test_make_branch_and_tree_with_format(self):
1945
 
        # we should be able to supply a format to make_branch_and_tree
1946
 
        self.make_branch_and_tree('a', format=bzrlib.bzrdir.BzrDirMetaFormat1())
1947
 
        self.assertIsInstance(bzrlib.controldir.ControlDir.open('a')._format,
1948
 
                              bzrlib.bzrdir.BzrDirMetaFormat1)
1949
 
 
1950
 
    def test_make_branch_and_memory_tree(self):
1951
 
        # we should be able to get a new branch and a mutable tree from
1952
 
        # TestCaseWithTransport
1953
 
        tree = self.make_branch_and_memory_tree('a')
1954
 
        self.assertIsInstance(tree, bzrlib.memorytree.MemoryTree)
1955
 
 
1956
 
    def test_make_tree_for_local_vfs_backed_transport(self):
1957
 
        # make_branch_and_tree has to use local branch and repositories
1958
 
        # when the vfs transport and local disk are colocated, even if
1959
 
        # a different transport is in use for url generation.
1960
 
        self.transport_server = test_server.FakeVFATServer
1961
 
        self.assertFalse(self.get_url('t1').startswith('file://'))
1962
 
        tree = self.make_branch_and_tree('t1')
1963
 
        base = tree.bzrdir.root_transport.base
1964
 
        self.assertStartsWith(base, 'file://')
1965
 
        self.assertEquals(tree.bzrdir.root_transport,
1966
 
                tree.branch.bzrdir.root_transport)
1967
 
        self.assertEquals(tree.bzrdir.root_transport,
1968
 
                tree.branch.repository.bzrdir.root_transport)
1969
 
 
1970
 
 
1971
 
class SelfTestHelper(object):
1972
 
 
1973
 
    def run_selftest(self, **kwargs):
1974
 
        """Run selftest returning its output."""
1975
 
        output = StringIO()
1976
 
        old_transport = bzrlib.tests.default_transport
1977
 
        old_root = tests.TestCaseWithMemoryTransport.TEST_ROOT
1978
 
        tests.TestCaseWithMemoryTransport.TEST_ROOT = None
1979
 
        try:
1980
 
            self.assertEqual(True, tests.selftest(stream=output, **kwargs))
1981
 
        finally:
1982
 
            bzrlib.tests.default_transport = old_transport
1983
 
            tests.TestCaseWithMemoryTransport.TEST_ROOT = old_root
1984
 
        output.seek(0)
1985
 
        return output
1986
 
 
1987
 
 
1988
 
class TestSelftest(tests.TestCase, SelfTestHelper):
1989
 
    """Tests of bzrlib.tests.selftest."""
1990
 
 
1991
 
    def test_selftest_benchmark_parameter_invokes_test_suite__benchmark__(self):
1992
 
        factory_called = []
1993
 
        def factory():
1994
 
            factory_called.append(True)
1995
 
            return TestUtil.TestSuite()
1996
 
        out = StringIO()
1997
 
        err = StringIO()
1998
 
        self.apply_redirected(out, err, None, bzrlib.tests.selftest,
1999
 
            test_suite_factory=factory)
2000
 
        self.assertEqual([True], factory_called)
2001
 
 
2002
 
    def factory(self):
2003
 
        """A test suite factory."""
2004
 
        class Test(tests.TestCase):
2005
 
            def a(self):
2006
 
                pass
2007
 
            def b(self):
2008
 
                pass
2009
 
            def c(self):
2010
 
                pass
2011
 
        return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
2012
 
 
2013
 
    def test_list_only(self):
2014
 
        output = self.run_selftest(test_suite_factory=self.factory,
2015
 
            list_only=True)
2016
 
        self.assertEqual(3, len(output.readlines()))
2017
 
 
2018
 
    def test_list_only_filtered(self):
2019
 
        output = self.run_selftest(test_suite_factory=self.factory,
2020
 
            list_only=True, pattern="Test.b")
2021
 
        self.assertEndsWith(output.getvalue(), "Test.b\n")
2022
 
        self.assertLength(1, output.readlines())
2023
 
 
2024
 
    def test_list_only_excludes(self):
2025
 
        output = self.run_selftest(test_suite_factory=self.factory,
2026
 
            list_only=True, exclude_pattern="Test.b")
2027
 
        self.assertNotContainsRe("Test.b", output.getvalue())
2028
 
        self.assertLength(2, output.readlines())
2029
 
 
2030
 
    def test_lsprof_tests(self):
2031
 
        self.requireFeature(features.lsprof_feature)
2032
 
        results = []
2033
 
        class Test(object):
2034
 
            def __call__(test, result):
2035
 
                test.run(result)
2036
 
            def run(test, result):
2037
 
                results.append(result)
2038
 
            def countTestCases(self):
2039
 
                return 1
2040
 
        self.run_selftest(test_suite_factory=Test, lsprof_tests=True)
2041
 
        self.assertLength(1, results)
2042
 
        self.assertIsInstance(results.pop(), ExtendedToOriginalDecorator)
2043
 
 
2044
 
    def test_random(self):
2045
 
        # test randomising by listing a number of tests.
2046
 
        output_123 = self.run_selftest(test_suite_factory=self.factory,
2047
 
            list_only=True, random_seed="123")
2048
 
        output_234 = self.run_selftest(test_suite_factory=self.factory,
2049
 
            list_only=True, random_seed="234")
2050
 
        self.assertNotEqual(output_123, output_234)
2051
 
        # "Randominzing test order..\n\n
2052
 
        self.assertLength(5, output_123.readlines())
2053
 
        self.assertLength(5, output_234.readlines())
2054
 
 
2055
 
    def test_random_reuse_is_same_order(self):
2056
 
        # test randomising by listing a number of tests.
2057
 
        expected = self.run_selftest(test_suite_factory=self.factory,
2058
 
            list_only=True, random_seed="123")
2059
 
        repeated = self.run_selftest(test_suite_factory=self.factory,
2060
 
            list_only=True, random_seed="123")
2061
 
        self.assertEqual(expected.getvalue(), repeated.getvalue())
2062
 
 
2063
 
    def test_runner_class(self):
2064
 
        self.requireFeature(features.subunit)
2065
 
        from subunit import ProtocolTestCase
2066
 
        stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
2067
 
            test_suite_factory=self.factory)
2068
 
        test = ProtocolTestCase(stream)
2069
 
        result = unittest.TestResult()
2070
 
        test.run(result)
2071
 
        self.assertEqual(3, result.testsRun)
2072
 
 
2073
 
    def test_starting_with_single_argument(self):
2074
 
        output = self.run_selftest(test_suite_factory=self.factory,
2075
 
            starting_with=['bzrlib.tests.test_selftest.Test.a'],
2076
 
            list_only=True)
2077
 
        self.assertEqual('bzrlib.tests.test_selftest.Test.a\n',
2078
 
            output.getvalue())
2079
 
 
2080
 
    def test_starting_with_multiple_argument(self):
2081
 
        output = self.run_selftest(test_suite_factory=self.factory,
2082
 
            starting_with=['bzrlib.tests.test_selftest.Test.a',
2083
 
                'bzrlib.tests.test_selftest.Test.b'],
2084
 
            list_only=True)
2085
 
        self.assertEqual('bzrlib.tests.test_selftest.Test.a\n'
2086
 
            'bzrlib.tests.test_selftest.Test.b\n',
2087
 
            output.getvalue())
2088
 
 
2089
 
    def check_transport_set(self, transport_server):
2090
 
        captured_transport = []
2091
 
        def seen_transport(a_transport):
2092
 
            captured_transport.append(a_transport)
2093
 
        class Capture(tests.TestCase):
2094
 
            def a(self):
2095
 
                seen_transport(bzrlib.tests.default_transport)
2096
 
        def factory():
2097
 
            return TestUtil.TestSuite([Capture("a")])
2098
 
        self.run_selftest(transport=transport_server, test_suite_factory=factory)
2099
 
        self.assertEqual(transport_server, captured_transport[0])
2100
 
 
2101
 
    def test_transport_sftp(self):
2102
 
        self.requireFeature(features.paramiko)
2103
 
        from bzrlib.tests import stub_sftp
2104
 
        self.check_transport_set(stub_sftp.SFTPAbsoluteServer)
2105
 
 
2106
 
    def test_transport_memory(self):
2107
 
        self.check_transport_set(memory.MemoryServer)
2108
 
 
2109
 
 
2110
 
class TestSelftestWithIdList(tests.TestCaseInTempDir, SelfTestHelper):
2111
 
    # Does IO: reads test.list
2112
 
 
2113
 
    def test_load_list(self):
2114
 
        # Provide a list with one test - this test.
2115
 
        test_id_line = '%s\n' % self.id()
2116
 
        self.build_tree_contents([('test.list', test_id_line)])
2117
 
        # And generate a list of the tests in  the suite.
2118
 
        stream = self.run_selftest(load_list='test.list', list_only=True)
2119
 
        self.assertEqual(test_id_line, stream.getvalue())
2120
 
 
2121
 
    def test_load_unknown(self):
2122
 
        # Provide a list with one test - this test.
2123
 
        # And generate a list of the tests in  the suite.
2124
 
        err = self.assertRaises(errors.NoSuchFile, self.run_selftest,
2125
 
            load_list='missing file name', list_only=True)
2126
 
 
2127
 
 
2128
 
class TestSubunitLogDetails(tests.TestCase, SelfTestHelper):
2129
 
 
2130
 
    _test_needs_features = [features.subunit]
2131
 
 
2132
 
    def run_subunit_stream(self, test_name):
2133
 
        from subunit import ProtocolTestCase
2134
 
        def factory():
2135
 
            return TestUtil.TestSuite([_get_test(test_name)])
2136
 
        stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
2137
 
            test_suite_factory=factory)
2138
 
        test = ProtocolTestCase(stream)
2139
 
        result = testtools.TestResult()
2140
 
        test.run(result)
2141
 
        content = stream.getvalue()
2142
 
        return content, result
2143
 
 
2144
 
    def test_fail_has_log(self):
2145
 
        content, result = self.run_subunit_stream('test_fail')
2146
 
        self.assertEqual(1, len(result.failures))
2147
 
        self.assertContainsRe(content, '(?m)^log$')
2148
 
        self.assertContainsRe(content, 'this test will fail')
2149
 
 
2150
 
    def test_error_has_log(self):
2151
 
        content, result = self.run_subunit_stream('test_error')
2152
 
        self.assertContainsRe(content, '(?m)^log$')
2153
 
        self.assertContainsRe(content, 'this test errored')
2154
 
 
2155
 
    def test_skip_has_no_log(self):
2156
 
        content, result = self.run_subunit_stream('test_skip')
2157
 
        self.assertNotContainsRe(content, '(?m)^log$')
2158
 
        self.assertNotContainsRe(content, 'this test will be skipped')
2159
 
        self.assertEqual(['reason'], result.skip_reasons.keys())
2160
 
        skips = result.skip_reasons['reason']
2161
 
        self.assertEqual(1, len(skips))
2162
 
        test = skips[0]
2163
 
        # RemotedTestCase doesn't preserve the "details"
2164
 
        ## self.assertFalse('log' in test.getDetails())
2165
 
 
2166
 
    def test_missing_feature_has_no_log(self):
2167
 
        content, result = self.run_subunit_stream('test_missing_feature')
2168
 
        self.assertNotContainsRe(content, '(?m)^log$')
2169
 
        self.assertNotContainsRe(content, 'missing the feature')
2170
 
        self.assertEqual(['_MissingFeature\n'], result.skip_reasons.keys())
2171
 
        skips = result.skip_reasons['_MissingFeature\n']
2172
 
        self.assertEqual(1, len(skips))
2173
 
        test = skips[0]
2174
 
        # RemotedTestCase doesn't preserve the "details"
2175
 
        ## self.assertFalse('log' in test.getDetails())
2176
 
 
2177
 
    def test_xfail_has_no_log(self):
2178
 
        content, result = self.run_subunit_stream('test_xfail')
2179
 
        self.assertNotContainsRe(content, '(?m)^log$')
2180
 
        self.assertNotContainsRe(content, 'test with expected failure')
2181
 
        self.assertEqual(1, len(result.expectedFailures))
2182
 
        result_content = result.expectedFailures[0][1]
2183
 
        self.assertNotContainsRe(result_content,
2184
 
            '(?m)^(?:Text attachment: )?log(?:$|: )')
2185
 
        self.assertNotContainsRe(result_content, 'test with expected failure')
2186
 
 
2187
 
    def test_unexpected_success_has_log(self):
2188
 
        content, result = self.run_subunit_stream('test_unexpected_success')
2189
 
        self.assertContainsRe(content, '(?m)^log$')
2190
 
        self.assertContainsRe(content, 'test with unexpected success')
2191
 
        # GZ 2011-05-18: Old versions of subunit treat unexpected success as a
2192
 
        #                success, if a min version check is added remove this
2193
 
        from subunit import TestProtocolClient as _Client
2194
 
        if _Client.addUnexpectedSuccess.im_func is _Client.addSuccess.im_func:
2195
 
            self.expectFailure('subunit treats "unexpectedSuccess"'
2196
 
                               ' as a plain success',
2197
 
                self.assertEqual, 1, len(result.unexpectedSuccesses))
2198
 
        self.assertEqual(1, len(result.unexpectedSuccesses))
2199
 
        test = result.unexpectedSuccesses[0]
2200
 
        # RemotedTestCase doesn't preserve the "details"
2201
 
        ## self.assertTrue('log' in test.getDetails())
2202
 
 
2203
 
    def test_success_has_no_log(self):
2204
 
        content, result = self.run_subunit_stream('test_success')
2205
 
        self.assertEqual(1, result.testsRun)
2206
 
        self.assertNotContainsRe(content, '(?m)^log$')
2207
 
        self.assertNotContainsRe(content, 'this test succeeds')
2208
 
 
2209
 
 
2210
 
class TestRunBzr(tests.TestCase):
2211
 
 
2212
 
    out = ''
2213
 
    err = ''
2214
 
 
2215
 
    def _run_bzr_core(self, argv, retcode=0, encoding=None, stdin=None,
2216
 
                         working_dir=None):
2217
 
        """Override _run_bzr_core to test how it is invoked by run_bzr.
2218
 
 
2219
 
        Attempts to run bzr from inside this class don't actually run it.
2220
 
 
2221
 
        We test how run_bzr actually invokes bzr in another location.  Here we
2222
 
        only need to test that it passes the right parameters to run_bzr.
2223
 
        """
2224
 
        self.argv = list(argv)
2225
 
        self.retcode = retcode
2226
 
        self.encoding = encoding
2227
 
        self.stdin = stdin
2228
 
        self.working_dir = working_dir
2229
 
        return self.retcode, self.out, self.err
2230
 
 
2231
 
    def test_run_bzr_error(self):
2232
 
        self.out = "It sure does!\n"
2233
 
        out, err = self.run_bzr_error(['^$'], ['rocks'], retcode=34)
2234
 
        self.assertEqual(['rocks'], self.argv)
2235
 
        self.assertEqual(34, self.retcode)
2236
 
        self.assertEqual('It sure does!\n', out)
2237
 
        self.assertEquals(out, self.out)
2238
 
        self.assertEqual('', err)
2239
 
        self.assertEquals(err, self.err)
2240
 
 
2241
 
    def test_run_bzr_error_regexes(self):
2242
 
        self.out = ''
2243
 
        self.err = "bzr: ERROR: foobarbaz is not versioned"
2244
 
        out, err = self.run_bzr_error(
2245
 
            ["bzr: ERROR: foobarbaz is not versioned"],
2246
 
            ['file-id', 'foobarbaz'])
2247
 
 
2248
 
    def test_encoding(self):
2249
 
        """Test that run_bzr passes encoding to _run_bzr_core"""
2250
 
        self.run_bzr('foo bar')
2251
 
        self.assertEqual(None, self.encoding)
2252
 
        self.assertEqual(['foo', 'bar'], self.argv)
2253
 
 
2254
 
        self.run_bzr('foo bar', encoding='baz')
2255
 
        self.assertEqual('baz', self.encoding)
2256
 
        self.assertEqual(['foo', 'bar'], self.argv)
2257
 
 
2258
 
    def test_retcode(self):
2259
 
        """Test that run_bzr passes retcode to _run_bzr_core"""
2260
 
        # Default is retcode == 0
2261
 
        self.run_bzr('foo bar')
2262
 
        self.assertEqual(0, self.retcode)
2263
 
        self.assertEqual(['foo', 'bar'], self.argv)
2264
 
 
2265
 
        self.run_bzr('foo bar', retcode=1)
2266
 
        self.assertEqual(1, self.retcode)
2267
 
        self.assertEqual(['foo', 'bar'], self.argv)
2268
 
 
2269
 
        self.run_bzr('foo bar', retcode=None)
2270
 
        self.assertEqual(None, self.retcode)
2271
 
        self.assertEqual(['foo', 'bar'], self.argv)
2272
 
 
2273
 
        self.run_bzr(['foo', 'bar'], retcode=3)
2274
 
        self.assertEqual(3, self.retcode)
2275
 
        self.assertEqual(['foo', 'bar'], self.argv)
2276
 
 
2277
 
    def test_stdin(self):
2278
 
        # test that the stdin keyword to run_bzr is passed through to
2279
 
        # _run_bzr_core as-is. We do this by overriding
2280
 
        # _run_bzr_core in this class, and then calling run_bzr,
2281
 
        # which is a convenience function for _run_bzr_core, so
2282
 
        # should invoke it.
2283
 
        self.run_bzr('foo bar', stdin='gam')
2284
 
        self.assertEqual('gam', self.stdin)
2285
 
        self.assertEqual(['foo', 'bar'], self.argv)
2286
 
 
2287
 
        self.run_bzr('foo bar', stdin='zippy')
2288
 
        self.assertEqual('zippy', self.stdin)
2289
 
        self.assertEqual(['foo', 'bar'], self.argv)
2290
 
 
2291
 
    def test_working_dir(self):
2292
 
        """Test that run_bzr passes working_dir to _run_bzr_core"""
2293
 
        self.run_bzr('foo bar')
2294
 
        self.assertEqual(None, self.working_dir)
2295
 
        self.assertEqual(['foo', 'bar'], self.argv)
2296
 
 
2297
 
        self.run_bzr('foo bar', working_dir='baz')
2298
 
        self.assertEqual('baz', self.working_dir)
2299
 
        self.assertEqual(['foo', 'bar'], self.argv)
2300
 
 
2301
 
    def test_reject_extra_keyword_arguments(self):
2302
 
        self.assertRaises(TypeError, self.run_bzr, "foo bar",
2303
 
                          error_regex=['error message'])
2304
 
 
2305
 
 
2306
 
class TestRunBzrCaptured(tests.TestCaseWithTransport):
2307
 
    # Does IO when testing the working_dir parameter.
2308
 
 
2309
 
    def apply_redirected(self, stdin=None, stdout=None, stderr=None,
2310
 
                         a_callable=None, *args, **kwargs):
2311
 
        self.stdin = stdin
2312
 
        self.factory_stdin = getattr(bzrlib.ui.ui_factory, "stdin", None)
2313
 
        self.factory = bzrlib.ui.ui_factory
2314
 
        self.working_dir = osutils.getcwd()
2315
 
        stdout.write('foo\n')
2316
 
        stderr.write('bar\n')
2317
 
        return 0
2318
 
 
2319
 
    def test_stdin(self):
2320
 
        # test that the stdin keyword to _run_bzr_core is passed through to
2321
 
        # apply_redirected as a StringIO. We do this by overriding
2322
 
        # apply_redirected in this class, and then calling _run_bzr_core,
2323
 
        # which calls apply_redirected.
2324
 
        self.run_bzr(['foo', 'bar'], stdin='gam')
2325
 
        self.assertEqual('gam', self.stdin.read())
2326
 
        self.assertTrue(self.stdin is self.factory_stdin)
2327
 
        self.run_bzr(['foo', 'bar'], stdin='zippy')
2328
 
        self.assertEqual('zippy', self.stdin.read())
2329
 
        self.assertTrue(self.stdin is self.factory_stdin)
2330
 
 
2331
 
    def test_ui_factory(self):
2332
 
        # each invocation of self.run_bzr should get its
2333
 
        # own UI factory, which is an instance of TestUIFactory,
2334
 
        # with stdin, stdout and stderr attached to the stdin,
2335
 
        # stdout and stderr of the invoked run_bzr
2336
 
        current_factory = bzrlib.ui.ui_factory
2337
 
        self.run_bzr(['foo'])
2338
 
        self.assertFalse(current_factory is self.factory)
2339
 
        self.assertNotEqual(sys.stdout, self.factory.stdout)
2340
 
        self.assertNotEqual(sys.stderr, self.factory.stderr)
2341
 
        self.assertEqual('foo\n', self.factory.stdout.getvalue())
2342
 
        self.assertEqual('bar\n', self.factory.stderr.getvalue())
2343
 
        self.assertIsInstance(self.factory, tests.TestUIFactory)
2344
 
 
2345
 
    def test_working_dir(self):
2346
 
        self.build_tree(['one/', 'two/'])
2347
 
        cwd = osutils.getcwd()
2348
 
 
2349
 
        # Default is to work in the current directory
2350
 
        self.run_bzr(['foo', 'bar'])
2351
 
        self.assertEqual(cwd, self.working_dir)
2352
 
 
2353
 
        self.run_bzr(['foo', 'bar'], working_dir=None)
2354
 
        self.assertEqual(cwd, self.working_dir)
2355
 
 
2356
 
        # The function should be run in the alternative directory
2357
 
        # but afterwards the current working dir shouldn't be changed
2358
 
        self.run_bzr(['foo', 'bar'], working_dir='one')
2359
 
        self.assertNotEqual(cwd, self.working_dir)
2360
 
        self.assertEndsWith(self.working_dir, 'one')
2361
 
        self.assertEqual(cwd, osutils.getcwd())
2362
 
 
2363
 
        self.run_bzr(['foo', 'bar'], working_dir='two')
2364
 
        self.assertNotEqual(cwd, self.working_dir)
2365
 
        self.assertEndsWith(self.working_dir, 'two')
2366
 
        self.assertEqual(cwd, osutils.getcwd())
2367
 
 
2368
 
 
2369
 
class StubProcess(object):
2370
 
    """A stub process for testing run_bzr_subprocess."""
2371
 
    
2372
 
    def __init__(self, out="", err="", retcode=0):
2373
 
        self.out = out
2374
 
        self.err = err
2375
 
        self.returncode = retcode
2376
 
 
2377
 
    def communicate(self):
2378
 
        return self.out, self.err
2379
 
 
2380
 
 
2381
 
class TestWithFakedStartBzrSubprocess(tests.TestCaseWithTransport):
2382
 
    """Base class for tests testing how we might run bzr."""
2383
 
 
2384
 
    def setUp(self):
2385
 
        tests.TestCaseWithTransport.setUp(self)
2386
 
        self.subprocess_calls = []
2387
 
 
2388
 
    def start_bzr_subprocess(self, process_args, env_changes=None,
2389
 
                             skip_if_plan_to_signal=False,
2390
 
                             working_dir=None,
2391
 
                             allow_plugins=False):
2392
 
        """capture what run_bzr_subprocess tries to do."""
2393
 
        self.subprocess_calls.append({'process_args':process_args,
2394
 
            'env_changes':env_changes,
2395
 
            'skip_if_plan_to_signal':skip_if_plan_to_signal,
2396
 
            'working_dir':working_dir, 'allow_plugins':allow_plugins})
2397
 
        return self.next_subprocess
2398
 
 
2399
 
 
2400
 
class TestRunBzrSubprocess(TestWithFakedStartBzrSubprocess):
2401
 
 
2402
 
    def assertRunBzrSubprocess(self, expected_args, process, *args, **kwargs):
2403
 
        """Run run_bzr_subprocess with args and kwargs using a stubbed process.
2404
 
 
2405
 
        Inside TestRunBzrSubprocessCommands we use a stub start_bzr_subprocess
2406
 
        that will return static results. This assertion method populates those
2407
 
        results and also checks the arguments run_bzr_subprocess generates.
2408
 
        """
2409
 
        self.next_subprocess = process
2410
 
        try:
2411
 
            result = self.run_bzr_subprocess(*args, **kwargs)
2412
 
        except:
2413
 
            self.next_subprocess = None
2414
 
            for key, expected in expected_args.iteritems():
2415
 
                self.assertEqual(expected, self.subprocess_calls[-1][key])
2416
 
            raise
2417
 
        else:
2418
 
            self.next_subprocess = None
2419
 
            for key, expected in expected_args.iteritems():
2420
 
                self.assertEqual(expected, self.subprocess_calls[-1][key])
2421
 
            return result
2422
 
 
2423
 
    def test_run_bzr_subprocess(self):
2424
 
        """The run_bzr_helper_external command behaves nicely."""
2425
 
        self.assertRunBzrSubprocess({'process_args':['--version']},
2426
 
            StubProcess(), '--version')
2427
 
        self.assertRunBzrSubprocess({'process_args':['--version']},
2428
 
            StubProcess(), ['--version'])
2429
 
        # retcode=None disables retcode checking
2430
 
        result = self.assertRunBzrSubprocess({},
2431
 
            StubProcess(retcode=3), '--version', retcode=None)
2432
 
        result = self.assertRunBzrSubprocess({},
2433
 
            StubProcess(out="is free software"), '--version')
2434
 
        self.assertContainsRe(result[0], 'is free software')
2435
 
        # Running a subcommand that is missing errors
2436
 
        self.assertRaises(AssertionError, self.assertRunBzrSubprocess,
2437
 
            {'process_args':['--versionn']}, StubProcess(retcode=3),
2438
 
            '--versionn')
2439
 
        # Unless it is told to expect the error from the subprocess
2440
 
        result = self.assertRunBzrSubprocess({},
2441
 
            StubProcess(retcode=3), '--versionn', retcode=3)
2442
 
        # Or to ignore retcode checking
2443
 
        result = self.assertRunBzrSubprocess({},
2444
 
            StubProcess(err="unknown command", retcode=3), '--versionn',
2445
 
            retcode=None)
2446
 
        self.assertContainsRe(result[1], 'unknown command')
2447
 
 
2448
 
    def test_env_change_passes_through(self):
2449
 
        self.assertRunBzrSubprocess(
2450
 
            {'env_changes':{'new':'value', 'changed':'newvalue', 'deleted':None}},
2451
 
            StubProcess(), '',
2452
 
            env_changes={'new':'value', 'changed':'newvalue', 'deleted':None})
2453
 
 
2454
 
    def test_no_working_dir_passed_as_None(self):
2455
 
        self.assertRunBzrSubprocess({'working_dir': None}, StubProcess(), '')
2456
 
 
2457
 
    def test_no_working_dir_passed_through(self):
2458
 
        self.assertRunBzrSubprocess({'working_dir': 'dir'}, StubProcess(), '',
2459
 
            working_dir='dir')
2460
 
 
2461
 
    def test_run_bzr_subprocess_no_plugins(self):
2462
 
        self.assertRunBzrSubprocess({'allow_plugins': False},
2463
 
            StubProcess(), '')
2464
 
 
2465
 
    def test_allow_plugins(self):
2466
 
        self.assertRunBzrSubprocess({'allow_plugins': True},
2467
 
            StubProcess(), '', allow_plugins=True)
2468
 
 
2469
 
 
2470
 
class TestFinishBzrSubprocess(TestWithFakedStartBzrSubprocess):
2471
 
 
2472
 
    def test_finish_bzr_subprocess_with_error(self):
2473
 
        """finish_bzr_subprocess allows specification of the desired exit code.
2474
 
        """
2475
 
        process = StubProcess(err="unknown command", retcode=3)
2476
 
        result = self.finish_bzr_subprocess(process, retcode=3)
2477
 
        self.assertEqual('', result[0])
2478
 
        self.assertContainsRe(result[1], 'unknown command')
2479
 
 
2480
 
    def test_finish_bzr_subprocess_ignoring_retcode(self):
2481
 
        """finish_bzr_subprocess allows the exit code to be ignored."""
2482
 
        process = StubProcess(err="unknown command", retcode=3)
2483
 
        result = self.finish_bzr_subprocess(process, retcode=None)
2484
 
        self.assertEqual('', result[0])
2485
 
        self.assertContainsRe(result[1], 'unknown command')
2486
 
 
2487
 
    def test_finish_subprocess_with_unexpected_retcode(self):
2488
 
        """finish_bzr_subprocess raises self.failureException if the retcode is
2489
 
        not the expected one.
2490
 
        """
2491
 
        process = StubProcess(err="unknown command", retcode=3)
2492
 
        self.assertRaises(self.failureException, self.finish_bzr_subprocess,
2493
 
                          process)
2494
 
 
2495
 
 
2496
 
class _DontSpawnProcess(Exception):
2497
 
    """A simple exception which just allows us to skip unnecessary steps"""
2498
 
 
2499
 
 
2500
 
class TestStartBzrSubProcess(tests.TestCase):
2501
 
    """Stub test start_bzr_subprocess."""
2502
 
 
2503
 
    def _subprocess_log_cleanup(self):
2504
 
        """Inhibits the base version as we don't produce a log file."""
2505
 
 
2506
 
    def _popen(self, *args, **kwargs):
2507
 
        """Override the base version to record the command that is run.
2508
 
 
2509
 
        From there we can ensure it is correct without spawning a real process.
2510
 
        """
2511
 
        self.check_popen_state()
2512
 
        self._popen_args = args
2513
 
        self._popen_kwargs = kwargs
2514
 
        raise _DontSpawnProcess()
2515
 
 
2516
 
    def check_popen_state(self):
2517
 
        """Replace to make assertions when popen is called."""
2518
 
 
2519
 
    def test_run_bzr_subprocess_no_plugins(self):
2520
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [])
2521
 
        command = self._popen_args[0]
2522
 
        self.assertEqual(sys.executable, command[0])
2523
 
        self.assertEqual(self.get_bzr_path(), command[1])
2524
 
        self.assertEqual(['--no-plugins'], command[2:])
2525
 
 
2526
 
    def test_allow_plugins(self):
2527
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2528
 
                          allow_plugins=True)
2529
 
        command = self._popen_args[0]
2530
 
        self.assertEqual([], command[2:])
2531
 
 
2532
 
    def test_set_env(self):
2533
 
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2534
 
        # set in the child
2535
 
        def check_environment():
2536
 
            self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2537
 
        self.check_popen_state = check_environment
2538
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2539
 
                          env_changes={'EXISTANT_ENV_VAR':'set variable'})
2540
 
        # not set in theparent
2541
 
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2542
 
 
2543
 
    def test_run_bzr_subprocess_env_del(self):
2544
 
        """run_bzr_subprocess can remove environment variables too."""
2545
 
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2546
 
        def check_environment():
2547
 
            self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2548
 
        os.environ['EXISTANT_ENV_VAR'] = 'set variable'
2549
 
        self.check_popen_state = check_environment
2550
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2551
 
                          env_changes={'EXISTANT_ENV_VAR':None})
2552
 
        # Still set in parent
2553
 
        self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2554
 
        del os.environ['EXISTANT_ENV_VAR']
2555
 
 
2556
 
    def test_env_del_missing(self):
2557
 
        self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2558
 
        def check_environment():
2559
 
            self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2560
 
        self.check_popen_state = check_environment
2561
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2562
 
                          env_changes={'NON_EXISTANT_ENV_VAR':None})
2563
 
 
2564
 
    def test_working_dir(self):
2565
 
        """Test that we can specify the working dir for the child"""
2566
 
        orig_getcwd = osutils.getcwd
2567
 
        orig_chdir = os.chdir
2568
 
        chdirs = []
2569
 
        def chdir(path):
2570
 
            chdirs.append(path)
2571
 
        self.overrideAttr(os, 'chdir', chdir)
2572
 
        def getcwd():
2573
 
            return 'current'
2574
 
        self.overrideAttr(osutils, 'getcwd', getcwd)
2575
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2576
 
                          working_dir='foo')
2577
 
        self.assertEqual(['foo', 'current'], chdirs)
2578
 
 
2579
 
    def test_get_bzr_path_with_cwd_bzrlib(self):
2580
 
        self.get_source_path = lambda: ""
2581
 
        self.overrideAttr(os.path, "isfile", lambda path: True)
2582
 
        self.assertEqual(self.get_bzr_path(), "bzr")
2583
 
 
2584
 
 
2585
 
class TestActuallyStartBzrSubprocess(tests.TestCaseWithTransport):
2586
 
    """Tests that really need to do things with an external bzr."""
2587
 
 
2588
 
    def test_start_and_stop_bzr_subprocess_send_signal(self):
2589
 
        """finish_bzr_subprocess raises self.failureException if the retcode is
2590
 
        not the expected one.
2591
 
        """
2592
 
        self.disable_missing_extensions_warning()
2593
 
        process = self.start_bzr_subprocess(['wait-until-signalled'],
2594
 
                                            skip_if_plan_to_signal=True)
2595
 
        self.assertEqual('running\n', process.stdout.readline())
2596
 
        result = self.finish_bzr_subprocess(process, send_signal=signal.SIGINT,
2597
 
                                            retcode=3)
2598
 
        self.assertEqual('', result[0])
2599
 
        self.assertEqual('bzr: interrupted\n', result[1])
2600
 
 
2601
 
 
2602
 
class TestSelftestFiltering(tests.TestCase):
2603
 
 
2604
 
    def setUp(self):
2605
 
        tests.TestCase.setUp(self)
2606
 
        self.suite = TestUtil.TestSuite()
2607
 
        self.loader = TestUtil.TestLoader()
2608
 
        self.suite.addTest(self.loader.loadTestsFromModule(
2609
 
            sys.modules['bzrlib.tests.test_selftest']))
2610
 
        self.all_names = _test_ids(self.suite)
2611
 
 
2612
 
    def test_condition_id_re(self):
2613
 
        test_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2614
 
            'test_condition_id_re')
2615
 
        filtered_suite = tests.filter_suite_by_condition(
2616
 
            self.suite, tests.condition_id_re('test_condition_id_re'))
2617
 
        self.assertEqual([test_name], _test_ids(filtered_suite))
2618
 
 
2619
 
    def test_condition_id_in_list(self):
2620
 
        test_names = ['bzrlib.tests.test_selftest.TestSelftestFiltering.'
2621
 
                      'test_condition_id_in_list']
2622
 
        id_list = tests.TestIdList(test_names)
2623
 
        filtered_suite = tests.filter_suite_by_condition(
2624
 
            self.suite, tests.condition_id_in_list(id_list))
2625
 
        my_pattern = 'TestSelftestFiltering.*test_condition_id_in_list'
2626
 
        re_filtered = tests.filter_suite_by_re(self.suite, my_pattern)
2627
 
        self.assertEqual(_test_ids(re_filtered), _test_ids(filtered_suite))
2628
 
 
2629
 
    def test_condition_id_startswith(self):
2630
 
        klass = 'bzrlib.tests.test_selftest.TestSelftestFiltering.'
2631
 
        start1 = klass + 'test_condition_id_starts'
2632
 
        start2 = klass + 'test_condition_id_in'
2633
 
        test_names = [ klass + 'test_condition_id_in_list',
2634
 
                      klass + 'test_condition_id_startswith',
2635
 
                     ]
2636
 
        filtered_suite = tests.filter_suite_by_condition(
2637
 
            self.suite, tests.condition_id_startswith([start1, start2]))
2638
 
        self.assertEqual(test_names, _test_ids(filtered_suite))
2639
 
 
2640
 
    def test_condition_isinstance(self):
2641
 
        filtered_suite = tests.filter_suite_by_condition(
2642
 
            self.suite, tests.condition_isinstance(self.__class__))
2643
 
        class_pattern = 'bzrlib.tests.test_selftest.TestSelftestFiltering.'
2644
 
        re_filtered = tests.filter_suite_by_re(self.suite, class_pattern)
2645
 
        self.assertEqual(_test_ids(re_filtered), _test_ids(filtered_suite))
2646
 
 
2647
 
    def test_exclude_tests_by_condition(self):
2648
 
        excluded_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2649
 
            'test_exclude_tests_by_condition')
2650
 
        filtered_suite = tests.exclude_tests_by_condition(self.suite,
2651
 
            lambda x:x.id() == excluded_name)
2652
 
        self.assertEqual(len(self.all_names) - 1,
2653
 
            filtered_suite.countTestCases())
2654
 
        self.assertFalse(excluded_name in _test_ids(filtered_suite))
2655
 
        remaining_names = list(self.all_names)
2656
 
        remaining_names.remove(excluded_name)
2657
 
        self.assertEqual(remaining_names, _test_ids(filtered_suite))
2658
 
 
2659
 
    def test_exclude_tests_by_re(self):
2660
 
        self.all_names = _test_ids(self.suite)
2661
 
        filtered_suite = tests.exclude_tests_by_re(self.suite,
2662
 
                                                   'exclude_tests_by_re')
2663
 
        excluded_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2664
 
            'test_exclude_tests_by_re')
2665
 
        self.assertEqual(len(self.all_names) - 1,
2666
 
            filtered_suite.countTestCases())
2667
 
        self.assertFalse(excluded_name in _test_ids(filtered_suite))
2668
 
        remaining_names = list(self.all_names)
2669
 
        remaining_names.remove(excluded_name)
2670
 
        self.assertEqual(remaining_names, _test_ids(filtered_suite))
2671
 
 
2672
 
    def test_filter_suite_by_condition(self):
2673
 
        test_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2674
 
            'test_filter_suite_by_condition')
2675
 
        filtered_suite = tests.filter_suite_by_condition(self.suite,
2676
 
            lambda x:x.id() == test_name)
2677
 
        self.assertEqual([test_name], _test_ids(filtered_suite))
2678
 
 
2679
 
    def test_filter_suite_by_re(self):
2680
 
        filtered_suite = tests.filter_suite_by_re(self.suite,
2681
 
                                                  'test_filter_suite_by_r')
2682
 
        filtered_names = _test_ids(filtered_suite)
2683
 
        self.assertEqual(filtered_names, ['bzrlib.tests.test_selftest.'
2684
 
            'TestSelftestFiltering.test_filter_suite_by_re'])
2685
 
 
2686
 
    def test_filter_suite_by_id_list(self):
2687
 
        test_list = ['bzrlib.tests.test_selftest.'
2688
 
                     'TestSelftestFiltering.test_filter_suite_by_id_list']
2689
 
        filtered_suite = tests.filter_suite_by_id_list(
2690
 
            self.suite, tests.TestIdList(test_list))
2691
 
        filtered_names = _test_ids(filtered_suite)
2692
 
        self.assertEqual(
2693
 
            filtered_names,
2694
 
            ['bzrlib.tests.test_selftest.'
2695
 
             'TestSelftestFiltering.test_filter_suite_by_id_list'])
2696
 
 
2697
 
    def test_filter_suite_by_id_startswith(self):
2698
 
        # By design this test may fail if another test is added whose name also
2699
 
        # begins with one of the start value used.
2700
 
        klass = 'bzrlib.tests.test_selftest.TestSelftestFiltering.'
2701
 
        start1 = klass + 'test_filter_suite_by_id_starts'
2702
 
        start2 = klass + 'test_filter_suite_by_id_li'
2703
 
        test_list = [klass + 'test_filter_suite_by_id_list',
2704
 
                     klass + 'test_filter_suite_by_id_startswith',
2705
 
                     ]
2706
 
        filtered_suite = tests.filter_suite_by_id_startswith(
2707
 
            self.suite, [start1, start2])
2708
 
        self.assertEqual(
2709
 
            test_list,
2710
 
            _test_ids(filtered_suite),
2711
 
            )
2712
 
 
2713
 
    def test_preserve_input(self):
2714
 
        # NB: Surely this is something in the stdlib to do this?
2715
 
        self.assertTrue(self.suite is tests.preserve_input(self.suite))
2716
 
        self.assertTrue("@#$" is tests.preserve_input("@#$"))
2717
 
 
2718
 
    def test_randomize_suite(self):
2719
 
        randomized_suite = tests.randomize_suite(self.suite)
2720
 
        # randomizing should not add or remove test names.
2721
 
        self.assertEqual(set(_test_ids(self.suite)),
2722
 
                         set(_test_ids(randomized_suite)))
2723
 
        # Technically, this *can* fail, because random.shuffle(list) can be
2724
 
        # equal to list. Trying multiple times just pushes the frequency back.
2725
 
        # As its len(self.all_names)!:1, the failure frequency should be low
2726
 
        # enough to ignore. RBC 20071021.
2727
 
        # It should change the order.
2728
 
        self.assertNotEqual(self.all_names, _test_ids(randomized_suite))
2729
 
        # But not the length. (Possibly redundant with the set test, but not
2730
 
        # necessarily.)
2731
 
        self.assertEqual(len(self.all_names), len(_test_ids(randomized_suite)))
2732
 
 
2733
 
    def test_split_suit_by_condition(self):
2734
 
        self.all_names = _test_ids(self.suite)
2735
 
        condition = tests.condition_id_re('test_filter_suite_by_r')
2736
 
        split_suite = tests.split_suite_by_condition(self.suite, condition)
2737
 
        filtered_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2738
 
            'test_filter_suite_by_re')
2739
 
        self.assertEqual([filtered_name], _test_ids(split_suite[0]))
2740
 
        self.assertFalse(filtered_name in _test_ids(split_suite[1]))
2741
 
        remaining_names = list(self.all_names)
2742
 
        remaining_names.remove(filtered_name)
2743
 
        self.assertEqual(remaining_names, _test_ids(split_suite[1]))
2744
 
 
2745
 
    def test_split_suit_by_re(self):
2746
 
        self.all_names = _test_ids(self.suite)
2747
 
        split_suite = tests.split_suite_by_re(self.suite,
2748
 
                                              'test_filter_suite_by_r')
2749
 
        filtered_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2750
 
            'test_filter_suite_by_re')
2751
 
        self.assertEqual([filtered_name], _test_ids(split_suite[0]))
2752
 
        self.assertFalse(filtered_name in _test_ids(split_suite[1]))
2753
 
        remaining_names = list(self.all_names)
2754
 
        remaining_names.remove(filtered_name)
2755
 
        self.assertEqual(remaining_names, _test_ids(split_suite[1]))
2756
 
 
2757
 
 
2758
 
class TestCheckTreeShape(tests.TestCaseWithTransport):
2759
 
 
2760
 
    def test_check_tree_shape(self):
2761
 
        files = ['a', 'b/', 'b/c']
2762
 
        tree = self.make_branch_and_tree('.')
2763
 
        self.build_tree(files)
2764
 
        tree.add(files)
2765
 
        tree.lock_read()
2766
 
        try:
2767
 
            self.check_tree_shape(tree, files)
2768
 
        finally:
2769
 
            tree.unlock()
2770
 
 
2771
 
 
2772
 
class TestBlackboxSupport(tests.TestCase):
2773
 
    """Tests for testsuite blackbox features."""
2774
 
 
2775
 
    def test_run_bzr_failure_not_caught(self):
2776
 
        # When we run bzr in blackbox mode, we want any unexpected errors to
2777
 
        # propagate up to the test suite so that it can show the error in the
2778
 
        # usual way, and we won't get a double traceback.
2779
 
        e = self.assertRaises(
2780
 
            AssertionError,
2781
 
            self.run_bzr, ['assert-fail'])
2782
 
        # make sure we got the real thing, not an error from somewhere else in
2783
 
        # the test framework
2784
 
        self.assertEquals('always fails', str(e))
2785
 
        # check that there's no traceback in the test log
2786
 
        self.assertNotContainsRe(self.get_log(), r'Traceback')
2787
 
 
2788
 
    def test_run_bzr_user_error_caught(self):
2789
 
        # Running bzr in blackbox mode, normal/expected/user errors should be
2790
 
        # caught in the regular way and turned into an error message plus exit
2791
 
        # code.
2792
 
        transport_server = memory.MemoryServer()
2793
 
        transport_server.start_server()
2794
 
        self.addCleanup(transport_server.stop_server)
2795
 
        url = transport_server.get_url()
2796
 
        self.permit_url(url)
2797
 
        out, err = self.run_bzr(["log", "%s/nonexistantpath" % url], retcode=3)
2798
 
        self.assertEqual(out, '')
2799
 
        self.assertContainsRe(err,
2800
 
            'bzr: ERROR: Not a branch: ".*nonexistantpath/".\n')
2801
 
 
2802
 
 
2803
 
class TestTestLoader(tests.TestCase):
2804
 
    """Tests for the test loader."""
2805
 
 
2806
 
    def _get_loader_and_module(self):
2807
 
        """Gets a TestLoader and a module with one test in it."""
2808
 
        loader = TestUtil.TestLoader()
2809
 
        module = {}
2810
 
        class Stub(tests.TestCase):
2811
 
            def test_foo(self):
2812
 
                pass
2813
 
        class MyModule(object):
2814
 
            pass
2815
 
        MyModule.a_class = Stub
2816
 
        module = MyModule()
2817
 
        return loader, module
2818
 
 
2819
 
    def test_module_no_load_tests_attribute_loads_classes(self):
2820
 
        loader, module = self._get_loader_and_module()
2821
 
        self.assertEqual(1, loader.loadTestsFromModule(module).countTestCases())
2822
 
 
2823
 
    def test_module_load_tests_attribute_gets_called(self):
2824
 
        loader, module = self._get_loader_and_module()
2825
 
        # 'self' is here because we're faking the module with a class. Regular
2826
 
        # load_tests do not need that :)
2827
 
        def load_tests(self, standard_tests, module, loader):
2828
 
            result = loader.suiteClass()
2829
 
            for test in tests.iter_suite_tests(standard_tests):
2830
 
                result.addTests([test, test])
2831
 
            return result
2832
 
        # add a load_tests() method which multiplies the tests from the module.
2833
 
        module.__class__.load_tests = load_tests
2834
 
        self.assertEqual(2, loader.loadTestsFromModule(module).countTestCases())
2835
 
 
2836
 
    def test_load_tests_from_module_name_smoke_test(self):
2837
 
        loader = TestUtil.TestLoader()
2838
 
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2839
 
        self.assertEquals(['bzrlib.tests.test_sampler.DemoTest.test_nothing'],
2840
 
                          _test_ids(suite))
2841
 
 
2842
 
    def test_load_tests_from_module_name_with_bogus_module_name(self):
2843
 
        loader = TestUtil.TestLoader()
2844
 
        self.assertRaises(ImportError, loader.loadTestsFromModuleName, 'bogus')
2845
 
 
2846
 
 
2847
 
class TestTestIdList(tests.TestCase):
2848
 
 
2849
 
    def _create_id_list(self, test_list):
2850
 
        return tests.TestIdList(test_list)
2851
 
 
2852
 
    def _create_suite(self, test_id_list):
2853
 
 
2854
 
        class Stub(tests.TestCase):
2855
 
            def test_foo(self):
2856
 
                pass
2857
 
 
2858
 
        def _create_test_id(id):
2859
 
            return lambda: id
2860
 
 
2861
 
        suite = TestUtil.TestSuite()
2862
 
        for id in test_id_list:
2863
 
            t  = Stub('test_foo')
2864
 
            t.id = _create_test_id(id)
2865
 
            suite.addTest(t)
2866
 
        return suite
2867
 
 
2868
 
    def _test_ids(self, test_suite):
2869
 
        """Get the ids for the tests in a test suite."""
2870
 
        return [t.id() for t in tests.iter_suite_tests(test_suite)]
2871
 
 
2872
 
    def test_empty_list(self):
2873
 
        id_list = self._create_id_list([])
2874
 
        self.assertEquals({}, id_list.tests)
2875
 
        self.assertEquals({}, id_list.modules)
2876
 
 
2877
 
    def test_valid_list(self):
2878
 
        id_list = self._create_id_list(
2879
 
            ['mod1.cl1.meth1', 'mod1.cl1.meth2',
2880
 
             'mod1.func1', 'mod1.cl2.meth2',
2881
 
             'mod1.submod1',
2882
 
             'mod1.submod2.cl1.meth1', 'mod1.submod2.cl2.meth2',
2883
 
             ])
2884
 
        self.assertTrue(id_list.refers_to('mod1'))
2885
 
        self.assertTrue(id_list.refers_to('mod1.submod1'))
2886
 
        self.assertTrue(id_list.refers_to('mod1.submod2'))
2887
 
        self.assertTrue(id_list.includes('mod1.cl1.meth1'))
2888
 
        self.assertTrue(id_list.includes('mod1.submod1'))
2889
 
        self.assertTrue(id_list.includes('mod1.func1'))
2890
 
 
2891
 
    def test_bad_chars_in_params(self):
2892
 
        id_list = self._create_id_list(['mod1.cl1.meth1(xx.yy)'])
2893
 
        self.assertTrue(id_list.refers_to('mod1'))
2894
 
        self.assertTrue(id_list.includes('mod1.cl1.meth1(xx.yy)'))
2895
 
 
2896
 
    def test_module_used(self):
2897
 
        id_list = self._create_id_list(['mod.class.meth'])
2898
 
        self.assertTrue(id_list.refers_to('mod'))
2899
 
        self.assertTrue(id_list.refers_to('mod.class'))
2900
 
        self.assertTrue(id_list.refers_to('mod.class.meth'))
2901
 
 
2902
 
    def test_test_suite_matches_id_list_with_unknown(self):
2903
 
        loader = TestUtil.TestLoader()
2904
 
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2905
 
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing',
2906
 
                     'bogus']
2907
 
        not_found, duplicates = tests.suite_matches_id_list(suite, test_list)
2908
 
        self.assertEquals(['bogus'], not_found)
2909
 
        self.assertEquals([], duplicates)
2910
 
 
2911
 
    def test_suite_matches_id_list_with_duplicates(self):
2912
 
        loader = TestUtil.TestLoader()
2913
 
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2914
 
        dupes = loader.suiteClass()
2915
 
        for test in tests.iter_suite_tests(suite):
2916
 
            dupes.addTest(test)
2917
 
            dupes.addTest(test) # Add it again
2918
 
 
2919
 
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing',]
2920
 
        not_found, duplicates = tests.suite_matches_id_list(
2921
 
            dupes, test_list)
2922
 
        self.assertEquals([], not_found)
2923
 
        self.assertEquals(['bzrlib.tests.test_sampler.DemoTest.test_nothing'],
2924
 
                          duplicates)
2925
 
 
2926
 
 
2927
 
class TestTestSuite(tests.TestCase):
2928
 
 
2929
 
    def test__test_suite_testmod_names(self):
2930
 
        # Test that a plausible list of test module names are returned
2931
 
        # by _test_suite_testmod_names.
2932
 
        test_list = tests._test_suite_testmod_names()
2933
 
        self.assertSubset([
2934
 
            'bzrlib.tests.blackbox',
2935
 
            'bzrlib.tests.per_transport',
2936
 
            'bzrlib.tests.test_selftest',
2937
 
            ],
2938
 
            test_list)
2939
 
 
2940
 
    def test__test_suite_modules_to_doctest(self):
2941
 
        # Test that a plausible list of modules to doctest is returned
2942
 
        # by _test_suite_modules_to_doctest.
2943
 
        test_list = tests._test_suite_modules_to_doctest()
2944
 
        if __doc__ is None:
2945
 
            # When docstrings are stripped, there are no modules to doctest
2946
 
            self.assertEqual([], test_list)
2947
 
            return
2948
 
        self.assertSubset([
2949
 
            'bzrlib.timestamp',
2950
 
            ],
2951
 
            test_list)
2952
 
 
2953
 
    def test_test_suite(self):
2954
 
        # test_suite() loads the entire test suite to operate. To avoid this
2955
 
        # overhead, and yet still be confident that things are happening,
2956
 
        # we temporarily replace two functions used by test_suite with 
2957
 
        # test doubles that supply a few sample tests to load, and check they
2958
 
        # are loaded.
2959
 
        calls = []
2960
 
        def testmod_names():
2961
 
            calls.append("testmod_names")
2962
 
            return [
2963
 
                'bzrlib.tests.blackbox.test_branch',
2964
 
                'bzrlib.tests.per_transport',
2965
 
                'bzrlib.tests.test_selftest',
2966
 
                ]
2967
 
        self.overrideAttr(tests, '_test_suite_testmod_names', testmod_names)
2968
 
        def doctests():
2969
 
            calls.append("modules_to_doctest")
2970
 
            if __doc__ is None:
2971
 
                return []
2972
 
            return ['bzrlib.timestamp']
2973
 
        self.overrideAttr(tests, '_test_suite_modules_to_doctest', doctests)
2974
 
        expected_test_list = [
2975
 
            # testmod_names
2976
 
            'bzrlib.tests.blackbox.test_branch.TestBranch.test_branch',
2977
 
            ('bzrlib.tests.per_transport.TransportTests'
2978
 
             '.test_abspath(LocalTransport,LocalURLServer)'),
2979
 
            'bzrlib.tests.test_selftest.TestTestSuite.test_test_suite',
2980
 
            # plugins can't be tested that way since selftest may be run with
2981
 
            # --no-plugins
2982
 
            ]
2983
 
        if __doc__ is not None:
2984
 
            expected_test_list.extend([
2985
 
                # modules_to_doctest
2986
 
                'bzrlib.timestamp.format_highres_date',
2987
 
                ])
2988
 
        suite = tests.test_suite()
2989
 
        self.assertEqual(set(["testmod_names", "modules_to_doctest"]),
2990
 
            set(calls))
2991
 
        self.assertSubset(expected_test_list, _test_ids(suite))
2992
 
 
2993
 
    def test_test_suite_list_and_start(self):
2994
 
        # We cannot test this at the same time as the main load, because we want
2995
 
        # to know that starting_with == None works. So a second load is
2996
 
        # incurred - note that the starting_with parameter causes a partial load
2997
 
        # rather than a full load so this test should be pretty quick.
2998
 
        test_list = ['bzrlib.tests.test_selftest.TestTestSuite.test_test_suite']
2999
 
        suite = tests.test_suite(test_list,
3000
 
                                 ['bzrlib.tests.test_selftest.TestTestSuite'])
3001
 
        # test_test_suite_list_and_start is not included 
3002
 
        self.assertEquals(test_list, _test_ids(suite))
3003
 
 
3004
 
 
3005
 
class TestLoadTestIdList(tests.TestCaseInTempDir):
3006
 
 
3007
 
    def _create_test_list_file(self, file_name, content):
3008
 
        fl = open(file_name, 'wt')
3009
 
        fl.write(content)
3010
 
        fl.close()
3011
 
 
3012
 
    def test_load_unknown(self):
3013
 
        self.assertRaises(errors.NoSuchFile,
3014
 
                          tests.load_test_id_list, 'i_do_not_exist')
3015
 
 
3016
 
    def test_load_test_list(self):
3017
 
        test_list_fname = 'test.list'
3018
 
        self._create_test_list_file(test_list_fname,
3019
 
                                    'mod1.cl1.meth1\nmod2.cl2.meth2\n')
3020
 
        tlist = tests.load_test_id_list(test_list_fname)
3021
 
        self.assertEquals(2, len(tlist))
3022
 
        self.assertEquals('mod1.cl1.meth1', tlist[0])
3023
 
        self.assertEquals('mod2.cl2.meth2', tlist[1])
3024
 
 
3025
 
    def test_load_dirty_file(self):
3026
 
        test_list_fname = 'test.list'
3027
 
        self._create_test_list_file(test_list_fname,
3028
 
                                    '  mod1.cl1.meth1\n\nmod2.cl2.meth2  \n'
3029
 
                                    'bar baz\n')
3030
 
        tlist = tests.load_test_id_list(test_list_fname)
3031
 
        self.assertEquals(4, len(tlist))
3032
 
        self.assertEquals('mod1.cl1.meth1', tlist[0])
3033
 
        self.assertEquals('', tlist[1])
3034
 
        self.assertEquals('mod2.cl2.meth2', tlist[2])
3035
 
        self.assertEquals('bar baz', tlist[3])
3036
 
 
3037
 
 
3038
 
class TestFilteredByModuleTestLoader(tests.TestCase):
3039
 
 
3040
 
    def _create_loader(self, test_list):
3041
 
        id_filter = tests.TestIdList(test_list)
3042
 
        loader = TestUtil.FilteredByModuleTestLoader(id_filter.refers_to)
3043
 
        return loader
3044
 
 
3045
 
    def test_load_tests(self):
3046
 
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing']
3047
 
        loader = self._create_loader(test_list)
3048
 
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
3049
 
        self.assertEquals(test_list, _test_ids(suite))
3050
 
 
3051
 
    def test_exclude_tests(self):
3052
 
        test_list = ['bogus']
3053
 
        loader = self._create_loader(test_list)
3054
 
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
3055
 
        self.assertEquals([], _test_ids(suite))
3056
 
 
3057
 
 
3058
 
class TestFilteredByNameStartTestLoader(tests.TestCase):
3059
 
 
3060
 
    def _create_loader(self, name_start):
3061
 
        def needs_module(name):
3062
 
            return name.startswith(name_start) or name_start.startswith(name)
3063
 
        loader = TestUtil.FilteredByModuleTestLoader(needs_module)
3064
 
        return loader
3065
 
 
3066
 
    def test_load_tests(self):
3067
 
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing']
3068
 
        loader = self._create_loader('bzrlib.tests.test_samp')
3069
 
 
3070
 
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
3071
 
        self.assertEquals(test_list, _test_ids(suite))
3072
 
 
3073
 
    def test_load_tests_inside_module(self):
3074
 
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing']
3075
 
        loader = self._create_loader('bzrlib.tests.test_sampler.Demo')
3076
 
 
3077
 
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
3078
 
        self.assertEquals(test_list, _test_ids(suite))
3079
 
 
3080
 
    def test_exclude_tests(self):
3081
 
        test_list = ['bogus']
3082
 
        loader = self._create_loader('bogus')
3083
 
 
3084
 
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
3085
 
        self.assertEquals([], _test_ids(suite))
3086
 
 
3087
 
 
3088
 
class TestTestPrefixRegistry(tests.TestCase):
3089
 
 
3090
 
    def _get_registry(self):
3091
 
        tp_registry = tests.TestPrefixAliasRegistry()
3092
 
        return tp_registry
3093
 
 
3094
 
    def test_register_new_prefix(self):
3095
 
        tpr = self._get_registry()
3096
 
        tpr.register('foo', 'fff.ooo.ooo')
3097
 
        self.assertEquals('fff.ooo.ooo', tpr.get('foo'))
3098
 
 
3099
 
    def test_register_existing_prefix(self):
3100
 
        tpr = self._get_registry()
3101
 
        tpr.register('bar', 'bbb.aaa.rrr')
3102
 
        tpr.register('bar', 'bBB.aAA.rRR')
3103
 
        self.assertEquals('bbb.aaa.rrr', tpr.get('bar'))
3104
 
        self.assertThat(self.get_log(),
3105
 
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR",
3106
 
                           doctest.ELLIPSIS))
3107
 
 
3108
 
    def test_get_unknown_prefix(self):
3109
 
        tpr = self._get_registry()
3110
 
        self.assertRaises(KeyError, tpr.get, 'I am not a prefix')
3111
 
 
3112
 
    def test_resolve_prefix(self):
3113
 
        tpr = self._get_registry()
3114
 
        tpr.register('bar', 'bb.aa.rr')
3115
 
        self.assertEquals('bb.aa.rr', tpr.resolve_alias('bar'))
3116
 
 
3117
 
    def test_resolve_unknown_alias(self):
3118
 
        tpr = self._get_registry()
3119
 
        self.assertRaises(errors.BzrCommandError,
3120
 
                          tpr.resolve_alias, 'I am not a prefix')
3121
 
 
3122
 
    def test_predefined_prefixes(self):
3123
 
        tpr = tests.test_prefix_alias_registry
3124
 
        self.assertEquals('bzrlib', tpr.resolve_alias('bzrlib'))
3125
 
        self.assertEquals('bzrlib.doc', tpr.resolve_alias('bd'))
3126
 
        self.assertEquals('bzrlib.utils', tpr.resolve_alias('bu'))
3127
 
        self.assertEquals('bzrlib.tests', tpr.resolve_alias('bt'))
3128
 
        self.assertEquals('bzrlib.tests.blackbox', tpr.resolve_alias('bb'))
3129
 
        self.assertEquals('bzrlib.plugins', tpr.resolve_alias('bp'))
3130
 
 
3131
 
 
3132
 
class TestThreadLeakDetection(tests.TestCase):
3133
 
    """Ensure when tests leak threads we detect and report it"""
3134
 
 
3135
 
    class LeakRecordingResult(tests.ExtendedTestResult):
3136
 
        def __init__(self):
3137
 
            tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
3138
 
            self.leaks = []
3139
 
        def _report_thread_leak(self, test, leaks, alive):
3140
 
            self.leaks.append((test, leaks))
3141
 
 
3142
 
    def test_testcase_without_addCleanups(self):
3143
 
        """Check old TestCase instances don't break with leak detection"""
3144
 
        class Test(unittest.TestCase):
3145
 
            def runTest(self):
3146
 
                pass
3147
 
        result = self.LeakRecordingResult()
3148
 
        test = Test()
3149
 
        result.startTestRun()
3150
 
        test.run(result)
3151
 
        result.stopTestRun()
3152
 
        self.assertEqual(result._tests_leaking_threads_count, 0)
3153
 
        self.assertEqual(result.leaks, [])
3154
 
        
3155
 
    def test_thread_leak(self):
3156
 
        """Ensure a thread that outlives the running of a test is reported
3157
 
 
3158
 
        Uses a thread that blocks on an event, and is started by the inner
3159
 
        test case. As the thread outlives the inner case's run, it should be
3160
 
        detected as a leak, but the event is then set so that the thread can
3161
 
        be safely joined in cleanup so it's not leaked for real.
3162
 
        """
3163
 
        event = threading.Event()
3164
 
        thread = threading.Thread(name="Leaker", target=event.wait)
3165
 
        class Test(tests.TestCase):
3166
 
            def test_leak(self):
3167
 
                thread.start()
3168
 
        result = self.LeakRecordingResult()
3169
 
        test = Test("test_leak")
3170
 
        self.addCleanup(thread.join)
3171
 
        self.addCleanup(event.set)
3172
 
        result.startTestRun()
3173
 
        test.run(result)
3174
 
        result.stopTestRun()
3175
 
        self.assertEqual(result._tests_leaking_threads_count, 1)
3176
 
        self.assertEqual(result._first_thread_leaker_id, test.id())
3177
 
        self.assertEqual(result.leaks, [(test, set([thread]))])
3178
 
        self.assertContainsString(result.stream.getvalue(), "leaking threads")
3179
 
 
3180
 
    def test_multiple_leaks(self):
3181
 
        """Check multiple leaks are blamed on the test cases at fault
3182
 
 
3183
 
        Same concept as the previous test, but has one inner test method that
3184
 
        leaks two threads, and one that doesn't leak at all.
3185
 
        """
3186
 
        event = threading.Event()
3187
 
        thread_a = threading.Thread(name="LeakerA", target=event.wait)
3188
 
        thread_b = threading.Thread(name="LeakerB", target=event.wait)
3189
 
        thread_c = threading.Thread(name="LeakerC", target=event.wait)
3190
 
        class Test(tests.TestCase):
3191
 
            def test_first_leak(self):
3192
 
                thread_b.start()
3193
 
            def test_second_no_leak(self):
3194
 
                pass
3195
 
            def test_third_leak(self):
3196
 
                thread_c.start()
3197
 
                thread_a.start()
3198
 
        result = self.LeakRecordingResult()
3199
 
        first_test = Test("test_first_leak")
3200
 
        third_test = Test("test_third_leak")
3201
 
        self.addCleanup(thread_a.join)
3202
 
        self.addCleanup(thread_b.join)
3203
 
        self.addCleanup(thread_c.join)
3204
 
        self.addCleanup(event.set)
3205
 
        result.startTestRun()
3206
 
        unittest.TestSuite(
3207
 
            [first_test, Test("test_second_no_leak"), third_test]
3208
 
            ).run(result)
3209
 
        result.stopTestRun()
3210
 
        self.assertEqual(result._tests_leaking_threads_count, 2)
3211
 
        self.assertEqual(result._first_thread_leaker_id, first_test.id())
3212
 
        self.assertEqual(result.leaks, [
3213
 
            (first_test, set([thread_b])),
3214
 
            (third_test, set([thread_a, thread_c]))])
3215
 
        self.assertContainsString(result.stream.getvalue(), "leaking threads")
3216
 
 
3217
 
 
3218
 
class TestPostMortemDebugging(tests.TestCase):
3219
 
    """Check post mortem debugging works when tests fail or error"""
3220
 
 
3221
 
    class TracebackRecordingResult(tests.ExtendedTestResult):
3222
 
        def __init__(self):
3223
 
            tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
3224
 
            self.postcode = None
3225
 
        def _post_mortem(self, tb=None):
3226
 
            """Record the code object at the end of the current traceback"""
3227
 
            tb = tb or sys.exc_info()[2]
3228
 
            if tb is not None:
3229
 
                next = tb.tb_next
3230
 
                while next is not None:
3231
 
                    tb = next
3232
 
                    next = next.tb_next
3233
 
                self.postcode = tb.tb_frame.f_code
3234
 
        def report_error(self, test, err):
3235
 
            pass
3236
 
        def report_failure(self, test, err):
3237
 
            pass
3238
 
 
3239
 
    def test_location_unittest_error(self):
3240
 
        """Needs right post mortem traceback with erroring unittest case"""
3241
 
        class Test(unittest.TestCase):
3242
 
            def runTest(self):
3243
 
                raise RuntimeError
3244
 
        result = self.TracebackRecordingResult()
3245
 
        Test().run(result)
3246
 
        self.assertEqual(result.postcode, Test.runTest.func_code)
3247
 
 
3248
 
    def test_location_unittest_failure(self):
3249
 
        """Needs right post mortem traceback with failing unittest case"""
3250
 
        class Test(unittest.TestCase):
3251
 
            def runTest(self):
3252
 
                raise self.failureException
3253
 
        result = self.TracebackRecordingResult()
3254
 
        Test().run(result)
3255
 
        self.assertEqual(result.postcode, Test.runTest.func_code)
3256
 
 
3257
 
    def test_location_bt_error(self):
3258
 
        """Needs right post mortem traceback with erroring bzrlib.tests case"""
3259
 
        class Test(tests.TestCase):
3260
 
            def test_error(self):
3261
 
                raise RuntimeError
3262
 
        result = self.TracebackRecordingResult()
3263
 
        Test("test_error").run(result)
3264
 
        self.assertEqual(result.postcode, Test.test_error.func_code)
3265
 
 
3266
 
    def test_location_bt_failure(self):
3267
 
        """Needs right post mortem traceback with failing bzrlib.tests case"""
3268
 
        class Test(tests.TestCase):
3269
 
            def test_failure(self):
3270
 
                raise self.failureException
3271
 
        result = self.TracebackRecordingResult()
3272
 
        Test("test_failure").run(result)
3273
 
        self.assertEqual(result.postcode, Test.test_failure.func_code)
3274
 
 
3275
 
    def test_env_var_triggers_post_mortem(self):
3276
 
        """Check pdb.post_mortem is called iff BZR_TEST_PDB is set"""
3277
 
        import pdb
3278
 
        result = tests.ExtendedTestResult(StringIO(), 0, 1)
3279
 
        post_mortem_calls = []
3280
 
        self.overrideAttr(pdb, "post_mortem", post_mortem_calls.append)
3281
 
        self.overrideEnv('BZR_TEST_PDB', None)
3282
 
        result._post_mortem(1)
3283
 
        self.overrideEnv('BZR_TEST_PDB', 'on')
3284
 
        result._post_mortem(2)
3285
 
        self.assertEqual([2], post_mortem_calls)
3286
 
 
3287
 
 
3288
 
class TestRunSuite(tests.TestCase):
3289
 
 
3290
 
    def test_runner_class(self):
3291
 
        """run_suite accepts and uses a runner_class keyword argument."""
3292
 
        class Stub(tests.TestCase):
3293
 
            def test_foo(self):
3294
 
                pass
3295
 
        suite = Stub("test_foo")
3296
 
        calls = []
3297
 
        class MyRunner(tests.TextTestRunner):
3298
 
            def run(self, test):
3299
 
                calls.append(test)
3300
 
                return tests.ExtendedTestResult(self.stream, self.descriptions,
3301
 
                                                self.verbosity)
3302
 
        tests.run_suite(suite, runner_class=MyRunner, stream=StringIO())
3303
 
        self.assertLength(1, calls)
3304
 
 
3305
 
 
3306
 
class _Selftest(object):
3307
 
    """Mixin for tests needing full selftest output"""
3308
 
 
3309
 
    def _inject_stream_into_subunit(self, stream):
3310
 
        """To be overridden by subclasses that run tests out of process"""
3311
 
 
3312
 
    def _run_selftest(self, **kwargs):
3313
 
        sio = StringIO()
3314
 
        self._inject_stream_into_subunit(sio)
3315
 
        tests.selftest(stream=sio, stop_on_failure=False, **kwargs)
3316
 
        return sio.getvalue()
3317
 
 
3318
 
 
3319
 
class _ForkedSelftest(_Selftest):
3320
 
    """Mixin for tests needing full selftest output with forked children"""
3321
 
 
3322
 
    _test_needs_features = [features.subunit]
3323
 
 
3324
 
    def _inject_stream_into_subunit(self, stream):
3325
 
        """Monkey-patch subunit so the extra output goes to stream not stdout
3326
 
 
3327
 
        Some APIs need rewriting so this kind of bogus hackery can be replaced
3328
 
        by passing the stream param from run_tests down into ProtocolTestCase.
3329
 
        """
3330
 
        from subunit import ProtocolTestCase
3331
 
        _original_init = ProtocolTestCase.__init__
3332
 
        def _init_with_passthrough(self, *args, **kwargs):
3333
 
            _original_init(self, *args, **kwargs)
3334
 
            self._passthrough = stream
3335
 
        self.overrideAttr(ProtocolTestCase, "__init__", _init_with_passthrough)
3336
 
 
3337
 
    def _run_selftest(self, **kwargs):
3338
 
        # GZ 2011-05-26: Add a PosixSystem feature so this check can go away
3339
 
        if getattr(os, "fork", None) is None:
3340
 
            raise tests.TestNotApplicable("Platform doesn't support forking")
3341
 
        # Make sure the fork code is actually invoked by claiming two cores
3342
 
        self.overrideAttr(osutils, "local_concurrency", lambda: 2)
3343
 
        kwargs.setdefault("suite_decorators", []).append(tests.fork_decorator)
3344
 
        return super(_ForkedSelftest, self)._run_selftest(**kwargs)
3345
 
 
3346
 
 
3347
 
class TestParallelFork(_ForkedSelftest, tests.TestCase):
3348
 
    """Check operation of --parallel=fork selftest option"""
3349
 
 
3350
 
    def test_error_in_child_during_fork(self):
3351
 
        """Error in a forked child during test setup should get reported"""
3352
 
        class Test(tests.TestCase):
3353
 
            def testMethod(self):
3354
 
                pass
3355
 
        # We don't care what, just break something that a child will run
3356
 
        self.overrideAttr(tests, "workaround_zealous_crypto_random", None)
3357
 
        out = self._run_selftest(test_suite_factory=Test)
3358
 
        # Lines from the tracebacks of the two child processes may be mixed
3359
 
        # together due to the way subunit parses and forwards the streams,
3360
 
        # so permit extra lines between each part of the error output.
3361
 
        self.assertContainsRe(out,
3362
 
            "Traceback.*:\n"
3363
 
            "(?:.*\n)*"
3364
 
            ".+ in fork_for_tests\n"
3365
 
            "(?:.*\n)*"
3366
 
            "\s*workaround_zealous_crypto_random\(\)\n"
3367
 
            "(?:.*\n)*"
3368
 
            "TypeError:")
3369
 
 
3370
 
 
3371
 
class TestUncollectedWarnings(_Selftest, tests.TestCase):
3372
 
    """Check a test case still alive after being run emits a warning"""
3373
 
 
3374
 
    class Test(tests.TestCase):
3375
 
        def test_pass(self):
3376
 
            pass
3377
 
        def test_self_ref(self):
3378
 
            self.also_self = self.test_self_ref
3379
 
        def test_skip(self):
3380
 
            self.skip("Don't need")
3381
 
 
3382
 
    def _get_suite(self):
3383
 
        return TestUtil.TestSuite([
3384
 
            self.Test("test_pass"),
3385
 
            self.Test("test_self_ref"),
3386
 
            self.Test("test_skip"),
3387
 
            ])
3388
 
 
3389
 
    def _run_selftest_with_suite(self, **kwargs):
3390
 
        old_flags = tests.selftest_debug_flags
3391
 
        tests.selftest_debug_flags = old_flags.union(["uncollected_cases"])
3392
 
        gc_on = gc.isenabled()
3393
 
        if gc_on:
3394
 
            gc.disable()
3395
 
        try:
3396
 
            output = self._run_selftest(test_suite_factory=self._get_suite,
3397
 
                **kwargs)
3398
 
        finally:
3399
 
            if gc_on:
3400
 
                gc.enable()
3401
 
            tests.selftest_debug_flags = old_flags
3402
 
        self.assertNotContainsRe(output, "Uncollected test case.*test_pass")
3403
 
        self.assertContainsRe(output, "Uncollected test case.*test_self_ref")
3404
 
        return output
3405
 
 
3406
 
    def test_testsuite(self):
3407
 
        self._run_selftest_with_suite()
3408
 
 
3409
 
    def test_pattern(self):
3410
 
        out = self._run_selftest_with_suite(pattern="test_(?:pass|self_ref)$")
3411
 
        self.assertNotContainsRe(out, "test_skip")
3412
 
 
3413
 
    def test_exclude_pattern(self):
3414
 
        out = self._run_selftest_with_suite(exclude_pattern="test_skip$")
3415
 
        self.assertNotContainsRe(out, "test_skip")
3416
 
 
3417
 
    def test_random_seed(self):
3418
 
        self._run_selftest_with_suite(random_seed="now")
3419
 
 
3420
 
    def test_matching_tests_first(self):
3421
 
        self._run_selftest_with_suite(matching_tests_first=True,
3422
 
            pattern="test_self_ref$")
3423
 
 
3424
 
    def test_starting_with_and_exclude(self):
3425
 
        out = self._run_selftest_with_suite(starting_with=["bt."],
3426
 
            exclude_pattern="test_skip$")
3427
 
        self.assertNotContainsRe(out, "test_skip")
3428
 
 
3429
 
    def test_additonal_decorator(self):
3430
 
        out = self._run_selftest_with_suite(
3431
 
            suite_decorators=[tests.TestDecorator])
3432
 
 
3433
 
 
3434
 
class TestUncollectedWarningsSubunit(TestUncollectedWarnings):
3435
 
    """Check warnings from tests staying alive are emitted with subunit"""
3436
 
 
3437
 
    _test_needs_features = [features.subunit]
3438
 
 
3439
 
    def _run_selftest_with_suite(self, **kwargs):
3440
 
        return TestUncollectedWarnings._run_selftest_with_suite(self,
3441
 
            runner_class=tests.SubUnitBzrRunner, **kwargs)
3442
 
 
3443
 
 
3444
 
class TestUncollectedWarningsForked(_ForkedSelftest, TestUncollectedWarnings):
3445
 
    """Check warnings from tests staying alive are emitted when forking"""
3446
 
 
3447
 
 
3448
 
class TestEnvironHandling(tests.TestCase):
3449
 
 
3450
 
    def test_overrideEnv_None_called_twice_doesnt_leak(self):
3451
 
        self.assertFalse('MYVAR' in os.environ)
3452
 
        self.overrideEnv('MYVAR', '42')
3453
 
        # We use an embedded test to make sure we fix the _captureVar bug
3454
 
        class Test(tests.TestCase):
3455
 
            def test_me(self):
3456
 
                # The first call save the 42 value
3457
 
                self.overrideEnv('MYVAR', None)
3458
 
                self.assertEquals(None, os.environ.get('MYVAR'))
3459
 
                # Make sure we can call it twice
3460
 
                self.overrideEnv('MYVAR', None)
3461
 
                self.assertEquals(None, os.environ.get('MYVAR'))
3462
 
        output = StringIO()
3463
 
        result = tests.TextTestResult(output, 0, 1)
3464
 
        Test('test_me').run(result)
3465
 
        if not result.wasStrictlySuccessful():
3466
 
            self.fail(output.getvalue())
3467
 
        # We get our value back
3468
 
        self.assertEquals('42', os.environ.get('MYVAR'))
3469
 
 
3470
 
 
3471
 
class TestIsolatedEnv(tests.TestCase):
3472
 
    """Test isolating tests from os.environ.
3473
 
 
3474
 
    Since we use tests that are already isolated from os.environ a bit of care
3475
 
    should be taken when designing the tests to avoid bootstrap side-effects.
3476
 
    The tests start an already clean os.environ which allow doing valid
3477
 
    assertions about which variables are present or not and design tests around
3478
 
    these assertions.
3479
 
    """
3480
 
 
3481
 
    class ScratchMonkey(tests.TestCase):
3482
 
 
3483
 
        def test_me(self):
3484
 
            pass
3485
 
 
3486
 
    def test_basics(self):
3487
 
        # Make sure we know the definition of BZR_HOME: not part of os.environ
3488
 
        # for tests.TestCase.
3489
 
        self.assertTrue('BZR_HOME' in tests.isolated_environ)
3490
 
        self.assertEquals(None, tests.isolated_environ['BZR_HOME'])
3491
 
        # Being part of isolated_environ, BZR_HOME should not appear here
3492
 
        self.assertFalse('BZR_HOME' in os.environ)
3493
 
        # Make sure we know the definition of LINES: part of os.environ for
3494
 
        # tests.TestCase
3495
 
        self.assertTrue('LINES' in tests.isolated_environ)
3496
 
        self.assertEquals('25', tests.isolated_environ['LINES'])
3497
 
        self.assertEquals('25', os.environ['LINES'])
3498
 
 
3499
 
    def test_injecting_unknown_variable(self):
3500
 
        # BZR_HOME is known to be absent from os.environ
3501
 
        test = self.ScratchMonkey('test_me')
3502
 
        tests.override_os_environ(test, {'BZR_HOME': 'foo'})
3503
 
        self.assertEquals('foo', os.environ['BZR_HOME'])
3504
 
        tests.restore_os_environ(test)
3505
 
        self.assertFalse('BZR_HOME' in os.environ)
3506
 
 
3507
 
    def test_injecting_known_variable(self):
3508
 
        test = self.ScratchMonkey('test_me')
3509
 
        # LINES is known to be present in os.environ
3510
 
        tests.override_os_environ(test, {'LINES': '42'})
3511
 
        self.assertEquals('42', os.environ['LINES'])
3512
 
        tests.restore_os_environ(test)
3513
 
        self.assertEquals('25', os.environ['LINES'])
3514
 
 
3515
 
    def test_deleting_variable(self):
3516
 
        test = self.ScratchMonkey('test_me')
3517
 
        # LINES is known to be present in os.environ
3518
 
        tests.override_os_environ(test, {'LINES': None})
3519
 
        self.assertTrue('LINES' not in os.environ)
3520
 
        tests.restore_os_environ(test)
3521
 
        self.assertEquals('25', os.environ['LINES'])
3522
 
 
3523
 
 
3524
 
class TestDocTestSuiteIsolation(tests.TestCase):
3525
 
    """Test that `tests.DocTestSuite` isolates doc tests from os.environ.
3526
 
 
3527
 
    Since tests.TestCase alreay provides an isolation from os.environ, we use
3528
 
    the clean environment as a base for testing. To precisely capture the
3529
 
    isolation provided by tests.DocTestSuite, we use doctest.DocTestSuite to
3530
 
    compare against.
3531
 
 
3532
 
    We want to make sure `tests.DocTestSuite` respect `tests.isolated_environ`,
3533
 
    not `os.environ` so each test overrides it to suit its needs.
3534
 
 
3535
 
    """
3536
 
 
3537
 
    def get_doctest_suite_for_string(self, klass, string):
3538
 
        class Finder(doctest.DocTestFinder):
3539
 
 
3540
 
            def find(*args, **kwargs):
3541
 
                test = doctest.DocTestParser().get_doctest(
3542
 
                    string, {}, 'foo', 'foo.py', 0)
3543
 
                return [test]
3544
 
 
3545
 
        suite = klass(test_finder=Finder())
3546
 
        return suite
3547
 
 
3548
 
    def run_doctest_suite_for_string(self, klass, string):
3549
 
        suite = self.get_doctest_suite_for_string(klass, string)
3550
 
        output = StringIO()
3551
 
        result = tests.TextTestResult(output, 0, 1)
3552
 
        suite.run(result)
3553
 
        return result, output
3554
 
 
3555
 
    def assertDocTestStringSucceds(self, klass, string):
3556
 
        result, output = self.run_doctest_suite_for_string(klass, string)
3557
 
        if not result.wasStrictlySuccessful():
3558
 
            self.fail(output.getvalue())
3559
 
 
3560
 
    def assertDocTestStringFails(self, klass, string):
3561
 
        result, output = self.run_doctest_suite_for_string(klass, string)
3562
 
        if result.wasStrictlySuccessful():
3563
 
            self.fail(output.getvalue())
3564
 
 
3565
 
    def test_injected_variable(self):
3566
 
        self.overrideAttr(tests, 'isolated_environ', {'LINES': '42'})
3567
 
        test = """
3568
 
            >>> import os
3569
 
            >>> os.environ['LINES']
3570
 
            '42'
3571
 
            """
3572
 
        # doctest.DocTestSuite fails as it sees '25'
3573
 
        self.assertDocTestStringFails(doctest.DocTestSuite, test)
3574
 
        # tests.DocTestSuite sees '42'
3575
 
        self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
3576
 
 
3577
 
    def test_deleted_variable(self):
3578
 
        self.overrideAttr(tests, 'isolated_environ', {'LINES': None})
3579
 
        test = """
3580
 
            >>> import os
3581
 
            >>> os.environ.get('LINES')
3582
 
            """
3583
 
        # doctest.DocTestSuite fails as it sees '25'
3584
 
        self.assertDocTestStringFails(doctest.DocTestSuite, test)
3585
 
        # tests.DocTestSuite sees None
3586
 
        self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
3587
 
 
3588
 
 
3589
 
class TestSelftestExcludePatterns(tests.TestCase):
3590
 
 
3591
 
    def setUp(self):
3592
 
        super(TestSelftestExcludePatterns, self).setUp()
3593
 
        self.overrideAttr(tests, 'test_suite', self.suite_factory)
3594
 
 
3595
 
    def suite_factory(self, keep_only=None, starting_with=None):
3596
 
        """A test suite factory with only a few tests."""
3597
 
        class Test(tests.TestCase):
3598
 
            def id(self):
3599
 
                # We don't need the full class path
3600
 
                return self._testMethodName
3601
 
            def a(self):
3602
 
                pass
3603
 
            def b(self):
3604
 
                pass
3605
 
            def c(self):
3606
 
                pass
3607
 
        return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
3608
 
 
3609
 
    def assertTestList(self, expected, *selftest_args):
3610
 
        # We rely on setUp installing the right test suite factory so we can
3611
 
        # test at the command level without loading the whole test suite
3612
 
        out, err = self.run_bzr(('selftest', '--list') + selftest_args)
3613
 
        actual = out.splitlines()
3614
 
        self.assertEquals(expected, actual)
3615
 
 
3616
 
    def test_full_list(self):
3617
 
        self.assertTestList(['a', 'b', 'c'])
3618
 
 
3619
 
    def test_single_exclude(self):
3620
 
        self.assertTestList(['b', 'c'], '-x', 'a')
3621
 
 
3622
 
    def test_mutiple_excludes(self):
3623
 
        self.assertTestList(['c'], '-x', 'a', '-x', 'b')
3624
 
 
3625
 
 
3626
 
class TestCounterHooks(tests.TestCase, SelfTestHelper):
3627
 
 
3628
 
    _test_needs_features = [features.subunit]
3629
 
 
3630
 
    def setUp(self):
3631
 
        super(TestCounterHooks, self).setUp()
3632
 
        class Test(tests.TestCase):
3633
 
 
3634
 
            def setUp(self):
3635
 
                super(Test, self).setUp()
3636
 
                self.hooks = hooks.Hooks()
3637
 
                self.hooks.add_hook('myhook', 'Foo bar blah', (2,4))
3638
 
                self.install_counter_hook(self.hooks, 'myhook')
3639
 
 
3640
 
            def no_hook(self):
3641
 
                pass
3642
 
 
3643
 
            def run_hook_once(self):
3644
 
                for hook in self.hooks['myhook']:
3645
 
                    hook(self)
3646
 
 
3647
 
        self.test_class = Test
3648
 
 
3649
 
    def assertHookCalls(self, expected_calls, test_name):
3650
 
        test = self.test_class(test_name)
3651
 
        result = unittest.TestResult()
3652
 
        test.run(result)
3653
 
        self.assertTrue(hasattr(test, '_counters'))
3654
 
        self.assertTrue(test._counters.has_key('myhook'))
3655
 
        self.assertEquals(expected_calls, test._counters['myhook'])
3656
 
 
3657
 
    def test_no_hook(self):
3658
 
        self.assertHookCalls(0, 'no_hook')
3659
 
 
3660
 
    def test_run_hook_once(self):
3661
 
        tt = features.testtools
3662
 
        if tt.module.__version__ < (0, 9, 8):
3663
 
            raise tests.TestSkipped('testtools-0.9.8 required for addDetail')
3664
 
        self.assertHookCalls(1, 'run_hook_once')