/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

Upgraded to the latest bzr.dev

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