/brz/remove-bazaar

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

« back to all changes in this revision

Viewing changes to bzrlib/selftest/__init__.py

  • Committer: Robert Collins
  • Date: 2005-10-19 10:11:57 UTC
  • mfrom: (1185.16.78)
  • mto: This revision was merged to the branch mainline in revision 1470.
  • Revision ID: robertc@robertcollins.net-20051019101157-17438d311e746b4f
mergeĀ fromĀ upstream

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005-2013, 2015, 2016 Canonical Ltd
2
 
#
 
1
# Copyright (C) 2005 by Canonical Ltd
 
2
 
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
5
5
# the Free Software Foundation; either version 2 of the License, or
6
6
# (at your option) any later version.
7
 
#
 
7
 
8
8
# This program is distributed in the hope that it will be useful,
9
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
11
# GNU General Public License for more details.
12
 
#
 
12
 
13
13
# You should have received a copy of the GNU General Public License
14
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
 
"""Testing framework extensions"""
18
 
 
19
 
from __future__ import absolute_import
20
 
 
21
 
# NOTE: Some classes in here use camelCaseNaming() rather than
22
 
# underscore_naming().  That's for consistency with unittest; it's not the
23
 
# general style of breezy.  Please continue that consistency when adding e.g.
24
 
# new assertFoo() methods.
25
 
 
26
 
import atexit
27
 
import codecs
28
 
import copy
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
16
 
 
17
 
 
18
from cStringIO import StringIO
29
19
import difflib
30
 
import doctest
31
20
import errno
32
 
import functools
33
 
from io import (
34
 
    BytesIO,
35
 
    StringIO,
36
 
    TextIOWrapper,
37
 
    )
38
 
import itertools
39
21
import logging
40
 
import math
41
22
import os
42
 
import platform
43
 
import pprint
44
 
import random
45
23
import re
46
 
import shlex
47
 
import site
48
 
import stat
49
 
import subprocess
 
24
import shutil
50
25
import sys
51
26
import tempfile
52
 
import threading
 
27
import unittest
53
28
import time
54
 
import traceback
55
 
import unittest
56
 
import warnings
57
 
 
58
 
import testtools
59
 
# nb: check this before importing anything else from within it
60
 
_testtools_version = getattr(testtools, '__version__', ())
61
 
if _testtools_version < (0, 9, 5):
62
 
    raise ImportError("need at least testtools 0.9.5: %s is %r"
63
 
                      % (testtools.__file__, _testtools_version))
64
 
from testtools import content
65
 
 
66
 
import breezy
67
 
from .. import (
68
 
    branchbuilder,
69
 
    controldir,
70
 
    commands as _mod_commands,
71
 
    config,
72
 
    i18n,
73
 
    debug,
74
 
    errors,
75
 
    hooks,
76
 
    lock as _mod_lock,
77
 
    lockdir,
78
 
    osutils,
79
 
    plugin as _mod_plugin,
80
 
    pyutils,
81
 
    ui,
82
 
    urlutils,
83
 
    registry,
84
 
    symbol_versioning,
85
 
    trace,
86
 
    transport as _mod_transport,
87
 
    workingtree,
88
 
    )
89
 
from breezy.bzr import (
90
 
    chk_map,
91
 
    )
92
 
try:
93
 
    import breezy.lsprof
94
 
except ImportError:
95
 
    # lsprof not available
96
 
    pass
97
 
from ..sixish import (
98
 
    int2byte,
99
 
    PY3,
100
 
    string_types,
101
 
    text_type,
102
 
    )
103
 
from ..bzr.smart import client, request
104
 
from ..transport import (
105
 
    memory,
106
 
    pathfilter,
107
 
    )
108
 
from ..tests import (
109
 
    fixtures,
110
 
    test_server,
111
 
    TestUtil,
112
 
    treeshape,
113
 
    ui_testing,
114
 
    )
115
 
 
116
 
# Mark this python module as being part of the implementation
117
 
# of unittest: this gives us better tracebacks where the last
118
 
# shown frame is the test code, not our assertXYZ.
119
 
__unittest = 1
120
 
 
121
 
default_transport = test_server.LocalURLServer
122
 
 
123
 
 
124
 
_unitialized_attr = object()
125
 
"""A sentinel needed to act as a default value in a method signature."""
126
 
 
127
 
 
128
 
# Subunit result codes, defined here to prevent a hard dependency on subunit.
129
 
SUBUNIT_SEEK_SET = 0
130
 
SUBUNIT_SEEK_CUR = 1
131
 
 
132
 
# These are intentionally brought into this namespace. That way plugins, etc
133
 
# can just "from breezy.tests import TestCase, TestLoader, etc"
134
 
TestSuite = TestUtil.TestSuite
135
 
TestLoader = TestUtil.TestLoader
136
 
 
137
 
# Tests should run in a clean and clearly defined environment. The goal is to
138
 
# keep them isolated from the running environment as mush as possible. The test
139
 
# framework ensures the variables defined below are set (or deleted if the
140
 
# value is None) before a test is run and reset to their original value after
141
 
# the test is run. Generally if some code depends on an environment variable,
142
 
# the tests should start without this variable in the environment. There are a
143
 
# few exceptions but you shouldn't violate this rule lightly.
144
 
isolated_environ = {
145
 
    'BRZ_HOME': None,
146
 
    'HOME': None,
147
 
    'GNUPGHOME': None,
148
 
    'XDG_CONFIG_HOME': None,
149
 
    # brz now uses the Win32 API and doesn't rely on APPDATA, but the
150
 
    # tests do check our impls match APPDATA
151
 
    'BRZ_EDITOR': None,  # test_msgeditor manipulates this variable
152
 
    'VISUAL': None,
153
 
    'EDITOR': None,
154
 
    'BRZ_EMAIL': None,
155
 
    'BZREMAIL': None,  # may still be present in the environment
156
 
    'EMAIL': 'jrandom@example.com',  # set EMAIL as brz does not guess
157
 
    'BRZ_PROGRESS_BAR': None,
158
 
    # Trap leaks to $XDG_CACHE_HOME/breezy/brz.log. This occurs when tests use
159
 
    # TestCase as a base class instead of TestCaseInTempDir. Tests inheriting
160
 
    # from TestCase should not use disk resources, BRZ_LOG is one.
161
 
    'BRZ_LOG': '/you-should-use-TestCaseInTempDir-if-you-need-a-log-file',
162
 
    'BRZ_PLUGIN_PATH': '-site',
163
 
    'BRZ_DISABLE_PLUGINS': None,
164
 
    'BRZ_PLUGINS_AT': None,
165
 
    'BRZ_CONCURRENCY': None,
166
 
    # Make sure that any text ui tests are consistent regardless of
167
 
    # the environment the test case is run in; you may want tests that
168
 
    # test other combinations.  'dumb' is a reasonable guess for tests
169
 
    # going to a pipe or a BytesIO.
170
 
    'TERM': 'dumb',
171
 
    'LINES': '25',
172
 
    'COLUMNS': '80',
173
 
    'BRZ_COLUMNS': '80',
174
 
    # Disable SSH Agent
175
 
    'SSH_AUTH_SOCK': None,
176
 
    # Proxies
177
 
    'http_proxy': None,
178
 
    'HTTP_PROXY': None,
179
 
    'https_proxy': None,
180
 
    'HTTPS_PROXY': None,
181
 
    'no_proxy': None,
182
 
    'NO_PROXY': None,
183
 
    'all_proxy': None,
184
 
    'ALL_PROXY': None,
185
 
    'BZR_REMOTE_PATH': None,
186
 
    # Generally speaking, we don't want apport reporting on crashes in
187
 
    # the test envirnoment unless we're specifically testing apport,
188
 
    # so that it doesn't leak into the real system environment.  We
189
 
    # use an env var so it propagates to subprocesses.
190
 
    'APPORT_DISABLE': '1',
191
 
    }
192
 
 
193
 
 
194
 
def override_os_environ(test, env=None):
195
 
    """Modify os.environ keeping a copy.
196
 
 
197
 
    :param test: A test instance
198
 
 
199
 
    :param env: A dict containing variable definitions to be installed
200
 
    """
201
 
    if env is None:
202
 
        env = isolated_environ
203
 
    test._original_os_environ = dict(**os.environ)
204
 
    for var in env:
205
 
        osutils.set_or_unset_env(var, env[var])
206
 
        if var not in test._original_os_environ:
207
 
            # The var is new, add it with a value of None, so
208
 
            # restore_os_environ will delete it
209
 
            test._original_os_environ[var] = None
210
 
 
211
 
 
212
 
def restore_os_environ(test):
213
 
    """Restore os.environ to its original state.
214
 
 
215
 
    :param test: A test instance previously passed to override_os_environ.
216
 
    """
217
 
    for var, value in test._original_os_environ.items():
218
 
        # Restore the original value (or delete it if the value has been set to
219
 
        # None in override_os_environ).
220
 
        osutils.set_or_unset_env(var, value)
221
 
 
222
 
 
223
 
def _clear__type_equality_funcs(test):
224
 
    """Cleanup bound methods stored on TestCase instances
225
 
 
226
 
    Clear the dict breaking a few (mostly) harmless cycles in the affected
227
 
    unittests released with Python 2.6 and initial Python 2.7 versions.
228
 
 
229
 
    For a few revisions between Python 2.7.1 and Python 2.7.2 that annoyingly
230
 
    shipped in Oneiric, an object with no clear method was used, hence the
231
 
    extra complications, see bug 809048 for details.
232
 
    """
233
 
    type_equality_funcs = getattr(test, "_type_equality_funcs", None)
234
 
    if type_equality_funcs is not None:
235
 
        tef_clear = getattr(type_equality_funcs, "clear", None)
236
 
        if tef_clear is None:
237
 
            tef_instance_dict = getattr(type_equality_funcs, "__dict__", None)
238
 
            if tef_instance_dict is not None:
239
 
                tef_clear = tef_instance_dict.clear
240
 
        if tef_clear is not None:
241
 
            tef_clear()
242
 
 
243
 
 
244
 
class ExtendedTestResult(testtools.TextTestResult):
245
 
    """Accepts, reports and accumulates the results of running tests.
246
 
 
247
 
    Compared to the unittest version this class adds support for
248
 
    profiling, benchmarking, stopping as soon as a test fails,  and
249
 
    skipping tests.  There are further-specialized subclasses for
250
 
    different types of display.
251
 
 
252
 
    When a test finishes, in whatever way, it calls one of the addSuccess,
253
 
    addFailure or addError methods.  These in turn may redirect to a more
254
 
    specific case for the special test results supported by our extended
255
 
    tests.
256
 
 
257
 
    Note that just one of these objects is fed the results from many tests.
258
 
    """
259
 
 
260
 
    stop_early = False
261
 
 
262
 
    def __init__(self, stream, descriptions, verbosity,
263
 
                 bench_history=None,
264
 
                 strict=False,
265
 
                 ):
266
 
        """Construct new TestResult.
267
 
 
268
 
        :param bench_history: Optionally, a writable file object to accumulate
269
 
            benchmark results.
270
 
        """
271
 
        testtools.TextTestResult.__init__(self, stream)
272
 
        if bench_history is not None:
273
 
            from breezy.version import _get_bzr_source_tree
274
 
            src_tree = _get_bzr_source_tree()
275
 
            if src_tree:
276
 
                try:
277
 
                    revision_id = src_tree.get_parent_ids()[0]
278
 
                except IndexError:
279
 
                    # XXX: if this is a brand new tree, do the same as if there
280
 
                    # is no branch.
281
 
                    revision_id = b''
282
 
            else:
283
 
                # XXX: If there's no branch, what should we do?
284
 
                revision_id = b''
285
 
            bench_history.write("--date %s %s\n" % (time.time(), revision_id))
286
 
        self._bench_history = bench_history
287
 
        self.ui = ui.ui_factory
288
 
        self.num_tests = 0
289
 
        self.error_count = 0
290
 
        self.failure_count = 0
291
 
        self.known_failure_count = 0
292
 
        self.skip_count = 0
293
 
        self.not_applicable_count = 0
294
 
        self.unsupported = {}
295
 
        self.count = 0
296
 
        self._overall_start_time = time.time()
297
 
        self._strict = strict
298
 
        self._first_thread_leaker_id = None
299
 
        self._tests_leaking_threads_count = 0
300
 
        self._traceback_from_test = None
301
 
 
302
 
    def stopTestRun(self):
303
 
        run = self.testsRun
304
 
        actionTaken = "Ran"
305
 
        stopTime = time.time()
306
 
        timeTaken = stopTime - self.startTime
307
 
        # GZ 2010-07-19: Seems testtools has no printErrors method, and though
308
 
        #                the parent class method is similar have to duplicate
309
 
        self._show_list('ERROR', self.errors)
310
 
        self._show_list('FAIL', self.failures)
311
 
        self.stream.write(self.sep2)
312
 
        self.stream.write("%s %d test%s in %.3fs\n\n" % (
313
 
            actionTaken, run, run != 1 and "s" or "", timeTaken))
314
 
        if not self.wasSuccessful():
315
 
            self.stream.write("FAILED (")
316
 
            failed, errored = map(len, (self.failures, self.errors))
317
 
            if failed:
318
 
                self.stream.write("failures=%d" % failed)
319
 
            if errored:
320
 
                if failed:
321
 
                    self.stream.write(", ")
322
 
                self.stream.write("errors=%d" % errored)
323
 
            if self.known_failure_count:
324
 
                if failed or errored:
325
 
                    self.stream.write(", ")
326
 
                self.stream.write("known_failure_count=%d" %
327
 
                                  self.known_failure_count)
328
 
            self.stream.write(")\n")
329
 
        else:
330
 
            if self.known_failure_count:
331
 
                self.stream.write("OK (known_failures=%d)\n" %
332
 
                                  self.known_failure_count)
333
 
            else:
334
 
                self.stream.write("OK\n")
335
 
        if self.skip_count > 0:
336
 
            skipped = self.skip_count
337
 
            self.stream.write('%d test%s skipped\n' %
338
 
                              (skipped, skipped != 1 and "s" or ""))
339
 
        if self.unsupported:
340
 
            for feature, count in sorted(self.unsupported.items()):
341
 
                self.stream.write("Missing feature '%s' skipped %d tests.\n" %
342
 
                                  (feature, count))
343
 
        if self._strict:
344
 
            ok = self.wasStrictlySuccessful()
345
 
        else:
346
 
            ok = self.wasSuccessful()
347
 
        if self._first_thread_leaker_id:
348
 
            self.stream.write(
349
 
                '%s is leaking threads among %d leaking tests.\n' % (
350
 
                    self._first_thread_leaker_id,
351
 
                    self._tests_leaking_threads_count))
352
 
            # We don't report the main thread as an active one.
353
 
            self.stream.write(
354
 
                '%d non-main threads were left active in the end.\n'
355
 
                % (len(self._active_threads) - 1))
356
 
 
357
 
    def getDescription(self, test):
358
 
        return test.id()
359
 
 
360
 
    def _extractBenchmarkTime(self, testCase, details=None):
361
 
        """Add a benchmark time for the current test case."""
362
 
        if details and 'benchtime' in details:
363
 
            return float(''.join(details['benchtime'].iter_bytes()))
364
 
        return getattr(testCase, "_benchtime", None)
365
 
 
366
 
    def _delta_to_float(self, a_timedelta, precision):
367
 
        # This calls ceiling to ensure that the most pessimistic view of time
368
 
        # taken is shown (rather than leaving it to the Python %f operator
369
 
        # to decide whether to round/floor/ceiling. This was added when we
370
 
        # had pyp3 test failures that suggest a floor was happening.
371
 
        shift = 10 ** precision
372
 
        return math.ceil(
373
 
            (a_timedelta.days * 86400.0 + a_timedelta.seconds +
374
 
             a_timedelta.microseconds / 1000000.0) * shift) / shift
375
 
 
376
 
    def _elapsedTestTimeString(self):
377
 
        """Return time string for overall time the current test has taken."""
378
 
        return self._formatTime(self._delta_to_float(
379
 
            self._now() - self._start_datetime, 3))
380
 
 
381
 
    def _testTimeString(self, testCase):
382
 
        benchmark_time = self._extractBenchmarkTime(testCase)
383
 
        if benchmark_time is not None:
384
 
            return self._formatTime(benchmark_time) + "*"
385
 
        else:
386
 
            return self._elapsedTestTimeString()
387
 
 
388
 
    def _formatTime(self, seconds):
389
 
        """Format seconds as milliseconds with leading spaces."""
390
 
        # some benchmarks can take thousands of seconds to run, so we need 8
391
 
        # places
392
 
        return "%8dms" % (1000 * seconds)
393
 
 
394
 
    def _shortened_test_description(self, test):
395
 
        what = test.id()
396
 
        what = re.sub(r'^breezy\.tests\.', '', what)
397
 
        return what
398
 
 
399
 
    # GZ 2010-10-04: Cloned tests may end up harmlessly calling this method
400
 
    #                multiple times in a row, because the handler is added for
401
 
    #                each test but the container list is shared between cases.
402
 
    #                See lp:498869 lp:625574 and lp:637725 for background.
403
 
    def _record_traceback_from_test(self, exc_info):
404
 
        """Store the traceback from passed exc_info tuple till"""
405
 
        self._traceback_from_test = exc_info[2]
 
29
 
 
30
import bzrlib.commands
 
31
import bzrlib.trace
 
32
import bzrlib.fetch
 
33
import bzrlib.osutils as osutils
 
34
from bzrlib.selftest import TestUtil
 
35
from bzrlib.selftest.TestUtil import TestLoader, TestSuite
 
36
from bzrlib.selftest.treeshape import build_tree_contents
 
37
 
 
38
MODULES_TO_TEST = []
 
39
MODULES_TO_DOCTEST = []
 
40
 
 
41
from logging import debug, warning, error
 
42
 
 
43
 
 
44
 
 
45
class EarlyStoppingTestResultAdapter(object):
 
46
    """An adapter for TestResult to stop at the first first failure or error"""
 
47
 
 
48
    def __init__(self, result):
 
49
        self._result = result
 
50
 
 
51
    def addError(self, test, err):
 
52
        self._result.addError(test, err)
 
53
        self._result.stop()
 
54
 
 
55
    def addFailure(self, test, err):
 
56
        self._result.addFailure(test, err)
 
57
        self._result.stop()
 
58
 
 
59
    def __getattr__(self, name):
 
60
        return getattr(self._result, name)
 
61
 
 
62
    def __setattr__(self, name, value):
 
63
        if name == '_result':
 
64
            object.__setattr__(self, name, value)
 
65
        return setattr(self._result, name, value)
 
66
 
 
67
 
 
68
class _MyResult(unittest._TextTestResult):
 
69
    """
 
70
    Custom TestResult.
 
71
 
 
72
    No special behaviour for now.
 
73
    """
 
74
 
 
75
    def _elapsedTime(self):
 
76
        return "(Took %.3fs)" % (time.time() - self._start_time)
406
77
 
407
78
    def startTest(self, test):
408
 
        super(ExtendedTestResult, self).startTest(test)
409
 
        if self.count == 0:
410
 
            self.startTests()
411
 
        self.count += 1
412
 
        self.report_test_start(test)
413
 
        test.number = self.count
414
 
        self._recordTestStartTime()
415
 
        # Make testtools cases give us the real traceback on failure
416
 
        addOnException = getattr(test, "addOnException", None)
417
 
        if addOnException is not None:
418
 
            addOnException(self._record_traceback_from_test)
419
 
        # Only check for thread leaks on breezy derived test cases
420
 
        if isinstance(test, TestCase):
421
 
            test.addCleanup(self._check_leaked_threads, test)
422
 
 
423
 
    def stopTest(self, test):
424
 
        super(ExtendedTestResult, self).stopTest(test)
425
 
        # Manually break cycles, means touching various private things but hey
426
 
        getDetails = getattr(test, "getDetails", None)
427
 
        if getDetails is not None:
428
 
            getDetails().clear()
429
 
        _clear__type_equality_funcs(test)
430
 
        self._traceback_from_test = None
431
 
 
432
 
    def startTests(self):
433
 
        self.report_tests_starting()
434
 
        self._active_threads = threading.enumerate()
435
 
 
436
 
    def _check_leaked_threads(self, test):
437
 
        """See if any threads have leaked since last call
438
 
 
439
 
        A sample of live threads is stored in the _active_threads attribute,
440
 
        when this method runs it compares the current live threads and any not
441
 
        in the previous sample are treated as having leaked.
442
 
        """
443
 
        now_active_threads = set(threading.enumerate())
444
 
        threads_leaked = now_active_threads.difference(self._active_threads)
445
 
        if threads_leaked:
446
 
            self._report_thread_leak(test, threads_leaked, now_active_threads)
447
 
            self._tests_leaking_threads_count += 1
448
 
            if self._first_thread_leaker_id is None:
449
 
                self._first_thread_leaker_id = test.id()
450
 
            self._active_threads = now_active_threads
451
 
 
452
 
    def _recordTestStartTime(self):
453
 
        """Record that a test has started."""
454
 
        self._start_datetime = self._now()
 
79
        unittest.TestResult.startTest(self, test)
 
80
        # TODO: Maybe show test.shortDescription somewhere?
 
81
        what = test.shortDescription() or test.id()        
 
82
        if self.showAll:
 
83
            self.stream.write('%-70.70s' % what)
 
84
        self.stream.flush()
 
85
        self._start_time = time.time()
455
86
 
456
87
    def addError(self, test, err):
457
 
        """Tell result that test finished with an error.
458
 
 
459
 
        Called from the TestCase run() method when the test
460
 
        fails with an unexpected error.
461
 
        """
462
 
        self._post_mortem(self._traceback_from_test or err[2])
463
 
        super(ExtendedTestResult, self).addError(test, err)
464
 
        self.error_count += 1
465
 
        self.report_error(test, err)
466
 
        if self.stop_early:
467
 
            self.stop()
 
88
        unittest.TestResult.addError(self, test, err)
 
89
        if self.showAll:
 
90
            self.stream.writeln("ERROR %s" % self._elapsedTime())
 
91
        elif self.dots:
 
92
            self.stream.write('E')
 
93
        self.stream.flush()
468
94
 
469
95
    def addFailure(self, test, err):
470
 
        """Tell result that test failed.
471
 
 
472
 
        Called from the TestCase run() method when the test
473
 
        fails because e.g. an assert() method failed.
474
 
        """
475
 
        self._post_mortem(self._traceback_from_test or err[2])
476
 
        super(ExtendedTestResult, self).addFailure(test, err)
477
 
        self.failure_count += 1
478
 
        self.report_failure(test, err)
479
 
        if self.stop_early:
480
 
            self.stop()
481
 
 
482
 
    def addSuccess(self, test, details=None):
483
 
        """Tell result that test completed successfully.
484
 
 
485
 
        Called from the TestCase run()
486
 
        """
487
 
        if self._bench_history is not None:
488
 
            benchmark_time = self._extractBenchmarkTime(test, details)
489
 
            if benchmark_time is not None:
490
 
                self._bench_history.write("%s %s\n" % (
491
 
                    self._formatTime(benchmark_time),
492
 
                    test.id()))
493
 
        self.report_success(test)
494
 
        super(ExtendedTestResult, self).addSuccess(test)
495
 
        test._log_contents = ''
496
 
 
497
 
    def addExpectedFailure(self, test, err):
498
 
        self.known_failure_count += 1
499
 
        self.report_known_failure(test, err)
500
 
 
501
 
    def addUnexpectedSuccess(self, test, details=None):
502
 
        """Tell result the test unexpectedly passed, counting as a failure
503
 
 
504
 
        When the minimum version of testtools required becomes 0.9.8 this
505
 
        can be updated to use the new handling there.
506
 
        """
507
 
        super(ExtendedTestResult, self).addFailure(test, details=details)
508
 
        self.failure_count += 1
509
 
        self.report_unexpected_success(test,
510
 
                                       "".join(details["reason"].iter_text()))
511
 
        if self.stop_early:
512
 
            self.stop()
513
 
 
514
 
    def addNotSupported(self, test, feature):
515
 
        """The test will not be run because of a missing feature.
516
 
        """
517
 
        # this can be called in two different ways: it may be that the
518
 
        # test started running, and then raised (through requireFeature)
519
 
        # UnavailableFeature.  Alternatively this method can be called
520
 
        # while probing for features before running the test code proper; in
521
 
        # that case we will see startTest and stopTest, but the test will
522
 
        # never actually run.
523
 
        self.unsupported.setdefault(str(feature), 0)
524
 
        self.unsupported[str(feature)] += 1
525
 
        self.report_unsupported(test, feature)
526
 
 
527
 
    def addSkip(self, test, reason):
528
 
        """A test has not run for 'reason'."""
529
 
        self.skip_count += 1
530
 
        self.report_skip(test, reason)
531
 
 
532
 
    def addNotApplicable(self, test, reason):
533
 
        self.not_applicable_count += 1
534
 
        self.report_not_applicable(test, reason)
535
 
 
536
 
    def _count_stored_tests(self):
537
 
        """Count of tests instances kept alive due to not succeeding"""
538
 
        return self.error_count + self.failure_count + self.known_failure_count
539
 
 
540
 
    def _post_mortem(self, tb=None):
541
 
        """Start a PDB post mortem session."""
542
 
        if os.environ.get('BRZ_TEST_PDB', None):
543
 
            import pdb
544
 
            pdb.post_mortem(tb)
545
 
 
546
 
    def progress(self, offset, whence):
547
 
        """The test is adjusting the count of tests to run."""
548
 
        if whence == SUBUNIT_SEEK_SET:
549
 
            self.num_tests = offset
550
 
        elif whence == SUBUNIT_SEEK_CUR:
551
 
            self.num_tests += offset
552
 
        else:
553
 
            raise errors.BzrError("Unknown whence %r" % whence)
554
 
 
555
 
    def report_tests_starting(self):
556
 
        """Display information before the test run begins"""
557
 
        if getattr(sys, 'frozen', None) is None:
558
 
            bzr_path = osutils.realpath(sys.argv[0])
559
 
        else:
560
 
            bzr_path = sys.executable
561
 
        self.stream.write(
562
 
            'brz selftest: %s\n' % (bzr_path,))
563
 
        self.stream.write(
564
 
            '   %s\n' % (
565
 
                breezy.__path__[0],))
566
 
        self.stream.write(
567
 
            '   bzr-%s python-%s %s\n' % (
568
 
                breezy.version_string,
569
 
                breezy._format_version_tuple(sys.version_info),
570
 
                platform.platform(aliased=1),
571
 
                ))
572
 
        self.stream.write('\n')
573
 
 
574
 
    def report_test_start(self, test):
575
 
        """Display information on the test just about to be run"""
576
 
 
577
 
    def _report_thread_leak(self, test, leaked_threads, active_threads):
578
 
        """Display information on a test that leaked one or more threads"""
579
 
        # GZ 2010-09-09: A leak summary reported separately from the general
580
 
        #                thread debugging would be nice. Tests under subunit
581
 
        #                need something not using stream, perhaps adding a
582
 
        #                testtools details object would be fitting.
583
 
        if 'threads' in selftest_debug_flags:
584
 
            self.stream.write('%s is leaking, active is now %d\n' %
585
 
                              (test.id(), len(active_threads)))
586
 
 
587
 
    def startTestRun(self):
588
 
        self.startTime = time.time()
589
 
 
590
 
    def report_success(self, test):
591
 
        pass
592
 
 
593
 
    def wasStrictlySuccessful(self):
594
 
        if self.unsupported or self.known_failure_count:
595
 
            return False
596
 
        return self.wasSuccessful()
597
 
 
598
 
 
599
 
class TextTestResult(ExtendedTestResult):
600
 
    """Displays progress and results of tests in text form"""
601
 
 
602
 
    def __init__(self, stream, descriptions, verbosity,
603
 
                 bench_history=None,
604
 
                 strict=None,
605
 
                 ):
606
 
        ExtendedTestResult.__init__(self, stream, descriptions, verbosity,
607
 
                                    bench_history, strict)
608
 
        self.pb = self.ui.nested_progress_bar()
609
 
        self.pb.show_pct = False
610
 
        self.pb.show_spinner = False
611
 
        self.pb.show_eta = False,
612
 
        self.pb.show_count = False
613
 
        self.pb.show_bar = False
614
 
        self.pb.update_latency = 0
615
 
        self.pb.show_transport_activity = False
616
 
 
617
 
    def stopTestRun(self):
618
 
        # called when the tests that are going to run have run
619
 
        self.pb.clear()
620
 
        self.pb.finished()
621
 
        super(TextTestResult, self).stopTestRun()
622
 
 
623
 
    def report_tests_starting(self):
624
 
        super(TextTestResult, self).report_tests_starting()
625
 
        self.pb.update('[test 0/%d] Starting' % (self.num_tests))
626
 
 
627
 
    def _progress_prefix_text(self):
628
 
        # the longer this text, the less space we have to show the test
629
 
        # name...
630
 
        a = '[%d' % self.count              # total that have been run
631
 
        # tests skipped as known not to be relevant are not important enough
632
 
        # to show here
633
 
        # if self.skip_count:
634
 
        #     a += ', %d skip' % self.skip_count
635
 
        # if self.known_failure_count:
636
 
        #     a += '+%dX' % self.known_failure_count
637
 
        if self.num_tests:
638
 
            a += '/%d' % self.num_tests
639
 
        a += ' in '
640
 
        runtime = time.time() - self._overall_start_time
641
 
        if runtime >= 60:
642
 
            a += '%dm%ds' % (runtime / 60, runtime % 60)
643
 
        else:
644
 
            a += '%ds' % runtime
645
 
        total_fail_count = self.error_count + self.failure_count
646
 
        if total_fail_count:
647
 
            a += ', %d failed' % total_fail_count
648
 
        # if self.unsupported:
649
 
        #     a += ', %d missing' % len(self.unsupported)
650
 
        a += ']'
651
 
        return a
652
 
 
653
 
    def report_test_start(self, test):
654
 
        self.pb.update(
655
 
            self._progress_prefix_text() +
656
 
            ' ' +
657
 
            self._shortened_test_description(test))
658
 
 
659
 
    def _test_description(self, test):
660
 
        return self._shortened_test_description(test)
661
 
 
662
 
    def report_error(self, test, err):
663
 
        self.stream.write('ERROR: %s\n    %s\n' % (
664
 
            self._test_description(test),
665
 
            err[1],
666
 
            ))
667
 
 
668
 
    def report_failure(self, test, err):
669
 
        self.stream.write('FAIL: %s\n    %s\n' % (
670
 
            self._test_description(test),
671
 
            err[1],
672
 
            ))
673
 
 
674
 
    def report_known_failure(self, test, err):
675
 
        pass
676
 
 
677
 
    def report_unexpected_success(self, test, reason):
678
 
        self.stream.write('FAIL: %s\n    %s: %s\n' % (
679
 
            self._test_description(test),
680
 
            "Unexpected success. Should have failed",
681
 
            reason,
682
 
            ))
683
 
 
684
 
    def report_skip(self, test, reason):
685
 
        pass
686
 
 
687
 
    def report_not_applicable(self, test, reason):
688
 
        pass
689
 
 
690
 
    def report_unsupported(self, test, feature):
691
 
        """test cannot be run because feature is missing."""
692
 
 
693
 
 
694
 
class VerboseTestResult(ExtendedTestResult):
695
 
    """Produce long output, with one line per test run plus times"""
696
 
 
697
 
    def _ellipsize_to_right(self, a_string, final_width):
698
 
        """Truncate and pad a string, keeping the right hand side"""
699
 
        if len(a_string) > final_width:
700
 
            result = '...' + a_string[3 - final_width:]
701
 
        else:
702
 
            result = a_string
703
 
        return result.ljust(final_width)
704
 
 
705
 
    def report_tests_starting(self):
706
 
        self.stream.write('running %d tests...\n' % self.num_tests)
707
 
        super(VerboseTestResult, self).report_tests_starting()
708
 
 
709
 
    def report_test_start(self, test):
710
 
        name = self._shortened_test_description(test)
711
 
        width = osutils.terminal_width()
712
 
        if width is not None:
713
 
            # width needs space for 6 char status, plus 1 for slash, plus an
714
 
            # 11-char time string, plus a trailing blank
715
 
            # when NUMBERED_DIRS: plus 5 chars on test number, plus 1 char on
716
 
            # space
717
 
            self.stream.write(self._ellipsize_to_right(name, width - 18))
718
 
        else:
719
 
            self.stream.write(name)
720
 
        self.stream.flush()
721
 
 
722
 
    def _error_summary(self, err):
723
 
        indent = ' ' * 4
724
 
        return '%s%s' % (indent, err[1])
725
 
 
726
 
    def report_error(self, test, err):
727
 
        self.stream.write('ERROR %s\n%s\n'
728
 
                          % (self._testTimeString(test),
729
 
                             self._error_summary(err)))
730
 
 
731
 
    def report_failure(self, test, err):
732
 
        self.stream.write(' FAIL %s\n%s\n'
733
 
                          % (self._testTimeString(test),
734
 
                             self._error_summary(err)))
735
 
 
736
 
    def report_known_failure(self, test, err):
737
 
        self.stream.write('XFAIL %s\n%s\n'
738
 
                          % (self._testTimeString(test),
739
 
                             self._error_summary(err)))
740
 
 
741
 
    def report_unexpected_success(self, test, reason):
742
 
        self.stream.write(' FAIL %s\n%s: %s\n'
743
 
                          % (self._testTimeString(test),
744
 
                             "Unexpected success. Should have failed",
745
 
                             reason))
746
 
 
747
 
    def report_success(self, test):
748
 
        self.stream.write('   OK %s\n' % self._testTimeString(test))
749
 
        for bench_called, stats in getattr(test, '_benchcalls', []):
750
 
            self.stream.write('LSProf output for %s(%s, %s)\n' % bench_called)
751
 
            stats.pprint(file=self.stream)
752
 
        # flush the stream so that we get smooth output. This verbose mode is
753
 
        # used to show the output in PQM.
754
 
        self.stream.flush()
755
 
 
756
 
    def report_skip(self, test, reason):
757
 
        self.stream.write(' SKIP %s\n%s\n'
758
 
                          % (self._testTimeString(test), reason))
759
 
 
760
 
    def report_not_applicable(self, test, reason):
761
 
        self.stream.write('  N/A %s\n    %s\n'
762
 
                          % (self._testTimeString(test), reason))
763
 
 
764
 
    def report_unsupported(self, test, feature):
765
 
        """test cannot be run because feature is missing."""
766
 
        self.stream.write("NODEP %s\n    The feature '%s' is not available.\n"
767
 
                          % (self._testTimeString(test), feature))
768
 
 
769
 
 
770
 
class TextTestRunner(object):
 
96
        unittest.TestResult.addFailure(self, test, err)
 
97
        if self.showAll:
 
98
            self.stream.writeln("FAIL %s" % self._elapsedTime())
 
99
        elif self.dots:
 
100
            self.stream.write('F')
 
101
        self.stream.flush()
 
102
 
 
103
    def addSuccess(self, test):
 
104
        if self.showAll:
 
105
            self.stream.writeln('OK %s' % self._elapsedTime())
 
106
        elif self.dots:
 
107
            self.stream.write('~')
 
108
        self.stream.flush()
 
109
        unittest.TestResult.addSuccess(self, test)
 
110
 
 
111
    def printErrorList(self, flavour, errors):
 
112
        for test, err in errors:
 
113
            self.stream.writeln(self.separator1)
 
114
            self.stream.writeln("%s: %s" % (flavour,self.getDescription(test)))
 
115
            if hasattr(test, '_get_log'):
 
116
                self.stream.writeln()
 
117
                self.stream.writeln('log from this test:')
 
118
                print >>self.stream, test._get_log()
 
119
            self.stream.writeln(self.separator2)
 
120
            self.stream.writeln("%s" % err)
 
121
 
 
122
 
 
123
class TextTestRunner(unittest.TextTestRunner):
771
124
    stop_on_failure = False
772
125
 
773
 
    def __init__(self,
774
 
                 stream=sys.stderr,
775
 
                 descriptions=0,
776
 
                 verbosity=1,
777
 
                 bench_history=None,
778
 
                 strict=False,
779
 
                 result_decorators=None,
780
 
                 ):
781
 
        """Create a TextTestRunner.
782
 
 
783
 
        :param result_decorators: An optional list of decorators to apply
784
 
            to the result object being used by the runner. Decorators are
785
 
            applied left to right - the first element in the list is the
786
 
            innermost decorator.
787
 
        """
788
 
        # stream may know claim to know to write unicode strings, but in older
789
 
        # pythons this goes sufficiently wrong that it is a bad idea. (
790
 
        # specifically a built in file with encoding 'UTF-8' will still try
791
 
        # to encode using ascii.
792
 
        new_encoding = osutils.get_terminal_encoding()
793
 
        codec = codecs.lookup(new_encoding)
794
 
        encode = codec.encode
795
 
        # GZ 2010-09-08: Really we don't want to be writing arbitrary bytes,
796
 
        #                so should swap to the plain codecs.StreamWriter
797
 
        stream = osutils.UnicodeOrBytesToBytesWriter(encode, stream,
798
 
                                                     "backslashreplace")
799
 
        stream.encoding = new_encoding
800
 
        self.stream = stream
801
 
        self.descriptions = descriptions
802
 
        self.verbosity = verbosity
803
 
        self._bench_history = bench_history
804
 
        self._strict = strict
805
 
        self._result_decorators = result_decorators or []
806
 
 
807
 
    def run(self, test):
808
 
        "Run the given test case or test suite."
809
 
        if self.verbosity == 1:
810
 
            result_class = TextTestResult
811
 
        elif self.verbosity >= 2:
812
 
            result_class = VerboseTestResult
813
 
        original_result = result_class(self.stream,
814
 
                                       self.descriptions,
815
 
                                       self.verbosity,
816
 
                                       bench_history=self._bench_history,
817
 
                                       strict=self._strict,
818
 
                                       )
819
 
        # Signal to result objects that look at stop early policy to stop,
820
 
        original_result.stop_early = self.stop_on_failure
821
 
        result = original_result
822
 
        for decorator in self._result_decorators:
823
 
            result = decorator(result)
824
 
            result.stop_early = self.stop_on_failure
825
 
        result.startTestRun()
826
 
        try:
827
 
            test.run(result)
828
 
        finally:
829
 
            result.stopTestRun()
830
 
        # higher level code uses our extended protocol to determine
831
 
        # what exit code to give.
832
 
        return original_result
 
126
    def _makeResult(self):
 
127
        result = _MyResult(self.stream, self.descriptions, self.verbosity)
 
128
        if self.stop_on_failure:
 
129
            result = EarlyStoppingTestResultAdapter(result)
 
130
        return result
833
131
 
834
132
 
835
133
def iter_suite_tests(suite):
836
134
    """Return all tests in a suite, recursing through nested suites"""
837
 
    if isinstance(suite, unittest.TestCase):
838
 
        yield suite
839
 
    elif isinstance(suite, unittest.TestSuite):
840
 
        for item in suite:
 
135
    for item in suite._tests:
 
136
        if isinstance(item, unittest.TestCase):
 
137
            yield item
 
138
        elif isinstance(item, unittest.TestSuite):
841
139
            for r in iter_suite_tests(item):
842
140
                yield r
843
 
    else:
844
 
        raise Exception('unknown type %r for object %r'
845
 
                        % (type(suite), suite))
846
 
 
847
 
 
848
 
TestSkipped = testtools.testcase.TestSkipped
849
 
 
850
 
 
851
 
class TestNotApplicable(TestSkipped):
852
 
    """A test is not applicable to the situation where it was run.
853
 
 
854
 
    This is only normally raised by parameterized tests, if they find that
855
 
    the instance they're constructed upon does not support one aspect
856
 
    of its interface.
857
 
    """
858
 
 
859
 
 
860
 
# traceback._some_str fails to format exceptions that have the default
861
 
# __str__ which does an implicit ascii conversion. However, repr() on those
862
 
# objects works, for all that its not quite what the doctor may have ordered.
863
 
def _clever_some_str(value):
864
 
    try:
865
 
        return str(value)
866
 
    except BaseException:
867
 
        try:
868
 
            return repr(value).replace('\\n', '\n')
869
 
        except BaseException:
870
 
            return '<unprintable %s object>' % type(value).__name__
871
 
 
872
 
 
873
 
traceback._some_str = _clever_some_str
874
 
 
875
 
 
876
 
# deprecated - use self.knownFailure(), or self.expectFailure.
877
 
KnownFailure = testtools.testcase._ExpectedFailure
878
 
 
879
 
 
880
 
class UnavailableFeature(Exception):
881
 
    """A feature required for this test was not available.
882
 
 
883
 
    This can be considered a specialised form of SkippedTest.
884
 
 
885
 
    The feature should be used to construct the exception.
886
 
    """
887
 
 
888
 
 
889
 
class StringIOWrapper(ui_testing.BytesIOWithEncoding):
890
 
 
891
 
    @symbol_versioning.deprecated_method(
892
 
        symbol_versioning.deprecated_in((3, 0)))
893
 
    def __init__(self, s=None):
894
 
        super(StringIOWrapper, self).__init__(s)
895
 
 
896
 
 
897
 
TestUIFactory = ui_testing.TestUIFactory
898
 
 
899
 
 
900
 
def isolated_doctest_setUp(test):
901
 
    override_os_environ(test)
902
 
    osutils.set_or_unset_env('BRZ_HOME', '/nonexistent')
903
 
    test._orig_ui_factory = ui.ui_factory
904
 
    ui.ui_factory = ui.SilentUIFactory()
905
 
 
906
 
 
907
 
def isolated_doctest_tearDown(test):
908
 
    restore_os_environ(test)
909
 
    ui.ui_factory = test._orig_ui_factory
910
 
 
911
 
 
912
 
def IsolatedDocTestSuite(*args, **kwargs):
913
 
    """Overrides doctest.DocTestSuite to handle isolation.
914
 
 
915
 
    The method is really a factory and users are expected to use it as such.
916
 
    """
917
 
 
918
 
    kwargs['setUp'] = isolated_doctest_setUp
919
 
    kwargs['tearDown'] = isolated_doctest_tearDown
920
 
    return doctest.DocTestSuite(*args, **kwargs)
921
 
 
922
 
 
923
 
class TestCase(testtools.TestCase):
924
 
    """Base class for brz unit tests.
925
 
 
926
 
    Tests that need access to disk resources should subclass
 
141
        else:
 
142
            raise Exception('unknown object %r inside test suite %r'
 
143
                            % (item, suite))
 
144
 
 
145
 
 
146
class TestSkipped(Exception):
 
147
    """Indicates that a test was intentionally skipped, rather than failing."""
 
148
    # XXX: Not used yet
 
149
 
 
150
 
 
151
class CommandFailed(Exception):
 
152
    pass
 
153
 
 
154
class TestCase(unittest.TestCase):
 
155
    """Base class for bzr unit tests.
 
156
    
 
157
    Tests that need access to disk resources should subclass 
927
158
    TestCaseInTempDir not TestCase.
928
159
 
929
160
    Error and debug log messages are redirected from their usual
930
161
    location into a temporary file, the contents of which can be
931
 
    retrieved by _get_log().  We use a real OS file, not an in-memory object,
932
 
    so that it can also capture file IO.  When the test completes this file
933
 
    is read into memory and removed from disk.
934
 
 
 
162
    retrieved by _get_log().
 
163
       
935
164
    There are also convenience functions to invoke bzr's command-line
936
 
    routine, and to build and check brz trees.
937
 
 
938
 
    In addition to the usual method of overriding tearDown(), this class also
939
 
    allows subclasses to register cleanup functions via addCleanup, which are
940
 
    run in order as the object is torn down.  It's less likely this will be
941
 
    accidentally overlooked.
942
 
    """
943
 
 
944
 
    _log_file = None
945
 
    # record lsprof data when performing benchmark calls.
946
 
    _gather_lsprof_in_benchmarks = False
947
 
 
948
 
    def __init__(self, methodName='testMethod'):
949
 
        super(TestCase, self).__init__(methodName)
950
 
        self._directory_isolation = True
951
 
        self.exception_handlers.insert(
952
 
            0, (UnavailableFeature, self._do_unsupported_or_skip))
953
 
        self.exception_handlers.insert(
954
 
            0, (TestNotApplicable, self._do_not_applicable))
 
165
    routine, and to build and check bzr trees."""
 
166
 
 
167
    BZRPATH = 'bzr'
 
168
    _log_file_name = None
955
169
 
956
170
    def setUp(self):
957
 
        super(TestCase, self).setUp()
958
 
 
959
 
        # At this point we're still accessing the config files in $BRZ_HOME (as
960
 
        # set by the user running selftest).
961
 
        timeout = config.GlobalStack().get('selftest.timeout')
962
 
        if timeout:
963
 
            timeout_fixture = fixtures.TimeoutFixture(timeout)
964
 
            timeout_fixture.setUp()
965
 
            self.addCleanup(timeout_fixture.cleanUp)
966
 
 
967
 
        for feature in getattr(self, '_test_needs_features', []):
968
 
            self.requireFeature(feature)
969
 
        self._cleanEnvironment()
970
 
 
971
 
        self.overrideAttr(breezy.get_global_state(), 'cmdline_overrides',
972
 
                          config.CommandLineStore())
973
 
 
974
 
        self._silenceUI()
975
 
        self._startLogFile()
976
 
        self._benchcalls = []
977
 
        self._benchtime = None
978
 
        self._clear_hooks()
979
 
        self._track_transports()
980
 
        self._track_locks()
981
 
        self._clear_debug_flags()
982
 
        # Isolate global verbosity level, to make sure it's reproducible
983
 
        # between tests.  We should get rid of this altogether: bug 656694. --
984
 
        # mbp 20101008
985
 
        self.overrideAttr(breezy.trace, '_verbosity_level', 0)
986
 
        self._log_files = set()
987
 
        # Each key in the ``_counters`` dict holds a value for a different
988
 
        # counter. When the test ends, addDetail() should be used to output the
989
 
        # counter values. This happens in install_counter_hook().
990
 
        self._counters = {}
991
 
        if 'config_stats' in selftest_debug_flags:
992
 
            self._install_config_stats_hooks()
993
 
        # Do not use i18n for tests (unless the test reverses this)
994
 
        i18n.disable_i18n()
995
 
 
996
 
    def debug(self):
997
 
        # debug a frame up.
998
 
        import pdb
999
 
        # The sys preserved stdin/stdout should allow blackbox tests debugging
1000
 
        pdb.Pdb(stdin=sys.__stdin__, stdout=sys.__stdout__
1001
 
                ).set_trace(sys._getframe().f_back)
1002
 
 
1003
 
    def discardDetail(self, name):
1004
 
        """Extend the addDetail, getDetails api so we can remove a detail.
1005
 
 
1006
 
        eg. brz always adds the 'log' detail at startup, but we don't want to
1007
 
        include it for skipped, xfail, etc tests.
1008
 
 
1009
 
        It is safe to call this for a detail that doesn't exist, in case this
1010
 
        gets called multiple times.
1011
 
        """
1012
 
        # We cheat. details is stored in __details which means we shouldn't
1013
 
        # touch it. but getDetails() returns the dict directly, so we can
1014
 
        # mutate it.
1015
 
        details = self.getDetails()
1016
 
        if name in details:
1017
 
            del details[name]
1018
 
 
1019
 
    def install_counter_hook(self, hooks, name, counter_name=None):
1020
 
        """Install a counting hook.
1021
 
 
1022
 
        Any hook can be counted as long as it doesn't need to return a value.
1023
 
 
1024
 
        :param hooks: Where the hook should be installed.
1025
 
 
1026
 
        :param name: The hook name that will be counted.
1027
 
 
1028
 
        :param counter_name: The counter identifier in ``_counters``, defaults
1029
 
            to ``name``.
1030
 
        """
1031
 
        _counters = self._counters  # Avoid closing over self
1032
 
        if counter_name is None:
1033
 
            counter_name = name
1034
 
        if counter_name in _counters:
1035
 
            raise AssertionError('%s is already used as a counter name'
1036
 
                                 % (counter_name,))
1037
 
        _counters[counter_name] = 0
1038
 
        self.addDetail(counter_name, content.Content(
1039
 
            content.UTF8_TEXT,
1040
 
            lambda: [b'%d' % (_counters[counter_name],)]))
1041
 
 
1042
 
        def increment_counter(*args, **kwargs):
1043
 
            _counters[counter_name] += 1
1044
 
        label = 'count %s calls' % (counter_name,)
1045
 
        hooks.install_named_hook(name, increment_counter, label)
1046
 
        self.addCleanup(hooks.uninstall_named_hook, name, label)
1047
 
 
1048
 
    def _install_config_stats_hooks(self):
1049
 
        """Install config hooks to count hook calls.
1050
 
 
1051
 
        """
1052
 
        for hook_name in ('get', 'set', 'remove', 'load', 'save'):
1053
 
            self.install_counter_hook(config.ConfigHooks, hook_name,
1054
 
                                      'config.%s' % (hook_name,))
1055
 
 
1056
 
        # The OldConfigHooks are private and need special handling to protect
1057
 
        # against recursive tests (tests that run other tests), so we just do
1058
 
        # manually what registering them into _builtin_known_hooks will provide
1059
 
        # us.
1060
 
        self.overrideAttr(config, 'OldConfigHooks', config._OldConfigHooks())
1061
 
        for hook_name in ('get', 'set', 'remove', 'load', 'save'):
1062
 
            self.install_counter_hook(config.OldConfigHooks, hook_name,
1063
 
                                      'old_config.%s' % (hook_name,))
1064
 
 
1065
 
    def _clear_debug_flags(self):
1066
 
        """Prevent externally set debug flags affecting tests.
1067
 
 
1068
 
        Tests that want to use debug flags can just set them in the
1069
 
        debug_flags set during setup/teardown.
1070
 
        """
1071
 
        # Start with a copy of the current debug flags we can safely modify.
1072
 
        self.overrideAttr(debug, 'debug_flags', set(debug.debug_flags))
1073
 
        if 'allow_debug' not in selftest_debug_flags:
1074
 
            debug.debug_flags.clear()
1075
 
        if 'disable_lock_checks' not in selftest_debug_flags:
1076
 
            debug.debug_flags.add('strict_locks')
1077
 
 
1078
 
    def _clear_hooks(self):
1079
 
        # prevent hooks affecting tests
1080
 
        known_hooks = hooks.known_hooks
1081
 
        self._preserved_hooks = {}
1082
 
        for key, (parent, name) in known_hooks.iter_parent_objects():
1083
 
            current_hooks = getattr(parent, name)
1084
 
            self._preserved_hooks[parent] = (name, current_hooks)
1085
 
        self._preserved_lazy_hooks = hooks._lazy_hooks
1086
 
        hooks._lazy_hooks = {}
1087
 
        self.addCleanup(self._restoreHooks)
1088
 
        for key, (parent, name) in known_hooks.iter_parent_objects():
1089
 
            factory = known_hooks.get(key)
1090
 
            setattr(parent, name, factory())
1091
 
        # this hook should always be installed
1092
 
        request._install_hook()
1093
 
 
1094
 
    def disable_directory_isolation(self):
1095
 
        """Turn off directory isolation checks."""
1096
 
        self._directory_isolation = False
1097
 
 
1098
 
    def enable_directory_isolation(self):
1099
 
        """Enable directory isolation checks."""
1100
 
        self._directory_isolation = True
1101
 
 
1102
 
    def _silenceUI(self):
1103
 
        """Turn off UI for duration of test"""
1104
 
        # by default the UI is off; tests can turn it on if they want it.
1105
 
        self.overrideAttr(ui, 'ui_factory', ui.SilentUIFactory())
1106
 
 
1107
 
    def _check_locks(self):
1108
 
        """Check that all lock take/release actions have been paired."""
1109
 
        # We always check for mismatched locks. If a mismatch is found, we
1110
 
        # fail unless -Edisable_lock_checks is supplied to selftest, in which
1111
 
        # case we just print a warning.
1112
 
        # unhook:
1113
 
        acquired_locks = [lock for action, lock in self._lock_actions
1114
 
                          if action == 'acquired']
1115
 
        released_locks = [lock for action, lock in self._lock_actions
1116
 
                          if action == 'released']
1117
 
        broken_locks = [lock for action, lock in self._lock_actions
1118
 
                        if action == 'broken']
1119
 
        # trivially, given the tests for lock acquistion and release, if we
1120
 
        # have as many in each list, it should be ok. Some lock tests also
1121
 
        # break some locks on purpose and should be taken into account by
1122
 
        # considering that breaking a lock is just a dirty way of releasing it.
1123
 
        if len(acquired_locks) != (len(released_locks) + len(broken_locks)):
1124
 
            message = (
1125
 
                'Different number of acquired and '
1126
 
                'released or broken locks.\n'
1127
 
                'acquired=%s\n'
1128
 
                'released=%s\n'
1129
 
                'broken=%s\n' %
1130
 
                (acquired_locks, released_locks, broken_locks))
1131
 
            if not self._lock_check_thorough:
1132
 
                # Rather than fail, just warn
1133
 
                print("Broken test %s: %s" % (self, message))
1134
 
                return
1135
 
            self.fail(message)
1136
 
 
1137
 
    def _track_locks(self):
1138
 
        """Track lock activity during tests."""
1139
 
        self._lock_actions = []
1140
 
        if 'disable_lock_checks' in selftest_debug_flags:
1141
 
            self._lock_check_thorough = False
1142
 
        else:
1143
 
            self._lock_check_thorough = True
1144
 
 
1145
 
        self.addCleanup(self._check_locks)
1146
 
        _mod_lock.Lock.hooks.install_named_hook('lock_acquired',
1147
 
                                                self._lock_acquired, None)
1148
 
        _mod_lock.Lock.hooks.install_named_hook('lock_released',
1149
 
                                                self._lock_released, None)
1150
 
        _mod_lock.Lock.hooks.install_named_hook('lock_broken',
1151
 
                                                self._lock_broken, None)
1152
 
 
1153
 
    def _lock_acquired(self, result):
1154
 
        self._lock_actions.append(('acquired', result))
1155
 
 
1156
 
    def _lock_released(self, result):
1157
 
        self._lock_actions.append(('released', result))
1158
 
 
1159
 
    def _lock_broken(self, result):
1160
 
        self._lock_actions.append(('broken', result))
1161
 
 
1162
 
    def permit_dir(self, name):
1163
 
        """Permit a directory to be used by this test. See permit_url."""
1164
 
        name_transport = _mod_transport.get_transport_from_path(name)
1165
 
        self.permit_url(name)
1166
 
        self.permit_url(name_transport.base)
1167
 
 
1168
 
    def permit_url(self, url):
1169
 
        """Declare that url is an ok url to use in this test.
1170
 
 
1171
 
        Do this for memory transports, temporary test directory etc.
1172
 
 
1173
 
        Do not do this for the current working directory, /tmp, or any other
1174
 
        preexisting non isolated url.
1175
 
        """
1176
 
        if not url.endswith('/'):
1177
 
            url += '/'
1178
 
        self._bzr_selftest_roots.append(url)
1179
 
 
1180
 
    def permit_source_tree_branch_repo(self):
1181
 
        """Permit the source tree brz is running from to be opened.
1182
 
 
1183
 
        Some code such as breezy.version attempts to read from the brz branch
1184
 
        that brz is executing from (if any). This method permits that directory
1185
 
        to be used in the test suite.
1186
 
        """
1187
 
        path = self.get_source_path()
1188
 
        self.record_directory_isolation()
1189
 
        try:
1190
 
            try:
1191
 
                workingtree.WorkingTree.open(path)
1192
 
            except (errors.NotBranchError, errors.NoWorkingTree):
1193
 
                raise TestSkipped('Needs a working tree of brz sources')
1194
 
        finally:
1195
 
            self.enable_directory_isolation()
1196
 
 
1197
 
    def _preopen_isolate_transport(self, transport):
1198
 
        """Check that all transport openings are done in the test work area."""
1199
 
        while isinstance(transport, pathfilter.PathFilteringTransport):
1200
 
            # Unwrap pathfiltered transports
1201
 
            transport = transport.server.backing_transport.clone(
1202
 
                transport._filter('.'))
1203
 
        url = transport.base
1204
 
        # ReadonlySmartTCPServer_for_testing decorates the backing transport
1205
 
        # urls it is given by prepending readonly+. This is appropriate as the
1206
 
        # client shouldn't know that the server is readonly (or not readonly).
1207
 
        # We could register all servers twice, with readonly+ prepending, but
1208
 
        # that makes for a long list; this is about the same but easier to
1209
 
        # read.
1210
 
        if url.startswith('readonly+'):
1211
 
            url = url[len('readonly+'):]
1212
 
        self._preopen_isolate_url(url)
1213
 
 
1214
 
    def _preopen_isolate_url(self, url):
1215
 
        if not self._directory_isolation:
1216
 
            return
1217
 
        if self._directory_isolation == 'record':
1218
 
            self._bzr_selftest_roots.append(url)
1219
 
            return
1220
 
        # This prevents all transports, including e.g. sftp ones backed on disk
1221
 
        # from working unless they are explicitly granted permission. We then
1222
 
        # depend on the code that sets up test transports to check that they
1223
 
        # are appropriately isolated and enable their use by calling
1224
 
        # self.permit_transport()
1225
 
        if not osutils.is_inside_any(self._bzr_selftest_roots, url):
1226
 
            raise errors.BzrError("Attempt to escape test isolation: %r %r"
1227
 
                                  % (url, self._bzr_selftest_roots))
1228
 
 
1229
 
    def record_directory_isolation(self):
1230
 
        """Gather accessed directories to permit later access.
1231
 
 
1232
 
        This is used for tests that access the branch brz is running from.
1233
 
        """
1234
 
        self._directory_isolation = "record"
1235
 
 
1236
 
    def start_server(self, transport_server, backing_server=None):
1237
 
        """Start transport_server for this test.
1238
 
 
1239
 
        This starts the server, registers a cleanup for it and permits the
1240
 
        server's urls to be used.
1241
 
        """
1242
 
        if backing_server is None:
1243
 
            transport_server.start_server()
1244
 
        else:
1245
 
            transport_server.start_server(backing_server)
1246
 
        self.addCleanup(transport_server.stop_server)
1247
 
        # Obtain a real transport because if the server supplies a password, it
1248
 
        # will be hidden from the base on the client side.
1249
 
        t = _mod_transport.get_transport_from_url(transport_server.get_url())
1250
 
        # Some transport servers effectively chroot the backing transport;
1251
 
        # others like SFTPServer don't - users of the transport can walk up the
1252
 
        # transport to read the entire backing transport. This wouldn't matter
1253
 
        # except that the workdir tests are given - and that they expect the
1254
 
        # server's url to point at - is one directory under the safety net. So
1255
 
        # Branch operations into the transport will attempt to walk up one
1256
 
        # directory. Chrooting all servers would avoid this but also mean that
1257
 
        # we wouldn't be testing directly against non-root urls. Alternatively
1258
 
        # getting the test framework to start the server with a backing server
1259
 
        # at the actual safety net directory would work too, but this then
1260
 
        # means that the self.get_url/self.get_transport methods would need
1261
 
        # to transform all their results. On balance its cleaner to handle it
1262
 
        # here, and permit a higher url when we have one of these transports.
1263
 
        if t.base.endswith('/work/'):
1264
 
            # we have safety net/test root/work
1265
 
            t = t.clone('../..')
1266
 
        elif isinstance(transport_server,
1267
 
                        test_server.SmartTCPServer_for_testing):
1268
 
            # The smart server adds a path similar to work, which is traversed
1269
 
            # up from by the client. But the server is chrooted - the actual
1270
 
            # backing transport is not escaped from, and VFS requests to the
1271
 
            # root will error (because they try to escape the chroot).
1272
 
            t2 = t.clone('..')
1273
 
            while t2.base != t.base:
1274
 
                t = t2
1275
 
                t2 = t.clone('..')
1276
 
        self.permit_url(t.base)
1277
 
 
1278
 
    def _track_transports(self):
1279
 
        """Install checks for transport usage."""
1280
 
        # TestCase has no safe place it can write to.
1281
 
        self._bzr_selftest_roots = []
1282
 
        # Currently the easiest way to be sure that nothing is going on is to
1283
 
        # hook into brz dir opening. This leaves a small window of error for
1284
 
        # transport tests, but they are well known, and we can improve on this
1285
 
        # step.
1286
 
        controldir.ControlDir.hooks.install_named_hook(
1287
 
            "pre_open", self._preopen_isolate_transport,
1288
 
            "Check brz directories are safe.")
 
171
        unittest.TestCase.setUp(self)
 
172
        self.oldenv = os.environ.get('HOME', None)
 
173
        os.environ['HOME'] = os.getcwd()
 
174
        self.bzr_email = os.environ.get('BZREMAIL')
 
175
        if self.bzr_email is not None:
 
176
            del os.environ['BZREMAIL']
 
177
        self.email = os.environ.get('EMAIL')
 
178
        if self.email is not None:
 
179
            del os.environ['EMAIL']
 
180
        bzrlib.trace.disable_default_logging()
 
181
        self._enable_file_logging()
1289
182
 
1290
183
    def _ndiff_strings(self, a, b):
1291
184
        """Return ndiff between two strings containing lines.
1292
 
 
 
185
        
1293
186
        A trailing newline is added if missing to make the strings
1294
187
        print properly."""
1295
 
        if b and not b.endswith('\n'):
 
188
        if b and b[-1] != '\n':
1296
189
            b += '\n'
1297
 
        if a and not a.endswith('\n'):
 
190
        if a and a[-1] != '\n':
1298
191
            a += '\n'
1299
192
        difflines = difflib.ndiff(a.splitlines(True),
1300
193
                                  b.splitlines(True),
1302
195
                                  charjunk=lambda x: False)
1303
196
        return ''.join(difflines)
1304
197
 
1305
 
    def assertEqual(self, a, b, message=''):
1306
 
        try:
1307
 
            if a == b:
1308
 
                return
1309
 
        except UnicodeError as e:
1310
 
            # If we can't compare without getting a UnicodeError, then
1311
 
            # obviously they are different
1312
 
            trace.mutter('UnicodeError: %s', e)
1313
 
        if message:
1314
 
            message += '\n'
1315
 
        raise AssertionError("%snot equal:\na = %s\nb = %s\n"
1316
 
                             % (message,
1317
 
                                pprint.pformat(a), pprint.pformat(b)))
1318
 
 
1319
 
    # FIXME: This is deprecated in unittest2 but plugins may still use it so we
1320
 
    # need a deprecation period for them -- vila 2016-02-01
1321
 
    assertEquals = assertEqual
1322
 
 
1323
 
    def assertEqualDiff(self, a, b, message=None):
 
198
    def assertEqualDiff(self, a, b):
1324
199
        """Assert two texts are equal, if not raise an exception.
1325
 
 
1326
 
        This is intended for use with multi-line strings where it can
 
200
        
 
201
        This is intended for use with multi-line strings where it can 
1327
202
        be hard to find the differences by eye.
1328
203
        """
1329
 
        # TODO: perhaps override assertEqual to call this for strings?
 
204
        # TODO: perhaps override assertEquals to call this for strings?
1330
205
        if a == b:
1331
206
            return
1332
 
        if message is None:
1333
 
            message = "texts not equal:\n"
1334
 
        if a + ('\n' if isinstance(a, text_type) else b'\n') == b:
1335
 
            message = 'first string is missing a final newline.\n'
1336
 
        if a == b + ('\n' if isinstance(b, text_type) else b'\n'):
1337
 
            message = 'second string is missing a final newline.\n'
1338
 
        raise AssertionError(message
1339
 
                             + self._ndiff_strings(
1340
 
                                 a if isinstance(a, text_type) else a.decode(),
1341
 
                                 b if isinstance(b, text_type) else b.decode()))
1342
 
 
1343
 
    def assertEqualMode(self, mode, mode_test):
1344
 
        self.assertEqual(mode, mode_test,
1345
 
                         'mode mismatch %o != %o' % (mode, mode_test))
1346
 
 
1347
 
    def assertEqualStat(self, expected, actual):
1348
 
        """assert that expected and actual are the same stat result.
1349
 
 
1350
 
        :param expected: A stat result.
1351
 
        :param actual: A stat result.
1352
 
        :raises AssertionError: If the expected and actual stat values differ
1353
 
            other than by atime.
1354
 
        """
1355
 
        self.assertEqual(expected.st_size, actual.st_size,
1356
 
                         'st_size did not match')
1357
 
        self.assertEqual(expected.st_mtime, actual.st_mtime,
1358
 
                         'st_mtime did not match')
1359
 
        self.assertEqual(expected.st_ctime, actual.st_ctime,
1360
 
                         'st_ctime did not match')
1361
 
        if sys.platform == 'win32':
1362
 
            # On Win32 both 'dev' and 'ino' cannot be trusted. In python2.4 it
1363
 
            # is 'dev' that varies, in python 2.5 (6?) it is st_ino that is
1364
 
            # odd. We just force it to always be 0 to avoid any problems.
1365
 
            self.assertEqual(0, expected.st_dev)
1366
 
            self.assertEqual(0, actual.st_dev)
1367
 
            self.assertEqual(0, expected.st_ino)
1368
 
            self.assertEqual(0, actual.st_ino)
1369
 
        else:
1370
 
            self.assertEqual(expected.st_dev, actual.st_dev,
1371
 
                             'st_dev did not match')
1372
 
            self.assertEqual(expected.st_ino, actual.st_ino,
1373
 
                             'st_ino did not match')
1374
 
        self.assertEqual(expected.st_mode, actual.st_mode,
1375
 
                         'st_mode did not match')
1376
 
 
1377
 
    def assertLength(self, length, obj_with_len):
1378
 
        """Assert that obj_with_len is of length length."""
1379
 
        if len(obj_with_len) != length:
1380
 
            self.fail("Incorrect length: wanted %d, got %d for %r" % (
1381
 
                length, len(obj_with_len), obj_with_len))
1382
 
 
1383
 
    def assertLogsError(self, exception_class, func, *args, **kwargs):
1384
 
        """Assert that `func(*args, **kwargs)` quietly logs a specific error.
1385
 
        """
1386
 
        captured = []
1387
 
        orig_log_exception_quietly = trace.log_exception_quietly
1388
 
        try:
1389
 
            def capture():
1390
 
                orig_log_exception_quietly()
1391
 
                captured.append(sys.exc_info()[1])
1392
 
            trace.log_exception_quietly = capture
1393
 
            func(*args, **kwargs)
1394
 
        finally:
1395
 
            trace.log_exception_quietly = orig_log_exception_quietly
1396
 
        self.assertLength(1, captured)
1397
 
        err = captured[0]
1398
 
        self.assertIsInstance(err, exception_class)
1399
 
        return err
1400
 
 
1401
 
    def assertPositive(self, val):
1402
 
        """Assert that val is greater than 0."""
1403
 
        self.assertTrue(val > 0, 'expected a positive value, but got %s' % val)
1404
 
 
1405
 
    def assertNegative(self, val):
1406
 
        """Assert that val is less than 0."""
1407
 
        self.assertTrue(val < 0, 'expected a negative value, but got %s' % val)
1408
 
 
1409
 
    def assertStartsWith(self, s, prefix):
1410
 
        if not s.startswith(prefix):
1411
 
            raise AssertionError(
1412
 
                'string %r does not start with %r' % (s, prefix))
1413
 
 
1414
 
    def assertEndsWith(self, s, suffix):
1415
 
        """Asserts that s ends with suffix."""
1416
 
        if not s.endswith(suffix):
1417
 
            raise AssertionError(
1418
 
                'string %r does not end with %r' % (s, suffix))
1419
 
 
1420
 
    def assertContainsRe(self, haystack, needle_re, flags=0):
 
207
        raise AssertionError("texts not equal:\n" + 
 
208
                             self._ndiff_strings(a, b))      
 
209
 
 
210
    def assertContainsRe(self, haystack, needle_re):
1421
211
        """Assert that a contains something matching a regular expression."""
1422
 
        if not re.search(needle_re, haystack, flags):
1423
 
            if (('\n' if isinstance(haystack, str) else b'\n') in haystack or
1424
 
                    len(haystack) > 60):
1425
 
                # a long string, format it in a more readable way
1426
 
                raise AssertionError(
1427
 
                    'pattern "%s" not found in\n"""\\\n%s"""\n'
 
212
        if not re.search(needle_re, haystack):
 
213
            raise AssertionError('pattern "%s" not found in "%s"'
1428
214
                    % (needle_re, haystack))
1429
 
            else:
1430
 
                raise AssertionError('pattern "%s" not found in "%s"'
1431
 
                                     % (needle_re, haystack))
1432
 
 
1433
 
    def assertNotContainsRe(self, haystack, needle_re, flags=0):
1434
 
        """Assert that a does not match a regular expression"""
1435
 
        if re.search(needle_re, haystack, flags):
1436
 
            raise AssertionError('pattern "%s" found in "%s"'
1437
 
                                 % (needle_re, haystack))
1438
 
 
1439
 
    def assertContainsString(self, haystack, needle):
1440
 
        if haystack.find(needle) == -1:
1441
 
            self.fail("string %r not found in '''%s'''" % (needle, haystack))
1442
 
 
1443
 
    def assertNotContainsString(self, haystack, needle):
1444
 
        if haystack.find(needle) != -1:
1445
 
            self.fail("string %r found in '''%s'''" % (needle, haystack))
1446
 
 
1447
 
    def assertSubset(self, sublist, superlist):
1448
 
        """Assert that every entry in sublist is present in superlist."""
1449
 
        missing = set(sublist) - set(superlist)
1450
 
        if len(missing) > 0:
1451
 
            raise AssertionError("value(s) %r not present in container %r" %
1452
 
                                 (missing, superlist))
1453
 
 
1454
 
    def assertListRaises(self, excClass, func, *args, **kwargs):
1455
 
        """Fail unless excClass is raised when the iterator from func is used.
1456
 
 
1457
 
        Many functions can return generators this makes sure
1458
 
        to wrap them in a list() call to make sure the whole generator
1459
 
        is run, and that the proper exception is raised.
1460
 
        """
1461
 
        try:
1462
 
            list(func(*args, **kwargs))
1463
 
        except excClass as e:
1464
 
            return e
1465
 
        else:
1466
 
            if getattr(excClass, '__name__', None) is not None:
1467
 
                excName = excClass.__name__
1468
 
            else:
1469
 
                excName = str(excClass)
1470
 
            raise self.failureException("%s not raised" % excName)
1471
 
 
1472
 
    def assertRaises(self, excClass, callableObj, *args, **kwargs):
1473
 
        """Assert that a callable raises a particular exception.
1474
 
 
1475
 
        :param excClass: As for the except statement, this may be either an
1476
 
            exception class, or a tuple of classes.
1477
 
        :param callableObj: A callable, will be passed ``*args`` and
1478
 
            ``**kwargs``.
1479
 
 
1480
 
        Returns the exception so that you can examine it.
1481
 
        """
1482
 
        try:
1483
 
            callableObj(*args, **kwargs)
1484
 
        except excClass as e:
1485
 
            return e
1486
 
        else:
1487
 
            if getattr(excClass, '__name__', None) is not None:
1488
 
                excName = excClass.__name__
1489
 
            else:
1490
 
                # probably a tuple
1491
 
                excName = str(excClass)
1492
 
            raise self.failureException("%s not raised" % excName)
1493
 
 
1494
 
    def assertIs(self, left, right, message=None):
1495
 
        if not (left is right):
1496
 
            if message is not None:
1497
 
                raise AssertionError(message)
1498
 
            else:
1499
 
                raise AssertionError("%r is not %r." % (left, right))
1500
 
 
1501
 
    def assertIsNot(self, left, right, message=None):
1502
 
        if (left is right):
1503
 
            if message is not None:
1504
 
                raise AssertionError(message)
1505
 
            else:
1506
 
                raise AssertionError("%r is %r." % (left, right))
1507
 
 
1508
 
    def assertTransportMode(self, transport, path, mode):
1509
 
        """Fail if a path does not have mode "mode".
1510
 
 
1511
 
        If modes are not supported on this transport, the assertion is ignored.
1512
 
        """
1513
 
        if not transport._can_roundtrip_unix_modebits():
1514
 
            return
1515
 
        path_stat = transport.stat(path)
1516
 
        actual_mode = stat.S_IMODE(path_stat.st_mode)
1517
 
        self.assertEqual(mode, actual_mode,
1518
 
                         'mode of %r incorrect (%s != %s)'
1519
 
                         % (path, oct(mode), oct(actual_mode)))
1520
 
 
1521
 
    def assertIsSameRealPath(self, path1, path2):
1522
 
        """Fail if path1 and path2 points to different files"""
1523
 
        self.assertEqual(osutils.realpath(path1),
1524
 
                         osutils.realpath(path2),
1525
 
                         "apparent paths:\na = %s\nb = %s\n," % (path1, path2))
1526
 
 
1527
 
    def assertIsInstance(self, obj, kls, msg=None):
1528
 
        """Fail if obj is not an instance of kls
1529
 
 
1530
 
        :param msg: Supplementary message to show if the assertion fails.
1531
 
        """
1532
 
        if not isinstance(obj, kls):
1533
 
            m = "%r is an instance of %s rather than %s" % (
1534
 
                obj, obj.__class__, kls)
1535
 
            if msg:
1536
 
                m += ": " + msg
1537
 
            self.fail(m)
1538
 
 
1539
 
    def assertFileEqual(self, content, path):
1540
 
        """Fail if path does not contain 'content'."""
1541
 
        self.assertPathExists(path)
1542
 
 
1543
 
        mode = 'r' + ('b' if isinstance(content, bytes) else '')
1544
 
        with open(path, mode) as f:
1545
 
            s = f.read()
1546
 
        self.assertEqualDiff(content, s)
1547
 
 
1548
 
    def assertDocstring(self, expected_docstring, obj):
1549
 
        """Fail if obj does not have expected_docstring"""
1550
 
        if __doc__ is None:
1551
 
            # With -OO the docstring should be None instead
1552
 
            self.assertIs(obj.__doc__, None)
1553
 
        else:
1554
 
            self.assertEqual(expected_docstring, obj.__doc__)
1555
 
 
1556
 
    def assertPathExists(self, path):
1557
 
        """Fail unless path or paths, which may be abs or relative, exist."""
1558
 
        # TODO(jelmer): Clean this up for pad.lv/1696545
1559
 
        if not isinstance(path, (bytes, str, text_type)):
1560
 
            for p in path:
1561
 
                self.assertPathExists(p)
1562
 
        else:
1563
 
            self.assertTrue(osutils.lexists(path),
1564
 
                            path + " does not exist")
1565
 
 
1566
 
    def assertPathDoesNotExist(self, path):
1567
 
        """Fail if path or paths, which may be abs or relative, exist."""
1568
 
        if not isinstance(path, (str, text_type)):
1569
 
            for p in path:
1570
 
                self.assertPathDoesNotExist(p)
1571
 
        else:
1572
 
            self.assertFalse(osutils.lexists(path),
1573
 
                             path + " exists")
1574
 
 
1575
 
    def _capture_deprecation_warnings(self, a_callable, *args, **kwargs):
1576
 
        """A helper for callDeprecated and applyDeprecated.
1577
 
 
1578
 
        :param a_callable: A callable to call.
1579
 
        :param args: The positional arguments for the callable
1580
 
        :param kwargs: The keyword arguments for the callable
1581
 
        :return: A tuple (warnings, result). result is the result of calling
1582
 
            a_callable(``*args``, ``**kwargs``).
1583
 
        """
1584
 
        local_warnings = []
1585
 
 
1586
 
        def capture_warnings(msg, cls=None, stacklevel=None):
1587
 
            # we've hooked into a deprecation specific callpath,
1588
 
            # only deprecations should getting sent via it.
1589
 
            self.assertEqual(cls, DeprecationWarning)
1590
 
            local_warnings.append(msg)
1591
 
        original_warning_method = symbol_versioning.warn
1592
 
        symbol_versioning.set_warning_method(capture_warnings)
1593
 
        try:
1594
 
            result = a_callable(*args, **kwargs)
1595
 
        finally:
1596
 
            symbol_versioning.set_warning_method(original_warning_method)
1597
 
        return (local_warnings, result)
1598
 
 
1599
 
    def applyDeprecated(self, deprecation_format, a_callable, *args, **kwargs):
1600
 
        """Call a deprecated callable without warning the user.
1601
 
 
1602
 
        Note that this only captures warnings raised by symbol_versioning.warn,
1603
 
        not other callers that go direct to the warning module.
1604
 
 
1605
 
        To test that a deprecated method raises an error, do something like
1606
 
        this (remember that both assertRaises and applyDeprecated delays *args
1607
 
        and **kwargs passing)::
1608
 
 
1609
 
            self.assertRaises(errors.ReservedId,
1610
 
                self.applyDeprecated,
1611
 
                deprecated_in((1, 5, 0)),
1612
 
                br.append_revision,
1613
 
                'current:')
1614
 
 
1615
 
        :param deprecation_format: The deprecation format that the callable
1616
 
            should have been deprecated with. This is the same type as the
1617
 
            parameter to deprecated_method/deprecated_function. If the
1618
 
            callable is not deprecated with this format, an assertion error
1619
 
            will be raised.
1620
 
        :param a_callable: A callable to call. This may be a bound method or
1621
 
            a regular function. It will be called with ``*args`` and
1622
 
            ``**kwargs``.
1623
 
        :param args: The positional arguments for the callable
1624
 
        :param kwargs: The keyword arguments for the callable
1625
 
        :return: The result of a_callable(``*args``, ``**kwargs``)
1626
 
        """
1627
 
        call_warnings, result = self._capture_deprecation_warnings(
1628
 
            a_callable, *args, **kwargs)
1629
 
        expected_first_warning = symbol_versioning.deprecation_string(
1630
 
            a_callable, deprecation_format)
1631
 
        if len(call_warnings) == 0:
1632
 
            self.fail("No deprecation warning generated by call to %s" %
1633
 
                      a_callable)
1634
 
        self.assertEqual(expected_first_warning, call_warnings[0])
1635
 
        return result
1636
 
 
1637
 
    def callCatchWarnings(self, fn, *args, **kw):
1638
 
        """Call a callable that raises python warnings.
1639
 
 
1640
 
        The caller's responsible for examining the returned warnings.
1641
 
 
1642
 
        If the callable raises an exception, the exception is not
1643
 
        caught and propagates up to the caller.  In that case, the list
1644
 
        of warnings is not available.
1645
 
 
1646
 
        :returns: ([warning_object, ...], fn_result)
1647
 
        """
1648
 
        # XXX: This is not perfect, because it completely overrides the
1649
 
        # warnings filters, and some code may depend on suppressing particular
1650
 
        # warnings.  It's the easiest way to insulate ourselves from -Werror,
1651
 
        # though.  -- Andrew, 20071062
1652
 
        wlist = []
1653
 
 
1654
 
        def _catcher(message, category, filename, lineno, file=None, line=None):
1655
 
            # despite the name, 'message' is normally(?) a Warning subclass
1656
 
            # instance
1657
 
            wlist.append(message)
1658
 
        saved_showwarning = warnings.showwarning
1659
 
        saved_filters = warnings.filters
1660
 
        try:
1661
 
            warnings.showwarning = _catcher
1662
 
            warnings.filters = []
1663
 
            result = fn(*args, **kw)
1664
 
        finally:
1665
 
            warnings.showwarning = saved_showwarning
1666
 
            warnings.filters = saved_filters
1667
 
        return wlist, result
1668
 
 
1669
 
    def callDeprecated(self, expected, callable, *args, **kwargs):
1670
 
        """Assert that a callable is deprecated in a particular way.
1671
 
 
1672
 
        This is a very precise test for unusual requirements. The
1673
 
        applyDeprecated helper function is probably more suited for most tests
1674
 
        as it allows you to simply specify the deprecation format being used
1675
 
        and will ensure that that is issued for the function being called.
1676
 
 
1677
 
        Note that this only captures warnings raised by symbol_versioning.warn,
1678
 
        not other callers that go direct to the warning module.  To catch
1679
 
        general warnings, use callCatchWarnings.
1680
 
 
1681
 
        :param expected: a list of the deprecation warnings expected, in order
1682
 
        :param callable: The callable to call
1683
 
        :param args: The positional arguments for the callable
1684
 
        :param kwargs: The keyword arguments for the callable
1685
 
        """
1686
 
        call_warnings, result = self._capture_deprecation_warnings(
1687
 
            callable, *args, **kwargs)
1688
 
        self.assertEqual(expected, call_warnings)
1689
 
        return result
1690
 
 
1691
 
    def _startLogFile(self):
1692
 
        """Setup a in-memory target for bzr and testcase log messages"""
1693
 
        pseudo_log_file = BytesIO()
1694
 
 
1695
 
        def _get_log_contents_for_weird_testtools_api():
1696
 
            return [pseudo_log_file.getvalue().decode(
1697
 
                "utf-8", "replace").encode("utf-8")]
1698
 
        self.addDetail(
1699
 
            "log", content.Content(
1700
 
                content.ContentType("text", "plain", {"charset": "utf8"}),
1701
 
                _get_log_contents_for_weird_testtools_api))
1702
 
        self._log_file = pseudo_log_file
1703
 
        self._log_memento = trace.push_log_file(self._log_file)
1704
 
        self.addCleanup(self._finishLogFile)
1705
 
 
1706
 
    def _finishLogFile(self):
1707
 
        """Flush and dereference the in-memory log for this testcase"""
1708
 
        if trace._trace_file:
1709
 
            # flush the log file, to get all content
1710
 
            trace._trace_file.flush()
1711
 
        trace.pop_log_file(self._log_memento)
1712
 
        # The logging module now tracks references for cleanup so discard ours
1713
 
        del self._log_memento
1714
 
 
1715
 
    def thisFailsStrictLockCheck(self):
1716
 
        """It is known that this test would fail with -Dstrict_locks.
1717
 
 
1718
 
        By default, all tests are run with strict lock checking unless
1719
 
        -Edisable_lock_checks is supplied. However there are some tests which
1720
 
        we know fail strict locks at this point that have not been fixed.
1721
 
        They should call this function to disable the strict checking.
1722
 
 
1723
 
        This should be used sparingly, it is much better to fix the locking
1724
 
        issues rather than papering over the problem by calling this function.
1725
 
        """
1726
 
        debug.debug_flags.discard('strict_locks')
1727
 
 
1728
 
    def overrideAttr(self, obj, attr_name, new=_unitialized_attr):
1729
 
        """Overrides an object attribute restoring it after the test.
1730
 
 
1731
 
        :note: This should be used with discretion; you should think about
1732
 
        whether it's better to make the code testable without monkey-patching.
1733
 
 
1734
 
        :param obj: The object that will be mutated.
1735
 
 
1736
 
        :param attr_name: The attribute name we want to preserve/override in
1737
 
            the object.
1738
 
 
1739
 
        :param new: The optional value we want to set the attribute to.
1740
 
 
1741
 
        :returns: The actual attr value.
1742
 
        """
1743
 
        # The actual value is captured by the call below
1744
 
        value = getattr(obj, attr_name, _unitialized_attr)
1745
 
        if value is _unitialized_attr:
1746
 
            # When the test completes, the attribute should not exist, but if
1747
 
            # we aren't setting a value, we don't need to do anything.
1748
 
            if new is not _unitialized_attr:
1749
 
                self.addCleanup(delattr, obj, attr_name)
1750
 
        else:
1751
 
            self.addCleanup(setattr, obj, attr_name, value)
1752
 
        if new is not _unitialized_attr:
1753
 
            setattr(obj, attr_name, new)
1754
 
        return value
1755
 
 
1756
 
    def overrideEnv(self, name, new):
1757
 
        """Set an environment variable, and reset it after the test.
1758
 
 
1759
 
        :param name: The environment variable name.
1760
 
 
1761
 
        :param new: The value to set the variable to. If None, the
1762
 
            variable is deleted from the environment.
1763
 
 
1764
 
        :returns: The actual variable value.
1765
 
        """
1766
 
        value = osutils.set_or_unset_env(name, new)
1767
 
        self.addCleanup(osutils.set_or_unset_env, name, value)
1768
 
        return value
1769
 
 
1770
 
    def recordCalls(self, obj, attr_name):
1771
 
        """Monkeypatch in a wrapper that will record calls.
1772
 
 
1773
 
        The monkeypatch is automatically removed when the test concludes.
1774
 
 
1775
 
        :param obj: The namespace holding the reference to be replaced;
1776
 
            typically a module, class, or object.
1777
 
        :param attr_name: A string for the name of the attribute to patch.
1778
 
        :returns: A list that will be extended with one item every time the
1779
 
            function is called, with a tuple of (args, kwargs).
1780
 
        """
1781
 
        calls = []
1782
 
 
1783
 
        def decorator(*args, **kwargs):
1784
 
            calls.append((args, kwargs))
1785
 
            return orig(*args, **kwargs)
1786
 
        orig = self.overrideAttr(obj, attr_name, decorator)
1787
 
        return calls
1788
 
 
1789
 
    def _cleanEnvironment(self):
1790
 
        for name, value in isolated_environ.items():
1791
 
            self.overrideEnv(name, value)
1792
 
 
1793
 
    def _restoreHooks(self):
1794
 
        for klass, (name, hooks) in self._preserved_hooks.items():
1795
 
            setattr(klass, name, hooks)
1796
 
        self._preserved_hooks.clear()
1797
 
        breezy.hooks._lazy_hooks = self._preserved_lazy_hooks
1798
 
        self._preserved_lazy_hooks.clear()
1799
 
 
1800
 
    def knownFailure(self, reason):
1801
 
        """Declare that this test fails for a known reason
1802
 
 
1803
 
        Tests that are known to fail should generally be using expectedFailure
1804
 
        with an appropriate reverse assertion if a change could cause the test
1805
 
        to start passing. Conversely if the test has no immediate prospect of
1806
 
        succeeding then using skip is more suitable.
1807
 
 
1808
 
        When this method is called while an exception is being handled, that
1809
 
        traceback will be used, otherwise a new exception will be thrown to
1810
 
        provide one but won't be reported.
1811
 
        """
1812
 
        self._add_reason(reason)
1813
 
        try:
1814
 
            exc_info = sys.exc_info()
1815
 
            if exc_info != (None, None, None):
1816
 
                self._report_traceback(exc_info)
1817
 
            else:
1818
 
                try:
1819
 
                    raise self.failureException(reason)
1820
 
                except self.failureException:
1821
 
                    exc_info = sys.exc_info()
1822
 
            # GZ 02-08-2011: Maybe cleanup this err.exc_info attribute too?
1823
 
            raise testtools.testcase._ExpectedFailure(exc_info)
1824
 
        finally:
1825
 
            del exc_info
1826
 
 
1827
 
    def _suppress_log(self):
1828
 
        """Remove the log info from details."""
1829
 
        self.discardDetail('log')
1830
 
 
1831
 
    def _do_skip(self, result, reason):
1832
 
        self._suppress_log()
1833
 
        addSkip = getattr(result, 'addSkip', None)
1834
 
        if not callable(addSkip):
1835
 
            result.addSuccess(result)
1836
 
        else:
1837
 
            addSkip(self, str(reason))
1838
 
 
1839
 
    @staticmethod
1840
 
    def _do_known_failure(self, result, e):
1841
 
        self._suppress_log()
1842
 
        err = sys.exc_info()
1843
 
        addExpectedFailure = getattr(result, 'addExpectedFailure', None)
1844
 
        if addExpectedFailure is not None:
1845
 
            addExpectedFailure(self, err)
1846
 
        else:
1847
 
            result.addSuccess(self)
1848
 
 
1849
 
    @staticmethod
1850
 
    def _do_not_applicable(self, result, e):
1851
 
        if not e.args:
1852
 
            reason = 'No reason given'
1853
 
        else:
1854
 
            reason = e.args[0]
1855
 
        self._suppress_log()
1856
 
        addNotApplicable = getattr(result, 'addNotApplicable', None)
1857
 
        if addNotApplicable is not None:
1858
 
            result.addNotApplicable(self, reason)
1859
 
        else:
1860
 
            self._do_skip(result, reason)
1861
 
 
1862
 
    @staticmethod
1863
 
    def _report_skip(self, result, err):
1864
 
        """Override the default _report_skip.
1865
 
 
1866
 
        We want to strip the 'log' detail. If we waint until _do_skip, it has
1867
 
        already been formatted into the 'reason' string, and we can't pull it
1868
 
        out again.
1869
 
        """
1870
 
        self._suppress_log()
1871
 
        super(TestCase, self)._report_skip(self, result, err)
1872
 
 
1873
 
    @staticmethod
1874
 
    def _report_expected_failure(self, result, err):
1875
 
        """Strip the log.
1876
 
 
1877
 
        See _report_skip for motivation.
1878
 
        """
1879
 
        self._suppress_log()
1880
 
        super(TestCase, self)._report_expected_failure(self, result, err)
1881
 
 
1882
 
    @staticmethod
1883
 
    def _do_unsupported_or_skip(self, result, e):
1884
 
        reason = e.args[0]
1885
 
        self._suppress_log()
1886
 
        addNotSupported = getattr(result, 'addNotSupported', None)
1887
 
        if addNotSupported is not None:
1888
 
            result.addNotSupported(self, reason)
1889
 
        else:
1890
 
            self._do_skip(result, reason)
1891
 
 
1892
 
    def time(self, callable, *args, **kwargs):
1893
 
        """Run callable and accrue the time it takes to the benchmark time.
1894
 
 
1895
 
        If lsprofiling is enabled (i.e. by --lsprof-time to brz selftest) then
1896
 
        this will cause lsprofile statistics to be gathered and stored in
1897
 
        self._benchcalls.
1898
 
        """
1899
 
        if self._benchtime is None:
1900
 
            self.addDetail('benchtime', content.Content(
1901
 
                content.UTF8_TEXT,
1902
 
                lambda: [str(self._benchtime).encode('utf-8')]))
1903
 
            self._benchtime = 0
1904
 
        start = time.time()
1905
 
        try:
1906
 
            if not self._gather_lsprof_in_benchmarks:
1907
 
                return callable(*args, **kwargs)
1908
 
            else:
1909
 
                # record this benchmark
1910
 
                ret, stats = breezy.lsprof.profile(callable, *args, **kwargs)
1911
 
                stats.sort()
1912
 
                self._benchcalls.append(((callable, args, kwargs), stats))
1913
 
                return ret
1914
 
        finally:
1915
 
            self._benchtime += time.time() - start
 
215
 
 
216
    def _enable_file_logging(self):
 
217
        fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
 
218
 
 
219
        self._log_file = os.fdopen(fileno, 'w+')
 
220
 
 
221
        hdlr = logging.StreamHandler(self._log_file)
 
222
        hdlr.setLevel(logging.DEBUG)
 
223
        hdlr.setFormatter(logging.Formatter('%(levelname)8s  %(message)s'))
 
224
        logging.getLogger('').addHandler(hdlr)
 
225
        logging.getLogger('').setLevel(logging.DEBUG)
 
226
        self._log_hdlr = hdlr
 
227
        debug('opened log file %s', name)
 
228
        
 
229
        self._log_file_name = name
 
230
 
 
231
    def tearDown(self):
 
232
        os.environ['HOME'] = self.oldenv
 
233
        if os.environ.get('BZREMAIL') is not None:
 
234
            del os.environ['BZREMAIL']
 
235
        if self.bzr_email is not None:
 
236
            os.environ['BZREMAIL'] = self.bzr_email
 
237
        if os.environ.get('EMAIL') is not None:
 
238
            del os.environ['EMAIL']
 
239
        if self.email is not None:
 
240
            os.environ['EMAIL'] = self.email
 
241
        logging.getLogger('').removeHandler(self._log_hdlr)
 
242
        bzrlib.trace.enable_default_logging()
 
243
        logging.debug('%s teardown', self.id())
 
244
        self._log_file.close()
 
245
        unittest.TestCase.tearDown(self)
1916
246
 
1917
247
    def log(self, *args):
1918
 
        trace.mutter(*args)
1919
 
 
1920
 
    def get_log(self):
1921
 
        """Get a unicode string containing the log from breezy.trace.
1922
 
 
1923
 
        Undecodable characters are replaced.
1924
 
        """
1925
 
        return u"".join(self.getDetails()['log'].iter_text())
1926
 
 
1927
 
    def requireFeature(self, feature):
1928
 
        """This test requires a specific feature is available.
1929
 
 
1930
 
        :raises UnavailableFeature: When feature is not available.
1931
 
        """
1932
 
        if not feature.available():
1933
 
            raise UnavailableFeature(feature)
1934
 
 
1935
 
    def _run_bzr_core(self, args, encoding, stdin, stdout, stderr,
1936
 
                      working_dir):
1937
 
        # Clear chk_map page cache, because the contents are likely to mask
1938
 
        # locking errors.
1939
 
        chk_map.clear_cache()
1940
 
 
1941
 
        self.log('run brz: %r', args)
1942
 
 
1943
 
        if PY3:
1944
 
            self._last_cmd_stdout = stdout
1945
 
            self._last_cmd_stderr = stderr
1946
 
        else:
1947
 
            self._last_cmd_stdout = codecs.getwriter(encoding)(stdout)
1948
 
            self._last_cmd_stderr = codecs.getwriter(encoding)(stderr)
1949
 
 
1950
 
        old_ui_factory = ui.ui_factory
1951
 
        ui.ui_factory = ui_testing.TestUIFactory(
1952
 
            stdin=stdin,
1953
 
            stdout=self._last_cmd_stdout,
1954
 
            stderr=self._last_cmd_stderr)
1955
 
 
1956
 
        cwd = None
1957
 
        if working_dir is not None:
1958
 
            cwd = osutils.getcwd()
1959
 
            os.chdir(working_dir)
1960
 
 
1961
 
        try:
1962
 
            with ui.ui_factory:
1963
 
                result = self.apply_redirected(
1964
 
                    ui.ui_factory.stdin,
1965
 
                    stdout, stderr,
1966
 
                    _mod_commands.run_bzr_catch_user_errors,
1967
 
                    args)
1968
 
        finally:
1969
 
            ui.ui_factory = old_ui_factory
1970
 
            if cwd is not None:
1971
 
                os.chdir(cwd)
1972
 
 
1973
 
        return result
1974
 
 
1975
 
    def run_bzr_raw(self, args, retcode=0, stdin=None, encoding=None,
1976
 
                    working_dir=None, error_regexes=[]):
1977
 
        """Invoke brz, as if it were run from the command line.
1978
 
 
1979
 
        The argument list should not include the brz program name - the
1980
 
        first argument is normally the brz command.  Arguments may be
1981
 
        passed in three ways:
1982
 
 
1983
 
        1- A list of strings, eg ["commit", "a"].  This is recommended
1984
 
        when the command contains whitespace or metacharacters, or
1985
 
        is built up at run time.
1986
 
 
1987
 
        2- A single string, eg "add a".  This is the most convenient
1988
 
        for hardcoded commands.
1989
 
 
1990
 
        This runs brz through the interface that catches and reports
1991
 
        errors, and with logging set to something approximating the
1992
 
        default, so that error reporting can be checked.
1993
 
 
1994
 
        This should be the main method for tests that want to exercise the
1995
 
        overall behavior of the brz application (rather than a unit test
1996
 
        or a functional test of the library.)
1997
 
 
1998
 
        This sends the stdout/stderr results into the test's log,
1999
 
        where it may be useful for debugging.  See also run_captured.
2000
 
 
2001
 
        :keyword stdin: A string to be used as stdin for the command.
2002
 
        :keyword retcode: The status code the command should return;
2003
 
            default 0.
2004
 
        :keyword working_dir: The directory to run the command in
2005
 
        :keyword error_regexes: A list of expected error messages.  If
2006
 
            specified they must be seen in the error output of the command.
2007
 
        """
2008
 
        if isinstance(args, string_types):
2009
 
            args = shlex.split(args)
2010
 
 
2011
 
        if encoding is None:
2012
 
            encoding = osutils.get_user_encoding()
2013
 
 
2014
 
        if sys.version_info[0] == 2:
2015
 
            wrapped_stdout = stdout = ui_testing.BytesIOWithEncoding()
2016
 
            wrapped_stderr = stderr = ui_testing.BytesIOWithEncoding()
2017
 
            stdout.encoding = stderr.encoding = encoding
2018
 
 
2019
 
            # FIXME: don't call into logging here
2020
 
            handler = trace.EncodedStreamHandler(
2021
 
                stderr, errors="replace")
2022
 
        else:
2023
 
            stdout = BytesIO()
2024
 
            stderr = BytesIO()
2025
 
            wrapped_stdout = TextIOWrapper(stdout, encoding)
2026
 
            wrapped_stderr = TextIOWrapper(stderr, encoding)
2027
 
            handler = logging.StreamHandler(wrapped_stderr)
2028
 
        handler.setLevel(logging.INFO)
2029
 
 
2030
 
        logger = logging.getLogger('')
2031
 
        logger.addHandler(handler)
2032
 
        try:
2033
 
            result = self._run_bzr_core(
2034
 
                args, encoding=encoding, stdin=stdin, stdout=wrapped_stdout,
2035
 
                stderr=wrapped_stderr, working_dir=working_dir,
2036
 
                )
2037
 
        finally:
2038
 
            logger.removeHandler(handler)
2039
 
 
2040
 
        if PY3:
2041
 
            wrapped_stdout.flush()
2042
 
            wrapped_stderr.flush()
2043
 
 
2044
 
        out = stdout.getvalue()
2045
 
        err = stderr.getvalue()
2046
 
        if out:
2047
 
            self.log('output:\n%r', out)
2048
 
        if err:
2049
 
            self.log('errors:\n%r', err)
2050
 
        if retcode is not None:
2051
 
            self.assertEqual(retcode, result,
2052
 
                             message='Unexpected return code')
2053
 
        self.assertIsInstance(error_regexes, (list, tuple))
2054
 
        for regex in error_regexes:
2055
 
            self.assertContainsRe(err, regex)
2056
 
        return out, err
2057
 
 
2058
 
    def run_bzr(self, args, retcode=0, stdin=None, encoding=None,
2059
 
                working_dir=None, error_regexes=[]):
2060
 
        """Invoke brz, as if it were run from the command line.
2061
 
 
2062
 
        The argument list should not include the brz program name - the
2063
 
        first argument is normally the brz command.  Arguments may be
2064
 
        passed in three ways:
2065
 
 
2066
 
        1- A list of strings, eg ["commit", "a"].  This is recommended
2067
 
        when the command contains whitespace or metacharacters, or
2068
 
        is built up at run time.
2069
 
 
2070
 
        2- A single string, eg "add a".  This is the most convenient
2071
 
        for hardcoded commands.
2072
 
 
2073
 
        This runs brz through the interface that catches and reports
2074
 
        errors, and with logging set to something approximating the
2075
 
        default, so that error reporting can be checked.
2076
 
 
2077
 
        This should be the main method for tests that want to exercise the
2078
 
        overall behavior of the brz application (rather than a unit test
2079
 
        or a functional test of the library.)
2080
 
 
2081
 
        This sends the stdout/stderr results into the test's log,
2082
 
        where it may be useful for debugging.  See also run_captured.
2083
 
 
2084
 
        :keyword stdin: A string to be used as stdin for the command.
2085
 
        :keyword retcode: The status code the command should return;
2086
 
            default 0.
2087
 
        :keyword working_dir: The directory to run the command in
2088
 
        :keyword error_regexes: A list of expected error messages.  If
2089
 
            specified they must be seen in the error output of the command.
2090
 
        """
2091
 
        if isinstance(args, string_types):
2092
 
            args = shlex.split(args)
2093
 
 
2094
 
        if encoding is None:
2095
 
            encoding = osutils.get_user_encoding()
2096
 
 
2097
 
        if sys.version_info[0] == 2:
2098
 
            stdout = ui_testing.BytesIOWithEncoding()
2099
 
            stderr = ui_testing.BytesIOWithEncoding()
2100
 
            stdout.encoding = stderr.encoding = encoding
2101
 
            # FIXME: don't call into logging here
2102
 
            handler = trace.EncodedStreamHandler(
2103
 
                stderr, errors="replace")
2104
 
        else:
2105
 
            stdout = ui_testing.StringIOWithEncoding()
2106
 
            stderr = ui_testing.StringIOWithEncoding()
2107
 
            stdout.encoding = stderr.encoding = encoding
2108
 
            handler = logging.StreamHandler(stream=stderr)
2109
 
        handler.setLevel(logging.INFO)
2110
 
 
2111
 
        logger = logging.getLogger('')
2112
 
        logger.addHandler(handler)
2113
 
 
2114
 
        try:
2115
 
            result = self._run_bzr_core(
2116
 
                args, encoding=encoding, stdin=stdin, stdout=stdout,
2117
 
                stderr=stderr, working_dir=working_dir)
2118
 
        finally:
2119
 
            logger.removeHandler(handler)
2120
 
 
2121
 
        out = stdout.getvalue()
2122
 
        err = stderr.getvalue()
2123
 
        if out:
2124
 
            self.log('output:\n%r', out)
2125
 
        if err:
2126
 
            self.log('errors:\n%r', err)
2127
 
        if retcode is not None:
2128
 
            self.assertEqual(retcode, result,
2129
 
                             message='Unexpected return code')
2130
 
        self.assertIsInstance(error_regexes, (list, tuple))
2131
 
        for regex in error_regexes:
2132
 
            self.assertContainsRe(err, regex)
2133
 
        return out, err
2134
 
 
2135
 
    def run_bzr_error(self, error_regexes, *args, **kwargs):
2136
 
        """Run brz, and check that stderr contains the supplied regexes
2137
 
 
2138
 
        :param error_regexes: Sequence of regular expressions which
2139
 
            must each be found in the error output. The relative ordering
2140
 
            is not enforced.
2141
 
        :param args: command-line arguments for brz
2142
 
        :param kwargs: Keyword arguments which are interpreted by run_brz
2143
 
            This function changes the default value of retcode to be 3,
2144
 
            since in most cases this is run when you expect brz to fail.
2145
 
 
2146
 
        :return: (out, err) The actual output of running the command (in case
2147
 
            you want to do more inspection)
2148
 
 
2149
 
        Examples of use::
2150
 
 
2151
 
            # Make sure that commit is failing because there is nothing to do
2152
 
            self.run_bzr_error(['no changes to commit'],
2153
 
                               ['commit', '-m', 'my commit comment'])
2154
 
            # Make sure --strict is handling an unknown file, rather than
2155
 
            # giving us the 'nothing to do' error
2156
 
            self.build_tree(['unknown'])
2157
 
            self.run_bzr_error(
2158
 
                ['Commit refused because there are unknown files'],
2159
 
                ['commit', --strict', '-m', 'my commit comment'])
2160
 
        """
2161
 
        kwargs.setdefault('retcode', 3)
2162
 
        kwargs['error_regexes'] = error_regexes
2163
 
        out, err = self.run_bzr(*args, **kwargs)
2164
 
        return out, err
2165
 
 
2166
 
    def run_bzr_subprocess(self, *args, **kwargs):
2167
 
        """Run brz in a subprocess for testing.
2168
 
 
2169
 
        This starts a new Python interpreter and runs brz in there.
2170
 
        This should only be used for tests that have a justifiable need for
2171
 
        this isolation: e.g. they are testing startup time, or signal
2172
 
        handling, or early startup code, etc.  Subprocess code can't be
2173
 
        profiled or debugged so easily.
2174
 
 
2175
 
        :keyword retcode: The status code that is expected.  Defaults to 0.  If
2176
 
            None is supplied, the status code is not checked.
2177
 
        :keyword env_changes: A dictionary which lists changes to environment
2178
 
            variables. A value of None will unset the env variable.
2179
 
            The values must be strings. The change will only occur in the
2180
 
            child, so you don't need to fix the environment after running.
2181
 
        :keyword universal_newlines: Convert CRLF => LF
2182
 
        :keyword allow_plugins: By default the subprocess is run with
2183
 
            --no-plugins to ensure test reproducibility. Also, it is possible
2184
 
            for system-wide plugins to create unexpected output on stderr,
2185
 
            which can cause unnecessary test failures.
2186
 
        """
2187
 
        env_changes = kwargs.get('env_changes', {})
2188
 
        working_dir = kwargs.get('working_dir', None)
2189
 
        allow_plugins = kwargs.get('allow_plugins', False)
2190
 
        if len(args) == 1:
2191
 
            if isinstance(args[0], list):
2192
 
                args = args[0]
2193
 
            elif isinstance(args[0], (str, text_type)):
2194
 
                args = list(shlex.split(args[0]))
2195
 
        else:
2196
 
            raise ValueError("passing varargs to run_bzr_subprocess")
2197
 
        process = self.start_bzr_subprocess(args, env_changes=env_changes,
2198
 
                                            working_dir=working_dir,
2199
 
                                            allow_plugins=allow_plugins)
2200
 
        # We distinguish between retcode=None and retcode not passed.
2201
 
        supplied_retcode = kwargs.get('retcode', 0)
2202
 
        return self.finish_bzr_subprocess(process, retcode=supplied_retcode,
2203
 
                                          universal_newlines=kwargs.get(
2204
 
                                              'universal_newlines', False),
2205
 
                                          process_args=args)
2206
 
 
2207
 
    def start_bzr_subprocess(self, process_args, env_changes=None,
2208
 
                             skip_if_plan_to_signal=False,
2209
 
                             working_dir=None,
2210
 
                             allow_plugins=False, stderr=subprocess.PIPE):
2211
 
        """Start brz in a subprocess for testing.
2212
 
 
2213
 
        This starts a new Python interpreter and runs brz in there.
2214
 
        This should only be used for tests that have a justifiable need for
2215
 
        this isolation: e.g. they are testing startup time, or signal
2216
 
        handling, or early startup code, etc.  Subprocess code can't be
2217
 
        profiled or debugged so easily.
2218
 
 
2219
 
        :param process_args: a list of arguments to pass to the brz executable,
2220
 
            for example ``['--version']``.
2221
 
        :param env_changes: A dictionary which lists changes to environment
2222
 
            variables. A value of None will unset the env variable.
2223
 
            The values must be strings. The change will only occur in the
2224
 
            child, so you don't need to fix the environment after running.
2225
 
        :param skip_if_plan_to_signal: raise TestSkipped when true and system
2226
 
            doesn't support signalling subprocesses.
2227
 
        :param allow_plugins: If False (default) pass --no-plugins to brz.
2228
 
        :param stderr: file to use for the subprocess's stderr.  Valid values
2229
 
            are those valid for the stderr argument of `subprocess.Popen`.
2230
 
            Default value is ``subprocess.PIPE``.
2231
 
 
2232
 
        :returns: Popen object for the started process.
2233
 
        """
2234
 
        if skip_if_plan_to_signal:
2235
 
            if os.name != "posix":
2236
 
                raise TestSkipped("Sending signals not supported")
2237
 
 
2238
 
        if env_changes is None:
2239
 
            env_changes = {}
2240
 
        # Because $HOME is set to a tempdir for the context of a test, modules
2241
 
        # installed in the user dir will not be found unless $PYTHONUSERBASE
2242
 
        # gets set to the computed directory of this parent process.
2243
 
        if site.USER_BASE is not None:
2244
 
            env_changes["PYTHONUSERBASE"] = site.USER_BASE
2245
 
        old_env = {}
2246
 
 
2247
 
        def cleanup_environment():
2248
 
            for env_var, value in env_changes.items():
2249
 
                old_env[env_var] = osutils.set_or_unset_env(env_var, value)
2250
 
 
2251
 
        def restore_environment():
2252
 
            for env_var, value in old_env.items():
2253
 
                osutils.set_or_unset_env(env_var, value)
2254
 
 
2255
 
        bzr_path = self.get_brz_path()
2256
 
 
2257
 
        cwd = None
2258
 
        if working_dir is not None:
2259
 
            cwd = osutils.getcwd()
2260
 
            os.chdir(working_dir)
2261
 
 
2262
 
        try:
2263
 
            # win32 subprocess doesn't support preexec_fn
2264
 
            # so we will avoid using it on all platforms, just to
2265
 
            # make sure the code path is used, and we don't break on win32
2266
 
            cleanup_environment()
2267
 
            # Include the subprocess's log file in the test details, in case
2268
 
            # the test fails due to an error in the subprocess.
2269
 
            self._add_subprocess_log(trace._get_brz_log_filename())
2270
 
            command = [sys.executable]
2271
 
            # frozen executables don't need the path to bzr
2272
 
            if getattr(sys, "frozen", None) is None:
2273
 
                command.append(bzr_path)
2274
 
            if not allow_plugins:
2275
 
                command.append('--no-plugins')
2276
 
            command.extend(process_args)
2277
 
            process = self._popen(command, stdin=subprocess.PIPE,
2278
 
                                  stdout=subprocess.PIPE,
2279
 
                                  stderr=stderr, bufsize=0)
2280
 
        finally:
2281
 
            restore_environment()
2282
 
            if cwd is not None:
2283
 
                os.chdir(cwd)
2284
 
 
2285
 
        return process
2286
 
 
2287
 
    def _add_subprocess_log(self, log_file_path):
2288
 
        if len(self._log_files) == 0:
2289
 
            # Register an addCleanup func.  We do this on the first call to
2290
 
            # _add_subprocess_log rather than in TestCase.setUp so that this
2291
 
            # addCleanup is registered after any cleanups for tempdirs that
2292
 
            # subclasses might create, which will probably remove the log file
2293
 
            # we want to read.
2294
 
            self.addCleanup(self._subprocess_log_cleanup)
2295
 
        # self._log_files is a set, so if a log file is reused we won't grab it
2296
 
        # twice.
2297
 
        self._log_files.add(log_file_path)
2298
 
 
2299
 
    def _subprocess_log_cleanup(self):
2300
 
        for count, log_file_path in enumerate(self._log_files):
2301
 
            # We use buffer_now=True to avoid holding the file open beyond
2302
 
            # the life of this function, which might interfere with e.g.
2303
 
            # cleaning tempdirs on Windows.
2304
 
            # XXX: Testtools 0.9.5 doesn't have the content_from_file helper
2305
 
            # detail_content = content.content_from_file(
2306
 
            #    log_file_path, buffer_now=True)
2307
 
            with open(log_file_path, 'rb') as log_file:
2308
 
                log_file_bytes = log_file.read()
2309
 
            detail_content = content.Content(
2310
 
                content.ContentType("text", "plain", {"charset": "utf8"}),
2311
 
                lambda: [log_file_bytes])
2312
 
            self.addDetail("start_bzr_subprocess-log-%d" % (count,),
2313
 
                           detail_content)
2314
 
 
2315
 
    def _popen(self, *args, **kwargs):
2316
 
        """Place a call to Popen.
2317
 
 
2318
 
        Allows tests to override this method to intercept the calls made to
2319
 
        Popen for introspection.
2320
 
        """
2321
 
        return subprocess.Popen(*args, **kwargs)
2322
 
 
2323
 
    def get_source_path(self):
2324
 
        """Return the path of the directory containing breezy."""
2325
 
        return os.path.dirname(os.path.dirname(breezy.__file__))
2326
 
 
2327
 
    def get_brz_path(self):
2328
 
        """Return the path of the 'brz' executable for this test suite."""
2329
 
        brz_path = os.path.join(self.get_source_path(), "brz")
2330
 
        if not os.path.isfile(brz_path):
2331
 
            # We are probably installed. Assume sys.argv is the right file
2332
 
            brz_path = sys.argv[0]
2333
 
        return brz_path
2334
 
 
2335
 
    def finish_bzr_subprocess(self, process, retcode=0, send_signal=None,
2336
 
                              universal_newlines=False, process_args=None):
2337
 
        """Finish the execution of process.
2338
 
 
2339
 
        :param process: the Popen object returned from start_bzr_subprocess.
2340
 
        :param retcode: The status code that is expected.  Defaults to 0.  If
2341
 
            None is supplied, the status code is not checked.
2342
 
        :param send_signal: an optional signal to send to the process.
2343
 
        :param universal_newlines: Convert CRLF => LF
2344
 
        :returns: (stdout, stderr)
2345
 
        """
2346
 
        if send_signal is not None:
2347
 
            os.kill(process.pid, send_signal)
2348
 
        out, err = process.communicate()
2349
 
 
2350
 
        if universal_newlines:
2351
 
            out = out.replace(b'\r\n', b'\n')
2352
 
            err = err.replace(b'\r\n', b'\n')
2353
 
 
2354
 
        if retcode is not None and retcode != process.returncode:
2355
 
            if process_args is None:
2356
 
                process_args = "(unknown args)"
2357
 
            trace.mutter('Output of brz %r:\n%s', process_args, out)
2358
 
            trace.mutter('Error for brz %r:\n%s', process_args, err)
2359
 
            self.fail('Command brz %r failed with retcode %d != %d'
2360
 
                      % (process_args, retcode, process.returncode))
2361
 
        return [out, err]
2362
 
 
2363
 
    def check_tree_shape(self, tree, shape):
2364
 
        """Compare a tree to a list of expected names.
 
248
        logging.debug(*args)
 
249
 
 
250
    def _get_log(self):
 
251
        """Return as a string the log for this test"""
 
252
        if self._log_file_name:
 
253
            return open(self._log_file_name).read()
 
254
        else:
 
255
            return ''
 
256
 
 
257
    def capture(self, cmd):
 
258
        """Shortcut that splits cmd into words, runs, and returns stdout"""
 
259
        return self.run_bzr_captured(cmd.split())[0]
 
260
 
 
261
    def run_bzr_captured(self, argv, retcode=0):
 
262
        """Invoke bzr and return (result, stdout, stderr).
 
263
 
 
264
        Useful for code that wants to check the contents of the
 
265
        output, the way error messages are presented, etc.
 
266
 
 
267
        This should be the main method for tests that want to exercise the
 
268
        overall behavior of the bzr application (rather than a unit test
 
269
        or a functional test of the library.)
 
270
 
 
271
        Much of the old code runs bzr by forking a new copy of Python, but
 
272
        that is slower, harder to debug, and generally not necessary.
 
273
 
 
274
        This runs bzr through the interface that catches and reports
 
275
        errors, and with logging set to something approximating the
 
276
        default, so that error reporting can be checked.
 
277
 
 
278
        argv -- arguments to invoke bzr
 
279
        retcode -- expected return code, or None for don't-care.
 
280
        """
 
281
        stdout = StringIO()
 
282
        stderr = StringIO()
 
283
        self.log('run bzr: %s', ' '.join(argv))
 
284
        handler = logging.StreamHandler(stderr)
 
285
        handler.setFormatter(bzrlib.trace.QuietFormatter())
 
286
        handler.setLevel(logging.INFO)
 
287
        logger = logging.getLogger('')
 
288
        logger.addHandler(handler)
 
289
        try:
 
290
            result = self.apply_redirected(None, stdout, stderr,
 
291
                                           bzrlib.commands.run_bzr_catch_errors,
 
292
                                           argv)
 
293
        finally:
 
294
            logger.removeHandler(handler)
 
295
        out = stdout.getvalue()
 
296
        err = stderr.getvalue()
 
297
        if out:
 
298
            self.log('output:\n%s', out)
 
299
        if err:
 
300
            self.log('errors:\n%s', err)
 
301
        if retcode is not None:
 
302
            self.assertEquals(result, retcode)
 
303
        return out, err
 
304
 
 
305
    def run_bzr(self, *args, **kwargs):
 
306
        """Invoke bzr, as if it were run from the command line.
 
307
 
 
308
        This should be the main method for tests that want to exercise the
 
309
        overall behavior of the bzr application (rather than a unit test
 
310
        or a functional test of the library.)
 
311
 
 
312
        This sends the stdout/stderr results into the test's log,
 
313
        where it may be useful for debugging.  See also run_captured.
 
314
        """
 
315
        retcode = kwargs.pop('retcode', 0)
 
316
        return self.run_bzr_captured(args, retcode)
 
317
 
 
318
    def check_inventory_shape(self, inv, shape):
 
319
        """Compare an inventory to a list of expected names.
2365
320
 
2366
321
        Fail if they are not precisely equal.
2367
322
        """
2368
323
        extras = []
2369
324
        shape = list(shape)             # copy
2370
 
        for path, ie in tree.iter_entries_by_dir():
 
325
        for path, ie in inv.entries():
2371
326
            name = path.replace('\\', '/')
2372
 
            if ie.kind == 'directory':
 
327
            if ie.kind == 'dir':
2373
328
                name = name + '/'
2374
 
            if name == "/":
2375
 
                pass  # ignore root entry
2376
 
            elif name in shape:
 
329
            if name in shape:
2377
330
                shape.remove(name)
2378
331
            else:
2379
332
                extras.append(name)
2390
343
        if not callable(a_callable):
2391
344
            raise ValueError("a_callable must be callable.")
2392
345
        if stdin is None:
2393
 
            stdin = BytesIO(b"")
 
346
            stdin = StringIO("")
2394
347
        if stdout is None:
2395
 
            if getattr(self, "_log_file", None) is not None:
 
348
            if hasattr(self, "_log_file"):
2396
349
                stdout = self._log_file
2397
350
            else:
2398
 
                if sys.version_info[0] == 2:
2399
 
                    stdout = BytesIO()
2400
 
                else:
2401
 
                    stdout = StringIO()
 
351
                stdout = StringIO()
2402
352
        if stderr is None:
2403
 
            if getattr(self, "_log_file", None is not None):
 
353
            if hasattr(self, "_log_file"):
2404
354
                stderr = self._log_file
2405
355
            else:
2406
 
                if sys.version_info[0] == 2:
2407
 
                    stderr = BytesIO()
2408
 
                else:
2409
 
                    stderr = StringIO()
 
356
                stderr = StringIO()
2410
357
        real_stdin = sys.stdin
2411
358
        real_stdout = sys.stdout
2412
359
        real_stderr = sys.stderr
2420
367
            sys.stderr = real_stderr
2421
368
            sys.stdin = real_stdin
2422
369
 
2423
 
    def reduceLockdirTimeout(self):
2424
 
        """Reduce the default lock timeout for the duration of the test, so that
2425
 
        if LockContention occurs during a test, it does so quickly.
2426
 
 
2427
 
        Tests that expect to provoke LockContention errors should call this.
2428
 
        """
2429
 
        self.overrideAttr(lockdir, '_DEFAULT_TIMEOUT_SECONDS', 0)
2430
 
 
2431
 
    def make_utf8_encoded_stringio(self, encoding_type=None):
2432
 
        """Return a wrapped BytesIO, that will encode text input to UTF-8."""
2433
 
        if encoding_type is None:
2434
 
            encoding_type = 'strict'
2435
 
        bio = BytesIO()
2436
 
        output_encoding = 'utf-8'
2437
 
        sio = codecs.getwriter(output_encoding)(bio, errors=encoding_type)
2438
 
        sio.encoding = output_encoding
2439
 
        return sio
2440
 
 
2441
 
    def disable_verb(self, verb):
2442
 
        """Disable a smart server verb for one test."""
2443
 
        from breezy.bzr.smart import request
2444
 
        request_handlers = request.request_handlers
2445
 
        orig_method = request_handlers.get(verb)
2446
 
        orig_info = request_handlers.get_info(verb)
2447
 
        request_handlers.remove(verb)
2448
 
        self.addCleanup(request_handlers.register, verb, orig_method,
2449
 
                        info=orig_info)
2450
 
 
2451
 
    def __hash__(self):
2452
 
        return id(self)
2453
 
 
2454
 
 
2455
 
class CapturedCall(object):
2456
 
    """A helper for capturing smart server calls for easy debug analysis."""
2457
 
 
2458
 
    def __init__(self, params, prefix_length):
2459
 
        """Capture the call with params and skip prefix_length stack frames."""
2460
 
        self.call = params
2461
 
        import traceback
2462
 
        # The last 5 frames are the __init__, the hook frame, and 3 smart
2463
 
        # client frames. Beyond this we could get more clever, but this is good
2464
 
        # enough for now.
2465
 
        stack = traceback.extract_stack()[prefix_length:-5]
2466
 
        self.stack = ''.join(traceback.format_list(stack))
2467
 
 
2468
 
    def __str__(self):
2469
 
        return self.call.method
2470
 
 
2471
 
    def __repr__(self):
2472
 
        return self.call.method
2473
 
 
2474
 
    def stack(self):
2475
 
        return self.stack
2476
 
 
2477
 
 
2478
 
class TestCaseWithMemoryTransport(TestCase):
2479
 
    """Common test class for tests that do not need disk resources.
2480
 
 
2481
 
    Tests that need disk resources should derive from TestCaseInTempDir
2482
 
    orTestCaseWithTransport.
2483
 
 
2484
 
    TestCaseWithMemoryTransport sets the TEST_ROOT variable for all brz tests.
2485
 
 
2486
 
    For TestCaseWithMemoryTransport the ``test_home_dir`` is set to the name of
2487
 
    a directory which does not exist. This serves to help ensure test isolation
2488
 
    is preserved. ``test_dir`` is set to the TEST_ROOT, as is cwd, because they
2489
 
    must exist. However, TestCaseWithMemoryTransport does not offer local file
2490
 
    defaults for the transport in tests, nor does it obey the command line
2491
 
    override, so tests that accidentally write to the common directory should
2492
 
    be rare.
2493
 
 
2494
 
    :cvar TEST_ROOT: Directory containing all temporary directories, plus a
2495
 
        ``.bzr`` directory that stops us ascending higher into the filesystem.
2496
 
    """
2497
 
 
2498
 
    TEST_ROOT = None
2499
 
    _TEST_NAME = 'test'
2500
 
 
2501
 
    def __init__(self, methodName='runTest'):
2502
 
        # allow test parameterization after test construction and before test
2503
 
        # execution. Variables that the parameterizer sets need to be
2504
 
        # ones that are not set by setUp, or setUp will trash them.
2505
 
        super(TestCaseWithMemoryTransport, self).__init__(methodName)
2506
 
        self.vfs_transport_factory = default_transport
2507
 
        self.transport_server = None
2508
 
        self.transport_readonly_server = None
2509
 
        self.__vfs_server = None
2510
 
 
2511
 
    def setUp(self):
2512
 
        super(TestCaseWithMemoryTransport, self).setUp()
2513
 
 
2514
 
        def _add_disconnect_cleanup(transport):
2515
 
            """Schedule disconnection of given transport at test cleanup
2516
 
 
2517
 
            This needs to happen for all connected transports or leaks occur.
2518
 
 
2519
 
            Note reconnections may mean we call disconnect multiple times per
2520
 
            transport which is suboptimal but seems harmless.
2521
 
            """
2522
 
            self.addCleanup(transport.disconnect)
2523
 
 
2524
 
        _mod_transport.Transport.hooks.install_named_hook(
2525
 
            'post_connect', _add_disconnect_cleanup, None)
2526
 
 
2527
 
        self._make_test_root()
2528
 
        self.addCleanup(os.chdir, osutils.getcwd())
2529
 
        self.makeAndChdirToTestDir()
2530
 
        self.overrideEnvironmentForTesting()
2531
 
        self.__readonly_server = None
2532
 
        self.__server = None
2533
 
        self.reduceLockdirTimeout()
2534
 
        # Each test may use its own config files even if the local config files
2535
 
        # don't actually exist. They'll rightly fail if they try to create them
2536
 
        # though.
2537
 
        self.overrideAttr(config, '_shared_stores', {})
2538
 
 
2539
 
    def get_transport(self, relpath=None):
2540
 
        """Return a writeable transport.
2541
 
 
2542
 
        This transport is for the test scratch space relative to
2543
 
        "self._test_root"
2544
 
 
2545
 
        :param relpath: a path relative to the base url.
2546
 
        """
2547
 
        t = _mod_transport.get_transport_from_url(self.get_url(relpath))
2548
 
        self.assertFalse(t.is_readonly())
2549
 
        return t
2550
 
 
2551
 
    def get_readonly_transport(self, relpath=None):
2552
 
        """Return a readonly transport for the test scratch space
2553
 
 
2554
 
        This can be used to test that operations which should only need
2555
 
        readonly access in fact do not try to write.
2556
 
 
2557
 
        :param relpath: a path relative to the base url.
2558
 
        """
2559
 
        t = _mod_transport.get_transport_from_url(
2560
 
            self.get_readonly_url(relpath))
2561
 
        self.assertTrue(t.is_readonly())
2562
 
        return t
2563
 
 
2564
 
    def create_transport_readonly_server(self):
2565
 
        """Create a transport server from class defined at init.
2566
 
 
2567
 
        This is mostly a hook for daughter classes.
2568
 
        """
2569
 
        return self.transport_readonly_server()
2570
 
 
2571
 
    def get_readonly_server(self):
2572
 
        """Get the server instance for the readonly transport
2573
 
 
2574
 
        This is useful for some tests with specific servers to do diagnostics.
2575
 
        """
2576
 
        if self.__readonly_server is None:
2577
 
            if self.transport_readonly_server is None:
2578
 
                # readonly decorator requested
2579
 
                self.__readonly_server = test_server.ReadonlyServer()
2580
 
            else:
2581
 
                # explicit readonly transport.
2582
 
                self.__readonly_server = (
2583
 
                    self.create_transport_readonly_server())
2584
 
            self.start_server(self.__readonly_server,
2585
 
                              self.get_vfs_only_server())
2586
 
        return self.__readonly_server
2587
 
 
2588
 
    def get_readonly_url(self, relpath=None):
2589
 
        """Get a URL for the readonly transport.
2590
 
 
2591
 
        This will either be backed by '.' or a decorator to the transport
2592
 
        used by self.get_url()
2593
 
        relpath provides for clients to get a path relative to the base url.
2594
 
        These should only be downwards relative, not upwards.
2595
 
        """
2596
 
        base = self.get_readonly_server().get_url()
2597
 
        return self._adjust_url(base, relpath)
2598
 
 
2599
 
    def get_vfs_only_server(self):
2600
 
        """Get the vfs only read/write server instance.
2601
 
 
2602
 
        This is useful for some tests with specific servers that need
2603
 
        diagnostics.
2604
 
 
2605
 
        For TestCaseWithMemoryTransport this is always a MemoryServer, and there
2606
 
        is no means to override it.
2607
 
        """
2608
 
        if self.__vfs_server is None:
2609
 
            self.__vfs_server = memory.MemoryServer()
2610
 
            self.start_server(self.__vfs_server)
2611
 
        return self.__vfs_server
2612
 
 
2613
 
    def get_server(self):
2614
 
        """Get the read/write server instance.
2615
 
 
2616
 
        This is useful for some tests with specific servers that need
2617
 
        diagnostics.
2618
 
 
2619
 
        This is built from the self.transport_server factory. If that is None,
2620
 
        then the self.get_vfs_server is returned.
2621
 
        """
2622
 
        if self.__server is None:
2623
 
            if (self.transport_server is None or self.transport_server is
2624
 
                    self.vfs_transport_factory):
2625
 
                self.__server = self.get_vfs_only_server()
2626
 
            else:
2627
 
                # bring up a decorated means of access to the vfs only server.
2628
 
                self.__server = self.transport_server()
2629
 
                self.start_server(self.__server, self.get_vfs_only_server())
2630
 
        return self.__server
2631
 
 
2632
 
    def _adjust_url(self, base, relpath):
2633
 
        """Get a URL (or maybe a path) for the readwrite transport.
2634
 
 
2635
 
        This will either be backed by '.' or to an equivalent non-file based
2636
 
        facility.
2637
 
        relpath provides for clients to get a path relative to the base url.
2638
 
        These should only be downwards relative, not upwards.
2639
 
        """
2640
 
        if relpath is not None and relpath != '.':
2641
 
            if not base.endswith('/'):
2642
 
                base = base + '/'
2643
 
            # XXX: Really base should be a url; we did after all call
2644
 
            # get_url()!  But sometimes it's just a path (from
2645
 
            # LocalAbspathServer), and it'd be wrong to append urlescaped data
2646
 
            # to a non-escaped local path.
2647
 
            if base.startswith('./') or base.startswith('/'):
2648
 
                base += relpath
2649
 
            else:
2650
 
                base += urlutils.escape(relpath)
2651
 
        return base
2652
 
 
2653
 
    def get_url(self, relpath=None):
2654
 
        """Get a URL (or maybe a path) for the readwrite transport.
2655
 
 
2656
 
        This will either be backed by '.' or to an equivalent non-file based
2657
 
        facility.
2658
 
        relpath provides for clients to get a path relative to the base url.
2659
 
        These should only be downwards relative, not upwards.
2660
 
        """
2661
 
        base = self.get_server().get_url()
2662
 
        return self._adjust_url(base, relpath)
2663
 
 
2664
 
    def get_vfs_only_url(self, relpath=None):
2665
 
        """Get a URL (or maybe a path for the plain old vfs transport.
2666
 
 
2667
 
        This will never be a smart protocol.  It always has all the
2668
 
        capabilities of the local filesystem, but it might actually be a
2669
 
        MemoryTransport or some other similar virtual filesystem.
2670
 
 
2671
 
        This is the backing transport (if any) of the server returned by
2672
 
        get_url and get_readonly_url.
2673
 
 
2674
 
        :param relpath: provides for clients to get a path relative to the base
2675
 
            url.  These should only be downwards relative, not upwards.
2676
 
        :return: A URL
2677
 
        """
2678
 
        base = self.get_vfs_only_server().get_url()
2679
 
        return self._adjust_url(base, relpath)
2680
 
 
2681
 
    def _create_safety_net(self):
2682
 
        """Make a fake bzr directory.
2683
 
 
2684
 
        This prevents any tests propagating up onto the TEST_ROOT directory's
2685
 
        real branch.
2686
 
        """
2687
 
        root = TestCaseWithMemoryTransport.TEST_ROOT
2688
 
        try:
2689
 
            # Make sure we get a readable and accessible home for brz.log
2690
 
            # and/or config files, and not fallback to weird defaults (see
2691
 
            # http://pad.lv/825027).
2692
 
            self.assertIs(None, os.environ.get('BRZ_HOME', None))
2693
 
            os.environ['BRZ_HOME'] = root
2694
 
            wt = controldir.ControlDir.create_standalone_workingtree(root)
2695
 
            del os.environ['BRZ_HOME']
2696
 
        except Exception as e:
2697
 
            self.fail("Fail to initialize the safety net: %r\n" % (e,))
2698
 
        # Hack for speed: remember the raw bytes of the dirstate file so that
2699
 
        # we don't need to re-open the wt to check it hasn't changed.
2700
 
        TestCaseWithMemoryTransport._SAFETY_NET_PRISTINE_DIRSTATE = (
2701
 
            wt.control_transport.get_bytes('dirstate'))
2702
 
 
2703
 
    def _check_safety_net(self):
2704
 
        """Check that the safety .bzr directory have not been touched.
2705
 
 
2706
 
        _make_test_root have created a .bzr directory to prevent tests from
2707
 
        propagating. This method ensures than a test did not leaked.
2708
 
        """
2709
 
        root = TestCaseWithMemoryTransport.TEST_ROOT
2710
 
        t = _mod_transport.get_transport_from_path(root)
2711
 
        self.permit_url(t.base)
2712
 
        if (t.get_bytes('.bzr/checkout/dirstate') !=
2713
 
                TestCaseWithMemoryTransport._SAFETY_NET_PRISTINE_DIRSTATE):
2714
 
            # The current test have modified the /bzr directory, we need to
2715
 
            # recreate a new one or all the followng tests will fail.
2716
 
            # If you need to inspect its content uncomment the following line
2717
 
            # import pdb; pdb.set_trace()
2718
 
            _rmtree_temp_dir(root + '/.bzr', test_id=self.id())
2719
 
            self._create_safety_net()
2720
 
            raise AssertionError('%s/.bzr should not be modified' % root)
2721
 
 
2722
 
    def _make_test_root(self):
2723
 
        if TestCaseWithMemoryTransport.TEST_ROOT is None:
2724
 
            # Watch out for tricky test dir (on OSX /tmp -> /private/tmp)
2725
 
            root = osutils.realpath(osutils.mkdtemp(prefix='testbzr-',
2726
 
                                                    suffix='.tmp'))
2727
 
            TestCaseWithMemoryTransport.TEST_ROOT = root
2728
 
 
2729
 
            self._create_safety_net()
2730
 
 
2731
 
            # The same directory is used by all tests, and we're not
2732
 
            # specifically told when all tests are finished.  This will do.
2733
 
            atexit.register(_rmtree_temp_dir, root)
2734
 
 
2735
 
        self.permit_dir(TestCaseWithMemoryTransport.TEST_ROOT)
2736
 
        self.addCleanup(self._check_safety_net)
2737
 
 
2738
 
    def makeAndChdirToTestDir(self):
2739
 
        """Create a temporary directories for this one test.
2740
 
 
2741
 
        This must set self.test_home_dir and self.test_dir and chdir to
2742
 
        self.test_dir.
2743
 
 
2744
 
        For TestCaseWithMemoryTransport we chdir to the TEST_ROOT for this
2745
 
        test.
2746
 
        """
2747
 
        os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
2748
 
        self.test_dir = TestCaseWithMemoryTransport.TEST_ROOT
2749
 
        self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
2750
 
        self.permit_dir(self.test_dir)
2751
 
 
2752
 
    def make_branch(self, relpath, format=None, name=None):
2753
 
        """Create a branch on the transport at relpath."""
2754
 
        repo = self.make_repository(relpath, format=format)
2755
 
        return repo.controldir.create_branch(append_revisions_only=False, name=name)
2756
 
 
2757
 
    def get_default_format(self):
2758
 
        return 'default'
2759
 
 
2760
 
    def resolve_format(self, format):
2761
 
        """Resolve an object to a ControlDir format object.
2762
 
 
2763
 
        The initial format object can either already be
2764
 
        a ControlDirFormat, None (for the default format),
2765
 
        or a string with the name of the control dir format.
2766
 
 
2767
 
        :param format: Object to resolve
2768
 
        :return A ControlDirFormat instance
2769
 
        """
2770
 
        if format is None:
2771
 
            format = self.get_default_format()
2772
 
        if isinstance(format, str):
2773
 
            format = controldir.format_registry.make_controldir(format)
2774
 
        return format
2775
 
 
2776
 
    def make_controldir(self, relpath, format=None):
2777
 
        try:
2778
 
            # might be a relative or absolute path
2779
 
            maybe_a_url = self.get_url(relpath)
2780
 
            segments = maybe_a_url.rsplit('/', 1)
2781
 
            t = _mod_transport.get_transport(maybe_a_url)
2782
 
            if len(segments) > 1 and segments[-1] not in ('', '.'):
2783
 
                t.ensure_base()
2784
 
            format = self.resolve_format(format)
2785
 
            return format.initialize_on_transport(t)
2786
 
        except errors.UninitializableFormat:
2787
 
            raise TestSkipped("Format %s is not initializable." % format)
2788
 
 
2789
 
    def make_repository(self, relpath, shared=None, format=None):
2790
 
        """Create a repository on our default transport at relpath.
2791
 
 
2792
 
        Note that relpath must be a relative path, not a full url.
2793
 
        """
2794
 
        # FIXME: If you create a remoterepository this returns the underlying
2795
 
        # real format, which is incorrect.  Actually we should make sure that
2796
 
        # RemoteBzrDir returns a RemoteRepository.
2797
 
        # maybe  mbp 20070410
2798
 
        made_control = self.make_controldir(relpath, format=format)
2799
 
        return made_control.create_repository(shared=shared)
2800
 
 
2801
 
    def make_smart_server(self, path, backing_server=None):
2802
 
        if backing_server is None:
2803
 
            backing_server = self.get_server()
2804
 
        smart_server = test_server.SmartTCPServer_for_testing()
2805
 
        self.start_server(smart_server, backing_server)
2806
 
        remote_transport = _mod_transport.get_transport_from_url(smart_server.get_url()
2807
 
                                                                 ).clone(path)
2808
 
        return remote_transport
2809
 
 
2810
 
    def make_branch_and_memory_tree(self, relpath, format=None):
2811
 
        """Create a branch on the default transport and a MemoryTree for it."""
2812
 
        b = self.make_branch(relpath, format=format)
2813
 
        return b.create_memorytree()
2814
 
 
2815
 
    def make_branch_builder(self, relpath, format=None):
2816
 
        branch = self.make_branch(relpath, format=format)
2817
 
        return branchbuilder.BranchBuilder(branch=branch)
2818
 
 
2819
 
    def overrideEnvironmentForTesting(self):
2820
 
        test_home_dir = self.test_home_dir
2821
 
        if not PY3 and isinstance(test_home_dir, text_type):
2822
 
            test_home_dir = test_home_dir.encode(sys.getfilesystemencoding())
2823
 
        self.overrideEnv('HOME', test_home_dir)
2824
 
        self.overrideEnv('BRZ_HOME', test_home_dir)
2825
 
        self.overrideEnv('GNUPGHOME', os.path.join(test_home_dir, '.gnupg'))
2826
 
 
2827
 
    def setup_smart_server_with_call_log(self):
2828
 
        """Sets up a smart server as the transport server with a call log."""
2829
 
        self.transport_server = test_server.SmartTCPServer_for_testing
2830
 
        self.hpss_connections = []
2831
 
        self.hpss_calls = []
2832
 
        import traceback
2833
 
        # Skip the current stack down to the caller of
2834
 
        # setup_smart_server_with_call_log
2835
 
        prefix_length = len(traceback.extract_stack()) - 2
2836
 
 
2837
 
        def capture_hpss_call(params):
2838
 
            self.hpss_calls.append(
2839
 
                CapturedCall(params, prefix_length))
2840
 
 
2841
 
        def capture_connect(transport):
2842
 
            self.hpss_connections.append(transport)
2843
 
        client._SmartClient.hooks.install_named_hook(
2844
 
            'call', capture_hpss_call, None)
2845
 
        _mod_transport.Transport.hooks.install_named_hook(
2846
 
            'post_connect', capture_connect, None)
2847
 
 
2848
 
    def reset_smart_call_log(self):
2849
 
        self.hpss_calls = []
2850
 
        self.hpss_connections = []
2851
 
 
2852
 
 
2853
 
class TestCaseInTempDir(TestCaseWithMemoryTransport):
 
370
 
 
371
BzrTestBase = TestCase
 
372
 
 
373
     
 
374
class TestCaseInTempDir(TestCase):
2854
375
    """Derived class that runs a test within a temporary directory.
2855
376
 
2856
377
    This is useful for tests that need to create a branch, etc.
2860
381
    All test cases create their own directory within that.  If the
2861
382
    tests complete successfully, the directory is removed.
2862
383
 
2863
 
    :ivar test_base_dir: The path of the top-level directory for this
2864
 
    test, which contains a home directory and a work directory.
2865
 
 
2866
 
    :ivar test_home_dir: An initially empty directory under test_base_dir
2867
 
    which is used as $HOME for this test.
2868
 
 
2869
 
    :ivar test_dir: A directory under test_base_dir used as the current
2870
 
    directory when the test proper is run.
 
384
    InTempDir is an old alias for FunctionalTestCase.
2871
385
    """
2872
386
 
 
387
    TEST_ROOT = None
 
388
    _TEST_NAME = 'test'
2873
389
    OVERRIDE_PYTHON = 'python'
2874
390
 
2875
 
    def setUp(self):
2876
 
        super(TestCaseInTempDir, self).setUp()
2877
 
        # Remove the protection set in isolated_environ, we have a proper
2878
 
        # access to disk resources now.
2879
 
        self.overrideEnv('BRZ_LOG', None)
2880
 
 
2881
391
    def check_file_contents(self, filename, expect):
2882
392
        self.log("check contents of file %s" % filename)
2883
 
        with open(filename, 'rb') as f:
2884
 
            contents = f.read()
 
393
        contents = file(filename, 'r').read()
2885
394
        if contents != expect:
2886
395
            self.log("expected: %r" % expect)
2887
396
            self.log("actually: %r" % contents)
2888
397
            self.fail("contents of %s not as expected" % filename)
2889
398
 
2890
 
    def _getTestDirPrefix(self):
2891
 
        # create a directory within the top level test directory
2892
 
        if sys.platform in ('win32', 'cygwin'):
2893
 
            name_prefix = re.sub('[<>*=+",:;_/\\-]', '_', self.id())
2894
 
            # windows is likely to have path-length limits so use a short name
2895
 
            name_prefix = name_prefix[-30:]
2896
 
        else:
2897
 
            name_prefix = re.sub('[/]', '_', self.id())
2898
 
        return name_prefix
2899
 
 
2900
 
    def makeAndChdirToTestDir(self):
2901
 
        """See TestCaseWithMemoryTransport.makeAndChdirToTestDir().
2902
 
 
2903
 
        For TestCaseInTempDir we create a temporary directory based on the test
2904
 
        name and then create two subdirs - test and home under it.
2905
 
        """
2906
 
        name_prefix = osutils.pathjoin(TestCaseWithMemoryTransport.TEST_ROOT,
2907
 
                                       self._getTestDirPrefix())
2908
 
        name = name_prefix
2909
 
        for i in range(100):
2910
 
            if os.path.exists(name):
2911
 
                name = name_prefix + '_' + str(i)
2912
 
            else:
2913
 
                # now create test and home directories within this dir
2914
 
                self.test_base_dir = name
2915
 
                self.addCleanup(self.deleteTestDir)
2916
 
                os.mkdir(self.test_base_dir)
2917
 
                break
2918
 
        self.permit_dir(self.test_base_dir)
2919
 
        # 'sprouting' and 'init' of a branch both walk up the tree to find
2920
 
        # stacking policy to honour; create a bzr dir with an unshared
2921
 
        # repository (but not a branch - our code would be trying to escape
2922
 
        # then!) to stop them, and permit it to be read.
2923
 
        # control = controldir.ControlDir.create(self.test_base_dir)
2924
 
        # control.create_repository()
2925
 
        self.test_home_dir = self.test_base_dir + '/home'
2926
 
        os.mkdir(self.test_home_dir)
2927
 
        self.test_dir = self.test_base_dir + '/work'
 
399
    def _make_test_root(self):
 
400
        if TestCaseInTempDir.TEST_ROOT is not None:
 
401
            return
 
402
        i = 0
 
403
        while True:
 
404
            root = 'test%04d.tmp' % i
 
405
            try:
 
406
                os.mkdir(root)
 
407
            except OSError, e:
 
408
                if e.errno == errno.EEXIST:
 
409
                    i += 1
 
410
                    continue
 
411
                else:
 
412
                    raise
 
413
            # successfully created
 
414
            TestCaseInTempDir.TEST_ROOT = os.path.abspath(root)
 
415
            break
 
416
        # make a fake bzr directory there to prevent any tests propagating
 
417
        # up onto the source directory's real branch
 
418
        os.mkdir(os.path.join(TestCaseInTempDir.TEST_ROOT, '.bzr'))
 
419
 
 
420
    def setUp(self):
 
421
        self._make_test_root()
 
422
        self._currentdir = os.getcwdu()
 
423
        short_id = self.id().replace('bzrlib.selftest.', '') \
 
424
                   .replace('__main__.', '')
 
425
        self.test_dir = os.path.join(self.TEST_ROOT, short_id)
2928
426
        os.mkdir(self.test_dir)
2929
427
        os.chdir(self.test_dir)
2930
 
        # put name of test inside
2931
 
        with open(self.test_base_dir + '/name', 'w') as f:
2932
 
            f.write(self.id())
2933
 
 
2934
 
    def deleteTestDir(self):
2935
 
        os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
2936
 
        _rmtree_temp_dir(self.test_base_dir, test_id=self.id())
2937
 
 
2938
 
    def build_tree(self, shape, line_endings='binary', transport=None):
 
428
        super(TestCaseInTempDir, self).setUp()
 
429
        
 
430
    def tearDown(self):
 
431
        os.chdir(self._currentdir)
 
432
        super(TestCaseInTempDir, self).tearDown()
 
433
 
 
434
    def build_tree(self, shape):
2939
435
        """Build a test tree according to a pattern.
2940
436
 
2941
437
        shape is a sequence of file specifications.  If the final
2942
438
        character is '/', a directory is created.
2943
439
 
2944
 
        This assumes that all the elements in the tree being built are new.
2945
 
 
2946
440
        This doesn't add anything to a branch.
2947
 
 
2948
 
        :type shape:    list or tuple.
2949
 
        :param line_endings: Either 'binary' or 'native'
2950
 
            in binary mode, exact contents are written in native mode, the
2951
 
            line endings match the default platform endings.
2952
 
        :param transport: A transport to write to, for building trees on VFS's.
2953
 
            If the transport is readonly or None, "." is opened automatically.
2954
 
        :return: None
2955
441
        """
2956
 
        if type(shape) not in (list, tuple):
2957
 
            raise AssertionError("Parameter 'shape' should be "
2958
 
                                 "a list or a tuple. Got %r instead" % (shape,))
2959
 
        # It's OK to just create them using forward slashes on windows.
2960
 
        if transport is None or transport.is_readonly():
2961
 
            transport = _mod_transport.get_transport_from_path(".")
 
442
        # XXX: It's OK to just create them using forward slashes on windows?
2962
443
        for name in shape:
2963
 
            self.assertIsInstance(name, (str, text_type))
 
444
            assert isinstance(name, basestring)
2964
445
            if name[-1] == '/':
2965
 
                transport.mkdir(urlutils.escape(name[:-1]))
2966
 
            else:
2967
 
                if line_endings == 'binary':
2968
 
                    end = b'\n'
2969
 
                elif line_endings == 'native':
2970
 
                    end = os.linesep.encode('ascii')
2971
 
                else:
2972
 
                    raise errors.BzrError(
2973
 
                        'Invalid line ending request %r' % line_endings)
2974
 
                content = b"contents of %s%s" % (name.encode('utf-8'), end)
2975
 
                transport.put_bytes_non_atomic(urlutils.escape(name), content)
2976
 
 
2977
 
    build_tree_contents = staticmethod(treeshape.build_tree_contents)
2978
 
 
2979
 
    def assertInWorkingTree(self, path, root_path='.', tree=None):
2980
 
        """Assert whether path or paths are in the WorkingTree"""
2981
 
        if tree is None:
2982
 
            tree = workingtree.WorkingTree.open(root_path)
2983
 
        if not isinstance(path, (str, text_type)):
2984
 
            for p in path:
2985
 
                self.assertInWorkingTree(p, tree=tree)
2986
 
        else:
2987
 
            self.assertTrue(tree.is_versioned(path),
2988
 
                            path + ' not in working tree.')
2989
 
 
2990
 
    def assertNotInWorkingTree(self, path, root_path='.', tree=None):
2991
 
        """Assert whether path or paths are not in the WorkingTree"""
2992
 
        if tree is None:
2993
 
            tree = workingtree.WorkingTree.open(root_path)
2994
 
        if not isinstance(path, (str, text_type)):
2995
 
            for p in path:
2996
 
                self.assertNotInWorkingTree(p, tree=tree)
2997
 
        else:
2998
 
            self.assertFalse(tree.is_versioned(
2999
 
                path), path + ' in working tree.')
3000
 
 
3001
 
 
3002
 
class TestCaseWithTransport(TestCaseInTempDir):
3003
 
    """A test case that provides get_url and get_readonly_url facilities.
3004
 
 
3005
 
    These back onto two transport servers, one for readonly access and one for
3006
 
    read write access.
3007
 
 
3008
 
    If no explicit class is provided for readonly access, a
3009
 
    ReadonlyTransportDecorator is used instead which allows the use of non disk
3010
 
    based read write transports.
3011
 
 
3012
 
    If an explicit class is provided for readonly access, that server and the
3013
 
    readwrite one must both define get_url() as resolving to os.getcwd().
3014
 
    """
3015
 
 
3016
 
    def setUp(self):
3017
 
        super(TestCaseWithTransport, self).setUp()
3018
 
        self.__vfs_server = None
3019
 
 
3020
 
    def get_vfs_only_server(self):
3021
 
        """See TestCaseWithMemoryTransport.
3022
 
 
3023
 
        This is useful for some tests with specific servers that need
3024
 
        diagnostics.
3025
 
        """
3026
 
        if self.__vfs_server is None:
3027
 
            self.__vfs_server = self.vfs_transport_factory()
3028
 
            self.start_server(self.__vfs_server)
3029
 
        return self.__vfs_server
3030
 
 
3031
 
    def make_branch_and_tree(self, relpath, format=None):
3032
 
        """Create a branch on the transport and a tree locally.
3033
 
 
3034
 
        If the transport is not a LocalTransport, the Tree can't be created on
3035
 
        the transport.  In that case if the vfs_transport_factory is
3036
 
        LocalURLServer the working tree is created in the local
3037
 
        directory backing the transport, and the returned tree's branch and
3038
 
        repository will also be accessed locally. Otherwise a lightweight
3039
 
        checkout is created and returned.
3040
 
 
3041
 
        We do this because we can't physically create a tree in the local
3042
 
        path, with a branch reference to the transport_factory url, and
3043
 
        a branch + repository in the vfs_transport, unless the vfs_transport
3044
 
        namespace is distinct from the local disk - the two branch objects
3045
 
        would collide. While we could construct a tree with its branch object
3046
 
        pointing at the transport_factory transport in memory, reopening it
3047
 
        would behaving unexpectedly, and has in the past caused testing bugs
3048
 
        when we tried to do it that way.
3049
 
 
3050
 
        :param format: The BzrDirFormat.
3051
 
        :returns: the WorkingTree.
3052
 
        """
3053
 
        # TODO: always use the local disk path for the working tree,
3054
 
        # this obviously requires a format that supports branch references
3055
 
        # so check for that by checking bzrdir.BzrDirFormat.get_default_format()
3056
 
        # RBC 20060208
3057
 
        format = self.resolve_format(format=format)
3058
 
        if not format.supports_workingtrees:
3059
 
            b = self.make_branch(relpath + '.branch', format=format)
3060
 
            return b.create_checkout(relpath, lightweight=True)
3061
 
        b = self.make_branch(relpath, format=format)
3062
 
        try:
3063
 
            return b.controldir.create_workingtree()
3064
 
        except errors.NotLocalUrl:
3065
 
            # We can only make working trees locally at the moment.  If the
3066
 
            # transport can't support them, then we keep the non-disk-backed
3067
 
            # branch and create a local checkout.
3068
 
            if self.vfs_transport_factory is test_server.LocalURLServer:
3069
 
                # the branch is colocated on disk, we cannot create a checkout.
3070
 
                # hopefully callers will expect this.
3071
 
                local_controldir = controldir.ControlDir.open(
3072
 
                    self.get_vfs_only_url(relpath))
3073
 
                wt = local_controldir.create_workingtree()
3074
 
                if wt.branch._format != b._format:
3075
 
                    wt._branch = b
3076
 
                    # Make sure that assigning to wt._branch fixes wt.branch,
3077
 
                    # in case the implementation details of workingtree objects
3078
 
                    # change.
3079
 
                    self.assertIs(b, wt.branch)
3080
 
                return wt
3081
 
            else:
3082
 
                return b.create_checkout(relpath, lightweight=True)
3083
 
 
3084
 
    def assertIsDirectory(self, relpath, transport):
3085
 
        """Assert that relpath within transport is a directory.
3086
 
 
3087
 
        This may not be possible on all transports; in that case it propagates
3088
 
        a TransportNotPossible.
3089
 
        """
3090
 
        try:
3091
 
            mode = transport.stat(relpath).st_mode
3092
 
        except errors.NoSuchFile:
3093
 
            self.fail("path %s is not a directory; no such file"
3094
 
                      % (relpath))
3095
 
        if not stat.S_ISDIR(mode):
3096
 
            self.fail("path %s is not a directory; has mode %#o"
3097
 
                      % (relpath, mode))
3098
 
 
3099
 
    def assertTreesEqual(self, left, right):
3100
 
        """Check that left and right have the same content and properties."""
3101
 
        # we use a tree delta to check for equality of the content, and we
3102
 
        # manually check for equality of other things such as the parents list.
3103
 
        self.assertEqual(left.get_parent_ids(), right.get_parent_ids())
3104
 
        differences = left.changes_from(right)
3105
 
        self.assertFalse(differences.has_changed(),
3106
 
                         "Trees %r and %r are different: %r" % (left, right, differences))
3107
 
 
3108
 
    def disable_missing_extensions_warning(self):
3109
 
        """Some tests expect a precise stderr content.
3110
 
 
3111
 
        There is no point in forcing them to duplicate the extension related
3112
 
        warning.
3113
 
        """
3114
 
        config.GlobalConfig().set_user_option(
3115
 
            'suppress_warnings', 'missing_extensions')
3116
 
 
3117
 
 
3118
 
class ChrootedTestCase(TestCaseWithTransport):
3119
 
    """A support class that provides readonly urls outside the local namespace.
3120
 
 
3121
 
    This is done by checking if self.transport_server is a MemoryServer. if it
3122
 
    is then we are chrooted already, if it is not then an HttpServer is used
3123
 
    for readonly urls.
3124
 
 
3125
 
    TODO RBC 20060127: make this an option to TestCaseWithTransport so it can
3126
 
                       be used without needed to redo it when a different
3127
 
                       subclass is in use ?
3128
 
    """
3129
 
 
3130
 
    def setUp(self):
3131
 
        from breezy.tests import http_server
3132
 
        super(ChrootedTestCase, self).setUp()
3133
 
        if not self.vfs_transport_factory == memory.MemoryServer:
3134
 
            self.transport_readonly_server = http_server.HttpServer
3135
 
 
3136
 
 
3137
 
def condition_id_re(pattern):
3138
 
    """Create a condition filter which performs a re check on a test's id.
3139
 
 
3140
 
    :param pattern: A regular expression string.
3141
 
    :return: A callable that returns True if the re matches.
3142
 
    """
3143
 
    filter_re = re.compile(pattern, 0)
3144
 
 
3145
 
    def condition(test):
3146
 
        test_id = test.id()
3147
 
        return filter_re.search(test_id)
3148
 
    return condition
3149
 
 
3150
 
 
3151
 
def condition_isinstance(klass_or_klass_list):
3152
 
    """Create a condition filter which returns isinstance(param, klass).
3153
 
 
3154
 
    :return: A callable which when called with one parameter obj return the
3155
 
        result of isinstance(obj, klass_or_klass_list).
3156
 
    """
3157
 
    def condition(obj):
3158
 
        return isinstance(obj, klass_or_klass_list)
3159
 
    return condition
3160
 
 
3161
 
 
3162
 
def condition_id_in_list(id_list):
3163
 
    """Create a condition filter which verify that test's id in a list.
3164
 
 
3165
 
    :param id_list: A TestIdList object.
3166
 
    :return: A callable that returns True if the test's id appears in the list.
3167
 
    """
3168
 
    def condition(test):
3169
 
        return id_list.includes(test.id())
3170
 
    return condition
3171
 
 
3172
 
 
3173
 
def condition_id_startswith(starts):
3174
 
    """Create a condition filter verifying that test's id starts with a string.
3175
 
 
3176
 
    :param starts: A list of string.
3177
 
    :return: A callable that returns True if the test's id starts with one of
3178
 
        the given strings.
3179
 
    """
3180
 
    def condition(test):
3181
 
        for start in starts:
3182
 
            if test.id().startswith(start):
3183
 
                return True
3184
 
        return False
3185
 
    return condition
3186
 
 
3187
 
 
3188
 
def exclude_tests_by_condition(suite, condition):
3189
 
    """Create a test suite which excludes some tests from suite.
3190
 
 
3191
 
    :param suite: The suite to get tests from.
3192
 
    :param condition: A callable whose result evaluates True when called with a
3193
 
        test case which should be excluded from the result.
3194
 
    :return: A suite which contains the tests found in suite that fail
3195
 
        condition.
3196
 
    """
3197
 
    result = []
3198
 
    for test in iter_suite_tests(suite):
3199
 
        if not condition(test):
3200
 
            result.append(test)
3201
 
    return TestUtil.TestSuite(result)
3202
 
 
3203
 
 
3204
 
def filter_suite_by_condition(suite, condition):
3205
 
    """Create a test suite by filtering another one.
3206
 
 
3207
 
    :param suite: The source suite.
3208
 
    :param condition: A callable whose result evaluates True when called with a
3209
 
        test case which should be included in the result.
3210
 
    :return: A suite which contains the tests found in suite that pass
3211
 
        condition.
3212
 
    """
3213
 
    result = []
3214
 
    for test in iter_suite_tests(suite):
3215
 
        if condition(test):
3216
 
            result.append(test)
3217
 
    return TestUtil.TestSuite(result)
 
446
                os.mkdir(name[:-1])
 
447
            else:
 
448
                f = file(name, 'wt')
 
449
                print >>f, "contents of", name
 
450
                f.close()
 
451
 
 
452
    def build_tree_contents(self, shape):
 
453
        bzrlib.selftest.build_tree_contents(shape)
 
454
 
 
455
    def failUnlessExists(self, path):
 
456
        """Fail unless path, which may be abs or relative, exists."""
 
457
        self.failUnless(osutils.lexists(path))
 
458
        
 
459
    def assertFileEqual(self, content, path):
 
460
        """Fail if path does not contain 'content'."""
 
461
        self.failUnless(osutils.lexists(path))
 
462
        self.assertEqualDiff(content, open(path, 'r').read())
 
463
        
 
464
 
 
465
class MetaTestLog(TestCase):
 
466
    def test_logging(self):
 
467
        """Test logs are captured when a test fails."""
 
468
        logging.info('an info message')
 
469
        warning('something looks dodgy...')
 
470
        logging.debug('hello, test is running')
 
471
        ## assert 0
3218
472
 
3219
473
 
3220
474
def filter_suite_by_re(suite, pattern):
3221
 
    """Create a test suite by filtering another one.
3222
 
 
3223
 
    :param suite:           the source suite
3224
 
    :param pattern:         pattern that names must match
3225
 
    :returns: the newly created suite
3226
 
    """
3227
 
    condition = condition_id_re(pattern)
3228
 
    result_suite = filter_suite_by_condition(suite, condition)
3229
 
    return result_suite
3230
 
 
3231
 
 
3232
 
def filter_suite_by_id_list(suite, test_id_list):
3233
 
    """Create a test suite by filtering another one.
3234
 
 
3235
 
    :param suite: The source suite.
3236
 
    :param test_id_list: A list of the test ids to keep as strings.
3237
 
    :returns: the newly created suite
3238
 
    """
3239
 
    condition = condition_id_in_list(test_id_list)
3240
 
    result_suite = filter_suite_by_condition(suite, condition)
3241
 
    return result_suite
3242
 
 
3243
 
 
3244
 
def filter_suite_by_id_startswith(suite, start):
3245
 
    """Create a test suite by filtering another one.
3246
 
 
3247
 
    :param suite: The source suite.
3248
 
    :param start: A list of string the test id must start with one of.
3249
 
    :returns: the newly created suite
3250
 
    """
3251
 
    condition = condition_id_startswith(start)
3252
 
    result_suite = filter_suite_by_condition(suite, condition)
3253
 
    return result_suite
3254
 
 
3255
 
 
3256
 
def exclude_tests_by_re(suite, pattern):
3257
 
    """Create a test suite which excludes some tests from suite.
3258
 
 
3259
 
    :param suite: The suite to get tests from.
3260
 
    :param pattern: A regular expression string. Test ids that match this
3261
 
        pattern will be excluded from the result.
3262
 
    :return: A TestSuite that contains all the tests from suite without the
3263
 
        tests that matched pattern. The order of tests is the same as it was in
3264
 
        suite.
3265
 
    """
3266
 
    return exclude_tests_by_condition(suite, condition_id_re(pattern))
3267
 
 
3268
 
 
3269
 
def preserve_input(something):
3270
 
    """A helper for performing test suite transformation chains.
3271
 
 
3272
 
    :param something: Anything you want to preserve.
3273
 
    :return: Something.
3274
 
    """
3275
 
    return something
3276
 
 
3277
 
 
3278
 
def randomize_suite(suite):
3279
 
    """Return a new TestSuite with suite's tests in random order.
3280
 
 
3281
 
    The tests in the input suite are flattened into a single suite in order to
3282
 
    accomplish this. Any nested TestSuites are removed to provide global
3283
 
    randomness.
3284
 
    """
3285
 
    tests = list(iter_suite_tests(suite))
3286
 
    random.shuffle(tests)
3287
 
    return TestUtil.TestSuite(tests)
3288
 
 
3289
 
 
3290
 
def split_suite_by_condition(suite, condition):
3291
 
    """Split a test suite into two by a condition.
3292
 
 
3293
 
    :param suite: The suite to split.
3294
 
    :param condition: The condition to match on. Tests that match this
3295
 
        condition are returned in the first test suite, ones that do not match
3296
 
        are in the second suite.
3297
 
    :return: A tuple of two test suites, where the first contains tests from
3298
 
        suite matching the condition, and the second contains the remainder
3299
 
        from suite. The order within each output suite is the same as it was in
3300
 
        suite.
3301
 
    """
3302
 
    matched = []
3303
 
    did_not_match = []
 
475
    result = TestUtil.TestSuite()
 
476
    filter_re = re.compile(pattern)
3304
477
    for test in iter_suite_tests(suite):
3305
 
        if condition(test):
3306
 
            matched.append(test)
3307
 
        else:
3308
 
            did_not_match.append(test)
3309
 
    return TestUtil.TestSuite(matched), TestUtil.TestSuite(did_not_match)
3310
 
 
3311
 
 
3312
 
def split_suite_by_re(suite, pattern):
3313
 
    """Split a test suite into two by a regular expression.
3314
 
 
3315
 
    :param suite: The suite to split.
3316
 
    :param pattern: A regular expression string. Test ids that match this
3317
 
        pattern will be in the first test suite returned, and the others in the
3318
 
        second test suite returned.
3319
 
    :return: A tuple of two test suites, where the first contains tests from
3320
 
        suite matching pattern, and the second contains the remainder from
3321
 
        suite. The order within each output suite is the same as it was in
3322
 
        suite.
3323
 
    """
3324
 
    return split_suite_by_condition(suite, condition_id_re(pattern))
 
478
        if filter_re.search(test.id()):
 
479
            result.addTest(test)
 
480
    return result
3325
481
 
3326
482
 
3327
483
def run_suite(suite, name='test', verbose=False, pattern=".*",
3328
 
              stop_on_failure=False,
3329
 
              transport=None, lsprof_timed=None, bench_history=None,
3330
 
              matching_tests_first=None,
3331
 
              list_only=False,
3332
 
              random_seed=None,
3333
 
              exclude_pattern=None,
3334
 
              strict=False,
3335
 
              runner_class=None,
3336
 
              suite_decorators=None,
3337
 
              stream=None,
3338
 
              result_decorators=None,
3339
 
              ):
3340
 
    """Run a test suite for brz selftest.
3341
 
 
3342
 
    :param runner_class: The class of runner to use. Must support the
3343
 
        constructor arguments passed by run_suite which are more than standard
3344
 
        python uses.
3345
 
    :return: A boolean indicating success.
3346
 
    """
3347
 
    TestCase._gather_lsprof_in_benchmarks = lsprof_timed
 
484
              stop_on_failure=False):
 
485
    TestCaseInTempDir._TEST_NAME = name
3348
486
    if verbose:
3349
487
        verbosity = 2
3350
488
    else:
3351
489
        verbosity = 1
3352
 
    if runner_class is None:
3353
 
        runner_class = TextTestRunner
3354
 
    if stream is None:
3355
 
        stream = sys.stdout
3356
 
    runner = runner_class(stream=stream,
3357
 
                          descriptions=0,
3358
 
                          verbosity=verbosity,
3359
 
                          bench_history=bench_history,
3360
 
                          strict=strict,
3361
 
                          result_decorators=result_decorators,
3362
 
                          )
3363
 
    runner.stop_on_failure = stop_on_failure
3364
 
    if isinstance(suite, unittest.TestSuite):
3365
 
        # Empty out _tests list of passed suite and populate new TestSuite
3366
 
        suite._tests[:], suite = [], TestSuite(suite)
3367
 
    # built in decorator factories:
3368
 
    decorators = [
3369
 
        random_order(random_seed, runner),
3370
 
        exclude_tests(exclude_pattern),
3371
 
        ]
3372
 
    if matching_tests_first:
3373
 
        decorators.append(tests_first(pattern))
3374
 
    else:
3375
 
        decorators.append(filter_tests(pattern))
3376
 
    if suite_decorators:
3377
 
        decorators.extend(suite_decorators)
3378
 
    # tell the result object how many tests will be running: (except if
3379
 
    # --parallel=fork is being used. Robert said he will provide a better
3380
 
    # progress design later -- vila 20090817)
3381
 
    if fork_decorator not in decorators:
3382
 
        decorators.append(CountingDecorator)
3383
 
    for decorator in decorators:
3384
 
        suite = decorator(suite)
3385
 
    if list_only:
3386
 
        # Done after test suite decoration to allow randomisation etc
3387
 
        # to take effect, though that is of marginal benefit.
3388
 
        if verbosity >= 2:
3389
 
            stream.write("Listing tests only ...\n")
3390
 
        if getattr(runner, 'list', None) is not None:
3391
 
            runner.list(suite)
3392
 
        else:
3393
 
            for t in iter_suite_tests(suite):
3394
 
                stream.write("%s\n" % (t.id()))
3395
 
        return True
 
490
    runner = TextTestRunner(stream=sys.stdout,
 
491
                            descriptions=0,
 
492
                            verbosity=verbosity)
 
493
    runner.stop_on_failure=stop_on_failure
 
494
    if pattern != '.*':
 
495
        suite = filter_suite_by_re(suite, pattern)
3396
496
    result = runner.run(suite)
3397
 
    if strict and getattr(result, 'wasStrictlySuccessful', False):
3398
 
        return result.wasStrictlySuccessful()
 
497
    # This is still a little bogus, 
 
498
    # but only a little. Folk not using our testrunner will
 
499
    # have to delete their temp directories themselves.
 
500
    if result.wasSuccessful():
 
501
        if TestCaseInTempDir.TEST_ROOT is not None:
 
502
            shutil.rmtree(TestCaseInTempDir.TEST_ROOT) 
3399
503
    else:
3400
 
        return result.wasSuccessful()
3401
 
 
3402
 
 
3403
 
# A registry where get() returns a suite decorator.
3404
 
parallel_registry = registry.Registry()
3405
 
 
3406
 
 
3407
 
def fork_decorator(suite):
3408
 
    if getattr(os, "fork", None) is None:
3409
 
        raise errors.BzrCommandError("platform does not support fork,"
3410
 
                                     " try --parallel=subprocess instead.")
3411
 
    concurrency = osutils.local_concurrency()
3412
 
    if concurrency == 1:
3413
 
        return suite
3414
 
    from testtools import ConcurrentTestSuite
3415
 
    return ConcurrentTestSuite(suite, fork_for_tests)
3416
 
 
3417
 
 
3418
 
parallel_registry.register('fork', fork_decorator)
3419
 
 
3420
 
 
3421
 
def subprocess_decorator(suite):
3422
 
    concurrency = osutils.local_concurrency()
3423
 
    if concurrency == 1:
3424
 
        return suite
3425
 
    from testtools import ConcurrentTestSuite
3426
 
    return ConcurrentTestSuite(suite, reinvoke_for_tests)
3427
 
 
3428
 
 
3429
 
parallel_registry.register('subprocess', subprocess_decorator)
3430
 
 
3431
 
 
3432
 
def exclude_tests(exclude_pattern):
3433
 
    """Return a test suite decorator that excludes tests."""
3434
 
    if exclude_pattern is None:
3435
 
        return identity_decorator
3436
 
 
3437
 
    def decorator(suite):
3438
 
        return ExcludeDecorator(suite, exclude_pattern)
3439
 
    return decorator
3440
 
 
3441
 
 
3442
 
def filter_tests(pattern):
3443
 
    if pattern == '.*':
3444
 
        return identity_decorator
3445
 
 
3446
 
    def decorator(suite):
3447
 
        return FilterTestsDecorator(suite, pattern)
3448
 
    return decorator
3449
 
 
3450
 
 
3451
 
def random_order(random_seed, runner):
3452
 
    """Return a test suite decorator factory for randomising tests order.
3453
 
 
3454
 
    :param random_seed: now, a string which casts to an integer, or an integer.
3455
 
    :param runner: A test runner with a stream attribute to report on.
3456
 
    """
3457
 
    if random_seed is None:
3458
 
        return identity_decorator
3459
 
 
3460
 
    def decorator(suite):
3461
 
        return RandomDecorator(suite, random_seed, runner.stream)
3462
 
    return decorator
3463
 
 
3464
 
 
3465
 
def tests_first(pattern):
3466
 
    if pattern == '.*':
3467
 
        return identity_decorator
3468
 
 
3469
 
    def decorator(suite):
3470
 
        return TestFirstDecorator(suite, pattern)
3471
 
    return decorator
3472
 
 
3473
 
 
3474
 
def identity_decorator(suite):
3475
 
    """Return suite."""
3476
 
    return suite
3477
 
 
3478
 
 
3479
 
class TestDecorator(TestUtil.TestSuite):
3480
 
    """A decorator for TestCase/TestSuite objects.
3481
 
 
3482
 
    Contains rather than flattening suite passed on construction
3483
 
    """
3484
 
 
3485
 
    def __init__(self, suite=None):
3486
 
        super(TestDecorator, self).__init__()
3487
 
        if suite is not None:
3488
 
            self.addTest(suite)
3489
 
 
3490
 
    # Don't need subclass run method with suite emptying
3491
 
    run = unittest.TestSuite.run
3492
 
 
3493
 
 
3494
 
class CountingDecorator(TestDecorator):
3495
 
    """A decorator which calls result.progress(self.countTestCases)."""
3496
 
 
3497
 
    def run(self, result):
3498
 
        progress_method = getattr(result, 'progress', None)
3499
 
        if callable(progress_method):
3500
 
            progress_method(self.countTestCases(), SUBUNIT_SEEK_SET)
3501
 
        return super(CountingDecorator, self).run(result)
3502
 
 
3503
 
 
3504
 
class ExcludeDecorator(TestDecorator):
3505
 
    """A decorator which excludes test matching an exclude pattern."""
3506
 
 
3507
 
    def __init__(self, suite, exclude_pattern):
3508
 
        super(ExcludeDecorator, self).__init__(
3509
 
            exclude_tests_by_re(suite, exclude_pattern))
3510
 
 
3511
 
 
3512
 
class FilterTestsDecorator(TestDecorator):
3513
 
    """A decorator which filters tests to those matching a pattern."""
3514
 
 
3515
 
    def __init__(self, suite, pattern):
3516
 
        super(FilterTestsDecorator, self).__init__(
3517
 
            filter_suite_by_re(suite, pattern))
3518
 
 
3519
 
 
3520
 
class RandomDecorator(TestDecorator):
3521
 
    """A decorator which randomises the order of its tests."""
3522
 
 
3523
 
    def __init__(self, suite, random_seed, stream):
3524
 
        random_seed = self.actual_seed(random_seed)
3525
 
        stream.write("Randomizing test order using seed %s\n\n" %
3526
 
                     (random_seed,))
3527
 
        # Initialise the random number generator.
3528
 
        random.seed(random_seed)
3529
 
        super(RandomDecorator, self).__init__(randomize_suite(suite))
3530
 
 
3531
 
    @staticmethod
3532
 
    def actual_seed(seed):
3533
 
        if seed == "now":
3534
 
            # We convert the seed to an integer to make it reuseable across
3535
 
            # invocations (because the user can reenter it).
3536
 
            return int(time.time())
3537
 
        else:
3538
 
            # Convert the seed to an integer if we can
3539
 
            try:
3540
 
                return int(seed)
3541
 
            except (TypeError, ValueError):
3542
 
                pass
3543
 
        return seed
3544
 
 
3545
 
 
3546
 
class TestFirstDecorator(TestDecorator):
3547
 
    """A decorator which moves named tests to the front."""
3548
 
 
3549
 
    def __init__(self, suite, pattern):
3550
 
        super(TestFirstDecorator, self).__init__()
3551
 
        self.addTests(split_suite_by_re(suite, pattern))
3552
 
 
3553
 
 
3554
 
def partition_tests(suite, count):
3555
 
    """Partition suite into count lists of tests."""
3556
 
    # This just assigns tests in a round-robin fashion.  On one hand this
3557
 
    # splits up blocks of related tests that might run faster if they shared
3558
 
    # resources, but on the other it avoids assigning blocks of slow tests to
3559
 
    # just one partition.  So the slowest partition shouldn't be much slower
3560
 
    # than the fastest.
3561
 
    partitions = [list() for i in range(count)]
3562
 
    tests = iter_suite_tests(suite)
3563
 
    for partition, test in zip(itertools.cycle(partitions), tests):
3564
 
        partition.append(test)
3565
 
    return partitions
3566
 
 
3567
 
 
3568
 
def workaround_zealous_crypto_random():
3569
 
    """Crypto.Random want to help us being secure, but we don't care here.
3570
 
 
3571
 
    This workaround some test failure related to the sftp server. Once paramiko
3572
 
    stop using the controversial API in Crypto.Random, we may get rid of it.
3573
 
    """
3574
 
    try:
3575
 
        from Crypto.Random import atfork
3576
 
        atfork()
3577
 
    except ImportError:
3578
 
        pass
3579
 
 
3580
 
 
3581
 
def fork_for_tests(suite):
3582
 
    """Take suite and start up one runner per CPU by forking()
3583
 
 
3584
 
    :return: An iterable of TestCase-like objects which can each have
3585
 
        run(result) called on them to feed tests to result.
3586
 
    """
3587
 
    concurrency = osutils.local_concurrency()
3588
 
    result = []
3589
 
    from subunit import ProtocolTestCase
3590
 
    from subunit.test_results import AutoTimingTestResultDecorator
3591
 
 
3592
 
    class TestInOtherProcess(ProtocolTestCase):
3593
 
        # Should be in subunit, I think. RBC.
3594
 
        def __init__(self, stream, pid):
3595
 
            ProtocolTestCase.__init__(self, stream)
3596
 
            self.pid = pid
3597
 
 
3598
 
        def run(self, result):
3599
 
            try:
3600
 
                ProtocolTestCase.run(self, result)
3601
 
            finally:
3602
 
                pid, status = os.waitpid(self.pid, 0)
3603
 
            # GZ 2011-10-18: If status is nonzero, should report to the result
3604
 
            #                that something went wrong.
3605
 
 
3606
 
    test_blocks = partition_tests(suite, concurrency)
3607
 
    # Clear the tests from the original suite so it doesn't keep them alive
3608
 
    suite._tests[:] = []
3609
 
    for process_tests in test_blocks:
3610
 
        process_suite = TestUtil.TestSuite(process_tests)
3611
 
        # Also clear each split list so new suite has only reference
3612
 
        process_tests[:] = []
3613
 
        c2pread, c2pwrite = os.pipe()
3614
 
        pid = os.fork()
3615
 
        if pid == 0:
3616
 
            try:
3617
 
                stream = os.fdopen(c2pwrite, 'wb', 0)
3618
 
                workaround_zealous_crypto_random()
3619
 
                try:
3620
 
                    import coverage
3621
 
                except ImportError:
3622
 
                    pass
3623
 
                else:
3624
 
                    coverage.process_startup()
3625
 
                os.close(c2pread)
3626
 
                # Leave stderr and stdout open so we can see test noise
3627
 
                # Close stdin so that the child goes away if it decides to
3628
 
                # read from stdin (otherwise its a roulette to see what
3629
 
                # child actually gets keystrokes for pdb etc).
3630
 
                sys.stdin.close()
3631
 
                subunit_result = AutoTimingTestResultDecorator(
3632
 
                    SubUnitBzrProtocolClientv1(stream))
3633
 
                process_suite.run(subunit_result)
3634
 
            except:
3635
 
                # Try and report traceback on stream, but exit with error even
3636
 
                # if stream couldn't be created or something else goes wrong.
3637
 
                # The traceback is formatted to a string and written in one go
3638
 
                # to avoid interleaving lines from multiple failing children.
3639
 
                tb = traceback.format_exc()
3640
 
                if isinstance(tb, text_type):
3641
 
                    tb = tb.encode('utf-8')
3642
 
                try:
3643
 
                    stream.write(tb)
3644
 
                finally:
3645
 
                    stream.flush()
3646
 
                    os._exit(1)
3647
 
            os._exit(0)
3648
 
        else:
3649
 
            os.close(c2pwrite)
3650
 
            stream = os.fdopen(c2pread, 'rb', 0)
3651
 
            test = TestInOtherProcess(stream, pid)
3652
 
            result.append(test)
3653
 
    return result
3654
 
 
3655
 
 
3656
 
def reinvoke_for_tests(suite):
3657
 
    """Take suite and start up one runner per CPU using subprocess().
3658
 
 
3659
 
    :return: An iterable of TestCase-like objects which can each have
3660
 
        run(result) called on them to feed tests to result.
3661
 
    """
3662
 
    concurrency = osutils.local_concurrency()
3663
 
    result = []
3664
 
    from subunit import ProtocolTestCase
3665
 
 
3666
 
    class TestInSubprocess(ProtocolTestCase):
3667
 
        def __init__(self, process, name):
3668
 
            ProtocolTestCase.__init__(self, process.stdout)
3669
 
            self.process = process
3670
 
            self.process.stdin.close()
3671
 
            self.name = name
3672
 
 
3673
 
        def run(self, result):
3674
 
            try:
3675
 
                ProtocolTestCase.run(self, result)
3676
 
            finally:
3677
 
                self.process.wait()
3678
 
                os.unlink(self.name)
3679
 
            # print "pid %d finished" % finished_process
3680
 
    test_blocks = partition_tests(suite, concurrency)
3681
 
    for process_tests in test_blocks:
3682
 
        # ugly; currently reimplement rather than reuses TestCase methods.
3683
 
        bzr_path = os.path.dirname(os.path.dirname(breezy.__file__)) + '/bzr'
3684
 
        if not os.path.isfile(bzr_path):
3685
 
            # We are probably installed. Assume sys.argv is the right file
3686
 
            bzr_path = sys.argv[0]
3687
 
        bzr_path = [bzr_path]
3688
 
        if sys.platform == "win32":
3689
 
            # if we're on windows, we can't execute the bzr script directly
3690
 
            bzr_path = [sys.executable] + bzr_path
3691
 
        fd, test_list_file_name = tempfile.mkstemp()
3692
 
        test_list_file = os.fdopen(fd, 'wb', 1)
3693
 
        for test in process_tests:
3694
 
            test_list_file.write(test.id() + '\n')
3695
 
        test_list_file.close()
3696
 
        try:
3697
 
            argv = bzr_path + ['selftest', '--load-list', test_list_file_name,
3698
 
                               '--subunit']
3699
 
            if '--no-plugins' in sys.argv:
3700
 
                argv.append('--no-plugins')
3701
 
            # stderr=subprocess.STDOUT would be ideal, but until we prevent
3702
 
            # noise on stderr it can interrupt the subunit protocol.
3703
 
            process = subprocess.Popen(argv, stdin=subprocess.PIPE,
3704
 
                                       stdout=subprocess.PIPE,
3705
 
                                       stderr=subprocess.PIPE,
3706
 
                                       bufsize=1)
3707
 
            test = TestInSubprocess(process, test_list_file_name)
3708
 
            result.append(test)
3709
 
        except:
3710
 
            os.unlink(test_list_file_name)
3711
 
            raise
3712
 
    return result
3713
 
 
3714
 
 
3715
 
class ProfileResult(testtools.ExtendedToOriginalDecorator):
3716
 
    """Generate profiling data for all activity between start and success.
3717
 
 
3718
 
    The profile data is appended to the test's _benchcalls attribute and can
3719
 
    be accessed by the forwarded-to TestResult.
3720
 
 
3721
 
    While it might be cleaner do accumulate this in stopTest, addSuccess is
3722
 
    where our existing output support for lsprof is, and this class aims to
3723
 
    fit in with that: while it could be moved it's not necessary to accomplish
3724
 
    test profiling, nor would it be dramatically cleaner.
3725
 
    """
3726
 
 
3727
 
    def startTest(self, test):
3728
 
        self.profiler = breezy.lsprof.BzrProfiler()
3729
 
        # Prevent deadlocks in tests that use lsprof: those tests will
3730
 
        # unavoidably fail.
3731
 
        breezy.lsprof.BzrProfiler.profiler_block = 0
3732
 
        self.profiler.start()
3733
 
        testtools.ExtendedToOriginalDecorator.startTest(self, test)
3734
 
 
3735
 
    def addSuccess(self, test):
3736
 
        stats = self.profiler.stop()
3737
 
        try:
3738
 
            calls = test._benchcalls
3739
 
        except AttributeError:
3740
 
            test._benchcalls = []
3741
 
            calls = test._benchcalls
3742
 
        calls.append(((test.id(), "", ""), stats))
3743
 
        testtools.ExtendedToOriginalDecorator.addSuccess(self, test)
3744
 
 
3745
 
    def stopTest(self, test):
3746
 
        testtools.ExtendedToOriginalDecorator.stopTest(self, test)
3747
 
        self.profiler = None
3748
 
 
3749
 
 
3750
 
# Controlled by "brz selftest -E=..." option
3751
 
# Currently supported:
3752
 
#   -Eallow_debug           Will no longer clear debug.debug_flags() so it
3753
 
#                           preserves any flags supplied at the command line.
3754
 
#   -Edisable_lock_checks   Turns errors in mismatched locks into simple prints
3755
 
#                           rather than failing tests. And no longer raise
3756
 
#                           LockContention when fctnl locks are not being used
3757
 
#                           with proper exclusion rules.
3758
 
#   -Ethreads               Will display thread ident at creation/join time to
3759
 
#                           help track thread leaks
3760
 
#   -Euncollected_cases     Display the identity of any test cases that weren't
3761
 
#                           deallocated after being completed.
3762
 
#   -Econfig_stats          Will collect statistics using addDetail
3763
 
selftest_debug_flags = set()
3764
 
 
3765
 
 
3766
 
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
3767
 
             transport=None,
3768
 
             test_suite_factory=None,
3769
 
             lsprof_timed=None,
3770
 
             bench_history=None,
3771
 
             matching_tests_first=None,
3772
 
             list_only=False,
3773
 
             random_seed=None,
3774
 
             exclude_pattern=None,
3775
 
             strict=False,
3776
 
             load_list=None,
3777
 
             debug_flags=None,
3778
 
             starting_with=None,
3779
 
             runner_class=None,
3780
 
             suite_decorators=None,
3781
 
             stream=None,
3782
 
             lsprof_tests=False,
3783
 
             ):
 
504
        print "Failed tests working directories are in '%s'\n" % TestCaseInTempDir.TEST_ROOT
 
505
    return result.wasSuccessful()
 
506
 
 
507
 
 
508
def selftest(verbose=False, pattern=".*", stop_on_failure=True):
3784
509
    """Run the whole test suite under the enhanced runner"""
3785
 
    # XXX: Very ugly way to do this...
3786
 
    # Disable warning about old formats because we don't want it to disturb
3787
 
    # any blackbox tests.
3788
 
    from breezy import repository
3789
 
    repository._deprecation_warning_done = True
3790
 
 
3791
 
    global default_transport
3792
 
    if transport is None:
3793
 
        transport = default_transport
3794
 
    old_transport = default_transport
3795
 
    default_transport = transport
3796
 
    global selftest_debug_flags
3797
 
    old_debug_flags = selftest_debug_flags
3798
 
    if debug_flags is not None:
3799
 
        selftest_debug_flags = set(debug_flags)
3800
 
    try:
3801
 
        if load_list is None:
3802
 
            keep_only = None
3803
 
        else:
3804
 
            keep_only = load_test_id_list(load_list)
3805
 
        if starting_with:
3806
 
            starting_with = [test_prefix_alias_registry.resolve_alias(start)
3807
 
                             for start in starting_with]
3808
 
            # Always consider 'unittest' an interesting name so that failed
3809
 
            # suites wrapped as test cases appear in the output.
3810
 
            starting_with.append('unittest')
3811
 
        if test_suite_factory is None:
3812
 
            # Reduce loading time by loading modules based on the starting_with
3813
 
            # patterns.
3814
 
            suite = test_suite(keep_only, starting_with)
3815
 
        else:
3816
 
            suite = test_suite_factory()
3817
 
        if starting_with:
3818
 
            # But always filter as requested.
3819
 
            suite = filter_suite_by_id_startswith(suite, starting_with)
3820
 
        result_decorators = []
3821
 
        if lsprof_tests:
3822
 
            result_decorators.append(ProfileResult)
3823
 
        return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
3824
 
                         stop_on_failure=stop_on_failure,
3825
 
                         transport=transport,
3826
 
                         lsprof_timed=lsprof_timed,
3827
 
                         bench_history=bench_history,
3828
 
                         matching_tests_first=matching_tests_first,
3829
 
                         list_only=list_only,
3830
 
                         random_seed=random_seed,
3831
 
                         exclude_pattern=exclude_pattern,
3832
 
                         strict=strict,
3833
 
                         runner_class=runner_class,
3834
 
                         suite_decorators=suite_decorators,
3835
 
                         stream=stream,
3836
 
                         result_decorators=result_decorators,
3837
 
                         )
3838
 
    finally:
3839
 
        default_transport = old_transport
3840
 
        selftest_debug_flags = old_debug_flags
3841
 
 
3842
 
 
3843
 
def load_test_id_list(file_name):
3844
 
    """Load a test id list from a text file.
3845
 
 
3846
 
    The format is one test id by line.  No special care is taken to impose
3847
 
    strict rules, these test ids are used to filter the test suite so a test id
3848
 
    that do not match an existing test will do no harm. This allows user to add
3849
 
    comments, leave blank lines, etc.
3850
 
    """
3851
 
    test_list = []
3852
 
    try:
3853
 
        ftest = open(file_name, 'rt')
3854
 
    except IOError as e:
3855
 
        if e.errno != errno.ENOENT:
3856
 
            raise
3857
 
        else:
3858
 
            raise errors.NoSuchFile(file_name)
3859
 
 
3860
 
    for test_name in ftest.readlines():
3861
 
        test_list.append(test_name.strip())
3862
 
    ftest.close()
3863
 
    return test_list
3864
 
 
3865
 
 
3866
 
def suite_matches_id_list(test_suite, id_list):
3867
 
    """Warns about tests not appearing or appearing more than once.
3868
 
 
3869
 
    :param test_suite: A TestSuite object.
3870
 
    :param test_id_list: The list of test ids that should be found in
3871
 
         test_suite.
3872
 
 
3873
 
    :return: (absents, duplicates) absents is a list containing the test found
3874
 
        in id_list but not in test_suite, duplicates is a list containing the
3875
 
        tests found multiple times in test_suite.
3876
 
 
3877
 
    When using a prefined test id list, it may occurs that some tests do not
3878
 
    exist anymore or that some tests use the same id. This function warns the
3879
 
    tester about potential problems in his workflow (test lists are volatile)
3880
 
    or in the test suite itself (using the same id for several tests does not
3881
 
    help to localize defects).
3882
 
    """
3883
 
    # Build a dict counting id occurrences
3884
 
    tests = dict()
3885
 
    for test in iter_suite_tests(test_suite):
3886
 
        id = test.id()
3887
 
        tests[id] = tests.get(id, 0) + 1
3888
 
 
3889
 
    not_found = []
3890
 
    duplicates = []
3891
 
    for id in id_list:
3892
 
        occurs = tests.get(id, 0)
3893
 
        if not occurs:
3894
 
            not_found.append(id)
3895
 
        elif occurs > 1:
3896
 
            duplicates.append(id)
3897
 
 
3898
 
    return not_found, duplicates
3899
 
 
3900
 
 
3901
 
class TestIdList(object):
3902
 
    """Test id list to filter a test suite.
3903
 
 
3904
 
    Relying on the assumption that test ids are built as:
3905
 
    <module>[.<class>.<method>][(<param>+)], <module> being in python dotted
3906
 
    notation, this class offers methods to :
3907
 
    - avoid building a test suite for modules not refered to in the test list,
3908
 
    - keep only the tests listed from the module test suite.
3909
 
    """
3910
 
 
3911
 
    def __init__(self, test_id_list):
3912
 
        # When a test suite needs to be filtered against us we compare test ids
3913
 
        # for equality, so a simple dict offers a quick and simple solution.
3914
 
        self.tests = dict().fromkeys(test_id_list, True)
3915
 
 
3916
 
        # While unittest.TestCase have ids like:
3917
 
        # <module>.<class>.<method>[(<param+)],
3918
 
        # doctest.DocTestCase can have ids like:
3919
 
        # <module>
3920
 
        # <module>.<class>
3921
 
        # <module>.<function>
3922
 
        # <module>.<class>.<method>
3923
 
 
3924
 
        # Since we can't predict a test class from its name only, we settle on
3925
 
        # a simple constraint: a test id always begins with its module name.
3926
 
 
3927
 
        modules = {}
3928
 
        for test_id in test_id_list:
3929
 
            parts = test_id.split('.')
3930
 
            mod_name = parts.pop(0)
3931
 
            modules[mod_name] = True
3932
 
            for part in parts:
3933
 
                mod_name += '.' + part
3934
 
                modules[mod_name] = True
3935
 
        self.modules = modules
3936
 
 
3937
 
    def refers_to(self, module_name):
3938
 
        """Is there tests for the module or one of its sub modules."""
3939
 
        return module_name in self.modules
3940
 
 
3941
 
    def includes(self, test_id):
3942
 
        return test_id in self.tests
3943
 
 
3944
 
 
3945
 
class TestPrefixAliasRegistry(registry.Registry):
3946
 
    """A registry for test prefix aliases.
3947
 
 
3948
 
    This helps implement shorcuts for the --starting-with selftest
3949
 
    option. Overriding existing prefixes is not allowed but not fatal (a
3950
 
    warning will be emitted).
3951
 
    """
3952
 
 
3953
 
    def register(self, key, obj, help=None, info=None,
3954
 
                 override_existing=False):
3955
 
        """See Registry.register.
3956
 
 
3957
 
        Trying to override an existing alias causes a warning to be emitted,
3958
 
        not a fatal execption.
3959
 
        """
3960
 
        try:
3961
 
            super(TestPrefixAliasRegistry, self).register(
3962
 
                key, obj, help=help, info=info, override_existing=False)
3963
 
        except KeyError:
3964
 
            actual = self.get(key)
3965
 
            trace.note(
3966
 
                'Test prefix alias %s is already used for %s, ignoring %s'
3967
 
                % (key, actual, obj))
3968
 
 
3969
 
    def resolve_alias(self, id_start):
3970
 
        """Replace the alias by the prefix in the given string.
3971
 
 
3972
 
        Using an unknown prefix is an error to help catching typos.
3973
 
        """
3974
 
        parts = id_start.split('.')
3975
 
        try:
3976
 
            parts[0] = self.get(parts[0])
3977
 
        except KeyError:
3978
 
            raise errors.BzrCommandError(
3979
 
                '%s is not a known test prefix alias' % parts[0])
3980
 
        return '.'.join(parts)
3981
 
 
3982
 
 
3983
 
test_prefix_alias_registry = TestPrefixAliasRegistry()
3984
 
"""Registry of test prefix aliases."""
3985
 
 
3986
 
 
3987
 
# This alias allows to detect typos ('bzrlin.') by making all valid test ids
3988
 
# appear prefixed ('breezy.' is "replaced" by 'breezy.').
3989
 
test_prefix_alias_registry.register('breezy', 'breezy')
3990
 
 
3991
 
# Obvious highest levels prefixes, feel free to add your own via a plugin
3992
 
test_prefix_alias_registry.register('bd', 'breezy.doc')
3993
 
test_prefix_alias_registry.register('bu', 'breezy.utils')
3994
 
test_prefix_alias_registry.register('bt', 'breezy.tests')
3995
 
test_prefix_alias_registry.register('bgt', 'breezy.git.tests')
3996
 
test_prefix_alias_registry.register('bb', 'breezy.tests.blackbox')
3997
 
test_prefix_alias_registry.register('bp', 'breezy.plugins')
3998
 
 
3999
 
 
4000
 
def _test_suite_testmod_names():
4001
 
    """Return the standard list of test module names to test."""
4002
 
    return [
4003
 
        'breezy.git.tests.test_blackbox',
4004
 
        'breezy.git.tests.test_builder',
4005
 
        'breezy.git.tests.test_branch',
4006
 
        'breezy.git.tests.test_cache',
4007
 
        'breezy.git.tests.test_dir',
4008
 
        'breezy.git.tests.test_fetch',
4009
 
        'breezy.git.tests.test_git_remote_helper',
4010
 
        'breezy.git.tests.test_mapping',
4011
 
        'breezy.git.tests.test_memorytree',
4012
 
        'breezy.git.tests.test_object_store',
4013
 
        'breezy.git.tests.test_pristine_tar',
4014
 
        'breezy.git.tests.test_push',
4015
 
        'breezy.git.tests.test_remote',
4016
 
        'breezy.git.tests.test_repository',
4017
 
        'breezy.git.tests.test_refs',
4018
 
        'breezy.git.tests.test_revspec',
4019
 
        'breezy.git.tests.test_roundtrip',
4020
 
        'breezy.git.tests.test_server',
4021
 
        'breezy.git.tests.test_transportgit',
4022
 
        'breezy.git.tests.test_unpeel_map',
4023
 
        'breezy.git.tests.test_urls',
4024
 
        'breezy.git.tests.test_workingtree',
4025
 
        'breezy.tests.blackbox',
4026
 
        'breezy.tests.commands',
4027
 
        'breezy.tests.per_branch',
4028
 
        'breezy.tests.per_bzrdir',
4029
 
        'breezy.tests.per_controldir',
4030
 
        'breezy.tests.per_controldir_colo',
4031
 
        'breezy.tests.per_foreign_vcs',
4032
 
        'breezy.tests.per_interrepository',
4033
 
        'breezy.tests.per_intertree',
4034
 
        'breezy.tests.per_inventory',
4035
 
        'breezy.tests.per_interbranch',
4036
 
        'breezy.tests.per_lock',
4037
 
        'breezy.tests.per_merger',
4038
 
        'breezy.tests.per_transport',
4039
 
        'breezy.tests.per_tree',
4040
 
        'breezy.tests.per_pack_repository',
4041
 
        'breezy.tests.per_repository',
4042
 
        'breezy.tests.per_repository_chk',
4043
 
        'breezy.tests.per_repository_reference',
4044
 
        'breezy.tests.per_repository_vf',
4045
 
        'breezy.tests.per_uifactory',
4046
 
        'breezy.tests.per_versionedfile',
4047
 
        'breezy.tests.per_workingtree',
4048
 
        'breezy.tests.test__annotator',
4049
 
        'breezy.tests.test__bencode',
4050
 
        'breezy.tests.test__btree_serializer',
4051
 
        'breezy.tests.test__chk_map',
4052
 
        'breezy.tests.test__dirstate_helpers',
4053
 
        'breezy.tests.test__groupcompress',
4054
 
        'breezy.tests.test__known_graph',
4055
 
        'breezy.tests.test__rio',
4056
 
        'breezy.tests.test__simple_set',
4057
 
        'breezy.tests.test__static_tuple',
4058
 
        'breezy.tests.test__walkdirs_win32',
4059
 
        'breezy.tests.test_ancestry',
4060
 
        'breezy.tests.test_annotate',
4061
 
        'breezy.tests.test_atomicfile',
4062
 
        'breezy.tests.test_bad_files',
4063
 
        'breezy.tests.test_bisect',
4064
 
        'breezy.tests.test_bisect_multi',
4065
 
        'breezy.tests.test_branch',
4066
 
        'breezy.tests.test_branchbuilder',
4067
 
        'breezy.tests.test_btree_index',
4068
 
        'breezy.tests.test_bugtracker',
4069
 
        'breezy.tests.test_bundle',
4070
 
        'breezy.tests.test_bzrdir',
4071
 
        'breezy.tests.test__chunks_to_lines',
4072
 
        'breezy.tests.test_cache_utf8',
4073
 
        'breezy.tests.test_chk_map',
4074
 
        'breezy.tests.test_chk_serializer',
4075
 
        'breezy.tests.test_chunk_writer',
4076
 
        'breezy.tests.test_clean_tree',
4077
 
        'breezy.tests.test_cleanup',
4078
 
        'breezy.tests.test_cmdline',
4079
 
        'breezy.tests.test_commands',
4080
 
        'breezy.tests.test_commit',
4081
 
        'breezy.tests.test_commit_merge',
4082
 
        'breezy.tests.test_config',
4083
 
        'breezy.tests.test_bedding',
4084
 
        'breezy.tests.test_conflicts',
4085
 
        'breezy.tests.test_controldir',
4086
 
        'breezy.tests.test_counted_lock',
4087
 
        'breezy.tests.test_crash',
4088
 
        'breezy.tests.test_decorators',
4089
 
        'breezy.tests.test_delta',
4090
 
        'breezy.tests.test_debug',
4091
 
        'breezy.tests.test_diff',
4092
 
        'breezy.tests.test_directory_service',
4093
 
        'breezy.tests.test_dirstate',
4094
 
        'breezy.tests.test_email_message',
4095
 
        'breezy.tests.test_eol_filters',
4096
 
        'breezy.tests.test_errors',
4097
 
        'breezy.tests.test_estimate_compressed_size',
4098
 
        'breezy.tests.test_export',
4099
 
        'breezy.tests.test_export_pot',
4100
 
        'breezy.tests.test_extract',
4101
 
        'breezy.tests.test_features',
4102
 
        'breezy.tests.test_fetch',
4103
 
        'breezy.tests.test_fetch_ghosts',
4104
 
        'breezy.tests.test_fixtures',
4105
 
        'breezy.tests.test_fifo_cache',
4106
 
        'breezy.tests.test_filters',
4107
 
        'breezy.tests.test_filter_tree',
4108
 
        'breezy.tests.test_foreign',
4109
 
        'breezy.tests.test_generate_docs',
4110
 
        'breezy.tests.test_generate_ids',
4111
 
        'breezy.tests.test_globbing',
4112
 
        'breezy.tests.test_gpg',
4113
 
        'breezy.tests.test_graph',
4114
 
        'breezy.tests.test_grep',
4115
 
        'breezy.tests.test_groupcompress',
4116
 
        'breezy.tests.test_hashcache',
4117
 
        'breezy.tests.test_help',
4118
 
        'breezy.tests.test_hooks',
4119
 
        'breezy.tests.test_http',
4120
 
        'breezy.tests.test_http_response',
4121
 
        'breezy.tests.test_https_ca_bundle',
4122
 
        'breezy.tests.test_https_urllib',
4123
 
        'breezy.tests.test_i18n',
4124
 
        'breezy.tests.test_identitymap',
4125
 
        'breezy.tests.test_ignores',
4126
 
        'breezy.tests.test_index',
4127
 
        'breezy.tests.test_import_tariff',
4128
 
        'breezy.tests.test_info',
4129
 
        'breezy.tests.test_inv',
4130
 
        'breezy.tests.test_inventory_delta',
4131
 
        'breezy.tests.test_knit',
4132
 
        'breezy.tests.test_lazy_import',
4133
 
        'breezy.tests.test_lazy_regex',
4134
 
        'breezy.tests.test_library_state',
4135
 
        'breezy.tests.test_location',
4136
 
        'breezy.tests.test_lock',
4137
 
        'breezy.tests.test_lockable_files',
4138
 
        'breezy.tests.test_lockdir',
4139
 
        'breezy.tests.test_log',
4140
 
        'breezy.tests.test_lru_cache',
4141
 
        'breezy.tests.test_lsprof',
4142
 
        'breezy.tests.test_mail_client',
4143
 
        'breezy.tests.test_matchers',
4144
 
        'breezy.tests.test_memorytree',
4145
 
        'breezy.tests.test_merge',
4146
 
        'breezy.tests.test_merge3',
4147
 
        'breezy.tests.test_mergeable',
4148
 
        'breezy.tests.test_merge_core',
4149
 
        'breezy.tests.test_merge_directive',
4150
 
        'breezy.tests.test_mergetools',
4151
 
        'breezy.tests.test_missing',
4152
 
        'breezy.tests.test_msgeditor',
4153
 
        'breezy.tests.test_multiparent',
4154
 
        'breezy.tests.test_multiwalker',
4155
 
        'breezy.tests.test_mutabletree',
4156
 
        'breezy.tests.test_nonascii',
4157
 
        'breezy.tests.test_options',
4158
 
        'breezy.tests.test_osutils',
4159
 
        'breezy.tests.test_osutils_encodings',
4160
 
        'breezy.tests.test_pack',
4161
 
        'breezy.tests.test_patch',
4162
 
        'breezy.tests.test_patches',
4163
 
        'breezy.tests.test_permissions',
4164
 
        'breezy.tests.test_plugins',
4165
 
        'breezy.tests.test_progress',
4166
 
        'breezy.tests.test_propose',
4167
 
        'breezy.tests.test_pyutils',
4168
 
        'breezy.tests.test_read_bundle',
4169
 
        'breezy.tests.test_reconcile',
4170
 
        'breezy.tests.test_reconfigure',
4171
 
        'breezy.tests.test_registry',
4172
 
        'breezy.tests.test_remote',
4173
 
        'breezy.tests.test_rename_map',
4174
 
        'breezy.tests.test_repository',
4175
 
        'breezy.tests.test_revert',
4176
 
        'breezy.tests.test_revision',
4177
 
        'breezy.tests.test_revisionspec',
4178
 
        'breezy.tests.test_revisiontree',
4179
 
        'breezy.tests.test_rio',
4180
 
        'breezy.tests.test_rules',
4181
 
        'breezy.tests.test_url_policy_open',
4182
 
        'breezy.tests.test_sampler',
4183
 
        'breezy.tests.test_scenarios',
4184
 
        'breezy.tests.test_script',
4185
 
        'breezy.tests.test_selftest',
4186
 
        'breezy.tests.test_serializer',
4187
 
        'breezy.tests.test_setup',
4188
 
        'breezy.tests.test_sftp_transport',
4189
 
        'breezy.tests.test_shelf',
4190
 
        'breezy.tests.test_shelf_ui',
4191
 
        'breezy.tests.test_smart',
4192
 
        'breezy.tests.test_smart_add',
4193
 
        'breezy.tests.test_smart_request',
4194
 
        'breezy.tests.test_smart_signals',
4195
 
        'breezy.tests.test_smart_transport',
4196
 
        'breezy.tests.test_smtp_connection',
4197
 
        'breezy.tests.test_source',
4198
 
        'breezy.tests.test_ssh_transport',
4199
 
        'breezy.tests.test_status',
4200
 
        'breezy.tests.test_strace',
4201
 
        'breezy.tests.test_subsume',
4202
 
        'breezy.tests.test_switch',
4203
 
        'breezy.tests.test_symbol_versioning',
4204
 
        'breezy.tests.test_tag',
4205
 
        'breezy.tests.test_test_server',
4206
 
        'breezy.tests.test_testament',
4207
 
        'breezy.tests.test_textfile',
4208
 
        'breezy.tests.test_textmerge',
4209
 
        'breezy.tests.test_cethread',
4210
 
        'breezy.tests.test_timestamp',
4211
 
        'breezy.tests.test_trace',
4212
 
        'breezy.tests.test_transactions',
4213
 
        'breezy.tests.test_transform',
4214
 
        'breezy.tests.test_transport',
4215
 
        'breezy.tests.test_transport_log',
4216
 
        'breezy.tests.test_tree',
4217
 
        'breezy.tests.test_treebuilder',
4218
 
        'breezy.tests.test_treeshape',
4219
 
        'breezy.tests.test_tsort',
4220
 
        'breezy.tests.test_tuned_gzip',
4221
 
        'breezy.tests.test_ui',
4222
 
        'breezy.tests.test_uncommit',
4223
 
        'breezy.tests.test_upgrade',
4224
 
        'breezy.tests.test_upgrade_stacked',
4225
 
        'breezy.tests.test_upstream_import',
4226
 
        'breezy.tests.test_urlutils',
4227
 
        'breezy.tests.test_utextwrap',
4228
 
        'breezy.tests.test_version',
4229
 
        'breezy.tests.test_version_info',
4230
 
        'breezy.tests.test_versionedfile',
4231
 
        'breezy.tests.test_vf_search',
4232
 
        'breezy.tests.test_views',
4233
 
        'breezy.tests.test_weave',
4234
 
        'breezy.tests.test_whitebox',
4235
 
        'breezy.tests.test_win32utils',
4236
 
        'breezy.tests.test_workingtree',
4237
 
        'breezy.tests.test_workingtree_4',
4238
 
        'breezy.tests.test_wsgi',
4239
 
        'breezy.tests.test_xml',
4240
 
        ]
4241
 
 
4242
 
 
4243
 
def _test_suite_modules_to_doctest():
4244
 
    """Return the list of modules to doctest."""
4245
 
    if __doc__ is None:
4246
 
        # GZ 2009-03-31: No docstrings with -OO so there's nothing to doctest
4247
 
        return []
4248
 
    return [
4249
 
        'breezy',
4250
 
        'breezy.branchbuilder',
4251
 
        'breezy.bzr.inventory',
4252
 
        'breezy.decorators',
4253
 
        'breezy.iterablefile',
4254
 
        'breezy.lockdir',
4255
 
        'breezy.merge3',
4256
 
        'breezy.option',
4257
 
        'breezy.pyutils',
4258
 
        'breezy.symbol_versioning',
4259
 
        'breezy.tests',
4260
 
        'breezy.tests.fixtures',
4261
 
        'breezy.timestamp',
4262
 
        'breezy.transport.http',
4263
 
        'breezy.version_info_formats.format_custom',
4264
 
        ]
4265
 
 
4266
 
 
4267
 
def test_suite(keep_only=None, starting_with=None):
4268
 
    """Build and return TestSuite for the whole of breezy.
4269
 
 
4270
 
    :param keep_only: A list of test ids limiting the suite returned.
4271
 
 
4272
 
    :param starting_with: An id limiting the suite returned to the tests
4273
 
         starting with it.
4274
 
 
4275
 
    This function can be replaced if you need to change the default test
4276
 
    suite on a global basis, but it is not encouraged.
4277
 
    """
4278
 
 
4279
 
    loader = TestUtil.TestLoader()
4280
 
 
4281
 
    if keep_only is not None:
4282
 
        id_filter = TestIdList(keep_only)
4283
 
    if starting_with:
4284
 
        # We take precedence over keep_only because *at loading time* using
4285
 
        # both options means we will load less tests for the same final result.
4286
 
        def interesting_module(name):
4287
 
            for start in starting_with:
4288
 
                # Either the module name starts with the specified string
4289
 
                # or it may contain tests starting with the specified string
4290
 
                if name.startswith(start) or start.startswith(name):
4291
 
                    return True
4292
 
            return False
4293
 
        loader = TestUtil.FilteredByModuleTestLoader(interesting_module)
4294
 
 
4295
 
    elif keep_only is not None:
4296
 
        loader = TestUtil.FilteredByModuleTestLoader(id_filter.refers_to)
4297
 
 
4298
 
        def interesting_module(name):
4299
 
            return id_filter.refers_to(name)
4300
 
 
4301
 
    else:
4302
 
        loader = TestUtil.TestLoader()
4303
 
 
4304
 
        def interesting_module(name):
4305
 
            # No filtering, all modules are interesting
4306
 
            return True
4307
 
 
4308
 
    suite = loader.suiteClass()
4309
 
 
4310
 
    # modules building their suite with loadTestsFromModuleNames
4311
 
    suite.addTest(loader.loadTestsFromModuleNames(_test_suite_testmod_names()))
4312
 
 
4313
 
    if not PY3:
4314
 
        suite.addTest(loader.loadTestsFromModuleNames(['breezy.doc']))
4315
 
 
4316
 
        # It's pretty much impossible to write readable doctests that work on
4317
 
        # both Python 2 and Python 3 because of their overreliance on
4318
 
        # consistent repr() return values.
4319
 
        # For now, just run doctests on Python 2 so we now they haven't broken.
4320
 
        for mod in _test_suite_modules_to_doctest():
4321
 
            if not interesting_module(mod):
4322
 
                # No tests to keep here, move along
4323
 
                continue
4324
 
            try:
4325
 
                # note that this really does mean "report only" -- doctest
4326
 
                # still runs the rest of the examples
4327
 
                doc_suite = IsolatedDocTestSuite(
4328
 
                    mod, optionflags=doctest.REPORT_ONLY_FIRST_FAILURE)
4329
 
            except ValueError as e:
4330
 
                print('**failed to get doctest for: %s\n%s' % (mod, e))
4331
 
                raise
4332
 
            if len(doc_suite._tests) == 0:
4333
 
                raise errors.BzrError("no doctests found in %s" % (mod,))
4334
 
            suite.addTest(doc_suite)
4335
 
 
4336
 
    default_encoding = sys.getdefaultencoding()
4337
 
    for name, plugin in _mod_plugin.plugins().items():
4338
 
        if not interesting_module(plugin.module.__name__):
4339
 
            continue
4340
 
        plugin_suite = plugin.test_suite()
4341
 
        # We used to catch ImportError here and turn it into just a warning,
4342
 
        # but really if you don't have --no-plugins this should be a failure.
4343
 
        # mbp 20080213 - see http://bugs.launchpad.net/bugs/189771
4344
 
        if plugin_suite is None:
4345
 
            plugin_suite = plugin.load_plugin_tests(loader)
4346
 
        if plugin_suite is not None:
4347
 
            suite.addTest(plugin_suite)
4348
 
        if default_encoding != sys.getdefaultencoding():
4349
 
            trace.warning(
4350
 
                'Plugin "%s" tried to reset default encoding to: %s', name,
4351
 
                sys.getdefaultencoding())
4352
 
            reload(sys)
4353
 
            sys.setdefaultencoding(default_encoding)
4354
 
 
4355
 
    if keep_only is not None:
4356
 
        # Now that the referred modules have loaded their tests, keep only the
4357
 
        # requested ones.
4358
 
        suite = filter_suite_by_id_list(suite, id_filter)
4359
 
        # Do some sanity checks on the id_list filtering
4360
 
        not_found, duplicates = suite_matches_id_list(suite, keep_only)
4361
 
        if starting_with:
4362
 
            # The tester has used both keep_only and starting_with, so he is
4363
 
            # already aware that some tests are excluded from the list, there
4364
 
            # is no need to tell him which.
4365
 
            pass
4366
 
        else:
4367
 
            # Some tests mentioned in the list are not in the test suite. The
4368
 
            # list may be out of date, report to the tester.
4369
 
            for id in not_found:
4370
 
                trace.warning('"%s" not found in the test suite', id)
4371
 
        for id in duplicates:
4372
 
            trace.warning('"%s" is used as an id by several tests', id)
4373
 
 
 
510
    return run_suite(test_suite(), 'testbzr', verbose=verbose, pattern=pattern,
 
511
                     stop_on_failure=stop_on_failure)
 
512
 
 
513
 
 
514
def test_suite():
 
515
    """Build and return TestSuite for the whole program."""
 
516
    import bzrlib.store, bzrlib.inventory, bzrlib.branch
 
517
    import bzrlib.osutils, bzrlib.merge3, bzrlib.plugin
 
518
    from doctest import DocTestSuite
 
519
 
 
520
    global MODULES_TO_TEST, MODULES_TO_DOCTEST
 
521
 
 
522
    testmod_names = \
 
523
                  ['bzrlib.selftest.MetaTestLog',
 
524
                   'bzrlib.selftest.testgpg',
 
525
                   'bzrlib.selftest.testidentitymap',
 
526
                   'bzrlib.selftest.testinv',
 
527
                   'bzrlib.selftest.test_ancestry',
 
528
                   'bzrlib.selftest.test_commit',
 
529
                   'bzrlib.selftest.test_commit_merge',
 
530
                   'bzrlib.selftest.testconfig',
 
531
                   'bzrlib.selftest.versioning',
 
532
                   'bzrlib.selftest.testmerge3',
 
533
                   'bzrlib.selftest.testmerge',
 
534
                   'bzrlib.selftest.testhashcache',
 
535
                   'bzrlib.selftest.teststatus',
 
536
                   'bzrlib.selftest.testlog',
 
537
                   'bzrlib.selftest.testrevisionnamespaces',
 
538
                   'bzrlib.selftest.testbranch',
 
539
                   'bzrlib.selftest.testrevision',
 
540
                   'bzrlib.selftest.test_revision_info',
 
541
                   'bzrlib.selftest.test_merge_core',
 
542
                   'bzrlib.selftest.test_smart_add',
 
543
                   'bzrlib.selftest.test_bad_files',
 
544
                   'bzrlib.selftest.testdiff',
 
545
                   'bzrlib.selftest.test_parent',
 
546
                   'bzrlib.selftest.test_xml',
 
547
                   'bzrlib.selftest.test_weave',
 
548
                   'bzrlib.selftest.testfetch',
 
549
                   'bzrlib.selftest.whitebox',
 
550
                   'bzrlib.selftest.teststore',
 
551
                   'bzrlib.selftest.blackbox',
 
552
                   'bzrlib.selftest.testsampler',
 
553
                   'bzrlib.selftest.testtransactions',
 
554
                   'bzrlib.selftest.testtransport',
 
555
                   'bzrlib.selftest.testgraph',
 
556
                   'bzrlib.selftest.testworkingtree',
 
557
                   'bzrlib.selftest.test_upgrade',
 
558
                   'bzrlib.selftest.test_conflicts',
 
559
                   'bzrlib.selftest.testtestament',
 
560
                   'bzrlib.selftest.testannotate',
 
561
                   'bzrlib.selftest.testrevprops',
 
562
                   'bzrlib.selftest.testoptions',
 
563
                   'bzrlib.selftest.testhttp',
 
564
                   'bzrlib.selftest.testnonascii',
 
565
                   ]
 
566
 
 
567
    for m in (bzrlib.store, bzrlib.inventory, bzrlib.branch,
 
568
              bzrlib.osutils, bzrlib.commands, bzrlib.merge3,
 
569
              bzrlib.errors,
 
570
              ):
 
571
        if m not in MODULES_TO_DOCTEST:
 
572
            MODULES_TO_DOCTEST.append(m)
 
573
 
 
574
    TestCase.BZRPATH = os.path.join(os.path.realpath(os.path.dirname(bzrlib.__path__[0])), 'bzr')
 
575
    print '%-30s %s' % ('bzr binary', TestCase.BZRPATH)
 
576
    print
 
577
    suite = TestSuite()
 
578
    suite.addTest(TestLoader().loadTestsFromNames(testmod_names))
 
579
    for m in MODULES_TO_TEST:
 
580
         suite.addTest(TestLoader().loadTestsFromModule(m))
 
581
    for m in (MODULES_TO_DOCTEST):
 
582
        suite.addTest(DocTestSuite(m))
 
583
    for p in bzrlib.plugin.all_plugins:
 
584
        if hasattr(p, 'test_suite'):
 
585
            suite.addTest(p.test_suite())
4374
586
    return suite
4375
587
 
4376
 
 
4377
 
def multiply_scenarios(*scenarios):
4378
 
    """Multiply two or more iterables of scenarios.
4379
 
 
4380
 
    It is safe to pass scenario generators or iterators.
4381
 
 
4382
 
    :returns: A list of compound scenarios: the cross-product of all
4383
 
        scenarios, with the names concatenated and the parameters
4384
 
        merged together.
4385
 
    """
4386
 
    return functools.reduce(_multiply_two_scenarios, map(list, scenarios))
4387
 
 
4388
 
 
4389
 
def _multiply_two_scenarios(scenarios_left, scenarios_right):
4390
 
    """Multiply two sets of scenarios.
4391
 
 
4392
 
    :returns: the cartesian product of the two sets of scenarios, that is
4393
 
        a scenario for every possible combination of a left scenario and a
4394
 
        right scenario.
4395
 
    """
4396
 
    return [
4397
 
        ('%s,%s' % (left_name, right_name),
4398
 
         dict(left_dict, **right_dict))
4399
 
        for left_name, left_dict in scenarios_left
4400
 
        for right_name, right_dict in scenarios_right]
4401
 
 
4402
 
 
4403
 
def multiply_tests(tests, scenarios, result):
4404
 
    """Multiply tests_list by scenarios into result.
4405
 
 
4406
 
    This is the core workhorse for test parameterisation.
4407
 
 
4408
 
    Typically the load_tests() method for a per-implementation test suite will
4409
 
    call multiply_tests and return the result.
4410
 
 
4411
 
    :param tests: The tests to parameterise.
4412
 
    :param scenarios: The scenarios to apply: pairs of (scenario_name,
4413
 
        scenario_param_dict).
4414
 
    :param result: A TestSuite to add created tests to.
4415
 
 
4416
 
    This returns the passed in result TestSuite with the cross product of all
4417
 
    the tests repeated once for each scenario.  Each test is adapted by adding
4418
 
    the scenario name at the end of its id(), and updating the test object's
4419
 
    __dict__ with the scenario_param_dict.
4420
 
 
4421
 
    >>> import breezy.tests.test_sampler
4422
 
    >>> r = multiply_tests(
4423
 
    ...     breezy.tests.test_sampler.DemoTest('test_nothing'),
4424
 
    ...     [('one', dict(param=1)),
4425
 
    ...      ('two', dict(param=2))],
4426
 
    ...     TestUtil.TestSuite())
4427
 
    >>> tests = list(iter_suite_tests(r))
4428
 
    >>> len(tests)
4429
 
    2
4430
 
    >>> tests[0].id()
4431
 
    'breezy.tests.test_sampler.DemoTest.test_nothing(one)'
4432
 
    >>> tests[0].param
4433
 
    1
4434
 
    >>> tests[1].param
4435
 
    2
4436
 
    """
4437
 
    for test in iter_suite_tests(tests):
4438
 
        apply_scenarios(test, scenarios, result)
4439
 
    return result
4440
 
 
4441
 
 
4442
 
def apply_scenarios(test, scenarios, result):
4443
 
    """Apply the scenarios in scenarios to test and add to result.
4444
 
 
4445
 
    :param test: The test to apply scenarios to.
4446
 
    :param scenarios: An iterable of scenarios to apply to test.
4447
 
    :return: result
4448
 
    :seealso: apply_scenario
4449
 
    """
4450
 
    for scenario in scenarios:
4451
 
        result.addTest(apply_scenario(test, scenario))
4452
 
    return result
4453
 
 
4454
 
 
4455
 
def apply_scenario(test, scenario):
4456
 
    """Copy test and apply scenario to it.
4457
 
 
4458
 
    :param test: A test to adapt.
4459
 
    :param scenario: A tuple describing the scenario.
4460
 
        The first element of the tuple is the new test id.
4461
 
        The second element is a dict containing attributes to set on the
4462
 
        test.
4463
 
    :return: The adapted test.
4464
 
    """
4465
 
    new_id = "%s(%s)" % (test.id(), scenario[0])
4466
 
    new_test = clone_test(test, new_id)
4467
 
    for name, value in scenario[1].items():
4468
 
        setattr(new_test, name, value)
4469
 
    return new_test
4470
 
 
4471
 
 
4472
 
def clone_test(test, new_id):
4473
 
    """Clone a test giving it a new id.
4474
 
 
4475
 
    :param test: The test to clone.
4476
 
    :param new_id: The id to assign to it.
4477
 
    :return: The new test.
4478
 
    """
4479
 
    new_test = copy.copy(test)
4480
 
    new_test.id = lambda: new_id
4481
 
    # XXX: Workaround <https://bugs.launchpad.net/testtools/+bug/637725>, which
4482
 
    # causes cloned tests to share the 'details' dict.  This makes it hard to
4483
 
    # read the test output for parameterized tests, because tracebacks will be
4484
 
    # associated with irrelevant tests.
4485
 
    try:
4486
 
        details = new_test._TestCase__details
4487
 
    except AttributeError:
4488
 
        # must be a different version of testtools than expected.  Do nothing.
4489
 
        pass
4490
 
    else:
4491
 
        # Reset the '__details' dict.
4492
 
        new_test._TestCase__details = {}
4493
 
    return new_test
4494
 
 
4495
 
 
4496
 
def permute_tests_for_extension(standard_tests, loader, py_module_name,
4497
 
                                ext_module_name):
4498
 
    """Helper for permutating tests against an extension module.
4499
 
 
4500
 
    This is meant to be used inside a modules 'load_tests()' function. It will
4501
 
    create 2 scenarios, and cause all tests in the 'standard_tests' to be run
4502
 
    against both implementations. Setting 'test.module' to the appropriate
4503
 
    module. See breezy.tests.test__chk_map.load_tests as an example.
4504
 
 
4505
 
    :param standard_tests: A test suite to permute
4506
 
    :param loader: A TestLoader
4507
 
    :param py_module_name: The python path to a python module that can always
4508
 
        be loaded, and will be considered the 'python' implementation. (eg
4509
 
        'breezy._chk_map_py')
4510
 
    :param ext_module_name: The python path to an extension module. If the
4511
 
        module cannot be loaded, a single test will be added, which notes that
4512
 
        the module is not available. If it can be loaded, all standard_tests
4513
 
        will be run against that module.
4514
 
    :return: (suite, feature) suite is a test-suite that has all the permuted
4515
 
        tests. feature is the Feature object that can be used to determine if
4516
 
        the module is available.
4517
 
    """
4518
 
 
4519
 
    from .features import ModuleAvailableFeature
4520
 
    py_module = pyutils.get_named_object(py_module_name)
4521
 
    scenarios = [
4522
 
        ('python', {'module': py_module}),
4523
 
    ]
4524
 
    suite = loader.suiteClass()
4525
 
    feature = ModuleAvailableFeature(ext_module_name)
4526
 
    if feature.available():
4527
 
        scenarios.append(('C', {'module': feature.module}))
4528
 
    else:
4529
 
        # the compiled module isn't available, so we add a failing test
4530
 
        class FailWithoutFeature(TestCase):
4531
 
            def test_fail(self):
4532
 
                self.requireFeature(feature)
4533
 
        suite.addTest(loader.loadTestsFromTestCase(FailWithoutFeature))
4534
 
    result = multiply_tests(standard_tests, scenarios, suite)
4535
 
    return result, feature
4536
 
 
4537
 
 
4538
 
def _rmtree_temp_dir(dirname, test_id=None):
4539
 
    # If LANG=C we probably have created some bogus paths
4540
 
    # which rmtree(unicode) will fail to delete
4541
 
    # so make sure we are using rmtree(str) to delete everything
4542
 
    # except on win32, where rmtree(str) will fail
4543
 
    # since it doesn't have the property of byte-stream paths
4544
 
    # (they are either ascii or mbcs)
4545
 
    if sys.platform == 'win32' and isinstance(dirname, bytes):
4546
 
        # make sure we are using the unicode win32 api
4547
 
        dirname = dirname.decode('mbcs')
4548
 
    else:
4549
 
        dirname = dirname.encode(sys.getfilesystemencoding())
4550
 
    try:
4551
 
        osutils.rmtree(dirname)
4552
 
    except OSError as e:
4553
 
        # We don't want to fail here because some useful display will be lost
4554
 
        # otherwise. Polluting the tmp dir is bad, but not giving all the
4555
 
        # possible info to the test runner is even worse.
4556
 
        if test_id is not None:
4557
 
            ui.ui_factory.clear_term()
4558
 
            sys.stderr.write('\nWhile running: %s\n' % (test_id,))
4559
 
        # Ugly, but the last thing we want here is fail, so bear with it.
4560
 
        printable_e = str(e).decode(osutils.get_user_encoding(), 'replace'
4561
 
                                    ).encode('ascii', 'replace')
4562
 
        sys.stderr.write('Unable to remove testing dir %s\n%s'
4563
 
                         % (os.path.basename(dirname), printable_e))
4564
 
 
4565
 
 
4566
 
def probe_unicode_in_user_encoding():
4567
 
    """Try to encode several unicode strings to use in unicode-aware tests.
4568
 
    Return first successfull match.
4569
 
 
4570
 
    :return:  (unicode value, encoded plain string value) or (None, None)
4571
 
    """
4572
 
    possible_vals = [u'm\xb5', u'\xe1', u'\u0410']
4573
 
    for uni_val in possible_vals:
4574
 
        try:
4575
 
            str_val = uni_val.encode(osutils.get_user_encoding())
4576
 
        except UnicodeEncodeError:
4577
 
            # Try a different character
4578
 
            pass
4579
 
        else:
4580
 
            return uni_val, str_val
4581
 
    return None, None
4582
 
 
4583
 
 
4584
 
def probe_bad_non_ascii(encoding):
4585
 
    """Try to find [bad] character with code [128..255]
4586
 
    that cannot be decoded to unicode in some encoding.
4587
 
    Return None if all non-ascii characters is valid
4588
 
    for given encoding.
4589
 
    """
4590
 
    for i in range(128, 256):
4591
 
        char = int2byte(i)
4592
 
        try:
4593
 
            char.decode(encoding)
4594
 
        except UnicodeDecodeError:
4595
 
            return char
4596
 
    return None
4597
 
 
4598
 
 
4599
 
# Only define SubUnitBzrRunner if subunit is available.
4600
 
try:
4601
 
    from subunit import TestProtocolClient
4602
 
    from subunit.test_results import AutoTimingTestResultDecorator
4603
 
 
4604
 
    class SubUnitBzrProtocolClientv1(TestProtocolClient):
4605
 
 
4606
 
        def stopTest(self, test):
4607
 
            super(SubUnitBzrProtocolClientv1, self).stopTest(test)
4608
 
            _clear__type_equality_funcs(test)
4609
 
 
4610
 
        def addSuccess(self, test, details=None):
4611
 
            # The subunit client always includes the details in the subunit
4612
 
            # stream, but we don't want to include it in ours.
4613
 
            if details is not None and 'log' in details:
4614
 
                del details['log']
4615
 
            return super(SubUnitBzrProtocolClientv1, self).addSuccess(
4616
 
                test, details)
4617
 
 
4618
 
    class SubUnitBzrRunnerv1(TextTestRunner):
4619
 
 
4620
 
        def run(self, test):
4621
 
            result = AutoTimingTestResultDecorator(
4622
 
                SubUnitBzrProtocolClientv1(self.stream))
4623
 
            test.run(result)
4624
 
            return result
4625
 
except ImportError:
4626
 
    pass
4627
 
 
4628
 
 
4629
 
try:
4630
 
    from subunit.run import SubunitTestRunner
4631
 
 
4632
 
    class SubUnitBzrRunnerv2(TextTestRunner, SubunitTestRunner):
4633
 
 
4634
 
        def __init__(self, stream=sys.stderr, descriptions=0, verbosity=1,
4635
 
                     bench_history=None, strict=False, result_decorators=None):
4636
 
            TextTestRunner.__init__(
4637
 
                self, stream=stream,
4638
 
                descriptions=descriptions, verbosity=verbosity,
4639
 
                bench_history=bench_history, strict=strict,
4640
 
                result_decorators=result_decorators)
4641
 
            SubunitTestRunner.__init__(self, verbosity=verbosity,
4642
 
                                       stream=stream)
4643
 
 
4644
 
        run = SubunitTestRunner.run
4645
 
except ImportError:
4646
 
    pass