~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Aaron Bentley
  • Date: 2005-07-26 14:06:11 UTC
  • mto: (1092.1.41) (1185.3.4) (974.1.47)
  • mto: This revision was merged to the branch mainline in revision 982.
  • Revision ID: abentley@panoramicfeedback.com-20050726140611-403e366f3c79c1f1
Fixed python invocation

Show diffs side-by-side

added added

removed removed

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