~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Matthäus G. Chajdas
  • Date: 2010-10-09 18:18:19 UTC
  • mto: (5484.1.1 trunk)
  • mto: This revision was merged to the branch mainline in revision 5485.
  • Revision ID: dev@anteru.net-20101009181819-fah4xnt3sn0e3pc2
Always update PATH in start_bzr.bat and write the proper path during installation.

start_bzr.bat now always sets PATH to point to Bzr, which allows start_bzr to work regardless if the user selected to update PATH or not.

Show diffs side-by-side

added added

removed removed

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