~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/__init__.py

  • Committer: Robert Collins
  • Date: 2005-10-18 13:11:57 UTC
  • mfrom: (1185.16.72) (0.2.1)
  • Revision ID: robertc@robertcollins.net-20051018131157-76a9970aa78e927e
Merged Martin.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005, 2006 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
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
 
18
 
# TODO: Perhaps there should be an API to find out if bzr running under the
19
 
# test suite -- some plugins might want to avoid making intrusive changes if
20
 
# this is the case.  However, we want behaviour under to test to diverge as
21
 
# little as possible, so this should be used rarely if it's added at all.
22
 
# (Suggestion from j-a-meinel, 2005-11-24)
23
 
 
24
 
# NOTE: Some classes in here use camelCaseNaming() rather than
25
 
# underscore_naming().  That's for consistency with unittest; it's not the
26
 
# general style of bzrlib.  Please continue that consistency when adding e.g.
27
 
# new assertFoo() methods.
28
 
 
29
 
import codecs
30
18
from cStringIO import StringIO
31
19
import difflib
32
 
import doctest
33
20
import errno
34
21
import logging
35
22
import os
36
23
import re
37
 
import shlex
38
 
import stat
39
 
from subprocess import Popen, PIPE
 
24
import shutil
40
25
import sys
41
26
import tempfile
42
27
import unittest
43
28
import time
44
29
 
45
 
 
46
 
from bzrlib import (
47
 
    bzrdir,
48
 
    debug,
49
 
    errors,
50
 
    memorytree,
51
 
    osutils,
52
 
    progress,
53
 
    urlutils,
54
 
    )
55
 
import bzrlib.branch
56
30
import bzrlib.commands
57
 
import bzrlib.bundle.serializer
58
 
import bzrlib.export
59
 
import bzrlib.inventory
60
 
import bzrlib.iterablefile
61
 
import bzrlib.lockdir
62
 
try:
63
 
    import bzrlib.lsprof
64
 
except ImportError:
65
 
    # lsprof not available
66
 
    pass
67
 
from bzrlib.merge import merge_inner
68
 
import bzrlib.merge3
69
 
import bzrlib.osutils
70
 
import bzrlib.plugin
71
 
from bzrlib.revision import common_ancestor
72
 
import bzrlib.store
73
 
from bzrlib import symbol_versioning
74
31
import bzrlib.trace
75
 
from bzrlib.transport import get_transport
76
 
import bzrlib.transport
77
 
from bzrlib.transport.local import LocalURLServer
78
 
from bzrlib.transport.memory import MemoryServer
79
 
from bzrlib.transport.readonly import ReadonlyServer
80
 
from bzrlib.trace import mutter, note
81
 
from bzrlib.tests import TestUtil
82
 
from bzrlib.tests.HttpServer import HttpServer
83
 
from bzrlib.tests.TestUtil import (
84
 
                          TestSuite,
85
 
                          TestLoader,
86
 
                          )
87
 
from bzrlib.tests.treeshape import build_tree_contents
88
 
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2
89
 
 
90
 
default_transport = LocalURLServer
 
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
91
37
 
92
38
MODULES_TO_TEST = []
93
 
MODULES_TO_DOCTEST = [
94
 
                      bzrlib.bundle.serializer,
95
 
                      bzrlib.errors,
96
 
                      bzrlib.export,
97
 
                      bzrlib.inventory,
98
 
                      bzrlib.iterablefile,
99
 
                      bzrlib.lockdir,
100
 
                      bzrlib.merge3,
101
 
                      bzrlib.option,
102
 
                      bzrlib.store,
103
 
                      ]
104
 
 
105
 
 
106
 
def packages_to_test():
107
 
    """Return a list of packages to test.
108
 
 
109
 
    The packages are not globally imported so that import failures are
110
 
    triggered when running selftest, not when importing the command.
111
 
    """
112
 
    import bzrlib.doc
113
 
    import bzrlib.tests.blackbox
114
 
    import bzrlib.tests.branch_implementations
115
 
    import bzrlib.tests.bzrdir_implementations
116
 
    import bzrlib.tests.interrepository_implementations
117
 
    import bzrlib.tests.interversionedfile_implementations
118
 
    import bzrlib.tests.intertree_implementations
119
 
    import bzrlib.tests.repository_implementations
120
 
    import bzrlib.tests.revisionstore_implementations
121
 
    import bzrlib.tests.tree_implementations
122
 
    import bzrlib.tests.workingtree_implementations
123
 
    return [
124
 
            bzrlib.doc,
125
 
            bzrlib.tests.blackbox,
126
 
            bzrlib.tests.branch_implementations,
127
 
            bzrlib.tests.bzrdir_implementations,
128
 
            bzrlib.tests.interrepository_implementations,
129
 
            bzrlib.tests.interversionedfile_implementations,
130
 
            bzrlib.tests.intertree_implementations,
131
 
            bzrlib.tests.repository_implementations,
132
 
            bzrlib.tests.revisionstore_implementations,
133
 
            bzrlib.tests.tree_implementations,
134
 
            bzrlib.tests.workingtree_implementations,
135
 
            ]
136
 
 
137
 
 
138
 
class ExtendedTestResult(unittest._TextTestResult):
139
 
    """Accepts, reports and accumulates the results of running tests.
140
 
 
141
 
    Compared to this unittest version this class adds support for profiling,
142
 
    benchmarking, stopping as soon as a test fails,  and skipping tests.
143
 
    There are further-specialized subclasses for different types of display.
144
 
    """
145
 
 
146
 
    stop_early = False
147
 
    
148
 
    def __init__(self, stream, descriptions, verbosity,
149
 
                 bench_history=None,
150
 
                 num_tests=None,
151
 
                 ):
152
 
        """Construct new TestResult.
153
 
 
154
 
        :param bench_history: Optionally, a writable file object to accumulate
155
 
            benchmark results.
156
 
        """
157
 
        unittest._TextTestResult.__init__(self, stream, descriptions, verbosity)
158
 
        if bench_history is not None:
159
 
            from bzrlib.version import _get_bzr_source_tree
160
 
            src_tree = _get_bzr_source_tree()
161
 
            if src_tree:
162
 
                try:
163
 
                    revision_id = src_tree.get_parent_ids()[0]
164
 
                except IndexError:
165
 
                    # XXX: if this is a brand new tree, do the same as if there
166
 
                    # is no branch.
167
 
                    revision_id = ''
168
 
            else:
169
 
                # XXX: If there's no branch, what should we do?
170
 
                revision_id = ''
171
 
            bench_history.write("--date %s %s\n" % (time.time(), revision_id))
172
 
        self._bench_history = bench_history
173
 
        self.ui = bzrlib.ui.ui_factory
174
 
        self.num_tests = num_tests
175
 
        self.error_count = 0
176
 
        self.failure_count = 0
177
 
        self.skip_count = 0
178
 
        self.count = 0
179
 
        self._overall_start_time = time.time()
180
 
    
181
 
    def extractBenchmarkTime(self, testCase):
182
 
        """Add a benchmark time for the current test case."""
183
 
        self._benchmarkTime = getattr(testCase, "_benchtime", None)
184
 
    
185
 
    def _elapsedTestTimeString(self):
186
 
        """Return a time string for the overall time the current test has taken."""
187
 
        return self._formatTime(time.time() - self._start_time)
188
 
 
189
 
    def _testTimeString(self):
190
 
        if self._benchmarkTime is not None:
191
 
            return "%s/%s" % (
192
 
                self._formatTime(self._benchmarkTime),
193
 
                self._elapsedTestTimeString())
194
 
        else:
195
 
            return "           %s" % self._elapsedTestTimeString()
196
 
 
197
 
    def _formatTime(self, seconds):
198
 
        """Format seconds as milliseconds with leading spaces."""
199
 
        # some benchmarks can take thousands of seconds to run, so we need 8
200
 
        # places
201
 
        return "%8dms" % (1000 * seconds)
202
 
 
203
 
    def _shortened_test_description(self, test):
204
 
        what = test.id()
205
 
        what = re.sub(r'^bzrlib\.(tests|benchmarks)\.', '', what)
206
 
        return what
 
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)
207
77
 
208
78
    def startTest(self, test):
209
79
        unittest.TestResult.startTest(self, test)
210
 
        self.report_test_start(test)
211
 
        self._recordTestStartTime()
212
 
 
213
 
    def _recordTestStartTime(self):
214
 
        """Record that a test has started."""
 
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()
215
85
        self._start_time = time.time()
216
86
 
217
87
    def addError(self, test, err):
218
 
        if isinstance(err[1], TestSkipped):
219
 
            return self.addSkipped(test, err)    
220
88
        unittest.TestResult.addError(self, test, err)
221
 
        # We can only do this if we have one of our TestCases, not if
222
 
        # we have a doctest.
223
 
        setKeepLogfile = getattr(test, 'setKeepLogfile', None)
224
 
        if setKeepLogfile is not None:
225
 
            setKeepLogfile()
226
 
        self.extractBenchmarkTime(test)
227
 
        self.report_error(test, err)
228
 
        if self.stop_early:
229
 
            self.stop()
 
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()
230
94
 
231
95
    def addFailure(self, test, err):
232
96
        unittest.TestResult.addFailure(self, test, err)
233
 
        # We can only do this if we have one of our TestCases, not if
234
 
        # we have a doctest.
235
 
        setKeepLogfile = getattr(test, 'setKeepLogfile', None)
236
 
        if setKeepLogfile is not None:
237
 
            setKeepLogfile()
238
 
        self.extractBenchmarkTime(test)
239
 
        self.report_failure(test, err)
240
 
        if self.stop_early:
241
 
            self.stop()
 
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()
242
102
 
243
103
    def addSuccess(self, test):
244
 
        self.extractBenchmarkTime(test)
245
 
        if self._bench_history is not None:
246
 
            if self._benchmarkTime is not None:
247
 
                self._bench_history.write("%s %s\n" % (
248
 
                    self._formatTime(self._benchmarkTime),
249
 
                    test.id()))
250
 
        self.report_success(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()
251
109
        unittest.TestResult.addSuccess(self, test)
252
110
 
253
 
    def addSkipped(self, test, skip_excinfo):
254
 
        self.extractBenchmarkTime(test)
255
 
        self.report_skip(test, skip_excinfo)
256
 
        # seems best to treat this as success from point-of-view of unittest
257
 
        # -- it actually does nothing so it barely matters :)
258
 
        try:
259
 
            test.tearDown()
260
 
        except KeyboardInterrupt:
261
 
            raise
262
 
        except:
263
 
            self.addError(test, test.__exc_info())
264
 
        else:
265
 
            unittest.TestResult.addSuccess(self, test)
266
 
 
267
111
    def printErrorList(self, flavour, errors):
268
112
        for test, err in errors:
269
113
            self.stream.writeln(self.separator1)
270
 
            self.stream.writeln("%s: %s" % (flavour, self.getDescription(test)))
271
 
            if getattr(test, '_get_log', None) is not None:
272
 
                print >>self.stream
273
 
                print >>self.stream, \
274
 
                        ('vvvv[log from %s]' % test.id()).ljust(78,'-')
 
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:')
275
118
                print >>self.stream, test._get_log()
276
 
                print >>self.stream, \
277
 
                        ('^^^^[log from %s]' % test.id()).ljust(78,'-')
278
119
            self.stream.writeln(self.separator2)
279
120
            self.stream.writeln("%s" % err)
280
121
 
281
 
    def finished(self):
282
 
        pass
283
 
 
284
 
    def report_cleaning_up(self):
285
 
        pass
286
 
 
287
 
    def report_success(self, test):
288
 
        pass
289
 
 
290
 
 
291
 
class TextTestResult(ExtendedTestResult):
292
 
    """Displays progress and results of tests in text form"""
293
 
 
294
 
    def __init__(self, *args, **kw):
295
 
        ExtendedTestResult.__init__(self, *args, **kw)
296
 
        self.pb = self.ui.nested_progress_bar()
297
 
        self.pb.show_pct = False
298
 
        self.pb.show_spinner = False
299
 
        self.pb.show_eta = False, 
300
 
        self.pb.show_count = False
301
 
        self.pb.show_bar = False
302
 
 
303
 
    def report_starting(self):
304
 
        self.pb.update('[test 0/%d] starting...' % (self.num_tests))
305
 
 
306
 
    def _progress_prefix_text(self):
307
 
        a = '[%d' % self.count
308
 
        if self.num_tests is not None:
309
 
            a +='/%d' % self.num_tests
310
 
        a += ' in %ds' % (time.time() - self._overall_start_time)
311
 
        if self.error_count:
312
 
            a += ', %d errors' % self.error_count
313
 
        if self.failure_count:
314
 
            a += ', %d failed' % self.failure_count
315
 
        if self.skip_count:
316
 
            a += ', %d skipped' % self.skip_count
317
 
        a += ']'
318
 
        return a
319
 
 
320
 
    def report_test_start(self, test):
321
 
        self.count += 1
322
 
        self.pb.update(
323
 
                self._progress_prefix_text()
324
 
                + ' ' 
325
 
                + self._shortened_test_description(test))
326
 
 
327
 
    def report_error(self, test, err):
328
 
        self.error_count += 1
329
 
        self.pb.note('ERROR: %s\n    %s\n', 
330
 
            self._shortened_test_description(test),
331
 
            err[1],
332
 
            )
333
 
 
334
 
    def report_failure(self, test, err):
335
 
        self.failure_count += 1
336
 
        self.pb.note('FAIL: %s\n    %s\n', 
337
 
            self._shortened_test_description(test),
338
 
            err[1],
339
 
            )
340
 
 
341
 
    def report_skip(self, test, skip_excinfo):
342
 
        self.skip_count += 1
343
 
        if False:
344
 
            # at the moment these are mostly not things we can fix
345
 
            # and so they just produce stipple; use the verbose reporter
346
 
            # to see them.
347
 
            if False:
348
 
                # show test and reason for skip
349
 
                self.pb.note('SKIP: %s\n    %s\n', 
350
 
                    self._shortened_test_description(test),
351
 
                    skip_excinfo[1])
352
 
            else:
353
 
                # since the class name was left behind in the still-visible
354
 
                # progress bar...
355
 
                self.pb.note('SKIP: %s', skip_excinfo[1])
356
 
 
357
 
    def report_cleaning_up(self):
358
 
        self.pb.update('cleaning up...')
359
 
 
360
 
    def finished(self):
361
 
        self.pb.finished()
362
 
 
363
 
 
364
 
class VerboseTestResult(ExtendedTestResult):
365
 
    """Produce long output, with one line per test run plus times"""
366
 
 
367
 
    def _ellipsize_to_right(self, a_string, final_width):
368
 
        """Truncate and pad a string, keeping the right hand side"""
369
 
        if len(a_string) > final_width:
370
 
            result = '...' + a_string[3-final_width:]
371
 
        else:
372
 
            result = a_string
373
 
        return result.ljust(final_width)
374
 
 
375
 
    def report_starting(self):
376
 
        self.stream.write('running %d tests...\n' % self.num_tests)
377
 
 
378
 
    def report_test_start(self, test):
379
 
        self.count += 1
380
 
        name = self._shortened_test_description(test)
381
 
        # width needs space for 6 char status, plus 1 for slash, plus 2 10-char
382
 
        # numbers, plus a trailing blank
383
 
        self.stream.write(self._ellipsize_to_right(name,
384
 
                            osutils.terminal_width()-30))
385
 
        self.stream.flush()
386
 
 
387
 
    def report_error(self, test, err):
388
 
        self.error_count += 1
389
 
        self.stream.writeln('ERROR %s\n    %s' 
390
 
                % (self._testTimeString(), err[1]))
391
 
 
392
 
    def report_failure(self, test, err):
393
 
        self.failure_count += 1
394
 
        self.stream.writeln(' FAIL %s\n    %s'
395
 
                % (self._testTimeString(), err[1]))
396
 
 
397
 
    def report_success(self, test):
398
 
        self.stream.writeln('   OK %s' % self._testTimeString())
399
 
        for bench_called, stats in getattr(test, '_benchcalls', []):
400
 
            self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
401
 
            stats.pprint(file=self.stream)
402
 
        self.stream.flush()
403
 
 
404
 
    def report_skip(self, test, skip_excinfo):
405
 
        print >>self.stream, ' SKIP %s' % self._testTimeString()
406
 
        print >>self.stream, '     %s' % skip_excinfo[1]
407
 
 
408
 
 
409
 
class TextTestRunner(object):
 
122
 
 
123
class TextTestRunner(unittest.TextTestRunner):
410
124
    stop_on_failure = False
411
125
 
412
 
    def __init__(self,
413
 
                 stream=sys.stderr,
414
 
                 descriptions=0,
415
 
                 verbosity=1,
416
 
                 keep_output=False,
417
 
                 bench_history=None):
418
 
        self.stream = unittest._WritelnDecorator(stream)
419
 
        self.descriptions = descriptions
420
 
        self.verbosity = verbosity
421
 
        self.keep_output = keep_output
422
 
        self._bench_history = bench_history
423
 
 
424
 
    def run(self, test):
425
 
        "Run the given test case or test suite."
426
 
        startTime = time.time()
427
 
        if self.verbosity == 1:
428
 
            result_class = TextTestResult
429
 
        elif self.verbosity >= 2:
430
 
            result_class = VerboseTestResult
431
 
        result = result_class(self.stream,
432
 
                              self.descriptions,
433
 
                              self.verbosity,
434
 
                              bench_history=self._bench_history,
435
 
                              num_tests=test.countTestCases(),
436
 
                              )
437
 
        result.stop_early = self.stop_on_failure
438
 
        result.report_starting()
439
 
        test.run(result)
440
 
        stopTime = time.time()
441
 
        timeTaken = stopTime - startTime
442
 
        result.printErrors()
443
 
        self.stream.writeln(result.separator2)
444
 
        run = result.testsRun
445
 
        self.stream.writeln("Ran %d test%s in %.3fs" %
446
 
                            (run, run != 1 and "s" or "", timeTaken))
447
 
        self.stream.writeln()
448
 
        if not result.wasSuccessful():
449
 
            self.stream.write("FAILED (")
450
 
            failed, errored = map(len, (result.failures, result.errors))
451
 
            if failed:
452
 
                self.stream.write("failures=%d" % failed)
453
 
            if errored:
454
 
                if failed: self.stream.write(", ")
455
 
                self.stream.write("errors=%d" % errored)
456
 
            self.stream.writeln(")")
457
 
        else:
458
 
            self.stream.writeln("OK")
459
 
        result.report_cleaning_up()
460
 
        # This is still a little bogus, 
461
 
        # but only a little. Folk not using our testrunner will
462
 
        # have to delete their temp directories themselves.
463
 
        test_root = TestCaseWithMemoryTransport.TEST_ROOT
464
 
        if result.wasSuccessful() or not self.keep_output:
465
 
            if test_root is not None:
466
 
                # If LANG=C we probably have created some bogus paths
467
 
                # which rmtree(unicode) will fail to delete
468
 
                # so make sure we are using rmtree(str) to delete everything
469
 
                # except on win32, where rmtree(str) will fail
470
 
                # since it doesn't have the property of byte-stream paths
471
 
                # (they are either ascii or mbcs)
472
 
                if sys.platform == 'win32':
473
 
                    # make sure we are using the unicode win32 api
474
 
                    test_root = unicode(test_root)
475
 
                else:
476
 
                    test_root = test_root.encode(
477
 
                        sys.getfilesystemencoding())
478
 
                osutils.rmtree(test_root)
479
 
        else:
480
 
            note("Failed tests working directories are in '%s'\n", test_root)
481
 
        TestCaseWithMemoryTransport.TEST_ROOT = None
482
 
        result.finished()
 
126
    def _makeResult(self):
 
127
        result = _MyResult(self.stream, self.descriptions, self.verbosity)
 
128
        if self.stop_on_failure:
 
129
            result = EarlyStoppingTestResultAdapter(result)
483
130
        return result
484
131
 
485
132
 
498
145
 
499
146
class TestSkipped(Exception):
500
147
    """Indicates that a test was intentionally skipped, rather than failing."""
 
148
    # XXX: Not used yet
501
149
 
502
150
 
503
151
class CommandFailed(Exception):
504
152
    pass
505
153
 
506
 
 
507
 
class StringIOWrapper(object):
508
 
    """A wrapper around cStringIO which just adds an encoding attribute.
509
 
    
510
 
    Internally we can check sys.stdout to see what the output encoding
511
 
    should be. However, cStringIO has no encoding attribute that we can
512
 
    set. So we wrap it instead.
513
 
    """
514
 
    encoding='ascii'
515
 
    _cstring = None
516
 
 
517
 
    def __init__(self, s=None):
518
 
        if s is not None:
519
 
            self.__dict__['_cstring'] = StringIO(s)
520
 
        else:
521
 
            self.__dict__['_cstring'] = StringIO()
522
 
 
523
 
    def __getattr__(self, name, getattr=getattr):
524
 
        return getattr(self.__dict__['_cstring'], name)
525
 
 
526
 
    def __setattr__(self, name, val):
527
 
        if name == 'encoding':
528
 
            self.__dict__['encoding'] = val
529
 
        else:
530
 
            return setattr(self._cstring, name, val)
531
 
 
532
 
 
533
154
class TestCase(unittest.TestCase):
534
155
    """Base class for bzr unit tests.
535
156
    
538
159
 
539
160
    Error and debug log messages are redirected from their usual
540
161
    location into a temporary file, the contents of which can be
541
 
    retrieved by _get_log().  We use a real OS file, not an in-memory object,
542
 
    so that it can also capture file IO.  When the test completes this file
543
 
    is read into memory and removed from disk.
 
162
    retrieved by _get_log().
544
163
       
545
164
    There are also convenience functions to invoke bzr's command-line
546
 
    routine, and to build and check bzr trees.
547
 
   
548
 
    In addition to the usual method of overriding tearDown(), this class also
549
 
    allows subclasses to register functions into the _cleanups list, which is
550
 
    run in order as the object is torn down.  It's less likely this will be
551
 
    accidentally overlooked.
552
 
    """
 
165
    routine, and to build and check bzr trees."""
553
166
 
 
167
    BZRPATH = 'bzr'
554
168
    _log_file_name = None
555
 
    _log_contents = ''
556
 
    _keep_log_file = False
557
 
    # record lsprof data when performing benchmark calls.
558
 
    _gather_lsprof_in_benchmarks = False
559
 
 
560
 
    def __init__(self, methodName='testMethod'):
561
 
        super(TestCase, self).__init__(methodName)
562
 
        self._cleanups = []
563
169
 
564
170
    def setUp(self):
565
171
        unittest.TestCase.setUp(self)
566
 
        self._cleanEnvironment()
 
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']
567
180
        bzrlib.trace.disable_default_logging()
568
 
        self._silenceUI()
569
 
        self._startLogFile()
570
 
        self._benchcalls = []
571
 
        self._benchtime = None
572
 
        # prevent hooks affecting tests
573
 
        self._preserved_hooks = bzrlib.branch.Branch.hooks
574
 
        self.addCleanup(self._restoreHooks)
575
 
        # this list of hooks must be kept in sync with the defaults
576
 
        # in branch.py
577
 
        bzrlib.branch.Branch.hooks = bzrlib.branch.BranchHooks()
578
 
 
579
 
    def _silenceUI(self):
580
 
        """Turn off UI for duration of test"""
581
 
        # by default the UI is off; tests can turn it on if they want it.
582
 
        saved = bzrlib.ui.ui_factory
583
 
        def _restore():
584
 
            bzrlib.ui.ui_factory = saved
585
 
        bzrlib.ui.ui_factory = bzrlib.ui.SilentUIFactory()
586
 
        self.addCleanup(_restore)
 
181
        self._enable_file_logging()
587
182
 
588
183
    def _ndiff_strings(self, a, b):
589
184
        """Return ndiff between two strings containing lines.
600
195
                                  charjunk=lambda x: False)
601
196
        return ''.join(difflines)
602
197
 
603
 
    def assertEqualDiff(self, a, b, message=None):
 
198
    def assertEqualDiff(self, a, b):
604
199
        """Assert two texts are equal, if not raise an exception.
605
200
        
606
201
        This is intended for use with multi-line strings where it can 
609
204
        # TODO: perhaps override assertEquals to call this for strings?
610
205
        if a == b:
611
206
            return
612
 
        if message is None:
613
 
            message = "texts not equal:\n"
614
 
        raise AssertionError(message + 
 
207
        raise AssertionError("texts not equal:\n" + 
615
208
                             self._ndiff_strings(a, b))      
616
 
        
617
 
    def assertEqualMode(self, mode, mode_test):
618
 
        self.assertEqual(mode, mode_test,
619
 
                         'mode mismatch %o != %o' % (mode, mode_test))
620
 
 
621
 
    def assertStartsWith(self, s, prefix):
622
 
        if not s.startswith(prefix):
623
 
            raise AssertionError('string %r does not start with %r' % (s, prefix))
624
 
 
625
 
    def assertEndsWith(self, s, suffix):
626
 
        """Asserts that s ends with suffix."""
627
 
        if not s.endswith(suffix):
628
 
            raise AssertionError('string %r does not end with %r' % (s, suffix))
629
209
 
630
210
    def assertContainsRe(self, haystack, needle_re):
631
211
        """Assert that a contains something matching a regular expression."""
633
213
            raise AssertionError('pattern "%s" not found in "%s"'
634
214
                    % (needle_re, haystack))
635
215
 
636
 
    def assertNotContainsRe(self, haystack, needle_re):
637
 
        """Assert that a does not match a regular expression"""
638
 
        if re.search(needle_re, haystack):
639
 
            raise AssertionError('pattern "%s" found in "%s"'
640
 
                    % (needle_re, haystack))
641
 
 
642
 
    def assertSubset(self, sublist, superlist):
643
 
        """Assert that every entry in sublist is present in superlist."""
644
 
        missing = []
645
 
        for entry in sublist:
646
 
            if entry not in superlist:
647
 
                missing.append(entry)
648
 
        if len(missing) > 0:
649
 
            raise AssertionError("value(s) %r not present in container %r" % 
650
 
                                 (missing, superlist))
651
 
 
652
 
    def assertListRaises(self, excClass, func, *args, **kwargs):
653
 
        """Fail unless excClass is raised when the iterator from func is used.
654
 
        
655
 
        Many functions can return generators this makes sure
656
 
        to wrap them in a list() call to make sure the whole generator
657
 
        is run, and that the proper exception is raised.
658
 
        """
659
 
        try:
660
 
            list(func(*args, **kwargs))
661
 
        except excClass:
662
 
            return
663
 
        else:
664
 
            if getattr(excClass,'__name__', None) is not None:
665
 
                excName = excClass.__name__
666
 
            else:
667
 
                excName = str(excClass)
668
 
            raise self.failureException, "%s not raised" % excName
669
 
 
670
 
    def assertIs(self, left, right, message=None):
671
 
        if not (left is right):
672
 
            if message is not None:
673
 
                raise AssertionError(message)
674
 
            else:
675
 
                raise AssertionError("%r is not %r." % (left, right))
676
 
 
677
 
    def assertIsNot(self, left, right, message=None):
678
 
        if (left is right):
679
 
            if message is not None:
680
 
                raise AssertionError(message)
681
 
            else:
682
 
                raise AssertionError("%r is %r." % (left, right))
683
 
 
684
 
    def assertTransportMode(self, transport, path, mode):
685
 
        """Fail if a path does not have mode mode.
686
 
        
687
 
        If modes are not supported on this transport, the assertion is ignored.
688
 
        """
689
 
        if not transport._can_roundtrip_unix_modebits():
690
 
            return
691
 
        path_stat = transport.stat(path)
692
 
        actual_mode = stat.S_IMODE(path_stat.st_mode)
693
 
        self.assertEqual(mode, actual_mode,
694
 
            'mode of %r incorrect (%o != %o)' % (path, mode, actual_mode))
695
 
 
696
 
    def assertIsInstance(self, obj, kls):
697
 
        """Fail if obj is not an instance of kls"""
698
 
        if not isinstance(obj, kls):
699
 
            self.fail("%r is an instance of %s rather than %s" % (
700
 
                obj, obj.__class__, kls))
701
 
 
702
 
    def _capture_warnings(self, a_callable, *args, **kwargs):
703
 
        """A helper for callDeprecated and applyDeprecated.
704
 
 
705
 
        :param a_callable: A callable to call.
706
 
        :param args: The positional arguments for the callable
707
 
        :param kwargs: The keyword arguments for the callable
708
 
        :return: A tuple (warnings, result). result is the result of calling
709
 
            a_callable(*args, **kwargs).
710
 
        """
711
 
        local_warnings = []
712
 
        def capture_warnings(msg, cls=None, stacklevel=None):
713
 
            # we've hooked into a deprecation specific callpath,
714
 
            # only deprecations should getting sent via it.
715
 
            self.assertEqual(cls, DeprecationWarning)
716
 
            local_warnings.append(msg)
717
 
        original_warning_method = symbol_versioning.warn
718
 
        symbol_versioning.set_warning_method(capture_warnings)
719
 
        try:
720
 
            result = a_callable(*args, **kwargs)
721
 
        finally:
722
 
            symbol_versioning.set_warning_method(original_warning_method)
723
 
        return (local_warnings, result)
724
 
 
725
 
    def applyDeprecated(self, deprecation_format, a_callable, *args, **kwargs):
726
 
        """Call a deprecated callable without warning the user.
727
 
 
728
 
        :param deprecation_format: The deprecation format that the callable
729
 
            should have been deprecated with. This is the same type as the 
730
 
            parameter to deprecated_method/deprecated_function. If the 
731
 
            callable is not deprecated with this format, an assertion error
732
 
            will be raised.
733
 
        :param a_callable: A callable to call. This may be a bound method or
734
 
            a regular function. It will be called with *args and **kwargs.
735
 
        :param args: The positional arguments for the callable
736
 
        :param kwargs: The keyword arguments for the callable
737
 
        :return: The result of a_callable(*args, **kwargs)
738
 
        """
739
 
        call_warnings, result = self._capture_warnings(a_callable,
740
 
            *args, **kwargs)
741
 
        expected_first_warning = symbol_versioning.deprecation_string(
742
 
            a_callable, deprecation_format)
743
 
        if len(call_warnings) == 0:
744
 
            self.fail("No assertion generated by call to %s" %
745
 
                a_callable)
746
 
        self.assertEqual(expected_first_warning, call_warnings[0])
747
 
        return result
748
 
 
749
 
    def callDeprecated(self, expected, callable, *args, **kwargs):
750
 
        """Assert that a callable is deprecated in a particular way.
751
 
 
752
 
        This is a very precise test for unusual requirements. The 
753
 
        applyDeprecated helper function is probably more suited for most tests
754
 
        as it allows you to simply specify the deprecation format being used
755
 
        and will ensure that that is issued for the function being called.
756
 
 
757
 
        :param expected: a list of the deprecation warnings expected, in order
758
 
        :param callable: The callable to call
759
 
        :param args: The positional arguments for the callable
760
 
        :param kwargs: The keyword arguments for the callable
761
 
        """
762
 
        call_warnings, result = self._capture_warnings(callable,
763
 
            *args, **kwargs)
764
 
        self.assertEqual(expected, call_warnings)
765
 
        return result
766
 
 
767
 
    def _startLogFile(self):
768
 
        """Send bzr and test log messages to a temporary file.
769
 
 
770
 
        The file is removed as the test is torn down.
771
 
        """
 
216
    def _enable_file_logging(self):
772
217
        fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
 
218
 
773
219
        self._log_file = os.fdopen(fileno, 'w+')
774
 
        self._log_nonce = bzrlib.trace.enable_test_log(self._log_file)
 
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
        
775
229
        self._log_file_name = name
776
 
        self.addCleanup(self._finishLogFile)
777
 
 
778
 
    def _finishLogFile(self):
779
 
        """Finished with the log file.
780
 
 
781
 
        Close the file and delete it, unless setKeepLogfile was called.
782
 
        """
783
 
        if self._log_file is None:
784
 
            return
785
 
        bzrlib.trace.disable_test_log(self._log_nonce)
 
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())
786
244
        self._log_file.close()
787
 
        self._log_file = None
788
 
        if not self._keep_log_file:
789
 
            os.remove(self._log_file_name)
790
 
            self._log_file_name = None
791
 
 
792
 
    def setKeepLogfile(self):
793
 
        """Make the logfile not be deleted when _finishLogFile is called."""
794
 
        self._keep_log_file = True
795
 
 
796
 
    def addCleanup(self, callable):
797
 
        """Arrange to run a callable when this case is torn down.
798
 
 
799
 
        Callables are run in the reverse of the order they are registered, 
800
 
        ie last-in first-out.
801
 
        """
802
 
        if callable in self._cleanups:
803
 
            raise ValueError("cleanup function %r already registered on %s" 
804
 
                    % (callable, self))
805
 
        self._cleanups.append(callable)
806
 
 
807
 
    def _cleanEnvironment(self):
808
 
        new_env = {
809
 
            'BZR_HOME': None, # Don't inherit BZR_HOME to all the tests.
810
 
            'HOME': os.getcwd(),
811
 
            'APPDATA': os.getcwd(),
812
 
            'BZR_EMAIL': None,
813
 
            'BZREMAIL': None, # may still be present in the environment
814
 
            'EMAIL': None,
815
 
            'BZR_PROGRESS_BAR': None,
816
 
            # Proxies
817
 
            'http_proxy': None,
818
 
            'HTTP_PROXY': None,
819
 
            'https_proxy': None,
820
 
            'HTTPS_PROXY': None,
821
 
            'no_proxy': None,
822
 
            'NO_PROXY': None,
823
 
            'all_proxy': None,
824
 
            'ALL_PROXY': None,
825
 
            # Nobody cares about these ones AFAIK. So far at
826
 
            # least. If you do (care), please update this comment
827
 
            # -- vila 20061212
828
 
            'ftp_proxy': None,
829
 
            'FTP_PROXY': None,
830
 
        }
831
 
        self.__old_env = {}
832
 
        self.addCleanup(self._restoreEnvironment)
833
 
        for name, value in new_env.iteritems():
834
 
            self._captureVar(name, value)
835
 
 
836
 
    def _captureVar(self, name, newvalue):
837
 
        """Set an environment variable, and reset it when finished."""
838
 
        self.__old_env[name] = osutils.set_or_unset_env(name, newvalue)
839
 
 
840
 
    def _restoreEnvironment(self):
841
 
        for name, value in self.__old_env.iteritems():
842
 
            osutils.set_or_unset_env(name, value)
843
 
 
844
 
    def _restoreHooks(self):
845
 
        bzrlib.branch.Branch.hooks = self._preserved_hooks
846
 
 
847
 
    def tearDown(self):
848
 
        self._runCleanups()
849
245
        unittest.TestCase.tearDown(self)
850
246
 
851
 
    def time(self, callable, *args, **kwargs):
852
 
        """Run callable and accrue the time it takes to the benchmark time.
853
 
        
854
 
        If lsprofiling is enabled (i.e. by --lsprof-time to bzr selftest) then
855
 
        this will cause lsprofile statistics to be gathered and stored in
856
 
        self._benchcalls.
857
 
        """
858
 
        if self._benchtime is None:
859
 
            self._benchtime = 0
860
 
        start = time.time()
861
 
        try:
862
 
            if not self._gather_lsprof_in_benchmarks:
863
 
                return callable(*args, **kwargs)
864
 
            else:
865
 
                # record this benchmark
866
 
                ret, stats = bzrlib.lsprof.profile(callable, *args, **kwargs)
867
 
                stats.sort()
868
 
                self._benchcalls.append(((callable, args, kwargs), stats))
869
 
                return ret
870
 
        finally:
871
 
            self._benchtime += time.time() - start
872
 
 
873
 
    def _runCleanups(self):
874
 
        """Run registered cleanup functions. 
875
 
 
876
 
        This should only be called from TestCase.tearDown.
877
 
        """
878
 
        # TODO: Perhaps this should keep running cleanups even if 
879
 
        # one of them fails?
880
 
        for cleanup_fn in reversed(self._cleanups):
881
 
            cleanup_fn()
882
 
 
883
247
    def log(self, *args):
884
 
        mutter(*args)
 
248
        logging.debug(*args)
885
249
 
886
 
    def _get_log(self, keep_log_file=False):
887
 
        """Return as a string the log for this test. If the file is still
888
 
        on disk and keep_log_file=False, delete the log file and store the
889
 
        content in self._log_contents."""
890
 
        # flush the log file, to get all content
891
 
        import bzrlib.trace
892
 
        bzrlib.trace._trace_file.flush()
893
 
        if self._log_contents:
894
 
            return self._log_contents
895
 
        if self._log_file_name is not None:
896
 
            logfile = open(self._log_file_name)
897
 
            try:
898
 
                log_contents = logfile.read()
899
 
            finally:
900
 
                logfile.close()
901
 
            if not keep_log_file:
902
 
                self._log_contents = log_contents
903
 
                os.remove(self._log_file_name)
904
 
            return log_contents
 
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()
905
254
        else:
906
 
            return "DELETED log file to reduce memory footprint"
 
255
            return ''
907
256
 
908
 
    def capture(self, cmd, retcode=0):
 
257
    def capture(self, cmd):
909
258
        """Shortcut that splits cmd into words, runs, and returns stdout"""
910
 
        return self.run_bzr_captured(cmd.split(), retcode=retcode)[0]
 
259
        return self.run_bzr_captured(cmd.split())[0]
911
260
 
912
 
    def run_bzr_captured(self, argv, retcode=0, encoding=None, stdin=None,
913
 
                         working_dir=None):
914
 
        """Invoke bzr and return (stdout, stderr).
 
261
    def run_bzr_captured(self, argv, retcode=0):
 
262
        """Invoke bzr and return (result, stdout, stderr).
915
263
 
916
264
        Useful for code that wants to check the contents of the
917
265
        output, the way error messages are presented, etc.
927
275
        errors, and with logging set to something approximating the
928
276
        default, so that error reporting can be checked.
929
277
 
930
 
        :param argv: arguments to invoke bzr
931
 
        :param retcode: expected return code, or None for don't-care.
932
 
        :param encoding: encoding for sys.stdout and sys.stderr
933
 
        :param stdin: A string to be used as stdin for the command.
934
 
        :param working_dir: Change to this directory before running
 
278
        argv -- arguments to invoke bzr
 
279
        retcode -- expected return code, or None for don't-care.
935
280
        """
936
 
        if encoding is None:
937
 
            encoding = bzrlib.user_encoding
938
 
        if stdin is not None:
939
 
            stdin = StringIO(stdin)
940
 
        stdout = StringIOWrapper()
941
 
        stderr = StringIOWrapper()
942
 
        stdout.encoding = encoding
943
 
        stderr.encoding = encoding
944
 
 
945
 
        self.log('run bzr: %r', argv)
946
 
        # FIXME: don't call into logging here
 
281
        stdout = StringIO()
 
282
        stderr = StringIO()
 
283
        self.log('run bzr: %s', ' '.join(argv))
947
284
        handler = logging.StreamHandler(stderr)
 
285
        handler.setFormatter(bzrlib.trace.QuietFormatter())
948
286
        handler.setLevel(logging.INFO)
949
287
        logger = logging.getLogger('')
950
288
        logger.addHandler(handler)
951
 
        old_ui_factory = bzrlib.ui.ui_factory
952
 
        bzrlib.ui.ui_factory = bzrlib.tests.blackbox.TestUIFactory(
953
 
            stdout=stdout,
954
 
            stderr=stderr)
955
 
        bzrlib.ui.ui_factory.stdin = stdin
956
 
 
957
 
        cwd = None
958
 
        if working_dir is not None:
959
 
            cwd = osutils.getcwd()
960
 
            os.chdir(working_dir)
961
 
 
962
289
        try:
963
 
            saved_debug_flags = frozenset(debug.debug_flags)
964
 
            debug.debug_flags.clear()
965
 
            try:
966
 
                result = self.apply_redirected(stdin, stdout, stderr,
967
 
                                               bzrlib.commands.run_bzr_catch_errors,
968
 
                                               argv)
969
 
            finally:
970
 
                debug.debug_flags.update(saved_debug_flags)
 
290
            result = self.apply_redirected(None, stdout, stderr,
 
291
                                           bzrlib.commands.run_bzr_catch_errors,
 
292
                                           argv)
971
293
        finally:
972
294
            logger.removeHandler(handler)
973
 
            bzrlib.ui.ui_factory = old_ui_factory
974
 
            if cwd is not None:
975
 
                os.chdir(cwd)
976
 
 
977
295
        out = stdout.getvalue()
978
296
        err = stderr.getvalue()
979
297
        if out:
980
 
            self.log('output:\n%r', out)
 
298
            self.log('output:\n%s', out)
981
299
        if err:
982
 
            self.log('errors:\n%r', err)
 
300
            self.log('errors:\n%s', err)
983
301
        if retcode is not None:
984
 
            self.assertEquals(retcode, result)
 
302
            self.assertEquals(result, retcode)
985
303
        return out, err
986
304
 
987
305
    def run_bzr(self, *args, **kwargs):
993
311
 
994
312
        This sends the stdout/stderr results into the test's log,
995
313
        where it may be useful for debugging.  See also run_captured.
996
 
 
997
 
        :param stdin: A string to be used as stdin for the command.
998
314
        """
999
315
        retcode = kwargs.pop('retcode', 0)
1000
 
        encoding = kwargs.pop('encoding', None)
1001
 
        stdin = kwargs.pop('stdin', None)
1002
 
        working_dir = kwargs.pop('working_dir', None)
1003
 
        return self.run_bzr_captured(args, retcode=retcode, encoding=encoding,
1004
 
                                     stdin=stdin, working_dir=working_dir)
1005
 
 
1006
 
    def run_bzr_decode(self, *args, **kwargs):
1007
 
        if 'encoding' in kwargs:
1008
 
            encoding = kwargs['encoding']
1009
 
        else:
1010
 
            encoding = bzrlib.user_encoding
1011
 
        return self.run_bzr(*args, **kwargs)[0].decode(encoding)
1012
 
 
1013
 
    def run_bzr_error(self, error_regexes, *args, **kwargs):
1014
 
        """Run bzr, and check that stderr contains the supplied regexes
1015
 
        
1016
 
        :param error_regexes: Sequence of regular expressions which 
1017
 
            must each be found in the error output. The relative ordering
1018
 
            is not enforced.
1019
 
        :param args: command-line arguments for bzr
1020
 
        :param kwargs: Keyword arguments which are interpreted by run_bzr
1021
 
            This function changes the default value of retcode to be 3,
1022
 
            since in most cases this is run when you expect bzr to fail.
1023
 
        :return: (out, err) The actual output of running the command (in case you
1024
 
                 want to do more inspection)
1025
 
 
1026
 
        Examples of use:
1027
 
            # Make sure that commit is failing because there is nothing to do
1028
 
            self.run_bzr_error(['no changes to commit'],
1029
 
                               'commit', '-m', 'my commit comment')
1030
 
            # Make sure --strict is handling an unknown file, rather than
1031
 
            # giving us the 'nothing to do' error
1032
 
            self.build_tree(['unknown'])
1033
 
            self.run_bzr_error(['Commit refused because there are unknown files'],
1034
 
                               'commit', '--strict', '-m', 'my commit comment')
1035
 
        """
1036
 
        kwargs.setdefault('retcode', 3)
1037
 
        out, err = self.run_bzr(*args, **kwargs)
1038
 
        for regex in error_regexes:
1039
 
            self.assertContainsRe(err, regex)
1040
 
        return out, err
1041
 
 
1042
 
    def run_bzr_subprocess(self, *args, **kwargs):
1043
 
        """Run bzr in a subprocess for testing.
1044
 
 
1045
 
        This starts a new Python interpreter and runs bzr in there. 
1046
 
        This should only be used for tests that have a justifiable need for
1047
 
        this isolation: e.g. they are testing startup time, or signal
1048
 
        handling, or early startup code, etc.  Subprocess code can't be 
1049
 
        profiled or debugged so easily.
1050
 
 
1051
 
        :param retcode: The status code that is expected.  Defaults to 0.  If
1052
 
            None is supplied, the status code is not checked.
1053
 
        :param env_changes: A dictionary which lists changes to environment
1054
 
            variables. A value of None will unset the env variable.
1055
 
            The values must be strings. The change will only occur in the
1056
 
            child, so you don't need to fix the environment after running.
1057
 
        :param universal_newlines: Convert CRLF => LF
1058
 
        :param allow_plugins: By default the subprocess is run with
1059
 
            --no-plugins to ensure test reproducibility. Also, it is possible
1060
 
            for system-wide plugins to create unexpected output on stderr,
1061
 
            which can cause unnecessary test failures.
1062
 
        """
1063
 
        env_changes = kwargs.get('env_changes', {})
1064
 
        working_dir = kwargs.get('working_dir', None)
1065
 
        allow_plugins = kwargs.get('allow_plugins', False)
1066
 
        process = self.start_bzr_subprocess(args, env_changes=env_changes,
1067
 
                                            working_dir=working_dir,
1068
 
                                            allow_plugins=allow_plugins)
1069
 
        # We distinguish between retcode=None and retcode not passed.
1070
 
        supplied_retcode = kwargs.get('retcode', 0)
1071
 
        return self.finish_bzr_subprocess(process, retcode=supplied_retcode,
1072
 
            universal_newlines=kwargs.get('universal_newlines', False),
1073
 
            process_args=args)
1074
 
 
1075
 
    def start_bzr_subprocess(self, process_args, env_changes=None,
1076
 
                             skip_if_plan_to_signal=False,
1077
 
                             working_dir=None,
1078
 
                             allow_plugins=False):
1079
 
        """Start bzr in a subprocess for testing.
1080
 
 
1081
 
        This starts a new Python interpreter and runs bzr in there.
1082
 
        This should only be used for tests that have a justifiable need for
1083
 
        this isolation: e.g. they are testing startup time, or signal
1084
 
        handling, or early startup code, etc.  Subprocess code can't be
1085
 
        profiled or debugged so easily.
1086
 
 
1087
 
        :param process_args: a list of arguments to pass to the bzr executable,
1088
 
            for example `['--version']`.
1089
 
        :param env_changes: A dictionary which lists changes to environment
1090
 
            variables. A value of None will unset the env variable.
1091
 
            The values must be strings. The change will only occur in the
1092
 
            child, so you don't need to fix the environment after running.
1093
 
        :param skip_if_plan_to_signal: raise TestSkipped when true and os.kill
1094
 
            is not available.
1095
 
        :param allow_plugins: If False (default) pass --no-plugins to bzr.
1096
 
 
1097
 
        :returns: Popen object for the started process.
1098
 
        """
1099
 
        if skip_if_plan_to_signal:
1100
 
            if not getattr(os, 'kill', None):
1101
 
                raise TestSkipped("os.kill not available.")
1102
 
 
1103
 
        if env_changes is None:
1104
 
            env_changes = {}
1105
 
        old_env = {}
1106
 
 
1107
 
        def cleanup_environment():
1108
 
            for env_var, value in env_changes.iteritems():
1109
 
                old_env[env_var] = osutils.set_or_unset_env(env_var, value)
1110
 
 
1111
 
        def restore_environment():
1112
 
            for env_var, value in old_env.iteritems():
1113
 
                osutils.set_or_unset_env(env_var, value)
1114
 
 
1115
 
        bzr_path = self.get_bzr_path()
1116
 
 
1117
 
        cwd = None
1118
 
        if working_dir is not None:
1119
 
            cwd = osutils.getcwd()
1120
 
            os.chdir(working_dir)
1121
 
 
1122
 
        try:
1123
 
            # win32 subprocess doesn't support preexec_fn
1124
 
            # so we will avoid using it on all platforms, just to
1125
 
            # make sure the code path is used, and we don't break on win32
1126
 
            cleanup_environment()
1127
 
            command = [sys.executable, bzr_path]
1128
 
            if not allow_plugins:
1129
 
                command.append('--no-plugins')
1130
 
            command.extend(process_args)
1131
 
            process = self._popen(command, stdin=PIPE, stdout=PIPE, stderr=PIPE)
1132
 
        finally:
1133
 
            restore_environment()
1134
 
            if cwd is not None:
1135
 
                os.chdir(cwd)
1136
 
 
1137
 
        return process
1138
 
 
1139
 
    def _popen(self, *args, **kwargs):
1140
 
        """Place a call to Popen.
1141
 
 
1142
 
        Allows tests to override this method to intercept the calls made to
1143
 
        Popen for introspection.
1144
 
        """
1145
 
        return Popen(*args, **kwargs)
1146
 
 
1147
 
    def get_bzr_path(self):
1148
 
        """Return the path of the 'bzr' executable for this test suite."""
1149
 
        bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
1150
 
        if not os.path.isfile(bzr_path):
1151
 
            # We are probably installed. Assume sys.argv is the right file
1152
 
            bzr_path = sys.argv[0]
1153
 
        return bzr_path
1154
 
 
1155
 
    def finish_bzr_subprocess(self, process, retcode=0, send_signal=None,
1156
 
                              universal_newlines=False, process_args=None):
1157
 
        """Finish the execution of process.
1158
 
 
1159
 
        :param process: the Popen object returned from start_bzr_subprocess.
1160
 
        :param retcode: The status code that is expected.  Defaults to 0.  If
1161
 
            None is supplied, the status code is not checked.
1162
 
        :param send_signal: an optional signal to send to the process.
1163
 
        :param universal_newlines: Convert CRLF => LF
1164
 
        :returns: (stdout, stderr)
1165
 
        """
1166
 
        if send_signal is not None:
1167
 
            os.kill(process.pid, send_signal)
1168
 
        out, err = process.communicate()
1169
 
 
1170
 
        if universal_newlines:
1171
 
            out = out.replace('\r\n', '\n')
1172
 
            err = err.replace('\r\n', '\n')
1173
 
 
1174
 
        if retcode is not None and retcode != process.returncode:
1175
 
            if process_args is None:
1176
 
                process_args = "(unknown args)"
1177
 
            mutter('Output of bzr %s:\n%s', process_args, out)
1178
 
            mutter('Error for bzr %s:\n%s', process_args, err)
1179
 
            self.fail('Command bzr %s failed with retcode %s != %s'
1180
 
                      % (process_args, retcode, process.returncode))
1181
 
        return [out, err]
 
316
        return self.run_bzr_captured(args, retcode)
1182
317
 
1183
318
    def check_inventory_shape(self, inv, shape):
1184
319
        """Compare an inventory to a list of expected names.
1210
345
        if stdin is None:
1211
346
            stdin = StringIO("")
1212
347
        if stdout is None:
1213
 
            if getattr(self, "_log_file", None) is not None:
 
348
            if hasattr(self, "_log_file"):
1214
349
                stdout = self._log_file
1215
350
            else:
1216
351
                stdout = StringIO()
1217
352
        if stderr is None:
1218
 
            if getattr(self, "_log_file", None is not None):
 
353
            if hasattr(self, "_log_file"):
1219
354
                stderr = self._log_file
1220
355
            else:
1221
356
                stderr = StringIO()
1232
367
            sys.stderr = real_stderr
1233
368
            sys.stdin = real_stdin
1234
369
 
1235
 
    @symbol_versioning.deprecated_method(symbol_versioning.zero_eleven)
1236
 
    def merge(self, branch_from, wt_to):
1237
 
        """A helper for tests to do a ui-less merge.
1238
 
 
1239
 
        This should move to the main library when someone has time to integrate
1240
 
        it in.
1241
 
        """
1242
 
        # minimal ui-less merge.
1243
 
        wt_to.branch.fetch(branch_from)
1244
 
        base_rev = common_ancestor(branch_from.last_revision(),
1245
 
                                   wt_to.branch.last_revision(),
1246
 
                                   wt_to.branch.repository)
1247
 
        merge_inner(wt_to.branch, branch_from.basis_tree(),
1248
 
                    wt_to.branch.repository.revision_tree(base_rev),
1249
 
                    this_tree=wt_to)
1250
 
        wt_to.add_parent_tree_id(branch_from.last_revision())
1251
 
 
1252
370
 
1253
371
BzrTestBase = TestCase
1254
372
 
1255
 
 
1256
 
class TestCaseWithMemoryTransport(TestCase):
1257
 
    """Common test class for tests that do not need disk resources.
1258
 
 
1259
 
    Tests that need disk resources should derive from TestCaseWithTransport.
1260
 
 
1261
 
    TestCaseWithMemoryTransport sets the TEST_ROOT variable for all bzr tests.
1262
 
 
1263
 
    For TestCaseWithMemoryTransport the test_home_dir is set to the name of
1264
 
    a directory which does not exist. This serves to help ensure test isolation
1265
 
    is preserved. test_dir is set to the TEST_ROOT, as is cwd, because they
1266
 
    must exist. However, TestCaseWithMemoryTransport does not offer local
1267
 
    file defaults for the transport in tests, nor does it obey the command line
1268
 
    override, so tests that accidentally write to the common directory should
1269
 
    be rare.
1270
 
    """
1271
 
 
1272
 
    TEST_ROOT = None
1273
 
    _TEST_NAME = 'test'
1274
 
 
1275
 
 
1276
 
    def __init__(self, methodName='runTest'):
1277
 
        # allow test parameterisation after test construction and before test
1278
 
        # execution. Variables that the parameteriser sets need to be 
1279
 
        # ones that are not set by setUp, or setUp will trash them.
1280
 
        super(TestCaseWithMemoryTransport, self).__init__(methodName)
1281
 
        self.transport_server = default_transport
1282
 
        self.transport_readonly_server = None
1283
 
 
1284
 
    def get_transport(self):
1285
 
        """Return a writeable transport for the test scratch space"""
1286
 
        t = get_transport(self.get_url())
1287
 
        self.assertFalse(t.is_readonly())
1288
 
        return t
1289
 
 
1290
 
    def get_readonly_transport(self):
1291
 
        """Return a readonly transport for the test scratch space
1292
 
        
1293
 
        This can be used to test that operations which should only need
1294
 
        readonly access in fact do not try to write.
1295
 
        """
1296
 
        t = get_transport(self.get_readonly_url())
1297
 
        self.assertTrue(t.is_readonly())
1298
 
        return t
1299
 
 
1300
 
    def create_transport_readonly_server(self):
1301
 
        """Create a transport server from class defined at init.
1302
 
 
1303
 
        This is mostly a hook for daughter classes.
1304
 
        """
1305
 
        return self.transport_readonly_server()
1306
 
 
1307
 
    def get_readonly_server(self):
1308
 
        """Get the server instance for the readonly transport
1309
 
 
1310
 
        This is useful for some tests with specific servers to do diagnostics.
1311
 
        """
1312
 
        if self.__readonly_server is None:
1313
 
            if self.transport_readonly_server is None:
1314
 
                # readonly decorator requested
1315
 
                # bring up the server
1316
 
                self.get_url()
1317
 
                self.__readonly_server = ReadonlyServer()
1318
 
                self.__readonly_server.setUp(self.__server)
1319
 
            else:
1320
 
                self.__readonly_server = self.create_transport_readonly_server()
1321
 
                self.__readonly_server.setUp()
1322
 
            self.addCleanup(self.__readonly_server.tearDown)
1323
 
        return self.__readonly_server
1324
 
 
1325
 
    def get_readonly_url(self, relpath=None):
1326
 
        """Get a URL for the readonly transport.
1327
 
 
1328
 
        This will either be backed by '.' or a decorator to the transport 
1329
 
        used by self.get_url()
1330
 
        relpath provides for clients to get a path relative to the base url.
1331
 
        These should only be downwards relative, not upwards.
1332
 
        """
1333
 
        base = self.get_readonly_server().get_url()
1334
 
        if relpath is not None:
1335
 
            if not base.endswith('/'):
1336
 
                base = base + '/'
1337
 
            base = base + relpath
1338
 
        return base
1339
 
 
1340
 
    def get_server(self):
1341
 
        """Get the read/write server instance.
1342
 
 
1343
 
        This is useful for some tests with specific servers that need
1344
 
        diagnostics.
1345
 
 
1346
 
        For TestCaseWithMemoryTransport this is always a MemoryServer, and there
1347
 
        is no means to override it.
1348
 
        """
1349
 
        if self.__server is None:
1350
 
            self.__server = MemoryServer()
1351
 
            self.__server.setUp()
1352
 
            self.addCleanup(self.__server.tearDown)
1353
 
        return self.__server
1354
 
 
1355
 
    def get_url(self, relpath=None):
1356
 
        """Get a URL (or maybe a path) for the readwrite transport.
1357
 
 
1358
 
        This will either be backed by '.' or to an equivalent non-file based
1359
 
        facility.
1360
 
        relpath provides for clients to get a path relative to the base url.
1361
 
        These should only be downwards relative, not upwards.
1362
 
        """
1363
 
        base = self.get_server().get_url()
1364
 
        if relpath is not None and relpath != '.':
1365
 
            if not base.endswith('/'):
1366
 
                base = base + '/'
1367
 
            # XXX: Really base should be a url; we did after all call
1368
 
            # get_url()!  But sometimes it's just a path (from
1369
 
            # LocalAbspathServer), and it'd be wrong to append urlescaped data
1370
 
            # to a non-escaped local path.
1371
 
            if base.startswith('./') or base.startswith('/'):
1372
 
                base += relpath
1373
 
            else:
1374
 
                base += urlutils.escape(relpath)
1375
 
        return base
1376
 
 
1377
 
    def _make_test_root(self):
1378
 
        if TestCaseWithMemoryTransport.TEST_ROOT is not None:
1379
 
            return
1380
 
        i = 0
1381
 
        while True:
1382
 
            root = u'test%04d.tmp' % i
1383
 
            try:
1384
 
                os.mkdir(root)
1385
 
            except OSError, e:
1386
 
                if e.errno == errno.EEXIST:
1387
 
                    i += 1
1388
 
                    continue
1389
 
                else:
1390
 
                    raise
1391
 
            # successfully created
1392
 
            TestCaseWithMemoryTransport.TEST_ROOT = osutils.abspath(root)
1393
 
            break
1394
 
        # make a fake bzr directory there to prevent any tests propagating
1395
 
        # up onto the source directory's real branch
1396
 
        bzrdir.BzrDir.create_standalone_workingtree(
1397
 
            TestCaseWithMemoryTransport.TEST_ROOT)
1398
 
 
1399
 
    def makeAndChdirToTestDir(self):
1400
 
        """Create a temporary directories for this one test.
1401
 
        
1402
 
        This must set self.test_home_dir and self.test_dir and chdir to
1403
 
        self.test_dir.
1404
 
        
1405
 
        For TestCaseWithMemoryTransport we chdir to the TEST_ROOT for this test.
1406
 
        """
1407
 
        os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
1408
 
        self.test_dir = TestCaseWithMemoryTransport.TEST_ROOT
1409
 
        self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
1410
 
        
1411
 
    def make_branch(self, relpath, format=None):
1412
 
        """Create a branch on the transport at relpath."""
1413
 
        repo = self.make_repository(relpath, format=format)
1414
 
        return repo.bzrdir.create_branch()
1415
 
 
1416
 
    def make_bzrdir(self, relpath, format=None):
1417
 
        try:
1418
 
            # might be a relative or absolute path
1419
 
            maybe_a_url = self.get_url(relpath)
1420
 
            segments = maybe_a_url.rsplit('/', 1)
1421
 
            t = get_transport(maybe_a_url)
1422
 
            if len(segments) > 1 and segments[-1] not in ('', '.'):
1423
 
                try:
1424
 
                    t.mkdir('.')
1425
 
                except errors.FileExists:
1426
 
                    pass
1427
 
            if isinstance(format, basestring):
1428
 
                format = bzrdir.format_registry.make_bzrdir(format)
1429
 
            elif format is None:
1430
 
                format = bzrlib.bzrdir.BzrDirFormat.get_default_format()
1431
 
            return format.initialize_on_transport(t)
1432
 
        except errors.UninitializableFormat:
1433
 
            raise TestSkipped("Format %s is not initializable." % format)
1434
 
 
1435
 
    def make_repository(self, relpath, shared=False, format=None):
1436
 
        """Create a repository on our default transport at relpath."""
1437
 
        made_control = self.make_bzrdir(relpath, format=format)
1438
 
        return made_control.create_repository(shared=shared)
1439
 
 
1440
 
    def make_branch_and_memory_tree(self, relpath, format=None):
1441
 
        """Create a branch on the default transport and a MemoryTree for it."""
1442
 
        b = self.make_branch(relpath, format=format)
1443
 
        return memorytree.MemoryTree.create_on_branch(b)
1444
 
 
1445
 
    def overrideEnvironmentForTesting(self):
1446
 
        os.environ['HOME'] = self.test_home_dir
1447
 
        os.environ['APPDATA'] = self.test_home_dir
1448
 
        
1449
 
    def setUp(self):
1450
 
        super(TestCaseWithMemoryTransport, self).setUp()
1451
 
        self._make_test_root()
1452
 
        _currentdir = os.getcwdu()
1453
 
        def _leaveDirectory():
1454
 
            os.chdir(_currentdir)
1455
 
        self.addCleanup(_leaveDirectory)
1456
 
        self.makeAndChdirToTestDir()
1457
 
        self.overrideEnvironmentForTesting()
1458
 
        self.__readonly_server = None
1459
 
        self.__server = None
1460
 
 
1461
373
     
1462
 
class TestCaseInTempDir(TestCaseWithMemoryTransport):
 
374
class TestCaseInTempDir(TestCase):
1463
375
    """Derived class that runs a test within a temporary directory.
1464
376
 
1465
377
    This is useful for tests that need to create a branch, etc.
1472
384
    InTempDir is an old alias for FunctionalTestCase.
1473
385
    """
1474
386
 
 
387
    TEST_ROOT = None
 
388
    _TEST_NAME = 'test'
1475
389
    OVERRIDE_PYTHON = 'python'
1476
390
 
1477
391
    def check_file_contents(self, filename, expect):
1482
396
            self.log("actually: %r" % contents)
1483
397
            self.fail("contents of %s not as expected" % filename)
1484
398
 
1485
 
    def makeAndChdirToTestDir(self):
1486
 
        """See TestCaseWithMemoryTransport.makeAndChdirToTestDir().
1487
 
        
1488
 
        For TestCaseInTempDir we create a temporary directory based on the test
1489
 
        name and then create two subdirs - test and home under it.
1490
 
        """
1491
 
        # shorten the name, to avoid test failures due to path length
1492
 
        short_id = self.id().replace('bzrlib.tests.', '') \
1493
 
                   .replace('__main__.', '')[-100:]
1494
 
        # it's possible the same test class is run several times for
1495
 
        # parameterized tests, so make sure the names don't collide.  
 
399
    def _make_test_root(self):
 
400
        if TestCaseInTempDir.TEST_ROOT is not None:
 
401
            return
1496
402
        i = 0
1497
403
        while True:
1498
 
            if i > 0:
1499
 
                candidate_dir = '%s/%s.%d' % (self.TEST_ROOT, short_id, i)
1500
 
            else:
1501
 
                candidate_dir = '%s/%s' % (self.TEST_ROOT, short_id)
1502
 
            if os.path.exists(candidate_dir):
1503
 
                i = i + 1
1504
 
                continue
1505
 
            else:
1506
 
                os.mkdir(candidate_dir)
1507
 
                self.test_home_dir = candidate_dir + '/home'
1508
 
                os.mkdir(self.test_home_dir)
1509
 
                self.test_dir = candidate_dir + '/work'
1510
 
                os.mkdir(self.test_dir)
1511
 
                os.chdir(self.test_dir)
1512
 
                break
1513
 
 
1514
 
    def build_tree(self, shape, line_endings='binary', transport=None):
 
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)
 
426
        os.mkdir(self.test_dir)
 
427
        os.chdir(self.test_dir)
 
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):
1515
435
        """Build a test tree according to a pattern.
1516
436
 
1517
437
        shape is a sequence of file specifications.  If the final
1518
438
        character is '/', a directory is created.
1519
439
 
1520
 
        This assumes that all the elements in the tree being built are new.
1521
 
 
1522
440
        This doesn't add anything to a branch.
1523
 
        :param line_endings: Either 'binary' or 'native'
1524
 
                             in binary mode, exact contents are written
1525
 
                             in native mode, the line endings match the
1526
 
                             default platform endings.
1527
 
 
1528
 
        :param transport: A transport to write to, for building trees on 
1529
 
                          VFS's. If the transport is readonly or None,
1530
 
                          "." is opened automatically.
1531
441
        """
1532
 
        # It's OK to just create them using forward slashes on windows.
1533
 
        if transport is None or transport.is_readonly():
1534
 
            transport = get_transport(".")
 
442
        # XXX: It's OK to just create them using forward slashes on windows?
1535
443
        for name in shape:
1536
 
            self.assert_(isinstance(name, basestring))
 
444
            assert isinstance(name, basestring)
1537
445
            if name[-1] == '/':
1538
 
                transport.mkdir(urlutils.escape(name[:-1]))
 
446
                os.mkdir(name[:-1])
1539
447
            else:
1540
 
                if line_endings == 'binary':
1541
 
                    end = '\n'
1542
 
                elif line_endings == 'native':
1543
 
                    end = os.linesep
1544
 
                else:
1545
 
                    raise errors.BzrError(
1546
 
                        'Invalid line ending request %r' % line_endings)
1547
 
                content = "contents of %s%s" % (name.encode('utf-8'), end)
1548
 
                transport.put_bytes_non_atomic(urlutils.escape(name), content)
 
448
                f = file(name, 'wt')
 
449
                print >>f, "contents of", name
 
450
                f.close()
1549
451
 
1550
452
    def build_tree_contents(self, shape):
1551
 
        build_tree_contents(shape)
 
453
        bzrlib.selftest.build_tree_contents(shape)
1552
454
 
 
455
    def failUnlessExists(self, path):
 
456
        """Fail unless path, which may be abs or relative, exists."""
 
457
        self.failUnless(osutils.lexists(path))
 
458
        
1553
459
    def assertFileEqual(self, content, path):
1554
460
        """Fail if path does not contain 'content'."""
1555
 
        self.failUnlessExists(path)
1556
 
        # TODO: jam 20060427 Shouldn't this be 'rb'?
 
461
        self.failUnless(osutils.lexists(path))
1557
462
        self.assertEqualDiff(content, open(path, 'r').read())
1558
 
 
1559
 
    def failUnlessExists(self, path):
1560
 
        """Fail unless path, which may be abs or relative, exists."""
1561
 
        self.failUnless(osutils.lexists(path),path+" does not exist")
1562
 
 
1563
 
    def failIfExists(self, path):
1564
 
        """Fail if path, which may be abs or relative, exists."""
1565
 
        self.failIf(osutils.lexists(path),path+" exists")
1566
 
 
1567
 
 
1568
 
class TestCaseWithTransport(TestCaseInTempDir):
1569
 
    """A test case that provides get_url and get_readonly_url facilities.
1570
 
 
1571
 
    These back onto two transport servers, one for readonly access and one for
1572
 
    read write access.
1573
 
 
1574
 
    If no explicit class is provided for readonly access, a
1575
 
    ReadonlyTransportDecorator is used instead which allows the use of non disk
1576
 
    based read write transports.
1577
 
 
1578
 
    If an explicit class is provided for readonly access, that server and the 
1579
 
    readwrite one must both define get_url() as resolving to os.getcwd().
1580
 
    """
1581
 
 
1582
 
    def create_transport_server(self):
1583
 
        """Create a transport server from class defined at init.
1584
 
 
1585
 
        This is mostly a hook for daughter classes.
1586
 
        """
1587
 
        return self.transport_server()
1588
 
 
1589
 
    def get_server(self):
1590
 
        """See TestCaseWithMemoryTransport.
1591
 
 
1592
 
        This is useful for some tests with specific servers that need
1593
 
        diagnostics.
1594
 
        """
1595
 
        if self.__server is None:
1596
 
            self.__server = self.create_transport_server()
1597
 
            self.__server.setUp()
1598
 
            self.addCleanup(self.__server.tearDown)
1599
 
        return self.__server
1600
 
 
1601
 
    def make_branch_and_tree(self, relpath, format=None):
1602
 
        """Create a branch on the transport and a tree locally.
1603
 
 
1604
 
        If the transport is not a LocalTransport, the Tree can't be created on
1605
 
        the transport.  In that case the working tree is created in the local
1606
 
        directory, and the returned tree's branch and repository will also be
1607
 
        accessed locally.
1608
 
 
1609
 
        This will fail if the original default transport for this test
1610
 
        case wasn't backed by the working directory, as the branch won't
1611
 
        be on disk for us to open it.  
1612
 
 
1613
 
        :param format: The BzrDirFormat.
1614
 
        :returns: the WorkingTree.
1615
 
        """
1616
 
        # TODO: always use the local disk path for the working tree,
1617
 
        # this obviously requires a format that supports branch references
1618
 
        # so check for that by checking bzrdir.BzrDirFormat.get_default_format()
1619
 
        # RBC 20060208
1620
 
        b = self.make_branch(relpath, format=format)
1621
 
        try:
1622
 
            return b.bzrdir.create_workingtree()
1623
 
        except errors.NotLocalUrl:
1624
 
            # We can only make working trees locally at the moment.  If the
1625
 
            # transport can't support them, then reopen the branch on a local
1626
 
            # transport, and create the working tree there.  
1627
 
            #
1628
 
            # Possibly we should instead keep
1629
 
            # the non-disk-backed branch and create a local checkout?
1630
 
            bd = bzrdir.BzrDir.open(relpath)
1631
 
            return bd.create_workingtree()
1632
 
 
1633
 
    def assertIsDirectory(self, relpath, transport):
1634
 
        """Assert that relpath within transport is a directory.
1635
 
 
1636
 
        This may not be possible on all transports; in that case it propagates
1637
 
        a TransportNotPossible.
1638
 
        """
1639
 
        try:
1640
 
            mode = transport.stat(relpath).st_mode
1641
 
        except errors.NoSuchFile:
1642
 
            self.fail("path %s is not a directory; no such file"
1643
 
                      % (relpath))
1644
 
        if not stat.S_ISDIR(mode):
1645
 
            self.fail("path %s is not a directory; has mode %#o"
1646
 
                      % (relpath, mode))
1647
 
 
1648
 
    def setUp(self):
1649
 
        super(TestCaseWithTransport, self).setUp()
1650
 
        self.__server = None
1651
 
 
1652
 
 
1653
 
class ChrootedTestCase(TestCaseWithTransport):
1654
 
    """A support class that provides readonly urls outside the local namespace.
1655
 
 
1656
 
    This is done by checking if self.transport_server is a MemoryServer. if it
1657
 
    is then we are chrooted already, if it is not then an HttpServer is used
1658
 
    for readonly urls.
1659
 
 
1660
 
    TODO RBC 20060127: make this an option to TestCaseWithTransport so it can
1661
 
                       be used without needed to redo it when a different 
1662
 
                       subclass is in use ?
1663
 
    """
1664
 
 
1665
 
    def setUp(self):
1666
 
        super(ChrootedTestCase, self).setUp()
1667
 
        if not self.transport_server == MemoryServer:
1668
 
            self.transport_readonly_server = HttpServer
 
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
1669
472
 
1670
473
 
1671
474
def filter_suite_by_re(suite, pattern):
1677
480
    return result
1678
481
 
1679
482
 
1680
 
def sort_suite_by_re(suite, pattern):
1681
 
    first = []
1682
 
    second = []
1683
 
    filter_re = re.compile(pattern)
1684
 
    for test in iter_suite_tests(suite):
1685
 
        if filter_re.search(test.id()):
1686
 
            first.append(test)
1687
 
        else:
1688
 
            second.append(test)
1689
 
    return TestUtil.TestSuite(first + second)
1690
 
 
1691
 
 
1692
483
def run_suite(suite, name='test', verbose=False, pattern=".*",
1693
 
              stop_on_failure=False, keep_output=False,
1694
 
              transport=None, lsprof_timed=None, bench_history=None,
1695
 
              matching_tests_first=None):
1696
 
    TestCase._gather_lsprof_in_benchmarks = lsprof_timed
 
484
              stop_on_failure=False):
 
485
    TestCaseInTempDir._TEST_NAME = name
1697
486
    if verbose:
1698
487
        verbosity = 2
1699
488
    else:
1700
489
        verbosity = 1
1701
490
    runner = TextTestRunner(stream=sys.stdout,
1702
491
                            descriptions=0,
1703
 
                            verbosity=verbosity,
1704
 
                            keep_output=keep_output,
1705
 
                            bench_history=bench_history)
 
492
                            verbosity=verbosity)
1706
493
    runner.stop_on_failure=stop_on_failure
1707
494
    if pattern != '.*':
1708
 
        if matching_tests_first:
1709
 
            suite = sort_suite_by_re(suite, pattern)
1710
 
        else:
1711
 
            suite = filter_suite_by_re(suite, pattern)
 
495
        suite = filter_suite_by_re(suite, pattern)
1712
496
    result = runner.run(suite)
 
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) 
 
503
    else:
 
504
        print "Failed tests working directories are in '%s'\n" % TestCaseInTempDir.TEST_ROOT
1713
505
    return result.wasSuccessful()
1714
506
 
1715
507
 
1716
 
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
1717
 
             keep_output=False,
1718
 
             transport=None,
1719
 
             test_suite_factory=None,
1720
 
             lsprof_timed=None,
1721
 
             bench_history=None,
1722
 
             matching_tests_first=None):
 
508
def selftest(verbose=False, pattern=".*", stop_on_failure=True):
1723
509
    """Run the whole test suite under the enhanced runner"""
1724
 
    # XXX: Very ugly way to do this...
1725
 
    # Disable warning about old formats because we don't want it to disturb
1726
 
    # any blackbox tests.
1727
 
    from bzrlib import repository
1728
 
    repository._deprecation_warning_done = True
1729
 
 
1730
 
    global default_transport
1731
 
    if transport is None:
1732
 
        transport = default_transport
1733
 
    old_transport = default_transport
1734
 
    default_transport = transport
1735
 
    try:
1736
 
        if test_suite_factory is None:
1737
 
            suite = test_suite()
1738
 
        else:
1739
 
            suite = test_suite_factory()
1740
 
        return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
1741
 
                     stop_on_failure=stop_on_failure, keep_output=keep_output,
1742
 
                     transport=transport,
1743
 
                     lsprof_timed=lsprof_timed,
1744
 
                     bench_history=bench_history,
1745
 
                     matching_tests_first=matching_tests_first)
1746
 
    finally:
1747
 
        default_transport = old_transport
 
510
    return run_suite(test_suite(), 'testbzr', verbose=verbose, pattern=pattern,
 
511
                     stop_on_failure=stop_on_failure)
1748
512
 
1749
513
 
1750
514
def test_suite():
1751
 
    """Build and return TestSuite for the whole of bzrlib.
1752
 
    
1753
 
    This function can be replaced if you need to change the default test
1754
 
    suite on a global basis, but it is not encouraged.
1755
 
    """
1756
 
    testmod_names = [
1757
 
                   'bzrlib.tests.test_ancestry',
1758
 
                   'bzrlib.tests.test_annotate',
1759
 
                   'bzrlib.tests.test_api',
1760
 
                   'bzrlib.tests.test_atomicfile',
1761
 
                   'bzrlib.tests.test_bad_files',
1762
 
                   'bzrlib.tests.test_branch',
1763
 
                   'bzrlib.tests.test_bundle',
1764
 
                   'bzrlib.tests.test_bzrdir',
1765
 
                   'bzrlib.tests.test_cache_utf8',
1766
 
                   'bzrlib.tests.test_commands',
1767
 
                   'bzrlib.tests.test_commit',
1768
 
                   'bzrlib.tests.test_commit_merge',
1769
 
                   'bzrlib.tests.test_config',
1770
 
                   'bzrlib.tests.test_conflicts',
1771
 
                   'bzrlib.tests.test_decorators',
1772
 
                   'bzrlib.tests.test_diff',
1773
 
                   'bzrlib.tests.test_doc_generate',
1774
 
                   'bzrlib.tests.test_errors',
1775
 
                   'bzrlib.tests.test_escaped_store',
1776
 
                   'bzrlib.tests.test_fetch',
1777
 
                   'bzrlib.tests.test_ftp_transport',
1778
 
                   'bzrlib.tests.test_generate_docs',
1779
 
                   'bzrlib.tests.test_generate_ids',
1780
 
                   'bzrlib.tests.test_globbing',
1781
 
                   'bzrlib.tests.test_gpg',
1782
 
                   'bzrlib.tests.test_graph',
1783
 
                   'bzrlib.tests.test_hashcache',
1784
 
                   'bzrlib.tests.test_http',
1785
 
                   'bzrlib.tests.test_http_response',
1786
 
                   'bzrlib.tests.test_identitymap',
1787
 
                   'bzrlib.tests.test_ignores',
1788
 
                   'bzrlib.tests.test_inv',
1789
 
                   'bzrlib.tests.test_knit',
1790
 
                   'bzrlib.tests.test_lazy_import',
1791
 
                   'bzrlib.tests.test_lazy_regex',
1792
 
                   'bzrlib.tests.test_lockdir',
1793
 
                   'bzrlib.tests.test_lockable_files',
1794
 
                   'bzrlib.tests.test_log',
1795
 
                   'bzrlib.tests.test_memorytree',
1796
 
                   'bzrlib.tests.test_merge',
1797
 
                   'bzrlib.tests.test_merge3',
1798
 
                   'bzrlib.tests.test_merge_core',
1799
 
                   'bzrlib.tests.test_missing',
1800
 
                   'bzrlib.tests.test_msgeditor',
1801
 
                   'bzrlib.tests.test_nonascii',
1802
 
                   'bzrlib.tests.test_options',
1803
 
                   'bzrlib.tests.test_osutils',
1804
 
                   'bzrlib.tests.test_osutils_encodings',
1805
 
                   'bzrlib.tests.test_patch',
1806
 
                   'bzrlib.tests.test_patches',
1807
 
                   'bzrlib.tests.test_permissions',
1808
 
                   'bzrlib.tests.test_plugins',
1809
 
                   'bzrlib.tests.test_progress',
1810
 
                   'bzrlib.tests.test_reconcile',
1811
 
                   'bzrlib.tests.test_registry',
1812
 
                   'bzrlib.tests.test_repository',
1813
 
                   'bzrlib.tests.test_revert',
1814
 
                   'bzrlib.tests.test_revision',
1815
 
                   'bzrlib.tests.test_revisionnamespaces',
1816
 
                   'bzrlib.tests.test_revisiontree',
1817
 
                   'bzrlib.tests.test_rio',
1818
 
                   'bzrlib.tests.test_sampler',
1819
 
                   'bzrlib.tests.test_selftest',
1820
 
                   'bzrlib.tests.test_setup',
1821
 
                   'bzrlib.tests.test_sftp_transport',
1822
 
                   'bzrlib.tests.test_smart_add',
1823
 
                   'bzrlib.tests.test_smart_transport',
1824
 
                   'bzrlib.tests.test_source',
1825
 
                   'bzrlib.tests.test_status',
1826
 
                   'bzrlib.tests.test_store',
1827
 
                   'bzrlib.tests.test_symbol_versioning',
1828
 
                   'bzrlib.tests.test_testament',
1829
 
                   'bzrlib.tests.test_textfile',
1830
 
                   'bzrlib.tests.test_textmerge',
1831
 
                   'bzrlib.tests.test_trace',
1832
 
                   'bzrlib.tests.test_transactions',
1833
 
                   'bzrlib.tests.test_transform',
1834
 
                   'bzrlib.tests.test_transport',
1835
 
                   'bzrlib.tests.test_tree',
1836
 
                   'bzrlib.tests.test_treebuilder',
1837
 
                   'bzrlib.tests.test_tsort',
1838
 
                   'bzrlib.tests.test_tuned_gzip',
1839
 
                   'bzrlib.tests.test_ui',
1840
 
                   'bzrlib.tests.test_upgrade',
1841
 
                   'bzrlib.tests.test_urlutils',
1842
 
                   'bzrlib.tests.test_versionedfile',
1843
 
                   'bzrlib.tests.test_version',
1844
 
                   'bzrlib.tests.test_version_info',
1845
 
                   'bzrlib.tests.test_weave',
1846
 
                   'bzrlib.tests.test_whitebox',
1847
 
                   'bzrlib.tests.test_workingtree',
1848
 
                   'bzrlib.tests.test_wsgi',
1849
 
                   'bzrlib.tests.test_xml',
 
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',
1850
565
                   ]
1851
 
    test_transport_implementations = [
1852
 
        'bzrlib.tests.test_transport_implementations',
1853
 
        'bzrlib.tests.test_read_bundle',
1854
 
        ]
1855
 
    suite = TestUtil.TestSuite()
1856
 
    loader = TestUtil.TestLoader()
1857
 
    suite.addTest(loader.loadTestsFromModuleNames(testmod_names))
1858
 
    from bzrlib.transport import TransportTestProviderAdapter
1859
 
    adapter = TransportTestProviderAdapter()
1860
 
    adapt_modules(test_transport_implementations, adapter, loader, suite)
1861
 
    for package in packages_to_test():
1862
 
        suite.addTest(package.test_suite())
 
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))
1863
579
    for m in MODULES_TO_TEST:
1864
 
        suite.addTest(loader.loadTestsFromModule(m))
1865
 
    for m in MODULES_TO_DOCTEST:
1866
 
        try:
1867
 
            suite.addTest(doctest.DocTestSuite(m))
1868
 
        except ValueError, e:
1869
 
            print '**failed to get doctest for: %s\n%s' %(m,e)
1870
 
            raise
1871
 
    for name, plugin in bzrlib.plugin.all_plugins().items():
1872
 
        if getattr(plugin, 'test_suite', None) is not None:
1873
 
            suite.addTest(plugin.test_suite())
 
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())
1874
586
    return suite
1875
587
 
1876
 
 
1877
 
def adapt_modules(mods_list, adapter, loader, suite):
1878
 
    """Adapt the modules in mods_list using adapter and add to suite."""
1879
 
    for test in iter_suite_tests(loader.loadTestsFromModuleNames(mods_list)):
1880
 
        suite.addTests(adapter.adapt(test))
1881
 
 
1882
 
 
1883
 
def clean_selftest_output(root=None, quiet=False):
1884
 
    """Remove all selftest output directories from root directory.
1885
 
 
1886
 
    :param  root:   root directory for clean
1887
 
                    (if ommitted or None then clean current directory).
1888
 
    :param  quiet:  suppress report about deleting directories
1889
 
    """
1890
 
    import re
1891
 
    import shutil
1892
 
 
1893
 
    re_dir = re.compile(r'''test\d\d\d\d\.tmp''')
1894
 
    if root is None:
1895
 
        root = u'.'
1896
 
    for i in os.listdir(root):
1897
 
        if os.path.isdir(i) and re_dir.match(i):
1898
 
            if not quiet:
1899
 
                print 'delete directory:', i
1900
 
            shutil.rmtree(i)