~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Martin Pool
  • Date: 2005-05-12 02:18:48 UTC
  • Revision ID: mbp@sourcefrog.net-20050512021848-d1a727373aee2c85
- WorkingTree loads statcache in constructor and holds
  it permanently

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.assertEqual(lines, ['XFAIL                   0ms', '    foo'])
743
 
    
744
 
    def test_text_report_known_failure(self):
745
 
        # text test output formatting
746
 
        pb = MockProgress()
747
 
        result = bzrlib.tests.TextTestResult(
748
 
            None,
749
 
            descriptions=0,
750
 
            verbosity=1,
751
 
            pb=pb,
752
 
            )
753
 
        test = self.get_passing_test()
754
 
        # this seeds the state to handle reporting the test.
755
 
        result.startTest(test)
756
 
        result.extractBenchmarkTime(test)
757
 
        # the err parameter has the shape:
758
 
        # (class, exception object, traceback)
759
 
        # KnownFailures dont get their tracebacks shown though, so we
760
 
        # can skip that.
761
 
        err = (KnownFailure, KnownFailure('foo'), None)
762
 
        result.report_known_failure(test, err)
763
 
        self.assertEqual(
764
 
            [
765
 
            ('update', '[1 in 0s] passing_test', None, None),
766
 
            ('note', 'XFAIL: %s\n%s\n', ('passing_test', err[1]))
767
 
            ],
768
 
            pb.calls)
769
 
        # known_failures should be printed in the summary, so if we run a test
770
 
        # after there are some known failures, the update prefix should match
771
 
        # this.
772
 
        result.known_failure_count = 3
773
 
        test.run(result)
774
 
        self.assertEqual(
775
 
            [
776
 
            ('update', '[2 in 0s, 3 known failures] passing_test', None, None),
777
 
            ],
778
 
            pb.calls[2:])
779
 
 
780
 
    def get_passing_test(self):
781
 
        """Return a test object that can't be run usefully."""
782
 
        def passing_test():
783
 
            pass
784
 
        return unittest.FunctionTestCase(passing_test)
785
 
 
786
 
    def test_add_not_supported(self):
787
 
        """Test the behaviour of invoking addNotSupported."""
788
 
        class InstrumentedTestResult(ExtendedTestResult):
789
 
            def report_test_start(self, test): pass
790
 
            def report_unsupported(self, test, feature):
791
 
                self._call = test, feature
792
 
        result = InstrumentedTestResult(None, None, None, None)
793
 
        test = SampleTestCase('_test_pass')
794
 
        feature = Feature()
795
 
        result.startTest(test)
796
 
        result.addNotSupported(test, feature)
797
 
        # it should invoke 'report_unsupported'.
798
 
        self.assertEqual(2, len(result._call))
799
 
        self.assertEqual(test, result._call[0])
800
 
        self.assertEqual(feature, result._call[1])
801
 
        # the result should be successful.
802
 
        self.assertTrue(result.wasSuccessful())
803
 
        # it should record the test against a count of tests not run due to
804
 
        # this feature.
805
 
        self.assertEqual(1, result.unsupported['Feature'])
806
 
        # and invoking it again should increment that counter
807
 
        result.addNotSupported(test, feature)
808
 
        self.assertEqual(2, result.unsupported['Feature'])
809
 
 
810
 
    def test_verbose_report_unsupported(self):
811
 
        # verbose test output formatting
812
 
        result_stream = StringIO()
813
 
        result = bzrlib.tests.VerboseTestResult(
814
 
            unittest._WritelnDecorator(result_stream),
815
 
            descriptions=0,
816
 
            verbosity=2,
817
 
            )
818
 
        test = self.get_passing_test()
819
 
        feature = Feature()
820
 
        result.startTest(test)
821
 
        result.extractBenchmarkTime(test)
822
 
        prefix = len(result_stream.getvalue())
823
 
        result.report_unsupported(test, feature)
824
 
        output = result_stream.getvalue()[prefix:]
825
 
        lines = output.splitlines()
826
 
        self.assertEqual(lines, ['NODEP                   0ms', "    The feature 'Feature' is not available."])
827
 
    
828
 
    def test_text_report_unsupported(self):
829
 
        # text test output formatting
830
 
        pb = MockProgress()
831
 
        result = bzrlib.tests.TextTestResult(
832
 
            None,
833
 
            descriptions=0,
834
 
            verbosity=1,
835
 
            pb=pb,
836
 
            )
837
 
        test = self.get_passing_test()
838
 
        feature = Feature()
839
 
        # this seeds the state to handle reporting the test.
840
 
        result.startTest(test)
841
 
        result.extractBenchmarkTime(test)
842
 
        result.report_unsupported(test, feature)
843
 
        # no output on unsupported features
844
 
        self.assertEqual(
845
 
            [('update', '[1 in 0s] passing_test', None, None)
846
 
            ],
847
 
            pb.calls)
848
 
        # the number of missing features should be printed in the progress
849
 
        # summary, so check for that.
850
 
        result.unsupported = {'foo':0, 'bar':0}
851
 
        test.run(result)
852
 
        self.assertEqual(
853
 
            [
854
 
            ('update', '[2 in 0s, 2 missing features] passing_test', None, None),
855
 
            ],
856
 
            pb.calls[1:])
857
 
    
858
 
    def test_unavailable_exception(self):
859
 
        """An UnavailableFeature being raised should invoke addNotSupported."""
860
 
        class InstrumentedTestResult(ExtendedTestResult):
861
 
 
862
 
            def report_test_start(self, test): pass
863
 
            def addNotSupported(self, test, feature):
864
 
                self._call = test, feature
865
 
        result = InstrumentedTestResult(None, None, None, None)
866
 
        feature = Feature()
867
 
        def test_function():
868
 
            raise UnavailableFeature(feature)
869
 
        test = unittest.FunctionTestCase(test_function)
870
 
        test.run(result)
871
 
        # it should invoke 'addNotSupported'.
872
 
        self.assertEqual(2, len(result._call))
873
 
        self.assertEqual(test, result._call[0])
874
 
        self.assertEqual(feature, result._call[1])
875
 
        # and not count as an error
876
 
        self.assertEqual(0, result.error_count)
877
 
 
878
 
 
879
 
class TestRunner(TestCase):
880
 
 
881
 
    def dummy_test(self):
882
 
        pass
883
 
 
884
 
    def run_test_runner(self, testrunner, test):
885
 
        """Run suite in testrunner, saving global state and restoring it.
886
 
 
887
 
        This current saves and restores:
888
 
        TestCaseInTempDir.TEST_ROOT
889
 
        
890
 
        There should be no tests in this file that use bzrlib.tests.TextTestRunner
891
 
        without using this convenience method, because of our use of global state.
892
 
        """
893
 
        old_root = TestCaseInTempDir.TEST_ROOT
894
 
        try:
895
 
            TestCaseInTempDir.TEST_ROOT = None
896
 
            return testrunner.run(test)
897
 
        finally:
898
 
            TestCaseInTempDir.TEST_ROOT = old_root
899
 
 
900
 
    def test_known_failure_failed_run(self):
901
 
        # run a test that generates a known failure which should be printed in
902
 
        # the final output when real failures occur.
903
 
        def known_failure_test():
904
 
            raise KnownFailure('failed')
905
 
        test = unittest.TestSuite()
906
 
        test.addTest(unittest.FunctionTestCase(known_failure_test))
907
 
        def failing_test():
908
 
            raise AssertionError('foo')
909
 
        test.addTest(unittest.FunctionTestCase(failing_test))
910
 
        stream = StringIO()
911
 
        runner = TextTestRunner(stream=stream)
912
 
        result = self.run_test_runner(runner, test)
913
 
        lines = stream.getvalue().splitlines()
914
 
        self.assertEqual([
915
 
            '',
916
 
            '======================================================================',
917
 
            'FAIL: unittest.FunctionTestCase (failing_test)',
918
 
            '----------------------------------------------------------------------',
919
 
            'Traceback (most recent call last):',
920
 
            '    raise AssertionError(\'foo\')',
921
 
            'AssertionError: foo',
922
 
            '',
923
 
            '----------------------------------------------------------------------',
924
 
            '',
925
 
            'FAILED (failures=1, known_failure_count=1)'],
926
 
            lines[0:5] + lines[6:10] + lines[11:])
927
 
 
928
 
    def test_known_failure_ok_run(self):
929
 
        # run a test that generates a known failure which should be printed in the final output.
930
 
        def known_failure_test():
931
 
            raise KnownFailure('failed')
932
 
        test = unittest.FunctionTestCase(known_failure_test)
933
 
        stream = StringIO()
934
 
        runner = TextTestRunner(stream=stream)
935
 
        result = self.run_test_runner(runner, test)
936
 
        self.assertEqual(
937
 
            '\n'
938
 
            '----------------------------------------------------------------------\n'
939
 
            'Ran 1 test in 0.000s\n'
940
 
            '\n'
941
 
            'OK (known_failures=1)\n',
942
 
            stream.getvalue())
943
 
 
944
 
    def test_skipped_test(self):
945
 
        # run a test that is skipped, and check the suite as a whole still
946
 
        # succeeds.
947
 
        # skipping_test must be hidden in here so it's not run as a real test
948
 
        def skipping_test():
949
 
            raise TestSkipped('test intentionally skipped')
950
 
 
951
 
        runner = TextTestRunner(stream=self._log_file, keep_output=True)
952
 
        test = unittest.FunctionTestCase(skipping_test)
953
 
        result = self.run_test_runner(runner, test)
954
 
        self.assertTrue(result.wasSuccessful())
955
 
 
956
 
    def test_skipped_from_setup(self):
957
 
        class SkippedSetupTest(TestCase):
958
 
 
959
 
            def setUp(self):
960
 
                self.counter = 1
961
 
                self.addCleanup(self.cleanup)
962
 
                raise TestSkipped('skipped setup')
963
 
 
964
 
            def test_skip(self):
965
 
                self.fail('test reached')
966
 
 
967
 
            def cleanup(self):
968
 
                self.counter -= 1
969
 
 
970
 
        runner = TextTestRunner(stream=self._log_file, keep_output=True)
971
 
        test = SkippedSetupTest('test_skip')
972
 
        result = self.run_test_runner(runner, test)
973
 
        self.assertTrue(result.wasSuccessful())
974
 
        # Check if cleanup was called the right number of times.
975
 
        self.assertEqual(0, test.counter)
976
 
 
977
 
    def test_skipped_from_test(self):
978
 
        class SkippedTest(TestCase):
979
 
 
980
 
            def setUp(self):
981
 
                self.counter = 1
982
 
                self.addCleanup(self.cleanup)
983
 
 
984
 
            def test_skip(self):
985
 
                raise TestSkipped('skipped test')
986
 
 
987
 
            def cleanup(self):
988
 
                self.counter -= 1
989
 
 
990
 
        runner = TextTestRunner(stream=self._log_file, keep_output=True)
991
 
        test = SkippedTest('test_skip')
992
 
        result = self.run_test_runner(runner, test)
993
 
        self.assertTrue(result.wasSuccessful())
994
 
        # Check if cleanup was called the right number of times.
995
 
        self.assertEqual(0, test.counter)
996
 
 
997
 
    def test_unsupported_features_listed(self):
998
 
        """When unsupported features are encountered they are detailed."""
999
 
        class Feature1(Feature):
1000
 
            def _probe(self): return False
1001
 
        class Feature2(Feature):
1002
 
            def _probe(self): return False
1003
 
        # create sample tests
1004
 
        test1 = SampleTestCase('_test_pass')
1005
 
        test1._test_needs_features = [Feature1()]
1006
 
        test2 = SampleTestCase('_test_pass')
1007
 
        test2._test_needs_features = [Feature2()]
1008
 
        test = unittest.TestSuite()
1009
 
        test.addTest(test1)
1010
 
        test.addTest(test2)
1011
 
        stream = StringIO()
1012
 
        runner = TextTestRunner(stream=stream)
1013
 
        result = self.run_test_runner(runner, test)
1014
 
        lines = stream.getvalue().splitlines()
1015
 
        self.assertEqual([
1016
 
            'OK',
1017
 
            "Missing feature 'Feature1' skipped 1 tests.",
1018
 
            "Missing feature 'Feature2' skipped 1 tests.",
1019
 
            ],
1020
 
            lines[-3:])
1021
 
 
1022
 
    def test_bench_history(self):
1023
 
        # tests that the running the benchmark produces a history file
1024
 
        # containing a timestamp and the revision id of the bzrlib source which
1025
 
        # was tested.
1026
 
        workingtree = _get_bzr_source_tree()
1027
 
        test = TestRunner('dummy_test')
1028
 
        output = StringIO()
1029
 
        runner = TextTestRunner(stream=self._log_file, bench_history=output)
1030
 
        result = self.run_test_runner(runner, test)
1031
 
        output_string = output.getvalue()
1032
 
        self.assertContainsRe(output_string, "--date [0-9.]+")
1033
 
        if workingtree is not None:
1034
 
            revision_id = workingtree.get_parent_ids()[0]
1035
 
            self.assertEndsWith(output_string.rstrip(), revision_id)
1036
 
 
1037
 
    def test_success_log_deleted(self):
1038
 
        """Successful tests have their log deleted"""
1039
 
 
1040
 
        class LogTester(TestCase):
1041
 
 
1042
 
            def test_success(self):
1043
 
                self.log('this will be removed\n')
1044
 
 
1045
 
        sio = cStringIO.StringIO()
1046
 
        runner = TextTestRunner(stream=sio)
1047
 
        test = LogTester('test_success')
1048
 
        result = self.run_test_runner(runner, test)
1049
 
 
1050
 
        log = test._get_log()
1051
 
        self.assertEqual("DELETED log file to reduce memory footprint", log)
1052
 
        self.assertEqual('', test._log_contents)
1053
 
        self.assertIs(None, test._log_file_name)
1054
 
 
1055
 
    def test_fail_log_kept(self):
1056
 
        """Failed tests have their log kept"""
1057
 
 
1058
 
        class LogTester(TestCase):
1059
 
 
1060
 
            def test_fail(self):
1061
 
                self.log('this will be kept\n')
1062
 
                self.fail('this test fails')
1063
 
 
1064
 
        sio = cStringIO.StringIO()
1065
 
        runner = TextTestRunner(stream=sio)
1066
 
        test = LogTester('test_fail')
1067
 
        result = self.run_test_runner(runner, test)
1068
 
 
1069
 
        text = sio.getvalue()
1070
 
        self.assertContainsRe(text, 'this will be kept')
1071
 
        self.assertContainsRe(text, 'this test fails')
1072
 
 
1073
 
        log = test._get_log()
1074
 
        self.assertContainsRe(log, 'this will be kept')
1075
 
        self.assertEqual(log, test._log_contents)
1076
 
 
1077
 
    def test_error_log_kept(self):
1078
 
        """Tests with errors have their log kept"""
1079
 
 
1080
 
        class LogTester(TestCase):
1081
 
 
1082
 
            def test_error(self):
1083
 
                self.log('this will be kept\n')
1084
 
                raise ValueError('random exception raised')
1085
 
 
1086
 
        sio = cStringIO.StringIO()
1087
 
        runner = TextTestRunner(stream=sio)
1088
 
        test = LogTester('test_error')
1089
 
        result = self.run_test_runner(runner, test)
1090
 
 
1091
 
        text = sio.getvalue()
1092
 
        self.assertContainsRe(text, 'this will be kept')
1093
 
        self.assertContainsRe(text, 'random exception raised')
1094
 
 
1095
 
        log = test._get_log()
1096
 
        self.assertContainsRe(log, 'this will be kept')
1097
 
        self.assertEqual(log, test._log_contents)
1098
 
 
1099
 
 
1100
 
class SampleTestCase(TestCase):
1101
 
 
1102
 
    def _test_pass(self):
1103
 
        pass
1104
 
 
1105
 
 
1106
 
class TestTestCase(TestCase):
1107
 
    """Tests that test the core bzrlib TestCase."""
1108
 
 
1109
 
    def inner_test(self):
1110
 
        # the inner child test
1111
 
        note("inner_test")
1112
 
 
1113
 
    def outer_child(self):
1114
 
        # the outer child test
1115
 
        note("outer_start")
1116
 
        self.inner_test = TestTestCase("inner_child")
1117
 
        result = bzrlib.tests.TextTestResult(self._log_file,
1118
 
                                        descriptions=0,
1119
 
                                        verbosity=1)
1120
 
        self.inner_test.run(result)
1121
 
        note("outer finish")
1122
 
 
1123
 
    def test_trace_nesting(self):
1124
 
        # this tests that each test case nests its trace facility correctly.
1125
 
        # we do this by running a test case manually. That test case (A)
1126
 
        # should setup a new log, log content to it, setup a child case (B),
1127
 
        # which should log independently, then case (A) should log a trailer
1128
 
        # and return.
1129
 
        # we do two nested children so that we can verify the state of the 
1130
 
        # logs after the outer child finishes is correct, which a bad clean
1131
 
        # up routine in tearDown might trigger a fault in our test with only
1132
 
        # one child, we should instead see the bad result inside our test with
1133
 
        # the two children.
1134
 
        # the outer child test
1135
 
        original_trace = bzrlib.trace._trace_file
1136
 
        outer_test = TestTestCase("outer_child")
1137
 
        result = bzrlib.tests.TextTestResult(self._log_file,
1138
 
                                        descriptions=0,
1139
 
                                        verbosity=1)
1140
 
        outer_test.run(result)
1141
 
        self.assertEqual(original_trace, bzrlib.trace._trace_file)
1142
 
 
1143
 
    def method_that_times_a_bit_twice(self):
1144
 
        # call self.time twice to ensure it aggregates
1145
 
        self.time(time.sleep, 0.007)
1146
 
        self.time(time.sleep, 0.007)
1147
 
 
1148
 
    def test_time_creates_benchmark_in_result(self):
1149
 
        """Test that the TestCase.time() method accumulates a benchmark time."""
1150
 
        sample_test = TestTestCase("method_that_times_a_bit_twice")
1151
 
        output_stream = StringIO()
1152
 
        result = bzrlib.tests.VerboseTestResult(
1153
 
            unittest._WritelnDecorator(output_stream),
1154
 
            descriptions=0,
1155
 
            verbosity=2,
1156
 
            num_tests=sample_test.countTestCases())
1157
 
        sample_test.run(result)
1158
 
        self.assertContainsRe(
1159
 
            output_stream.getvalue(),
1160
 
            r"\d+ms/ +\d+ms\n$")
1161
 
 
1162
 
    def test_hooks_sanitised(self):
1163
 
        """The bzrlib hooks should be sanitised by setUp."""
1164
 
        self.assertEqual(bzrlib.branch.BranchHooks(),
1165
 
            bzrlib.branch.Branch.hooks)
1166
 
        self.assertEqual(bzrlib.smart.server.SmartServerHooks(),
1167
 
            bzrlib.smart.server.SmartTCPServer.hooks)
1168
 
 
1169
 
    def test__gather_lsprof_in_benchmarks(self):
1170
 
        """When _gather_lsprof_in_benchmarks is on, accumulate profile data.
1171
 
        
1172
 
        Each self.time() call is individually and separately profiled.
1173
 
        """
1174
 
        try:
1175
 
            import bzrlib.lsprof
1176
 
        except ImportError:
1177
 
            raise TestSkipped("lsprof not installed.")
1178
 
        # overrides the class member with an instance member so no cleanup 
1179
 
        # needed.
1180
 
        self._gather_lsprof_in_benchmarks = True
1181
 
        self.time(time.sleep, 0.000)
1182
 
        self.time(time.sleep, 0.003)
1183
 
        self.assertEqual(2, len(self._benchcalls))
1184
 
        self.assertEqual((time.sleep, (0.000,), {}), self._benchcalls[0][0])
1185
 
        self.assertEqual((time.sleep, (0.003,), {}), self._benchcalls[1][0])
1186
 
        self.assertIsInstance(self._benchcalls[0][1], bzrlib.lsprof.Stats)
1187
 
        self.assertIsInstance(self._benchcalls[1][1], bzrlib.lsprof.Stats)
1188
 
 
1189
 
    def test_knownFailure(self):
1190
 
        """Self.knownFailure() should raise a KnownFailure exception."""
1191
 
        self.assertRaises(KnownFailure, self.knownFailure, "A Failure")
1192
 
 
1193
 
    def test_requireFeature_available(self):
1194
 
        """self.requireFeature(available) is a no-op."""
1195
 
        class Available(Feature):
1196
 
            def _probe(self):return True
1197
 
        feature = Available()
1198
 
        self.requireFeature(feature)
1199
 
 
1200
 
    def test_requireFeature_unavailable(self):
1201
 
        """self.requireFeature(unavailable) raises UnavailableFeature."""
1202
 
        class Unavailable(Feature):
1203
 
            def _probe(self):return False
1204
 
        feature = Unavailable()
1205
 
        self.assertRaises(UnavailableFeature, self.requireFeature, feature)
1206
 
 
1207
 
    def test_run_no_parameters(self):
1208
 
        test = SampleTestCase('_test_pass')
1209
 
        test.run()
1210
 
    
1211
 
    def test_run_enabled_unittest_result(self):
1212
 
        """Test we revert to regular behaviour when the test is enabled."""
1213
 
        test = SampleTestCase('_test_pass')
1214
 
        class EnabledFeature(object):
1215
 
            def available(self):
1216
 
                return True
1217
 
        test._test_needs_features = [EnabledFeature()]
1218
 
        result = unittest.TestResult()
1219
 
        test.run(result)
1220
 
        self.assertEqual(1, result.testsRun)
1221
 
        self.assertEqual([], result.errors)
1222
 
        self.assertEqual([], result.failures)
1223
 
 
1224
 
    def test_run_disabled_unittest_result(self):
1225
 
        """Test our compatability for disabled tests with unittest results."""
1226
 
        test = SampleTestCase('_test_pass')
1227
 
        class DisabledFeature(object):
1228
 
            def available(self):
1229
 
                return False
1230
 
        test._test_needs_features = [DisabledFeature()]
1231
 
        result = unittest.TestResult()
1232
 
        test.run(result)
1233
 
        self.assertEqual(1, result.testsRun)
1234
 
        self.assertEqual([], result.errors)
1235
 
        self.assertEqual([], result.failures)
1236
 
 
1237
 
    def test_run_disabled_supporting_result(self):
1238
 
        """Test disabled tests behaviour with support aware results."""
1239
 
        test = SampleTestCase('_test_pass')
1240
 
        class DisabledFeature(object):
1241
 
            def available(self):
1242
 
                return False
1243
 
        the_feature = DisabledFeature()
1244
 
        test._test_needs_features = [the_feature]
1245
 
        class InstrumentedTestResult(unittest.TestResult):
1246
 
            def __init__(self):
1247
 
                unittest.TestResult.__init__(self)
1248
 
                self.calls = []
1249
 
            def startTest(self, test):
1250
 
                self.calls.append(('startTest', test))
1251
 
            def stopTest(self, test):
1252
 
                self.calls.append(('stopTest', test))
1253
 
            def addNotSupported(self, test, feature):
1254
 
                self.calls.append(('addNotSupported', test, feature))
1255
 
        result = InstrumentedTestResult()
1256
 
        test.run(result)
1257
 
        self.assertEqual([
1258
 
            ('startTest', test),
1259
 
            ('addNotSupported', test, the_feature),
1260
 
            ('stopTest', test),
1261
 
            ],
1262
 
            result.calls)
1263
 
 
1264
 
 
1265
 
@symbol_versioning.deprecated_function(zero_eleven)
1266
 
def sample_deprecated_function():
1267
 
    """A deprecated function to test applyDeprecated with."""
1268
 
    return 2
1269
 
 
1270
 
 
1271
 
def sample_undeprecated_function(a_param):
1272
 
    """A undeprecated function to test applyDeprecated with."""
1273
 
 
1274
 
 
1275
 
class ApplyDeprecatedHelper(object):
1276
 
    """A helper class for ApplyDeprecated tests."""
1277
 
 
1278
 
    @symbol_versioning.deprecated_method(zero_eleven)
1279
 
    def sample_deprecated_method(self, param_one):
1280
 
        """A deprecated method for testing with."""
1281
 
        return param_one
1282
 
 
1283
 
    def sample_normal_method(self):
1284
 
        """A undeprecated method."""
1285
 
 
1286
 
    @symbol_versioning.deprecated_method(zero_ten)
1287
 
    def sample_nested_deprecation(self):
1288
 
        return sample_deprecated_function()
1289
 
 
1290
 
 
1291
 
class TestExtraAssertions(TestCase):
1292
 
    """Tests for new test assertions in bzrlib test suite"""
1293
 
 
1294
 
    def test_assert_isinstance(self):
1295
 
        self.assertIsInstance(2, int)
1296
 
        self.assertIsInstance(u'', basestring)
1297
 
        self.assertRaises(AssertionError, self.assertIsInstance, None, int)
1298
 
        self.assertRaises(AssertionError, self.assertIsInstance, 23.3, int)
1299
 
 
1300
 
    def test_assertEndsWith(self):
1301
 
        self.assertEndsWith('foo', 'oo')
1302
 
        self.assertRaises(AssertionError, self.assertEndsWith, 'o', 'oo')
1303
 
 
1304
 
    def test_applyDeprecated_not_deprecated(self):
1305
 
        sample_object = ApplyDeprecatedHelper()
1306
 
        # calling an undeprecated callable raises an assertion
1307
 
        self.assertRaises(AssertionError, self.applyDeprecated, zero_eleven,
1308
 
            sample_object.sample_normal_method)
1309
 
        self.assertRaises(AssertionError, self.applyDeprecated, zero_eleven,
1310
 
            sample_undeprecated_function, "a param value")
1311
 
        # calling a deprecated callable (function or method) with the wrong
1312
 
        # expected deprecation fails.
1313
 
        self.assertRaises(AssertionError, self.applyDeprecated, zero_ten,
1314
 
            sample_object.sample_deprecated_method, "a param value")
1315
 
        self.assertRaises(AssertionError, self.applyDeprecated, zero_ten,
1316
 
            sample_deprecated_function)
1317
 
        # calling a deprecated callable (function or method) with the right
1318
 
        # expected deprecation returns the functions result.
1319
 
        self.assertEqual("a param value", self.applyDeprecated(zero_eleven,
1320
 
            sample_object.sample_deprecated_method, "a param value"))
1321
 
        self.assertEqual(2, self.applyDeprecated(zero_eleven,
1322
 
            sample_deprecated_function))
1323
 
        # calling a nested deprecation with the wrong deprecation version
1324
 
        # fails even if a deeper nested function was deprecated with the 
1325
 
        # supplied version.
1326
 
        self.assertRaises(AssertionError, self.applyDeprecated,
1327
 
            zero_eleven, sample_object.sample_nested_deprecation)
1328
 
        # calling a nested deprecation with the right deprecation value
1329
 
        # returns the calls result.
1330
 
        self.assertEqual(2, self.applyDeprecated(zero_ten,
1331
 
            sample_object.sample_nested_deprecation))
1332
 
 
1333
 
    def test_callDeprecated(self):
1334
 
        def testfunc(be_deprecated, result=None):
1335
 
            if be_deprecated is True:
1336
 
                symbol_versioning.warn('i am deprecated', DeprecationWarning, 
1337
 
                                       stacklevel=1)
1338
 
            return result
1339
 
        result = self.callDeprecated(['i am deprecated'], testfunc, True)
1340
 
        self.assertIs(None, result)
1341
 
        result = self.callDeprecated([], testfunc, False, 'result')
1342
 
        self.assertEqual('result', result)
1343
 
        self.callDeprecated(['i am deprecated'], testfunc, be_deprecated=True)
1344
 
        self.callDeprecated([], testfunc, be_deprecated=False)
1345
 
 
1346
 
 
1347
 
class TestConvenienceMakers(TestCaseWithTransport):
1348
 
    """Test for the make_* convenience functions."""
1349
 
 
1350
 
    def test_make_branch_and_tree_with_format(self):
1351
 
        # we should be able to supply a format to make_branch_and_tree
1352
 
        self.make_branch_and_tree('a', format=bzrlib.bzrdir.BzrDirMetaFormat1())
1353
 
        self.make_branch_and_tree('b', format=bzrlib.bzrdir.BzrDirFormat6())
1354
 
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('a')._format,
1355
 
                              bzrlib.bzrdir.BzrDirMetaFormat1)
1356
 
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('b')._format,
1357
 
                              bzrlib.bzrdir.BzrDirFormat6)
1358
 
 
1359
 
    def test_make_branch_and_memory_tree(self):
1360
 
        # we should be able to get a new branch and a mutable tree from
1361
 
        # TestCaseWithTransport
1362
 
        tree = self.make_branch_and_memory_tree('a')
1363
 
        self.assertIsInstance(tree, bzrlib.memorytree.MemoryTree)
1364
 
 
1365
 
 
1366
 
class TestSFTPMakeBranchAndTree(TestCaseWithSFTPServer):
1367
 
 
1368
 
    def test_make_tree_for_sftp_branch(self):
1369
 
        """Transports backed by local directories create local trees."""
1370
 
 
1371
 
        tree = self.make_branch_and_tree('t1')
1372
 
        base = tree.bzrdir.root_transport.base
1373
 
        self.failIf(base.startswith('sftp'),
1374
 
                'base %r is on sftp but should be local' % base)
1375
 
        self.assertEquals(tree.bzrdir.root_transport,
1376
 
                tree.branch.bzrdir.root_transport)
1377
 
        self.assertEquals(tree.bzrdir.root_transport,
1378
 
                tree.branch.repository.bzrdir.root_transport)
1379
 
 
1380
 
 
1381
 
class TestSelftest(TestCase):
1382
 
    """Tests of bzrlib.tests.selftest."""
1383
 
 
1384
 
    def test_selftest_benchmark_parameter_invokes_test_suite__benchmark__(self):
1385
 
        factory_called = []
1386
 
        def factory():
1387
 
            factory_called.append(True)
1388
 
            return TestSuite()
1389
 
        out = StringIO()
1390
 
        err = StringIO()
1391
 
        self.apply_redirected(out, err, None, bzrlib.tests.selftest, 
1392
 
            test_suite_factory=factory)
1393
 
        self.assertEqual([True], factory_called)
1394
 
 
1395
 
 
1396
 
class TestSelftestCleanOutput(TestCaseInTempDir):
1397
 
 
1398
 
    def test_clean_output(self):
1399
 
        # test functionality of clean_selftest_output()
1400
 
        self.build_tree(['test0000.tmp/', 'test0001.tmp/',
1401
 
                         'bzrlib/', 'tests/',
1402
 
                         'bzr', 'setup.py', 'test9999.tmp'])
1403
 
 
1404
 
        root = os.getcwdu()
1405
 
        before = os.listdir(root)
1406
 
        before.sort()
1407
 
        self.assertEquals(['bzr','bzrlib','setup.py',
1408
 
                           'test0000.tmp','test0001.tmp',
1409
 
                           'test9999.tmp','tests'],
1410
 
                           before)
1411
 
        clean_selftest_output(root, quiet=True)
1412
 
        after = os.listdir(root)
1413
 
        after.sort()
1414
 
        self.assertEquals(['bzr','bzrlib','setup.py',
1415
 
                           'test9999.tmp','tests'],
1416
 
                           after)
1417
 
 
1418
 
    def test_clean_readonly(self):
1419
 
        # test for delete read-only files
1420
 
        self.build_tree(['test0000.tmp/', 'test0000.tmp/foo'])
1421
 
        osutils.make_readonly('test0000.tmp/foo')
1422
 
        root = os.getcwdu()
1423
 
        before = os.listdir(root);  before.sort()
1424
 
        self.assertEquals(['test0000.tmp'], before)
1425
 
        clean_selftest_output(root, quiet=True)
1426
 
        after = os.listdir(root);   after.sort()
1427
 
        self.assertEquals([], after)
1428
 
 
1429
 
 
1430
 
class TestKnownFailure(TestCase):
1431
 
 
1432
 
    def test_known_failure(self):
1433
 
        """Check that KnownFailure is defined appropriately."""
1434
 
        # a KnownFailure is an assertion error for compatability with unaware
1435
 
        # runners.
1436
 
        self.assertIsInstance(KnownFailure(""), AssertionError)
1437
 
 
1438
 
    def test_expect_failure(self):
1439
 
        try:
1440
 
            self.expectFailure("Doomed to failure", self.assertTrue, False)
1441
 
        except KnownFailure, e:
1442
 
            self.assertEqual('Doomed to failure', e.args[0])
1443
 
        try:
1444
 
            self.expectFailure("Doomed to failure", self.assertTrue, True)
1445
 
        except AssertionError, e:
1446
 
            self.assertEqual('Unexpected success.  Should have failed:'
1447
 
                             ' Doomed to failure', e.args[0])
1448
 
        else:
1449
 
            self.fail('Assertion not raised')
1450
 
 
1451
 
 
1452
 
class TestFeature(TestCase):
1453
 
 
1454
 
    def test_caching(self):
1455
 
        """Feature._probe is called by the feature at most once."""
1456
 
        class InstrumentedFeature(Feature):
1457
 
            def __init__(self):
1458
 
                Feature.__init__(self)
1459
 
                self.calls = []
1460
 
            def _probe(self):
1461
 
                self.calls.append('_probe')
1462
 
                return False
1463
 
        feature = InstrumentedFeature()
1464
 
        feature.available()
1465
 
        self.assertEqual(['_probe'], feature.calls)
1466
 
        feature.available()
1467
 
        self.assertEqual(['_probe'], feature.calls)
1468
 
 
1469
 
    def test_named_str(self):
1470
 
        """Feature.__str__ should thunk to feature_name()."""
1471
 
        class NamedFeature(Feature):
1472
 
            def feature_name(self):
1473
 
                return 'symlinks'
1474
 
        feature = NamedFeature()
1475
 
        self.assertEqual('symlinks', str(feature))
1476
 
 
1477
 
    def test_default_str(self):
1478
 
        """Feature.__str__ should default to __class__.__name__."""
1479
 
        class NamedFeature(Feature):
1480
 
            pass
1481
 
        feature = NamedFeature()
1482
 
        self.assertEqual('NamedFeature', str(feature))
1483
 
 
1484
 
 
1485
 
class TestUnavailableFeature(TestCase):
1486
 
 
1487
 
    def test_access_feature(self):
1488
 
        feature = Feature()
1489
 
        exception = UnavailableFeature(feature)
1490
 
        self.assertIs(feature, exception.args[0])