/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

Use ROOT_ID when the repository supports old clients (Bug #107168)

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright (C) 2005, 2006, 2007 Canonical Ltd
 
2
#
 
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.
 
7
#
 
8
# This program is distributed in the hope that it will be useful,
 
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
11
# GNU General Public License for more details.
 
12
#
 
13
# You should have received a copy of the GNU General Public License
 
14
# along with this program; if not, write to the Free Software
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
16
 
 
17
"""Tests for the test framework."""
 
18
 
 
19
import cStringIO
 
20
import os
 
21
from StringIO import StringIO
 
22
import sys
 
23
import time
 
24
import unittest
 
25
import warnings
 
26
 
 
27
import bzrlib
 
28
from bzrlib import (
 
29
    bzrdir,
 
30
    errors,
 
31
    memorytree,
 
32
    osutils,
 
33
    repository,
 
34
    symbol_versioning,
 
35
    )
 
36
from bzrlib.progress import _BaseProgressBar
 
37
from bzrlib.repofmt import weaverepo
 
38
from bzrlib.symbol_versioning import zero_ten, zero_eleven
 
39
from bzrlib.tests import (
 
40
                          ChrootedTestCase,
 
41
                          ExtendedTestResult,
 
42
                          Feature,
 
43
                          KnownFailure,
 
44
                          TestCase,
 
45
                          TestCaseInTempDir,
 
46
                          TestCaseWithMemoryTransport,
 
47
                          TestCaseWithTransport,
 
48
                          TestSkipped,
 
49
                          TestSuite,
 
50
                          TextTestRunner,
 
51
                          UnavailableFeature,
 
52
                          clean_selftest_output,
 
53
                          )
 
54
from bzrlib.tests.test_sftp_transport import TestCaseWithSFTPServer
 
55
from bzrlib.tests.TestUtil import _load_module_by_name
 
56
from bzrlib.trace import note
 
57
from bzrlib.transport.memory import MemoryServer, MemoryTransport
 
58
from bzrlib.version import _get_bzr_source_tree
 
59
 
 
60
 
 
61
class SelftestTests(TestCase):
 
62
 
 
63
    def test_import_tests(self):
 
64
        mod = _load_module_by_name('bzrlib.tests.test_selftest')
 
65
        self.assertEqual(mod.SelftestTests, SelftestTests)
 
66
 
 
67
    def test_import_test_failure(self):
 
68
        self.assertRaises(ImportError,
 
69
                          _load_module_by_name,
 
70
                          'bzrlib.no-name-yet')
 
71
 
 
72
class MetaTestLog(TestCase):
 
73
 
 
74
    def test_logging(self):
 
75
        """Test logs are captured when a test fails."""
 
76
        self.log('a test message')
 
77
        self._log_file.flush()
 
78
        self.assertContainsRe(self._get_log(keep_log_file=True),
 
79
                              'a test message\n')
 
80
 
 
81
 
 
82
class TestTreeShape(TestCaseInTempDir):
 
83
 
 
84
    def test_unicode_paths(self):
 
85
        filename = u'hell\u00d8'
 
86
        try:
 
87
            self.build_tree_contents([(filename, 'contents of hello')])
 
88
        except UnicodeEncodeError:
 
89
            raise TestSkipped("can't build unicode working tree in "
 
90
                "filesystem encoding %s" % sys.getfilesystemencoding())
 
91
        self.failUnlessExists(filename)
 
92
 
 
93
 
 
94
class TestTransportProviderAdapter(TestCase):
 
95
    """A group of tests that test the transport implementation adaption core.
 
96
 
 
97
    This is a meta test that the tests are applied to all available 
 
98
    transports.
 
99
 
 
100
    This will be generalised in the future which is why it is in this 
 
101
    test file even though it is specific to transport tests at the moment.
 
102
    """
 
103
 
 
104
    def test_get_transport_permutations(self):
 
105
        # this checks that we the module get_test_permutations call
 
106
        # is made by the adapter get_transport_test_permitations method.
 
107
        class MockModule(object):
 
108
            def get_test_permutations(self):
 
109
                return sample_permutation
 
110
        sample_permutation = [(1,2), (3,4)]
 
111
        from bzrlib.transport import TransportTestProviderAdapter
 
112
        adapter = TransportTestProviderAdapter()
 
113
        self.assertEqual(sample_permutation,
 
114
                         adapter.get_transport_test_permutations(MockModule()))
 
115
 
 
116
    def test_adapter_checks_all_modules(self):
 
117
        # this checks that the adapter returns as many permurtations as
 
118
        # there are in all the registered# transport modules for there
 
119
        # - we assume if this matches its probably doing the right thing
 
120
        # especially in combination with the tests for setting the right
 
121
        # classes below.
 
122
        from bzrlib.transport import (TransportTestProviderAdapter,
 
123
                                      _get_transport_modules
 
124
                                      )
 
125
        modules = _get_transport_modules()
 
126
        permutation_count = 0
 
127
        for module in modules:
 
128
            try:
 
129
                permutation_count += len(reduce(getattr, 
 
130
                    (module + ".get_test_permutations").split('.')[1:],
 
131
                     __import__(module))())
 
132
            except errors.DependencyNotPresent:
 
133
                pass
 
134
        input_test = TestTransportProviderAdapter(
 
135
            "test_adapter_sets_transport_class")
 
136
        adapter = TransportTestProviderAdapter()
 
137
        self.assertEqual(permutation_count,
 
138
                         len(list(iter(adapter.adapt(input_test)))))
 
139
 
 
140
    def test_adapter_sets_transport_class(self):
 
141
        # Check that the test adapter inserts a transport and server into the
 
142
        # generated test.
 
143
        #
 
144
        # This test used to know about all the possible transports and the
 
145
        # order they were returned but that seems overly brittle (mbp
 
146
        # 20060307)
 
147
        input_test = TestTransportProviderAdapter(
 
148
            "test_adapter_sets_transport_class")
 
149
        from bzrlib.transport import TransportTestProviderAdapter
 
150
        suite = TransportTestProviderAdapter().adapt(input_test)
 
151
        tests = list(iter(suite))
 
152
        self.assertTrue(len(tests) > 6)
 
153
        # there are at least that many builtin transports
 
154
        one_test = tests[0]
 
155
        self.assertTrue(issubclass(one_test.transport_class, 
 
156
                                   bzrlib.transport.Transport))
 
157
        self.assertTrue(issubclass(one_test.transport_server, 
 
158
                                   bzrlib.transport.Server))
 
159
 
 
160
 
 
161
class TestBranchProviderAdapter(TestCase):
 
162
    """A group of tests that test the branch implementation test adapter."""
 
163
 
 
164
    def test_adapted_tests(self):
 
165
        # check that constructor parameters are passed through to the adapted
 
166
        # test.
 
167
        from bzrlib.branch import BranchTestProviderAdapter
 
168
        input_test = TestBranchProviderAdapter(
 
169
            "test_adapted_tests")
 
170
        server1 = "a"
 
171
        server2 = "b"
 
172
        formats = [("c", "C"), ("d", "D")]
 
173
        adapter = BranchTestProviderAdapter(server1, server2, formats)
 
174
        suite = adapter.adapt(input_test)
 
175
        tests = list(iter(suite))
 
176
        self.assertEqual(2, len(tests))
 
177
        self.assertEqual(tests[0].branch_format, formats[0][0])
 
178
        self.assertEqual(tests[0].bzrdir_format, formats[0][1])
 
179
        self.assertEqual(tests[0].transport_server, server1)
 
180
        self.assertEqual(tests[0].transport_readonly_server, server2)
 
181
        self.assertEqual(tests[1].branch_format, formats[1][0])
 
182
        self.assertEqual(tests[1].bzrdir_format, formats[1][1])
 
183
        self.assertEqual(tests[1].transport_server, server1)
 
184
        self.assertEqual(tests[1].transport_readonly_server, server2)
 
185
 
 
186
 
 
187
class TestBzrDirProviderAdapter(TestCase):
 
188
    """A group of tests that test the bzr dir implementation test adapter."""
 
189
 
 
190
    def test_adapted_tests(self):
 
191
        # check that constructor parameters are passed through to the adapted
 
192
        # test.
 
193
        from bzrlib.bzrdir import BzrDirTestProviderAdapter
 
194
        input_test = TestBzrDirProviderAdapter(
 
195
            "test_adapted_tests")
 
196
        server1 = "a"
 
197
        server2 = "b"
 
198
        formats = ["c", "d"]
 
199
        adapter = BzrDirTestProviderAdapter(server1, server2, formats)
 
200
        suite = adapter.adapt(input_test)
 
201
        tests = list(iter(suite))
 
202
        self.assertEqual(2, len(tests))
 
203
        self.assertEqual(tests[0].bzrdir_format, formats[0])
 
204
        self.assertEqual(tests[0].transport_server, server1)
 
205
        self.assertEqual(tests[0].transport_readonly_server, server2)
 
206
        self.assertEqual(tests[1].bzrdir_format, formats[1])
 
207
        self.assertEqual(tests[1].transport_server, server1)
 
208
        self.assertEqual(tests[1].transport_readonly_server, server2)
 
209
 
 
210
 
 
211
class TestRepositoryProviderAdapter(TestCase):
 
212
    """A group of tests that test the repository implementation test adapter."""
 
213
 
 
214
    def test_adapted_tests(self):
 
215
        # check that constructor parameters are passed through to the adapted
 
216
        # test.
 
217
        from bzrlib.repository import RepositoryTestProviderAdapter
 
218
        input_test = TestRepositoryProviderAdapter(
 
219
            "test_adapted_tests")
 
220
        server1 = "a"
 
221
        server2 = "b"
 
222
        formats = [("c", "C"), ("d", "D")]
 
223
        adapter = RepositoryTestProviderAdapter(server1, server2, formats)
 
224
        suite = adapter.adapt(input_test)
 
225
        tests = list(iter(suite))
 
226
        self.assertEqual(2, len(tests))
 
227
        self.assertEqual(tests[0].bzrdir_format, formats[0][1])
 
228
        self.assertEqual(tests[0].repository_format, formats[0][0])
 
229
        self.assertEqual(tests[0].transport_server, server1)
 
230
        self.assertEqual(tests[0].transport_readonly_server, server2)
 
231
        self.assertEqual(tests[1].bzrdir_format, formats[1][1])
 
232
        self.assertEqual(tests[1].repository_format, formats[1][0])
 
233
        self.assertEqual(tests[1].transport_server, server1)
 
234
        self.assertEqual(tests[1].transport_readonly_server, server2)
 
235
 
 
236
 
 
237
class TestInterRepositoryProviderAdapter(TestCase):
 
238
    """A group of tests that test the InterRepository test adapter."""
 
239
 
 
240
    def test_adapted_tests(self):
 
241
        # check that constructor parameters are passed through to the adapted
 
242
        # test.
 
243
        from bzrlib.repository import InterRepositoryTestProviderAdapter
 
244
        input_test = TestInterRepositoryProviderAdapter(
 
245
            "test_adapted_tests")
 
246
        server1 = "a"
 
247
        server2 = "b"
 
248
        formats = [(str, "C1", "C2"), (int, "D1", "D2")]
 
249
        adapter = InterRepositoryTestProviderAdapter(server1, server2, formats)
 
250
        suite = adapter.adapt(input_test)
 
251
        tests = list(iter(suite))
 
252
        self.assertEqual(2, len(tests))
 
253
        self.assertEqual(tests[0].interrepo_class, formats[0][0])
 
254
        self.assertEqual(tests[0].repository_format, formats[0][1])
 
255
        self.assertEqual(tests[0].repository_format_to, formats[0][2])
 
256
        self.assertEqual(tests[0].transport_server, server1)
 
257
        self.assertEqual(tests[0].transport_readonly_server, server2)
 
258
        self.assertEqual(tests[1].interrepo_class, formats[1][0])
 
259
        self.assertEqual(tests[1].repository_format, formats[1][1])
 
260
        self.assertEqual(tests[1].repository_format_to, formats[1][2])
 
261
        self.assertEqual(tests[1].transport_server, server1)
 
262
        self.assertEqual(tests[1].transport_readonly_server, server2)
 
263
 
 
264
 
 
265
class TestInterVersionedFileProviderAdapter(TestCase):
 
266
    """A group of tests that test the InterVersionedFile test adapter."""
 
267
 
 
268
    def test_adapted_tests(self):
 
269
        # check that constructor parameters are passed through to the adapted
 
270
        # test.
 
271
        from bzrlib.versionedfile import InterVersionedFileTestProviderAdapter
 
272
        input_test = TestInterRepositoryProviderAdapter(
 
273
            "test_adapted_tests")
 
274
        server1 = "a"
 
275
        server2 = "b"
 
276
        formats = [(str, "C1", "C2"), (int, "D1", "D2")]
 
277
        adapter = InterVersionedFileTestProviderAdapter(server1, server2, formats)
 
278
        suite = adapter.adapt(input_test)
 
279
        tests = list(iter(suite))
 
280
        self.assertEqual(2, len(tests))
 
281
        self.assertEqual(tests[0].interversionedfile_class, formats[0][0])
 
282
        self.assertEqual(tests[0].versionedfile_factory, formats[0][1])
 
283
        self.assertEqual(tests[0].versionedfile_factory_to, formats[0][2])
 
284
        self.assertEqual(tests[0].transport_server, server1)
 
285
        self.assertEqual(tests[0].transport_readonly_server, server2)
 
286
        self.assertEqual(tests[1].interversionedfile_class, formats[1][0])
 
287
        self.assertEqual(tests[1].versionedfile_factory, formats[1][1])
 
288
        self.assertEqual(tests[1].versionedfile_factory_to, formats[1][2])
 
289
        self.assertEqual(tests[1].transport_server, server1)
 
290
        self.assertEqual(tests[1].transport_readonly_server, server2)
 
291
 
 
292
 
 
293
class TestRevisionStoreProviderAdapter(TestCase):
 
294
    """A group of tests that test the RevisionStore test adapter."""
 
295
 
 
296
    def test_adapted_tests(self):
 
297
        # check that constructor parameters are passed through to the adapted
 
298
        # test.
 
299
        from bzrlib.store.revision import RevisionStoreTestProviderAdapter
 
300
        input_test = TestRevisionStoreProviderAdapter(
 
301
            "test_adapted_tests")
 
302
        # revision stores need a store factory - i.e. RevisionKnit
 
303
        #, a readonly and rw transport 
 
304
        # transport servers:
 
305
        server1 = "a"
 
306
        server2 = "b"
 
307
        store_factories = ["c", "d"]
 
308
        adapter = RevisionStoreTestProviderAdapter(server1, server2, store_factories)
 
309
        suite = adapter.adapt(input_test)
 
310
        tests = list(iter(suite))
 
311
        self.assertEqual(2, len(tests))
 
312
        self.assertEqual(tests[0].store_factory, store_factories[0][0])
 
313
        self.assertEqual(tests[0].transport_server, server1)
 
314
        self.assertEqual(tests[0].transport_readonly_server, server2)
 
315
        self.assertEqual(tests[1].store_factory, store_factories[1][0])
 
316
        self.assertEqual(tests[1].transport_server, server1)
 
317
        self.assertEqual(tests[1].transport_readonly_server, server2)
 
318
 
 
319
 
 
320
class TestWorkingTreeProviderAdapter(TestCase):
 
321
    """A group of tests that test the workingtree implementation test adapter."""
 
322
 
 
323
    def test_adapted_tests(self):
 
324
        # check that constructor parameters are passed through to the adapted
 
325
        # test.
 
326
        from bzrlib.workingtree import WorkingTreeTestProviderAdapter
 
327
        input_test = TestWorkingTreeProviderAdapter(
 
328
            "test_adapted_tests")
 
329
        server1 = "a"
 
330
        server2 = "b"
 
331
        formats = [("c", "C"), ("d", "D")]
 
332
        adapter = WorkingTreeTestProviderAdapter(server1, server2, formats)
 
333
        suite = adapter.adapt(input_test)
 
334
        tests = list(iter(suite))
 
335
        self.assertEqual(2, len(tests))
 
336
        self.assertEqual(tests[0].workingtree_format, formats[0][0])
 
337
        self.assertEqual(tests[0].bzrdir_format, formats[0][1])
 
338
        self.assertEqual(tests[0].transport_server, server1)
 
339
        self.assertEqual(tests[0].transport_readonly_server, server2)
 
340
        self.assertEqual(tests[1].workingtree_format, formats[1][0])
 
341
        self.assertEqual(tests[1].bzrdir_format, formats[1][1])
 
342
        self.assertEqual(tests[1].transport_server, server1)
 
343
        self.assertEqual(tests[1].transport_readonly_server, server2)
 
344
 
 
345
 
 
346
class TestTreeProviderAdapter(TestCase):
 
347
    """Test the setup of tree_implementation tests."""
 
348
 
 
349
    def test_adapted_tests(self):
 
350
        # the tree implementation adapter is meant to setup one instance for
 
351
        # each working tree format, and one additional instance that will
 
352
        # use the default wt format, but create a revision tree for the tests.
 
353
        # this means that the wt ones should have the workingtree_to_test_tree
 
354
        # attribute set to 'return_parameter' and the revision one set to
 
355
        # revision_tree_from_workingtree.
 
356
 
 
357
        from bzrlib.tests.tree_implementations import (
 
358
            TreeTestProviderAdapter,
 
359
            return_parameter,
 
360
            revision_tree_from_workingtree
 
361
            )
 
362
        from bzrlib.workingtree import WorkingTreeFormat, WorkingTreeFormat3
 
363
        input_test = TestTreeProviderAdapter(
 
364
            "test_adapted_tests")
 
365
        server1 = "a"
 
366
        server2 = "b"
 
367
        formats = [("c", "C"), ("d", "D")]
 
368
        adapter = TreeTestProviderAdapter(server1, server2, formats)
 
369
        suite = adapter.adapt(input_test)
 
370
        tests = list(iter(suite))
 
371
        self.assertEqual(4, len(tests))
 
372
        # this must match the default format setp up in
 
373
        # TreeTestProviderAdapter.adapt
 
374
        default_format = WorkingTreeFormat3
 
375
        self.assertEqual(tests[0].workingtree_format, formats[0][0])
 
376
        self.assertEqual(tests[0].bzrdir_format, formats[0][1])
 
377
        self.assertEqual(tests[0].transport_server, server1)
 
378
        self.assertEqual(tests[0].transport_readonly_server, server2)
 
379
        self.assertEqual(tests[0].workingtree_to_test_tree, return_parameter)
 
380
        self.assertEqual(tests[1].workingtree_format, formats[1][0])
 
381
        self.assertEqual(tests[1].bzrdir_format, formats[1][1])
 
382
        self.assertEqual(tests[1].transport_server, server1)
 
383
        self.assertEqual(tests[1].transport_readonly_server, server2)
 
384
        self.assertEqual(tests[1].workingtree_to_test_tree, return_parameter)
 
385
        self.assertIsInstance(tests[2].workingtree_format, default_format)
 
386
        #self.assertEqual(tests[2].bzrdir_format,
 
387
        #                 default_format._matchingbzrdir)
 
388
        self.assertEqual(tests[2].transport_server, server1)
 
389
        self.assertEqual(tests[2].transport_readonly_server, server2)
 
390
        self.assertEqual(tests[2].workingtree_to_test_tree,
 
391
            revision_tree_from_workingtree)
 
392
 
 
393
 
 
394
class TestInterTreeProviderAdapter(TestCase):
 
395
    """A group of tests that test the InterTreeTestAdapter."""
 
396
 
 
397
    def test_adapted_tests(self):
 
398
        # check that constructor parameters are passed through to the adapted
 
399
        # test.
 
400
        # for InterTree tests we want the machinery to bring up two trees in
 
401
        # each instance: the base one, and the one we are interacting with.
 
402
        # because each optimiser can be direction specific, we need to test
 
403
        # each optimiser in its chosen direction.
 
404
        # unlike the TestProviderAdapter we dont want to automatically add a
 
405
        # parameterised one for WorkingTree - the optimisers will tell us what
 
406
        # ones to add.
 
407
        from bzrlib.tests.tree_implementations import (
 
408
            return_parameter,
 
409
            revision_tree_from_workingtree
 
410
            )
 
411
        from bzrlib.tests.intertree_implementations import (
 
412
            InterTreeTestProviderAdapter,
 
413
            )
 
414
        from bzrlib.workingtree import WorkingTreeFormat2, WorkingTreeFormat3
 
415
        input_test = TestInterTreeProviderAdapter(
 
416
            "test_adapted_tests")
 
417
        server1 = "a"
 
418
        server2 = "b"
 
419
        format1 = WorkingTreeFormat2()
 
420
        format2 = WorkingTreeFormat3()
 
421
        formats = [(str, format1, format2, "converter1"),
 
422
            (int, format2, format1, "converter2")]
 
423
        adapter = InterTreeTestProviderAdapter(server1, server2, formats)
 
424
        suite = adapter.adapt(input_test)
 
425
        tests = list(iter(suite))
 
426
        self.assertEqual(2, len(tests))
 
427
        self.assertEqual(tests[0].intertree_class, formats[0][0])
 
428
        self.assertEqual(tests[0].workingtree_format, formats[0][1])
 
429
        self.assertEqual(tests[0].workingtree_format_to, formats[0][2])
 
430
        self.assertEqual(tests[0].mutable_trees_to_test_trees, formats[0][3])
 
431
        self.assertEqual(tests[0].workingtree_to_test_tree, return_parameter)
 
432
        self.assertEqual(tests[0].transport_server, server1)
 
433
        self.assertEqual(tests[0].transport_readonly_server, server2)
 
434
        self.assertEqual(tests[1].intertree_class, formats[1][0])
 
435
        self.assertEqual(tests[1].workingtree_format, formats[1][1])
 
436
        self.assertEqual(tests[1].workingtree_format_to, formats[1][2])
 
437
        self.assertEqual(tests[1].mutable_trees_to_test_trees, formats[1][3])
 
438
        self.assertEqual(tests[1].workingtree_to_test_tree, return_parameter)
 
439
        self.assertEqual(tests[1].transport_server, server1)
 
440
        self.assertEqual(tests[1].transport_readonly_server, server2)
 
441
 
 
442
 
 
443
class TestTestCaseInTempDir(TestCaseInTempDir):
 
444
 
 
445
    def test_home_is_not_working(self):
 
446
        self.assertNotEqual(self.test_dir, self.test_home_dir)
 
447
        cwd = osutils.getcwd()
 
448
        self.assertEqual(self.test_dir, cwd)
 
449
        self.assertEqual(self.test_home_dir, os.environ['HOME'])
 
450
 
 
451
 
 
452
class TestTestCaseWithMemoryTransport(TestCaseWithMemoryTransport):
 
453
 
 
454
    def test_home_is_non_existant_dir_under_root(self):
 
455
        """The test_home_dir for TestCaseWithMemoryTransport is missing.
 
456
 
 
457
        This is because TestCaseWithMemoryTransport is for tests that do not
 
458
        need any disk resources: they should be hooked into bzrlib in such a 
 
459
        way that no global settings are being changed by the test (only a 
 
460
        few tests should need to do that), and having a missing dir as home is
 
461
        an effective way to ensure that this is the case.
 
462
        """
 
463
        self.assertEqual(self.TEST_ROOT + "/MemoryTransportMissingHomeDir",
 
464
            self.test_home_dir)
 
465
        self.assertEqual(self.test_home_dir, os.environ['HOME'])
 
466
        
 
467
    def test_cwd_is_TEST_ROOT(self):
 
468
        self.assertEqual(self.test_dir, self.TEST_ROOT)
 
469
        cwd = osutils.getcwd()
 
470
        self.assertEqual(self.test_dir, cwd)
 
471
 
 
472
    def test_make_branch_and_memory_tree(self):
 
473
        """In TestCaseWithMemoryTransport we should not make the branch on disk.
 
474
 
 
475
        This is hard to comprehensively robustly test, so we settle for making
 
476
        a branch and checking no directory was created at its relpath.
 
477
        """
 
478
        tree = self.make_branch_and_memory_tree('dir')
 
479
        # Guard against regression into MemoryTransport leaking
 
480
        # files to disk instead of keeping them in memory.
 
481
        self.failIf(osutils.lexists('dir'))
 
482
        self.assertIsInstance(tree, memorytree.MemoryTree)
 
483
 
 
484
    def test_make_branch_and_memory_tree_with_format(self):
 
485
        """make_branch_and_memory_tree should accept a format option."""
 
486
        format = bzrdir.BzrDirMetaFormat1()
 
487
        format.repository_format = weaverepo.RepositoryFormat7()
 
488
        tree = self.make_branch_and_memory_tree('dir', format=format)
 
489
        # Guard against regression into MemoryTransport leaking
 
490
        # files to disk instead of keeping them in memory.
 
491
        self.failIf(osutils.lexists('dir'))
 
492
        self.assertIsInstance(tree, memorytree.MemoryTree)
 
493
        self.assertEqual(format.repository_format.__class__,
 
494
            tree.branch.repository._format.__class__)
 
495
 
 
496
 
 
497
class TestTestCaseWithTransport(TestCaseWithTransport):
 
498
    """Tests for the convenience functions TestCaseWithTransport introduces."""
 
499
 
 
500
    def test_get_readonly_url_none(self):
 
501
        from bzrlib.transport import get_transport
 
502
        from bzrlib.transport.memory import MemoryServer
 
503
        from bzrlib.transport.readonly import ReadonlyTransportDecorator
 
504
        self.vfs_transport_factory = MemoryServer
 
505
        self.transport_readonly_server = None
 
506
        # calling get_readonly_transport() constructs a decorator on the url
 
507
        # for the server
 
508
        url = self.get_readonly_url()
 
509
        url2 = self.get_readonly_url('foo/bar')
 
510
        t = get_transport(url)
 
511
        t2 = get_transport(url2)
 
512
        self.failUnless(isinstance(t, ReadonlyTransportDecorator))
 
513
        self.failUnless(isinstance(t2, ReadonlyTransportDecorator))
 
514
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
 
515
 
 
516
    def test_get_readonly_url_http(self):
 
517
        from bzrlib.tests.HttpServer import HttpServer
 
518
        from bzrlib.transport import get_transport
 
519
        from bzrlib.transport.local import LocalURLServer
 
520
        from bzrlib.transport.http import HttpTransportBase
 
521
        self.transport_server = LocalURLServer
 
522
        self.transport_readonly_server = HttpServer
 
523
        # calling get_readonly_transport() gives us a HTTP server instance.
 
524
        url = self.get_readonly_url()
 
525
        url2 = self.get_readonly_url('foo/bar')
 
526
        # the transport returned may be any HttpTransportBase subclass
 
527
        t = get_transport(url)
 
528
        t2 = get_transport(url2)
 
529
        self.failUnless(isinstance(t, HttpTransportBase))
 
530
        self.failUnless(isinstance(t2, HttpTransportBase))
 
531
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
 
532
 
 
533
    def test_is_directory(self):
 
534
        """Test assertIsDirectory assertion"""
 
535
        t = self.get_transport()
 
536
        self.build_tree(['a_dir/', 'a_file'], transport=t)
 
537
        self.assertIsDirectory('a_dir', t)
 
538
        self.assertRaises(AssertionError, self.assertIsDirectory, 'a_file', t)
 
539
        self.assertRaises(AssertionError, self.assertIsDirectory, 'not_here', t)
 
540
 
 
541
 
 
542
class TestTestCaseTransports(TestCaseWithTransport):
 
543
 
 
544
    def setUp(self):
 
545
        super(TestTestCaseTransports, self).setUp()
 
546
        self.vfs_transport_factory = MemoryServer
 
547
 
 
548
    def test_make_bzrdir_preserves_transport(self):
 
549
        t = self.get_transport()
 
550
        result_bzrdir = self.make_bzrdir('subdir')
 
551
        self.assertIsInstance(result_bzrdir.transport, 
 
552
                              MemoryTransport)
 
553
        # should not be on disk, should only be in memory
 
554
        self.failIfExists('subdir')
 
555
 
 
556
 
 
557
class TestChrootedTest(ChrootedTestCase):
 
558
 
 
559
    def test_root_is_root(self):
 
560
        from bzrlib.transport import get_transport
 
561
        t = get_transport(self.get_readonly_url())
 
562
        url = t.base
 
563
        self.assertEqual(url, t.clone('..').base)
 
564
 
 
565
 
 
566
class MockProgress(_BaseProgressBar):
 
567
    """Progress-bar standin that records calls.
 
568
 
 
569
    Useful for testing pb using code.
 
570
    """
 
571
 
 
572
    def __init__(self):
 
573
        _BaseProgressBar.__init__(self)
 
574
        self.calls = []
 
575
 
 
576
    def tick(self):
 
577
        self.calls.append(('tick',))
 
578
 
 
579
    def update(self, msg=None, current=None, total=None):
 
580
        self.calls.append(('update', msg, current, total))
 
581
 
 
582
    def clear(self):
 
583
        self.calls.append(('clear',))
 
584
 
 
585
    def note(self, msg, *args):
 
586
        self.calls.append(('note', msg, args))
 
587
 
 
588
 
 
589
class TestTestResult(TestCase):
 
590
 
 
591
    def test_elapsed_time_with_benchmarking(self):
 
592
        result = bzrlib.tests.TextTestResult(self._log_file,
 
593
                                        descriptions=0,
 
594
                                        verbosity=1,
 
595
                                        )
 
596
        result._recordTestStartTime()
 
597
        time.sleep(0.003)
 
598
        result.extractBenchmarkTime(self)
 
599
        timed_string = result._testTimeString()
 
600
        # without explicit benchmarking, we should get a simple time.
 
601
        self.assertContainsRe(timed_string, "^ *[ 1-9][0-9]ms$")
 
602
        # if a benchmark time is given, we want a x of y style result.
 
603
        self.time(time.sleep, 0.001)
 
604
        result.extractBenchmarkTime(self)
 
605
        timed_string = result._testTimeString()
 
606
        self.assertContainsRe(timed_string, "^ *[ 1-9][0-9]ms/ *[ 1-9][0-9]ms$")
 
607
        # extracting the time from a non-bzrlib testcase sets to None
 
608
        result._recordTestStartTime()
 
609
        result.extractBenchmarkTime(
 
610
            unittest.FunctionTestCase(self.test_elapsed_time_with_benchmarking))
 
611
        timed_string = result._testTimeString()
 
612
        self.assertContainsRe(timed_string, "^ *[ 1-9][0-9]ms$")
 
613
        # cheat. Yes, wash thy mouth out with soap.
 
614
        self._benchtime = None
 
615
 
 
616
    def test_assigned_benchmark_file_stores_date(self):
 
617
        output = StringIO()
 
618
        result = bzrlib.tests.TextTestResult(self._log_file,
 
619
                                        descriptions=0,
 
620
                                        verbosity=1,
 
621
                                        bench_history=output
 
622
                                        )
 
623
        output_string = output.getvalue()
 
624
        
 
625
        # if you are wondering about the regexp please read the comment in
 
626
        # test_bench_history (bzrlib.tests.test_selftest.TestRunner)
 
627
        # XXX: what comment?  -- Andrew Bennetts
 
628
        self.assertContainsRe(output_string, "--date [0-9.]+")
 
629
 
 
630
    def test_benchhistory_records_test_times(self):
 
631
        result_stream = StringIO()
 
632
        result = bzrlib.tests.TextTestResult(
 
633
            self._log_file,
 
634
            descriptions=0,
 
635
            verbosity=1,
 
636
            bench_history=result_stream
 
637
            )
 
638
 
 
639
        # we want profile a call and check that its test duration is recorded
 
640
        # make a new test instance that when run will generate a benchmark
 
641
        example_test_case = TestTestResult("_time_hello_world_encoding")
 
642
        # execute the test, which should succeed and record times
 
643
        example_test_case.run(result)
 
644
        lines = result_stream.getvalue().splitlines()
 
645
        self.assertEqual(2, len(lines))
 
646
        self.assertContainsRe(lines[1],
 
647
            " *[0-9]+ms bzrlib.tests.test_selftest.TestTestResult"
 
648
            "._time_hello_world_encoding")
 
649
 
 
650
    def _time_hello_world_encoding(self):
 
651
        """Profile two sleep calls
 
652
        
 
653
        This is used to exercise the test framework.
 
654
        """
 
655
        self.time(unicode, 'hello', errors='replace')
 
656
        self.time(unicode, 'world', errors='replace')
 
657
 
 
658
    def test_lsprofiling(self):
 
659
        """Verbose test result prints lsprof statistics from test cases."""
 
660
        try:
 
661
            import bzrlib.lsprof
 
662
        except ImportError:
 
663
            raise TestSkipped("lsprof not installed.")
 
664
        result_stream = StringIO()
 
665
        result = bzrlib.tests.VerboseTestResult(
 
666
            unittest._WritelnDecorator(result_stream),
 
667
            descriptions=0,
 
668
            verbosity=2,
 
669
            )
 
670
        # we want profile a call of some sort and check it is output by
 
671
        # addSuccess. We dont care about addError or addFailure as they
 
672
        # are not that interesting for performance tuning.
 
673
        # make a new test instance that when run will generate a profile
 
674
        example_test_case = TestTestResult("_time_hello_world_encoding")
 
675
        example_test_case._gather_lsprof_in_benchmarks = True
 
676
        # execute the test, which should succeed and record profiles
 
677
        example_test_case.run(result)
 
678
        # lsprofile_something()
 
679
        # if this worked we want 
 
680
        # LSProf output for <built in function unicode> (['hello'], {'errors': 'replace'})
 
681
        #    CallCount    Recursive    Total(ms)   Inline(ms) module:lineno(function)
 
682
        # (the lsprof header)
 
683
        # ... an arbitrary number of lines
 
684
        # and the function call which is time.sleep.
 
685
        #           1        0            ???         ???       ???(sleep) 
 
686
        # and then repeated but with 'world', rather than 'hello'.
 
687
        # this should appear in the output stream of our test result.
 
688
        output = result_stream.getvalue()
 
689
        self.assertContainsRe(output,
 
690
            r"LSProf output for <type 'unicode'>\(\('hello',\), {'errors': 'replace'}\)")
 
691
        self.assertContainsRe(output,
 
692
            r" *CallCount *Recursive *Total\(ms\) *Inline\(ms\) *module:lineno\(function\)\n")
 
693
        self.assertContainsRe(output,
 
694
            r"( +1 +0 +0\.\d+ +0\.\d+ +<method 'disable' of '_lsprof\.Profiler' objects>\n)?")
 
695
        self.assertContainsRe(output,
 
696
            r"LSProf output for <type 'unicode'>\(\('world',\), {'errors': 'replace'}\)\n")
 
697
 
 
698
    def test_known_failure(self):
 
699
        """A KnownFailure being raised should trigger several result actions."""
 
700
        class InstrumentedTestResult(ExtendedTestResult):
 
701
 
 
702
            def report_test_start(self, test): pass
 
703
            def report_known_failure(self, test, err):
 
704
                self._call = test, err
 
705
        result = InstrumentedTestResult(None, None, None, None)
 
706
        def test_function():
 
707
            raise KnownFailure('failed!')
 
708
        test = unittest.FunctionTestCase(test_function)
 
709
        test.run(result)
 
710
        # it should invoke 'report_known_failure'.
 
711
        self.assertEqual(2, len(result._call))
 
712
        self.assertEqual(test, result._call[0])
 
713
        self.assertEqual(KnownFailure, result._call[1][0])
 
714
        self.assertIsInstance(result._call[1][1], KnownFailure)
 
715
        # we dont introspec the traceback, if the rest is ok, it would be
 
716
        # exceptional for it not to be.
 
717
        # it should update the known_failure_count on the object.
 
718
        self.assertEqual(1, result.known_failure_count)
 
719
        # the result should be successful.
 
720
        self.assertTrue(result.wasSuccessful())
 
721
 
 
722
    def test_verbose_report_known_failure(self):
 
723
        # verbose test output formatting
 
724
        result_stream = StringIO()
 
725
        result = bzrlib.tests.VerboseTestResult(
 
726
            unittest._WritelnDecorator(result_stream),
 
727
            descriptions=0,
 
728
            verbosity=2,
 
729
            )
 
730
        test = self.get_passing_test()
 
731
        result.startTest(test)
 
732
        result.extractBenchmarkTime(test)
 
733
        prefix = len(result_stream.getvalue())
 
734
        # the err parameter has the shape:
 
735
        # (class, exception object, traceback)
 
736
        # KnownFailures dont get their tracebacks shown though, so we
 
737
        # can skip that.
 
738
        err = (KnownFailure, KnownFailure('foo'), None)
 
739
        result.report_known_failure(test, err)
 
740
        output = result_stream.getvalue()[prefix:]
 
741
        lines = output.splitlines()
 
742
        self.assertContainsRe(lines[0], r'XFAIL *\d+ms$')
 
743
        self.assertEqual(lines[1], '    foo')
 
744
        self.assertEqual(2, len(lines))
 
745
 
 
746
    def test_text_report_known_failure(self):
 
747
        # text test output formatting
 
748
        pb = MockProgress()
 
749
        result = bzrlib.tests.TextTestResult(
 
750
            None,
 
751
            descriptions=0,
 
752
            verbosity=1,
 
753
            pb=pb,
 
754
            )
 
755
        test = self.get_passing_test()
 
756
        # this seeds the state to handle reporting the test.
 
757
        result.startTest(test)
 
758
        result.extractBenchmarkTime(test)
 
759
        # the err parameter has the shape:
 
760
        # (class, exception object, traceback)
 
761
        # KnownFailures dont get their tracebacks shown though, so we
 
762
        # can skip that.
 
763
        err = (KnownFailure, KnownFailure('foo'), None)
 
764
        result.report_known_failure(test, err)
 
765
        self.assertEqual(
 
766
            [
 
767
            ('update', '[1 in 0s] passing_test', None, None),
 
768
            ('note', 'XFAIL: %s\n%s\n', ('passing_test', err[1]))
 
769
            ],
 
770
            pb.calls)
 
771
        # known_failures should be printed in the summary, so if we run a test
 
772
        # after there are some known failures, the update prefix should match
 
773
        # this.
 
774
        result.known_failure_count = 3
 
775
        test.run(result)
 
776
        self.assertEqual(
 
777
            [
 
778
            ('update', '[2 in 0s, 3 known failures] passing_test', None, None),
 
779
            ],
 
780
            pb.calls[2:])
 
781
 
 
782
    def get_passing_test(self):
 
783
        """Return a test object that can't be run usefully."""
 
784
        def passing_test():
 
785
            pass
 
786
        return unittest.FunctionTestCase(passing_test)
 
787
 
 
788
    def test_add_not_supported(self):
 
789
        """Test the behaviour of invoking addNotSupported."""
 
790
        class InstrumentedTestResult(ExtendedTestResult):
 
791
            def report_test_start(self, test): pass
 
792
            def report_unsupported(self, test, feature):
 
793
                self._call = test, feature
 
794
        result = InstrumentedTestResult(None, None, None, None)
 
795
        test = SampleTestCase('_test_pass')
 
796
        feature = Feature()
 
797
        result.startTest(test)
 
798
        result.addNotSupported(test, feature)
 
799
        # it should invoke 'report_unsupported'.
 
800
        self.assertEqual(2, len(result._call))
 
801
        self.assertEqual(test, result._call[0])
 
802
        self.assertEqual(feature, result._call[1])
 
803
        # the result should be successful.
 
804
        self.assertTrue(result.wasSuccessful())
 
805
        # it should record the test against a count of tests not run due to
 
806
        # this feature.
 
807
        self.assertEqual(1, result.unsupported['Feature'])
 
808
        # and invoking it again should increment that counter
 
809
        result.addNotSupported(test, feature)
 
810
        self.assertEqual(2, result.unsupported['Feature'])
 
811
 
 
812
    def test_verbose_report_unsupported(self):
 
813
        # verbose test output formatting
 
814
        result_stream = StringIO()
 
815
        result = bzrlib.tests.VerboseTestResult(
 
816
            unittest._WritelnDecorator(result_stream),
 
817
            descriptions=0,
 
818
            verbosity=2,
 
819
            )
 
820
        test = self.get_passing_test()
 
821
        feature = Feature()
 
822
        result.startTest(test)
 
823
        result.extractBenchmarkTime(test)
 
824
        prefix = len(result_stream.getvalue())
 
825
        result.report_unsupported(test, feature)
 
826
        output = result_stream.getvalue()[prefix:]
 
827
        lines = output.splitlines()
 
828
        self.assertEqual(lines, ['NODEP                   0ms', "    The feature 'Feature' is not available."])
 
829
    
 
830
    def test_text_report_unsupported(self):
 
831
        # text test output formatting
 
832
        pb = MockProgress()
 
833
        result = bzrlib.tests.TextTestResult(
 
834
            None,
 
835
            descriptions=0,
 
836
            verbosity=1,
 
837
            pb=pb,
 
838
            )
 
839
        test = self.get_passing_test()
 
840
        feature = Feature()
 
841
        # this seeds the state to handle reporting the test.
 
842
        result.startTest(test)
 
843
        result.extractBenchmarkTime(test)
 
844
        result.report_unsupported(test, feature)
 
845
        # no output on unsupported features
 
846
        self.assertEqual(
 
847
            [('update', '[1 in 0s] passing_test', None, None)
 
848
            ],
 
849
            pb.calls)
 
850
        # the number of missing features should be printed in the progress
 
851
        # summary, so check for that.
 
852
        result.unsupported = {'foo':0, 'bar':0}
 
853
        test.run(result)
 
854
        self.assertEqual(
 
855
            [
 
856
            ('update', '[2 in 0s, 2 missing features] passing_test', None, None),
 
857
            ],
 
858
            pb.calls[1:])
 
859
    
 
860
    def test_unavailable_exception(self):
 
861
        """An UnavailableFeature being raised should invoke addNotSupported."""
 
862
        class InstrumentedTestResult(ExtendedTestResult):
 
863
 
 
864
            def report_test_start(self, test): pass
 
865
            def addNotSupported(self, test, feature):
 
866
                self._call = test, feature
 
867
        result = InstrumentedTestResult(None, None, None, None)
 
868
        feature = Feature()
 
869
        def test_function():
 
870
            raise UnavailableFeature(feature)
 
871
        test = unittest.FunctionTestCase(test_function)
 
872
        test.run(result)
 
873
        # it should invoke 'addNotSupported'.
 
874
        self.assertEqual(2, len(result._call))
 
875
        self.assertEqual(test, result._call[0])
 
876
        self.assertEqual(feature, result._call[1])
 
877
        # and not count as an error
 
878
        self.assertEqual(0, result.error_count)
 
879
 
 
880
 
 
881
class TestRunner(TestCase):
 
882
 
 
883
    def dummy_test(self):
 
884
        pass
 
885
 
 
886
    def run_test_runner(self, testrunner, test):
 
887
        """Run suite in testrunner, saving global state and restoring it.
 
888
 
 
889
        This current saves and restores:
 
890
        TestCaseInTempDir.TEST_ROOT
 
891
        
 
892
        There should be no tests in this file that use bzrlib.tests.TextTestRunner
 
893
        without using this convenience method, because of our use of global state.
 
894
        """
 
895
        old_root = TestCaseInTempDir.TEST_ROOT
 
896
        try:
 
897
            TestCaseInTempDir.TEST_ROOT = None
 
898
            return testrunner.run(test)
 
899
        finally:
 
900
            TestCaseInTempDir.TEST_ROOT = old_root
 
901
 
 
902
    def test_known_failure_failed_run(self):
 
903
        # run a test that generates a known failure which should be printed in
 
904
        # the final output when real failures occur.
 
905
        def known_failure_test():
 
906
            raise KnownFailure('failed')
 
907
        test = unittest.TestSuite()
 
908
        test.addTest(unittest.FunctionTestCase(known_failure_test))
 
909
        def failing_test():
 
910
            raise AssertionError('foo')
 
911
        test.addTest(unittest.FunctionTestCase(failing_test))
 
912
        stream = StringIO()
 
913
        runner = TextTestRunner(stream=stream)
 
914
        result = self.run_test_runner(runner, test)
 
915
        lines = stream.getvalue().splitlines()
 
916
        self.assertEqual([
 
917
            '',
 
918
            '======================================================================',
 
919
            'FAIL: unittest.FunctionTestCase (failing_test)',
 
920
            '----------------------------------------------------------------------',
 
921
            'Traceback (most recent call last):',
 
922
            '    raise AssertionError(\'foo\')',
 
923
            'AssertionError: foo',
 
924
            '',
 
925
            '----------------------------------------------------------------------',
 
926
            '',
 
927
            'FAILED (failures=1, known_failure_count=1)'],
 
928
            lines[0:5] + lines[6:10] + lines[11:])
 
929
 
 
930
    def test_known_failure_ok_run(self):
 
931
        # run a test that generates a known failure which should be printed in the final output.
 
932
        def known_failure_test():
 
933
            raise KnownFailure('failed')
 
934
        test = unittest.FunctionTestCase(known_failure_test)
 
935
        stream = StringIO()
 
936
        runner = TextTestRunner(stream=stream)
 
937
        result = self.run_test_runner(runner, test)
 
938
        self.assertContainsRe(stream.getvalue(),
 
939
            '\n'
 
940
            '-*\n'
 
941
            'Ran 1 test in .*\n'
 
942
            '\n'
 
943
            'OK \\(known_failures=1\\)\n')
 
944
 
 
945
    def test_skipped_test(self):
 
946
        # run a test that is skipped, and check the suite as a whole still
 
947
        # succeeds.
 
948
        # skipping_test must be hidden in here so it's not run as a real test
 
949
        def skipping_test():
 
950
            raise TestSkipped('test intentionally skipped')
 
951
 
 
952
        runner = TextTestRunner(stream=self._log_file, keep_output=True)
 
953
        test = unittest.FunctionTestCase(skipping_test)
 
954
        result = self.run_test_runner(runner, test)
 
955
        self.assertTrue(result.wasSuccessful())
 
956
 
 
957
    def test_skipped_from_setup(self):
 
958
        class SkippedSetupTest(TestCase):
 
959
 
 
960
            def setUp(self):
 
961
                self.counter = 1
 
962
                self.addCleanup(self.cleanup)
 
963
                raise TestSkipped('skipped setup')
 
964
 
 
965
            def test_skip(self):
 
966
                self.fail('test reached')
 
967
 
 
968
            def cleanup(self):
 
969
                self.counter -= 1
 
970
 
 
971
        runner = TextTestRunner(stream=self._log_file, keep_output=True)
 
972
        test = SkippedSetupTest('test_skip')
 
973
        result = self.run_test_runner(runner, test)
 
974
        self.assertTrue(result.wasSuccessful())
 
975
        # Check if cleanup was called the right number of times.
 
976
        self.assertEqual(0, test.counter)
 
977
 
 
978
    def test_skipped_from_test(self):
 
979
        class SkippedTest(TestCase):
 
980
 
 
981
            def setUp(self):
 
982
                self.counter = 1
 
983
                self.addCleanup(self.cleanup)
 
984
 
 
985
            def test_skip(self):
 
986
                raise TestSkipped('skipped test')
 
987
 
 
988
            def cleanup(self):
 
989
                self.counter -= 1
 
990
 
 
991
        runner = TextTestRunner(stream=self._log_file, keep_output=True)
 
992
        test = SkippedTest('test_skip')
 
993
        result = self.run_test_runner(runner, test)
 
994
        self.assertTrue(result.wasSuccessful())
 
995
        # Check if cleanup was called the right number of times.
 
996
        self.assertEqual(0, test.counter)
 
997
 
 
998
    def test_unsupported_features_listed(self):
 
999
        """When unsupported features are encountered they are detailed."""
 
1000
        class Feature1(Feature):
 
1001
            def _probe(self): return False
 
1002
        class Feature2(Feature):
 
1003
            def _probe(self): return False
 
1004
        # create sample tests
 
1005
        test1 = SampleTestCase('_test_pass')
 
1006
        test1._test_needs_features = [Feature1()]
 
1007
        test2 = SampleTestCase('_test_pass')
 
1008
        test2._test_needs_features = [Feature2()]
 
1009
        test = unittest.TestSuite()
 
1010
        test.addTest(test1)
 
1011
        test.addTest(test2)
 
1012
        stream = StringIO()
 
1013
        runner = TextTestRunner(stream=stream)
 
1014
        result = self.run_test_runner(runner, test)
 
1015
        lines = stream.getvalue().splitlines()
 
1016
        self.assertEqual([
 
1017
            'OK',
 
1018
            "Missing feature 'Feature1' skipped 1 tests.",
 
1019
            "Missing feature 'Feature2' skipped 1 tests.",
 
1020
            ],
 
1021
            lines[-3:])
 
1022
 
 
1023
    def test_bench_history(self):
 
1024
        # tests that the running the benchmark produces a history file
 
1025
        # containing a timestamp and the revision id of the bzrlib source which
 
1026
        # was tested.
 
1027
        workingtree = _get_bzr_source_tree()
 
1028
        test = TestRunner('dummy_test')
 
1029
        output = StringIO()
 
1030
        runner = TextTestRunner(stream=self._log_file, bench_history=output)
 
1031
        result = self.run_test_runner(runner, test)
 
1032
        output_string = output.getvalue()
 
1033
        self.assertContainsRe(output_string, "--date [0-9.]+")
 
1034
        if workingtree is not None:
 
1035
            revision_id = workingtree.get_parent_ids()[0]
 
1036
            self.assertEndsWith(output_string.rstrip(), revision_id)
 
1037
 
 
1038
    def test_success_log_deleted(self):
 
1039
        """Successful tests have their log deleted"""
 
1040
 
 
1041
        class LogTester(TestCase):
 
1042
 
 
1043
            def test_success(self):
 
1044
                self.log('this will be removed\n')
 
1045
 
 
1046
        sio = cStringIO.StringIO()
 
1047
        runner = TextTestRunner(stream=sio)
 
1048
        test = LogTester('test_success')
 
1049
        result = self.run_test_runner(runner, test)
 
1050
 
 
1051
        log = test._get_log()
 
1052
        self.assertEqual("DELETED log file to reduce memory footprint", log)
 
1053
        self.assertEqual('', test._log_contents)
 
1054
        self.assertIs(None, test._log_file_name)
 
1055
 
 
1056
    def test_fail_log_kept(self):
 
1057
        """Failed tests have their log kept"""
 
1058
 
 
1059
        class LogTester(TestCase):
 
1060
 
 
1061
            def test_fail(self):
 
1062
                self.log('this will be kept\n')
 
1063
                self.fail('this test fails')
 
1064
 
 
1065
        sio = cStringIO.StringIO()
 
1066
        runner = TextTestRunner(stream=sio)
 
1067
        test = LogTester('test_fail')
 
1068
        result = self.run_test_runner(runner, test)
 
1069
 
 
1070
        text = sio.getvalue()
 
1071
        self.assertContainsRe(text, 'this will be kept')
 
1072
        self.assertContainsRe(text, 'this test fails')
 
1073
 
 
1074
        log = test._get_log()
 
1075
        self.assertContainsRe(log, 'this will be kept')
 
1076
        self.assertEqual(log, test._log_contents)
 
1077
 
 
1078
    def test_error_log_kept(self):
 
1079
        """Tests with errors have their log kept"""
 
1080
 
 
1081
        class LogTester(TestCase):
 
1082
 
 
1083
            def test_error(self):
 
1084
                self.log('this will be kept\n')
 
1085
                raise ValueError('random exception raised')
 
1086
 
 
1087
        sio = cStringIO.StringIO()
 
1088
        runner = TextTestRunner(stream=sio)
 
1089
        test = LogTester('test_error')
 
1090
        result = self.run_test_runner(runner, test)
 
1091
 
 
1092
        text = sio.getvalue()
 
1093
        self.assertContainsRe(text, 'this will be kept')
 
1094
        self.assertContainsRe(text, 'random exception raised')
 
1095
 
 
1096
        log = test._get_log()
 
1097
        self.assertContainsRe(log, 'this will be kept')
 
1098
        self.assertEqual(log, test._log_contents)
 
1099
 
 
1100
 
 
1101
class SampleTestCase(TestCase):
 
1102
 
 
1103
    def _test_pass(self):
 
1104
        pass
 
1105
 
 
1106
 
 
1107
class TestTestCase(TestCase):
 
1108
    """Tests that test the core bzrlib TestCase."""
 
1109
 
 
1110
    def inner_test(self):
 
1111
        # the inner child test
 
1112
        note("inner_test")
 
1113
 
 
1114
    def outer_child(self):
 
1115
        # the outer child test
 
1116
        note("outer_start")
 
1117
        self.inner_test = TestTestCase("inner_child")
 
1118
        result = bzrlib.tests.TextTestResult(self._log_file,
 
1119
                                        descriptions=0,
 
1120
                                        verbosity=1)
 
1121
        self.inner_test.run(result)
 
1122
        note("outer finish")
 
1123
 
 
1124
    def test_trace_nesting(self):
 
1125
        # this tests that each test case nests its trace facility correctly.
 
1126
        # we do this by running a test case manually. That test case (A)
 
1127
        # should setup a new log, log content to it, setup a child case (B),
 
1128
        # which should log independently, then case (A) should log a trailer
 
1129
        # and return.
 
1130
        # we do two nested children so that we can verify the state of the 
 
1131
        # logs after the outer child finishes is correct, which a bad clean
 
1132
        # up routine in tearDown might trigger a fault in our test with only
 
1133
        # one child, we should instead see the bad result inside our test with
 
1134
        # the two children.
 
1135
        # the outer child test
 
1136
        original_trace = bzrlib.trace._trace_file
 
1137
        outer_test = TestTestCase("outer_child")
 
1138
        result = bzrlib.tests.TextTestResult(self._log_file,
 
1139
                                        descriptions=0,
 
1140
                                        verbosity=1)
 
1141
        outer_test.run(result)
 
1142
        self.assertEqual(original_trace, bzrlib.trace._trace_file)
 
1143
 
 
1144
    def method_that_times_a_bit_twice(self):
 
1145
        # call self.time twice to ensure it aggregates
 
1146
        self.time(time.sleep, 0.007)
 
1147
        self.time(time.sleep, 0.007)
 
1148
 
 
1149
    def test_time_creates_benchmark_in_result(self):
 
1150
        """Test that the TestCase.time() method accumulates a benchmark time."""
 
1151
        sample_test = TestTestCase("method_that_times_a_bit_twice")
 
1152
        output_stream = StringIO()
 
1153
        result = bzrlib.tests.VerboseTestResult(
 
1154
            unittest._WritelnDecorator(output_stream),
 
1155
            descriptions=0,
 
1156
            verbosity=2,
 
1157
            num_tests=sample_test.countTestCases())
 
1158
        sample_test.run(result)
 
1159
        self.assertContainsRe(
 
1160
            output_stream.getvalue(),
 
1161
            r"\d+ms/ +\d+ms\n$")
 
1162
 
 
1163
    def test_hooks_sanitised(self):
 
1164
        """The bzrlib hooks should be sanitised by setUp."""
 
1165
        self.assertEqual(bzrlib.branch.BranchHooks(),
 
1166
            bzrlib.branch.Branch.hooks)
 
1167
        self.assertEqual(bzrlib.smart.server.SmartServerHooks(),
 
1168
            bzrlib.smart.server.SmartTCPServer.hooks)
 
1169
 
 
1170
    def test__gather_lsprof_in_benchmarks(self):
 
1171
        """When _gather_lsprof_in_benchmarks is on, accumulate profile data.
 
1172
        
 
1173
        Each self.time() call is individually and separately profiled.
 
1174
        """
 
1175
        try:
 
1176
            import bzrlib.lsprof
 
1177
        except ImportError:
 
1178
            raise TestSkipped("lsprof not installed.")
 
1179
        # overrides the class member with an instance member so no cleanup 
 
1180
        # needed.
 
1181
        self._gather_lsprof_in_benchmarks = True
 
1182
        self.time(time.sleep, 0.000)
 
1183
        self.time(time.sleep, 0.003)
 
1184
        self.assertEqual(2, len(self._benchcalls))
 
1185
        self.assertEqual((time.sleep, (0.000,), {}), self._benchcalls[0][0])
 
1186
        self.assertEqual((time.sleep, (0.003,), {}), self._benchcalls[1][0])
 
1187
        self.assertIsInstance(self._benchcalls[0][1], bzrlib.lsprof.Stats)
 
1188
        self.assertIsInstance(self._benchcalls[1][1], bzrlib.lsprof.Stats)
 
1189
 
 
1190
    def test_knownFailure(self):
 
1191
        """Self.knownFailure() should raise a KnownFailure exception."""
 
1192
        self.assertRaises(KnownFailure, self.knownFailure, "A Failure")
 
1193
 
 
1194
    def test_requireFeature_available(self):
 
1195
        """self.requireFeature(available) is a no-op."""
 
1196
        class Available(Feature):
 
1197
            def _probe(self):return True
 
1198
        feature = Available()
 
1199
        self.requireFeature(feature)
 
1200
 
 
1201
    def test_requireFeature_unavailable(self):
 
1202
        """self.requireFeature(unavailable) raises UnavailableFeature."""
 
1203
        class Unavailable(Feature):
 
1204
            def _probe(self):return False
 
1205
        feature = Unavailable()
 
1206
        self.assertRaises(UnavailableFeature, self.requireFeature, feature)
 
1207
 
 
1208
    def test_run_no_parameters(self):
 
1209
        test = SampleTestCase('_test_pass')
 
1210
        test.run()
 
1211
    
 
1212
    def test_run_enabled_unittest_result(self):
 
1213
        """Test we revert to regular behaviour when the test is enabled."""
 
1214
        test = SampleTestCase('_test_pass')
 
1215
        class EnabledFeature(object):
 
1216
            def available(self):
 
1217
                return True
 
1218
        test._test_needs_features = [EnabledFeature()]
 
1219
        result = unittest.TestResult()
 
1220
        test.run(result)
 
1221
        self.assertEqual(1, result.testsRun)
 
1222
        self.assertEqual([], result.errors)
 
1223
        self.assertEqual([], result.failures)
 
1224
 
 
1225
    def test_run_disabled_unittest_result(self):
 
1226
        """Test our compatability for disabled tests with unittest results."""
 
1227
        test = SampleTestCase('_test_pass')
 
1228
        class DisabledFeature(object):
 
1229
            def available(self):
 
1230
                return False
 
1231
        test._test_needs_features = [DisabledFeature()]
 
1232
        result = unittest.TestResult()
 
1233
        test.run(result)
 
1234
        self.assertEqual(1, result.testsRun)
 
1235
        self.assertEqual([], result.errors)
 
1236
        self.assertEqual([], result.failures)
 
1237
 
 
1238
    def test_run_disabled_supporting_result(self):
 
1239
        """Test disabled tests behaviour with support aware results."""
 
1240
        test = SampleTestCase('_test_pass')
 
1241
        class DisabledFeature(object):
 
1242
            def available(self):
 
1243
                return False
 
1244
        the_feature = DisabledFeature()
 
1245
        test._test_needs_features = [the_feature]
 
1246
        class InstrumentedTestResult(unittest.TestResult):
 
1247
            def __init__(self):
 
1248
                unittest.TestResult.__init__(self)
 
1249
                self.calls = []
 
1250
            def startTest(self, test):
 
1251
                self.calls.append(('startTest', test))
 
1252
            def stopTest(self, test):
 
1253
                self.calls.append(('stopTest', test))
 
1254
            def addNotSupported(self, test, feature):
 
1255
                self.calls.append(('addNotSupported', test, feature))
 
1256
        result = InstrumentedTestResult()
 
1257
        test.run(result)
 
1258
        self.assertEqual([
 
1259
            ('startTest', test),
 
1260
            ('addNotSupported', test, the_feature),
 
1261
            ('stopTest', test),
 
1262
            ],
 
1263
            result.calls)
 
1264
 
 
1265
 
 
1266
@symbol_versioning.deprecated_function(zero_eleven)
 
1267
def sample_deprecated_function():
 
1268
    """A deprecated function to test applyDeprecated with."""
 
1269
    return 2
 
1270
 
 
1271
 
 
1272
def sample_undeprecated_function(a_param):
 
1273
    """A undeprecated function to test applyDeprecated with."""
 
1274
 
 
1275
 
 
1276
class ApplyDeprecatedHelper(object):
 
1277
    """A helper class for ApplyDeprecated tests."""
 
1278
 
 
1279
    @symbol_versioning.deprecated_method(zero_eleven)
 
1280
    def sample_deprecated_method(self, param_one):
 
1281
        """A deprecated method for testing with."""
 
1282
        return param_one
 
1283
 
 
1284
    def sample_normal_method(self):
 
1285
        """A undeprecated method."""
 
1286
 
 
1287
    @symbol_versioning.deprecated_method(zero_ten)
 
1288
    def sample_nested_deprecation(self):
 
1289
        return sample_deprecated_function()
 
1290
 
 
1291
 
 
1292
class TestExtraAssertions(TestCase):
 
1293
    """Tests for new test assertions in bzrlib test suite"""
 
1294
 
 
1295
    def test_assert_isinstance(self):
 
1296
        self.assertIsInstance(2, int)
 
1297
        self.assertIsInstance(u'', basestring)
 
1298
        self.assertRaises(AssertionError, self.assertIsInstance, None, int)
 
1299
        self.assertRaises(AssertionError, self.assertIsInstance, 23.3, int)
 
1300
 
 
1301
    def test_assertEndsWith(self):
 
1302
        self.assertEndsWith('foo', 'oo')
 
1303
        self.assertRaises(AssertionError, self.assertEndsWith, 'o', 'oo')
 
1304
 
 
1305
    def test_applyDeprecated_not_deprecated(self):
 
1306
        sample_object = ApplyDeprecatedHelper()
 
1307
        # calling an undeprecated callable raises an assertion
 
1308
        self.assertRaises(AssertionError, self.applyDeprecated, zero_eleven,
 
1309
            sample_object.sample_normal_method)
 
1310
        self.assertRaises(AssertionError, self.applyDeprecated, zero_eleven,
 
1311
            sample_undeprecated_function, "a param value")
 
1312
        # calling a deprecated callable (function or method) with the wrong
 
1313
        # expected deprecation fails.
 
1314
        self.assertRaises(AssertionError, self.applyDeprecated, zero_ten,
 
1315
            sample_object.sample_deprecated_method, "a param value")
 
1316
        self.assertRaises(AssertionError, self.applyDeprecated, zero_ten,
 
1317
            sample_deprecated_function)
 
1318
        # calling a deprecated callable (function or method) with the right
 
1319
        # expected deprecation returns the functions result.
 
1320
        self.assertEqual("a param value", self.applyDeprecated(zero_eleven,
 
1321
            sample_object.sample_deprecated_method, "a param value"))
 
1322
        self.assertEqual(2, self.applyDeprecated(zero_eleven,
 
1323
            sample_deprecated_function))
 
1324
        # calling a nested deprecation with the wrong deprecation version
 
1325
        # fails even if a deeper nested function was deprecated with the 
 
1326
        # supplied version.
 
1327
        self.assertRaises(AssertionError, self.applyDeprecated,
 
1328
            zero_eleven, sample_object.sample_nested_deprecation)
 
1329
        # calling a nested deprecation with the right deprecation value
 
1330
        # returns the calls result.
 
1331
        self.assertEqual(2, self.applyDeprecated(zero_ten,
 
1332
            sample_object.sample_nested_deprecation))
 
1333
 
 
1334
    def test_callDeprecated(self):
 
1335
        def testfunc(be_deprecated, result=None):
 
1336
            if be_deprecated is True:
 
1337
                symbol_versioning.warn('i am deprecated', DeprecationWarning, 
 
1338
                                       stacklevel=1)
 
1339
            return result
 
1340
        result = self.callDeprecated(['i am deprecated'], testfunc, True)
 
1341
        self.assertIs(None, result)
 
1342
        result = self.callDeprecated([], testfunc, False, 'result')
 
1343
        self.assertEqual('result', result)
 
1344
        self.callDeprecated(['i am deprecated'], testfunc, be_deprecated=True)
 
1345
        self.callDeprecated([], testfunc, be_deprecated=False)
 
1346
 
 
1347
 
 
1348
class TestConvenienceMakers(TestCaseWithTransport):
 
1349
    """Test for the make_* convenience functions."""
 
1350
 
 
1351
    def test_make_branch_and_tree_with_format(self):
 
1352
        # we should be able to supply a format to make_branch_and_tree
 
1353
        self.make_branch_and_tree('a', format=bzrlib.bzrdir.BzrDirMetaFormat1())
 
1354
        self.make_branch_and_tree('b', format=bzrlib.bzrdir.BzrDirFormat6())
 
1355
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('a')._format,
 
1356
                              bzrlib.bzrdir.BzrDirMetaFormat1)
 
1357
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('b')._format,
 
1358
                              bzrlib.bzrdir.BzrDirFormat6)
 
1359
 
 
1360
    def test_make_branch_and_memory_tree(self):
 
1361
        # we should be able to get a new branch and a mutable tree from
 
1362
        # TestCaseWithTransport
 
1363
        tree = self.make_branch_and_memory_tree('a')
 
1364
        self.assertIsInstance(tree, bzrlib.memorytree.MemoryTree)
 
1365
 
 
1366
 
 
1367
class TestSFTPMakeBranchAndTree(TestCaseWithSFTPServer):
 
1368
 
 
1369
    def test_make_tree_for_sftp_branch(self):
 
1370
        """Transports backed by local directories create local trees."""
 
1371
 
 
1372
        tree = self.make_branch_and_tree('t1')
 
1373
        base = tree.bzrdir.root_transport.base
 
1374
        self.failIf(base.startswith('sftp'),
 
1375
                'base %r is on sftp but should be local' % base)
 
1376
        self.assertEquals(tree.bzrdir.root_transport,
 
1377
                tree.branch.bzrdir.root_transport)
 
1378
        self.assertEquals(tree.bzrdir.root_transport,
 
1379
                tree.branch.repository.bzrdir.root_transport)
 
1380
 
 
1381
 
 
1382
class TestSelftest(TestCase):
 
1383
    """Tests of bzrlib.tests.selftest."""
 
1384
 
 
1385
    def test_selftest_benchmark_parameter_invokes_test_suite__benchmark__(self):
 
1386
        factory_called = []
 
1387
        def factory():
 
1388
            factory_called.append(True)
 
1389
            return TestSuite()
 
1390
        out = StringIO()
 
1391
        err = StringIO()
 
1392
        self.apply_redirected(out, err, None, bzrlib.tests.selftest, 
 
1393
            test_suite_factory=factory)
 
1394
        self.assertEqual([True], factory_called)
 
1395
 
 
1396
 
 
1397
class TestSelftestCleanOutput(TestCaseInTempDir):
 
1398
 
 
1399
    def test_clean_output(self):
 
1400
        # test functionality of clean_selftest_output()
 
1401
        self.build_tree(['test0000.tmp/', 'test0001.tmp/',
 
1402
                         'bzrlib/', 'tests/',
 
1403
                         'bzr', 'setup.py', 'test9999.tmp'])
 
1404
 
 
1405
        root = os.getcwdu()
 
1406
        before = os.listdir(root)
 
1407
        before.sort()
 
1408
        self.assertEquals(['bzr','bzrlib','setup.py',
 
1409
                           'test0000.tmp','test0001.tmp',
 
1410
                           'test9999.tmp','tests'],
 
1411
                           before)
 
1412
        clean_selftest_output(root, quiet=True)
 
1413
        after = os.listdir(root)
 
1414
        after.sort()
 
1415
        self.assertEquals(['bzr','bzrlib','setup.py',
 
1416
                           'test9999.tmp','tests'],
 
1417
                           after)
 
1418
 
 
1419
    def test_clean_readonly(self):
 
1420
        # test for delete read-only files
 
1421
        self.build_tree(['test0000.tmp/', 'test0000.tmp/foo'])
 
1422
        osutils.make_readonly('test0000.tmp/foo')
 
1423
        root = os.getcwdu()
 
1424
        before = os.listdir(root);  before.sort()
 
1425
        self.assertEquals(['test0000.tmp'], before)
 
1426
        clean_selftest_output(root, quiet=True)
 
1427
        after = os.listdir(root);   after.sort()
 
1428
        self.assertEquals([], after)
 
1429
 
 
1430
 
 
1431
class TestKnownFailure(TestCase):
 
1432
 
 
1433
    def test_known_failure(self):
 
1434
        """Check that KnownFailure is defined appropriately."""
 
1435
        # a KnownFailure is an assertion error for compatability with unaware
 
1436
        # runners.
 
1437
        self.assertIsInstance(KnownFailure(""), AssertionError)
 
1438
 
 
1439
    def test_expect_failure(self):
 
1440
        try:
 
1441
            self.expectFailure("Doomed to failure", self.assertTrue, False)
 
1442
        except KnownFailure, e:
 
1443
            self.assertEqual('Doomed to failure', e.args[0])
 
1444
        try:
 
1445
            self.expectFailure("Doomed to failure", self.assertTrue, True)
 
1446
        except AssertionError, e:
 
1447
            self.assertEqual('Unexpected success.  Should have failed:'
 
1448
                             ' Doomed to failure', e.args[0])
 
1449
        else:
 
1450
            self.fail('Assertion not raised')
 
1451
 
 
1452
 
 
1453
class TestFeature(TestCase):
 
1454
 
 
1455
    def test_caching(self):
 
1456
        """Feature._probe is called by the feature at most once."""
 
1457
        class InstrumentedFeature(Feature):
 
1458
            def __init__(self):
 
1459
                Feature.__init__(self)
 
1460
                self.calls = []
 
1461
            def _probe(self):
 
1462
                self.calls.append('_probe')
 
1463
                return False
 
1464
        feature = InstrumentedFeature()
 
1465
        feature.available()
 
1466
        self.assertEqual(['_probe'], feature.calls)
 
1467
        feature.available()
 
1468
        self.assertEqual(['_probe'], feature.calls)
 
1469
 
 
1470
    def test_named_str(self):
 
1471
        """Feature.__str__ should thunk to feature_name()."""
 
1472
        class NamedFeature(Feature):
 
1473
            def feature_name(self):
 
1474
                return 'symlinks'
 
1475
        feature = NamedFeature()
 
1476
        self.assertEqual('symlinks', str(feature))
 
1477
 
 
1478
    def test_default_str(self):
 
1479
        """Feature.__str__ should default to __class__.__name__."""
 
1480
        class NamedFeature(Feature):
 
1481
            pass
 
1482
        feature = NamedFeature()
 
1483
        self.assertEqual('NamedFeature', str(feature))
 
1484
 
 
1485
 
 
1486
class TestUnavailableFeature(TestCase):
 
1487
 
 
1488
    def test_access_feature(self):
 
1489
        feature = Feature()
 
1490
        exception = UnavailableFeature(feature)
 
1491
        self.assertIs(feature, exception.args[0])