~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/__init__.py

  • Committer: Martin Pool
  • Date: 2005-05-06 03:20:15 UTC
  • Revision ID: mbp@sourcefrog.net-20050506032014-decf4918803147d2
- split out notes on storing annotations in revfiles

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005, 2006 by Canonical Ltd
2
 
#
3
 
# This program is free software; you can redistribute it and/or modify
4
 
# it under the terms of the GNU General Public License as published by
5
 
# the Free Software Foundation; either version 2 of the License, or
6
 
# (at your option) any later version.
7
 
#
8
 
# This program is distributed in the hope that it will be useful,
9
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
 
# GNU General Public License for more details.
12
 
#
13
 
# You should have received a copy of the GNU General Public License
14
 
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
 
 
17
 
 
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
 
from cStringIO import StringIO
31
 
import difflib
32
 
import doctest
33
 
import errno
34
 
import logging
35
 
import os
36
 
import re
37
 
import shlex
38
 
import stat
39
 
from subprocess import Popen, PIPE
40
 
import sys
41
 
import tempfile
42
 
import unittest
43
 
import time
44
 
 
45
 
 
46
 
from bzrlib import memorytree
47
 
import bzrlib.branch
48
 
import bzrlib.bzrdir as bzrdir
49
 
import bzrlib.commands
50
 
import bzrlib.bundle.serializer
51
 
import bzrlib.errors as errors
52
 
import bzrlib.inventory
53
 
import bzrlib.iterablefile
54
 
import bzrlib.lockdir
55
 
try:
56
 
    import bzrlib.lsprof
57
 
except ImportError:
58
 
    # lsprof not available
59
 
    pass
60
 
from bzrlib.merge import merge_inner
61
 
import bzrlib.merge3
62
 
import bzrlib.osutils
63
 
import bzrlib.osutils as osutils
64
 
import bzrlib.plugin
65
 
import bzrlib.progress as progress
66
 
from bzrlib.revision import common_ancestor
67
 
import bzrlib.store
68
 
from bzrlib import symbol_versioning
69
 
import bzrlib.trace
70
 
from bzrlib.transport import get_transport
71
 
import bzrlib.transport
72
 
from bzrlib.transport.local import LocalRelpathServer
73
 
from bzrlib.transport.readonly import ReadonlyServer
74
 
from bzrlib.trace import mutter
75
 
from bzrlib.tests import TestUtil
76
 
from bzrlib.tests.TestUtil import (
77
 
                          TestSuite,
78
 
                          TestLoader,
79
 
                          )
80
 
from bzrlib.tests.treeshape import build_tree_contents
81
 
import bzrlib.urlutils as urlutils
82
 
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2
83
 
 
84
 
default_transport = LocalRelpathServer
85
 
 
86
 
MODULES_TO_TEST = []
87
 
MODULES_TO_DOCTEST = [
88
 
                      bzrlib.branch,
89
 
                      bzrlib.bundle.serializer,
90
 
                      bzrlib.commands,
91
 
                      bzrlib.errors,
92
 
                      bzrlib.inventory,
93
 
                      bzrlib.iterablefile,
94
 
                      bzrlib.lockdir,
95
 
                      bzrlib.merge3,
96
 
                      bzrlib.option,
97
 
                      bzrlib.osutils,
98
 
                      bzrlib.store,
99
 
                      bzrlib.transport,
100
 
                      ]
101
 
 
102
 
 
103
 
def packages_to_test():
104
 
    """Return a list of packages to test.
105
 
 
106
 
    The packages are not globally imported so that import failures are
107
 
    triggered when running selftest, not when importing the command.
108
 
    """
109
 
    import bzrlib.doc
110
 
    import bzrlib.tests.blackbox
111
 
    import bzrlib.tests.branch_implementations
112
 
    import bzrlib.tests.bzrdir_implementations
113
 
    import bzrlib.tests.interrepository_implementations
114
 
    import bzrlib.tests.interversionedfile_implementations
115
 
    import bzrlib.tests.intertree_implementations
116
 
    import bzrlib.tests.repository_implementations
117
 
    import bzrlib.tests.revisionstore_implementations
118
 
    import bzrlib.tests.tree_implementations
119
 
    import bzrlib.tests.workingtree_implementations
120
 
    return [
121
 
            bzrlib.doc,
122
 
            bzrlib.tests.blackbox,
123
 
            bzrlib.tests.branch_implementations,
124
 
            bzrlib.tests.bzrdir_implementations,
125
 
            bzrlib.tests.interrepository_implementations,
126
 
            bzrlib.tests.interversionedfile_implementations,
127
 
            bzrlib.tests.intertree_implementations,
128
 
            bzrlib.tests.repository_implementations,
129
 
            bzrlib.tests.revisionstore_implementations,
130
 
            bzrlib.tests.tree_implementations,
131
 
            bzrlib.tests.workingtree_implementations,
132
 
            ]
133
 
 
134
 
 
135
 
class _MyResult(unittest._TextTestResult):
136
 
    """Custom TestResult.
137
 
 
138
 
    Shows output in a different format, including displaying runtime for tests.
139
 
    """
140
 
    stop_early = False
141
 
    
142
 
    def __init__(self, stream, descriptions, verbosity, pb=None,
143
 
                 bench_history=None):
144
 
        """Construct new TestResult.
145
 
 
146
 
        :param bench_history: Optionally, a writable file object to accumulate
147
 
            benchmark results.
148
 
        """
149
 
        unittest._TextTestResult.__init__(self, stream, descriptions, verbosity)
150
 
        self.pb = pb
151
 
        if bench_history is not None:
152
 
            from bzrlib.version import _get_bzr_source_tree
153
 
            src_tree = _get_bzr_source_tree()
154
 
            if src_tree:
155
 
                try:
156
 
                    revision_id = src_tree.get_parent_ids()[0]
157
 
                except IndexError:
158
 
                    # XXX: if this is a brand new tree, do the same as if there
159
 
                    # is no branch.
160
 
                    revision_id = ''
161
 
            else:
162
 
                # XXX: If there's no branch, what should we do?
163
 
                revision_id = ''
164
 
            bench_history.write("--date %s %s\n" % (time.time(), revision_id))
165
 
        self._bench_history = bench_history
166
 
    
167
 
    def extractBenchmarkTime(self, testCase):
168
 
        """Add a benchmark time for the current test case."""
169
 
        self._benchmarkTime = getattr(testCase, "_benchtime", None)
170
 
    
171
 
    def _elapsedTestTimeString(self):
172
 
        """Return a time string for the overall time the current test has taken."""
173
 
        return self._formatTime(time.time() - self._start_time)
174
 
 
175
 
    def _testTimeString(self):
176
 
        if self._benchmarkTime is not None:
177
 
            return "%s/%s" % (
178
 
                self._formatTime(self._benchmarkTime),
179
 
                self._elapsedTestTimeString())
180
 
        else:
181
 
            return "      %s" % self._elapsedTestTimeString()
182
 
 
183
 
    def _formatTime(self, seconds):
184
 
        """Format seconds as milliseconds with leading spaces."""
185
 
        return "%5dms" % (1000 * seconds)
186
 
 
187
 
    def _ellipsise_unimportant_words(self, a_string, final_width,
188
 
                                   keep_start=False):
189
 
        """Add ellipses (sp?) for overly long strings.
190
 
        
191
 
        :param keep_start: If true preserve the start of a_string rather
192
 
                           than the end of it.
193
 
        """
194
 
        if keep_start:
195
 
            if len(a_string) > final_width:
196
 
                result = a_string[:final_width-3] + '...'
197
 
            else:
198
 
                result = a_string
199
 
        else:
200
 
            if len(a_string) > final_width:
201
 
                result = '...' + a_string[3-final_width:]
202
 
            else:
203
 
                result = a_string
204
 
        return result.ljust(final_width)
205
 
 
206
 
    def startTest(self, test):
207
 
        unittest.TestResult.startTest(self, test)
208
 
        # In a short description, the important words are in
209
 
        # the beginning, but in an id, the important words are
210
 
        # at the end
211
 
        SHOW_DESCRIPTIONS = False
212
 
 
213
 
        if not self.showAll and self.dots and self.pb is not None:
214
 
            final_width = 13
215
 
        else:
216
 
            final_width = osutils.terminal_width()
217
 
            final_width = final_width - 15 - 8
218
 
        what = None
219
 
        if SHOW_DESCRIPTIONS:
220
 
            what = test.shortDescription()
221
 
            if what:
222
 
                what = self._ellipsise_unimportant_words(what, final_width, keep_start=True)
223
 
        if what is None:
224
 
            what = test.id()
225
 
            if what.startswith('bzrlib.tests.'):
226
 
                what = what[13:]
227
 
            what = self._ellipsise_unimportant_words(what, final_width)
228
 
        if self.showAll:
229
 
            self.stream.write(what)
230
 
        elif self.dots and self.pb is not None:
231
 
            self.pb.update(what, self.testsRun - 1, None)
232
 
        self.stream.flush()
233
 
        self._recordTestStartTime()
234
 
 
235
 
    def _recordTestStartTime(self):
236
 
        """Record that a test has started."""
237
 
        self._start_time = time.time()
238
 
 
239
 
    def addError(self, test, err):
240
 
        if isinstance(err[1], TestSkipped):
241
 
            return self.addSkipped(test, err)    
242
 
        unittest.TestResult.addError(self, test, err)
243
 
        self.extractBenchmarkTime(test)
244
 
        if self.showAll:
245
 
            self.stream.writeln("ERROR %s" % self._testTimeString())
246
 
        elif self.dots and self.pb is None:
247
 
            self.stream.write('E')
248
 
        elif self.dots:
249
 
            self.pb.update(self._ellipsise_unimportant_words('ERROR', 13), self.testsRun, None)
250
 
            self.pb.note(self._ellipsise_unimportant_words(
251
 
                            test.id() + ': ERROR',
252
 
                            osutils.terminal_width()))
253
 
        self.stream.flush()
254
 
        if self.stop_early:
255
 
            self.stop()
256
 
 
257
 
    def addFailure(self, test, err):
258
 
        unittest.TestResult.addFailure(self, test, err)
259
 
        self.extractBenchmarkTime(test)
260
 
        if self.showAll:
261
 
            self.stream.writeln(" FAIL %s" % self._testTimeString())
262
 
        elif self.dots and self.pb is None:
263
 
            self.stream.write('F')
264
 
        elif self.dots:
265
 
            self.pb.update(self._ellipsise_unimportant_words('FAIL', 13), self.testsRun, None)
266
 
            self.pb.note(self._ellipsise_unimportant_words(
267
 
                            test.id() + ': FAIL',
268
 
                            osutils.terminal_width()))
269
 
        self.stream.flush()
270
 
        if self.stop_early:
271
 
            self.stop()
272
 
 
273
 
    def addSuccess(self, test):
274
 
        self.extractBenchmarkTime(test)
275
 
        if self._bench_history is not None:
276
 
            if self._benchmarkTime is not None:
277
 
                self._bench_history.write("%s %s\n" % (
278
 
                    self._formatTime(self._benchmarkTime),
279
 
                    test.id()))
280
 
        if self.showAll:
281
 
            self.stream.writeln('   OK %s' % self._testTimeString())
282
 
            for bench_called, stats in getattr(test, '_benchcalls', []):
283
 
                self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
284
 
                stats.pprint(file=self.stream)
285
 
        elif self.dots and self.pb is None:
286
 
            self.stream.write('~')
287
 
        elif self.dots:
288
 
            self.pb.update(self._ellipsise_unimportant_words('OK', 13), self.testsRun, None)
289
 
        self.stream.flush()
290
 
        unittest.TestResult.addSuccess(self, test)
291
 
 
292
 
    def addSkipped(self, test, skip_excinfo):
293
 
        self.extractBenchmarkTime(test)
294
 
        if self.showAll:
295
 
            print >>self.stream, ' SKIP %s' % self._testTimeString()
296
 
            print >>self.stream, '     %s' % skip_excinfo[1]
297
 
        elif self.dots and self.pb is None:
298
 
            self.stream.write('S')
299
 
        elif self.dots:
300
 
            self.pb.update(self._ellipsise_unimportant_words('SKIP', 13), self.testsRun, None)
301
 
        self.stream.flush()
302
 
        # seems best to treat this as success from point-of-view of unittest
303
 
        # -- it actually does nothing so it barely matters :)
304
 
        try:
305
 
            test.tearDown()
306
 
        except KeyboardInterrupt:
307
 
            raise
308
 
        except:
309
 
            self.addError(test, test.__exc_info())
310
 
        else:
311
 
            unittest.TestResult.addSuccess(self, test)
312
 
 
313
 
    def printErrorList(self, flavour, errors):
314
 
        for test, err in errors:
315
 
            self.stream.writeln(self.separator1)
316
 
            self.stream.writeln("%s: %s" % (flavour, self.getDescription(test)))
317
 
            if getattr(test, '_get_log', None) is not None:
318
 
                print >>self.stream
319
 
                print >>self.stream, \
320
 
                        ('vvvv[log from %s]' % test.id()).ljust(78,'-')
321
 
                print >>self.stream, test._get_log()
322
 
                print >>self.stream, \
323
 
                        ('^^^^[log from %s]' % test.id()).ljust(78,'-')
324
 
            self.stream.writeln(self.separator2)
325
 
            self.stream.writeln("%s" % err)
326
 
 
327
 
 
328
 
class TextTestRunner(object):
329
 
    stop_on_failure = False
330
 
 
331
 
    def __init__(self,
332
 
                 stream=sys.stderr,
333
 
                 descriptions=0,
334
 
                 verbosity=1,
335
 
                 keep_output=False,
336
 
                 pb=None,
337
 
                 bench_history=None):
338
 
        self.stream = unittest._WritelnDecorator(stream)
339
 
        self.descriptions = descriptions
340
 
        self.verbosity = verbosity
341
 
        self.keep_output = keep_output
342
 
        self.pb = pb
343
 
        self._bench_history = bench_history
344
 
 
345
 
    def _makeResult(self):
346
 
        result = _MyResult(self.stream,
347
 
                           self.descriptions,
348
 
                           self.verbosity,
349
 
                           pb=self.pb,
350
 
                           bench_history=self._bench_history)
351
 
        result.stop_early = self.stop_on_failure
352
 
        return result
353
 
 
354
 
    def run(self, test):
355
 
        "Run the given test case or test suite."
356
 
        result = self._makeResult()
357
 
        startTime = time.time()
358
 
        if self.pb is not None:
359
 
            self.pb.update('Running tests', 0, test.countTestCases())
360
 
        test.run(result)
361
 
        stopTime = time.time()
362
 
        timeTaken = stopTime - startTime
363
 
        result.printErrors()
364
 
        self.stream.writeln(result.separator2)
365
 
        run = result.testsRun
366
 
        self.stream.writeln("Ran %d test%s in %.3fs" %
367
 
                            (run, run != 1 and "s" or "", timeTaken))
368
 
        self.stream.writeln()
369
 
        if not result.wasSuccessful():
370
 
            self.stream.write("FAILED (")
371
 
            failed, errored = map(len, (result.failures, result.errors))
372
 
            if failed:
373
 
                self.stream.write("failures=%d" % failed)
374
 
            if errored:
375
 
                if failed: self.stream.write(", ")
376
 
                self.stream.write("errors=%d" % errored)
377
 
            self.stream.writeln(")")
378
 
        else:
379
 
            self.stream.writeln("OK")
380
 
        if self.pb is not None:
381
 
            self.pb.update('Cleaning up', 0, 1)
382
 
        # This is still a little bogus, 
383
 
        # but only a little. Folk not using our testrunner will
384
 
        # have to delete their temp directories themselves.
385
 
        test_root = TestCaseInTempDir.TEST_ROOT
386
 
        if result.wasSuccessful() or not self.keep_output:
387
 
            if test_root is not None:
388
 
                # If LANG=C we probably have created some bogus paths
389
 
                # which rmtree(unicode) will fail to delete
390
 
                # so make sure we are using rmtree(str) to delete everything
391
 
                # except on win32, where rmtree(str) will fail
392
 
                # since it doesn't have the property of byte-stream paths
393
 
                # (they are either ascii or mbcs)
394
 
                if sys.platform == 'win32':
395
 
                    # make sure we are using the unicode win32 api
396
 
                    test_root = unicode(test_root)
397
 
                else:
398
 
                    test_root = test_root.encode(
399
 
                        sys.getfilesystemencoding())
400
 
                osutils.rmtree(test_root)
401
 
        else:
402
 
            if self.pb is not None:
403
 
                self.pb.note("Failed tests working directories are in '%s'\n",
404
 
                             test_root)
405
 
            else:
406
 
                self.stream.writeln(
407
 
                    "Failed tests working directories are in '%s'\n" %
408
 
                    test_root)
409
 
        TestCaseInTempDir.TEST_ROOT = None
410
 
        if self.pb is not None:
411
 
            self.pb.clear()
412
 
        return result
413
 
 
414
 
 
415
 
def iter_suite_tests(suite):
416
 
    """Return all tests in a suite, recursing through nested suites"""
417
 
    for item in suite._tests:
418
 
        if isinstance(item, unittest.TestCase):
419
 
            yield item
420
 
        elif isinstance(item, unittest.TestSuite):
421
 
            for r in iter_suite_tests(item):
422
 
                yield r
423
 
        else:
424
 
            raise Exception('unknown object %r inside test suite %r'
425
 
                            % (item, suite))
426
 
 
427
 
 
428
 
class TestSkipped(Exception):
429
 
    """Indicates that a test was intentionally skipped, rather than failing."""
430
 
 
431
 
 
432
 
class CommandFailed(Exception):
433
 
    pass
434
 
 
435
 
 
436
 
class StringIOWrapper(object):
437
 
    """A wrapper around cStringIO which just adds an encoding attribute.
438
 
    
439
 
    Internally we can check sys.stdout to see what the output encoding
440
 
    should be. However, cStringIO has no encoding attribute that we can
441
 
    set. So we wrap it instead.
442
 
    """
443
 
    encoding='ascii'
444
 
    _cstring = None
445
 
 
446
 
    def __init__(self, s=None):
447
 
        if s is not None:
448
 
            self.__dict__['_cstring'] = StringIO(s)
449
 
        else:
450
 
            self.__dict__['_cstring'] = StringIO()
451
 
 
452
 
    def __getattr__(self, name, getattr=getattr):
453
 
        return getattr(self.__dict__['_cstring'], name)
454
 
 
455
 
    def __setattr__(self, name, val):
456
 
        if name == 'encoding':
457
 
            self.__dict__['encoding'] = val
458
 
        else:
459
 
            return setattr(self._cstring, name, val)
460
 
 
461
 
 
462
 
class TestCase(unittest.TestCase):
463
 
    """Base class for bzr unit tests.
464
 
    
465
 
    Tests that need access to disk resources should subclass 
466
 
    TestCaseInTempDir not TestCase.
467
 
 
468
 
    Error and debug log messages are redirected from their usual
469
 
    location into a temporary file, the contents of which can be
470
 
    retrieved by _get_log().  We use a real OS file, not an in-memory object,
471
 
    so that it can also capture file IO.  When the test completes this file
472
 
    is read into memory and removed from disk.
473
 
       
474
 
    There are also convenience functions to invoke bzr's command-line
475
 
    routine, and to build and check bzr trees.
476
 
   
477
 
    In addition to the usual method of overriding tearDown(), this class also
478
 
    allows subclasses to register functions into the _cleanups list, which is
479
 
    run in order as the object is torn down.  It's less likely this will be
480
 
    accidentally overlooked.
481
 
    """
482
 
 
483
 
    _log_file_name = None
484
 
    _log_contents = ''
485
 
    # record lsprof data when performing benchmark calls.
486
 
    _gather_lsprof_in_benchmarks = False
487
 
 
488
 
    def __init__(self, methodName='testMethod'):
489
 
        super(TestCase, self).__init__(methodName)
490
 
        self._cleanups = []
491
 
 
492
 
    def setUp(self):
493
 
        unittest.TestCase.setUp(self)
494
 
        self._cleanEnvironment()
495
 
        bzrlib.trace.disable_default_logging()
496
 
        self._startLogFile()
497
 
        self._benchcalls = []
498
 
        self._benchtime = None
499
 
 
500
 
    def _ndiff_strings(self, a, b):
501
 
        """Return ndiff between two strings containing lines.
502
 
        
503
 
        A trailing newline is added if missing to make the strings
504
 
        print properly."""
505
 
        if b and b[-1] != '\n':
506
 
            b += '\n'
507
 
        if a and a[-1] != '\n':
508
 
            a += '\n'
509
 
        difflines = difflib.ndiff(a.splitlines(True),
510
 
                                  b.splitlines(True),
511
 
                                  linejunk=lambda x: False,
512
 
                                  charjunk=lambda x: False)
513
 
        return ''.join(difflines)
514
 
 
515
 
    def assertEqualDiff(self, a, b, message=None):
516
 
        """Assert two texts are equal, if not raise an exception.
517
 
        
518
 
        This is intended for use with multi-line strings where it can 
519
 
        be hard to find the differences by eye.
520
 
        """
521
 
        # TODO: perhaps override assertEquals to call this for strings?
522
 
        if a == b:
523
 
            return
524
 
        if message is None:
525
 
            message = "texts not equal:\n"
526
 
        raise AssertionError(message + 
527
 
                             self._ndiff_strings(a, b))      
528
 
        
529
 
    def assertEqualMode(self, mode, mode_test):
530
 
        self.assertEqual(mode, mode_test,
531
 
                         'mode mismatch %o != %o' % (mode, mode_test))
532
 
 
533
 
    def assertStartsWith(self, s, prefix):
534
 
        if not s.startswith(prefix):
535
 
            raise AssertionError('string %r does not start with %r' % (s, prefix))
536
 
 
537
 
    def assertEndsWith(self, s, suffix):
538
 
        """Asserts that s ends with suffix."""
539
 
        if not s.endswith(suffix):
540
 
            raise AssertionError('string %r does not end with %r' % (s, suffix))
541
 
 
542
 
    def assertContainsRe(self, haystack, needle_re):
543
 
        """Assert that a contains something matching a regular expression."""
544
 
        if not re.search(needle_re, haystack):
545
 
            raise AssertionError('pattern "%s" not found in "%s"'
546
 
                    % (needle_re, haystack))
547
 
 
548
 
    def assertNotContainsRe(self, haystack, needle_re):
549
 
        """Assert that a does not match a regular expression"""
550
 
        if re.search(needle_re, haystack):
551
 
            raise AssertionError('pattern "%s" found in "%s"'
552
 
                    % (needle_re, haystack))
553
 
 
554
 
    def assertSubset(self, sublist, superlist):
555
 
        """Assert that every entry in sublist is present in superlist."""
556
 
        missing = []
557
 
        for entry in sublist:
558
 
            if entry not in superlist:
559
 
                missing.append(entry)
560
 
        if len(missing) > 0:
561
 
            raise AssertionError("value(s) %r not present in container %r" % 
562
 
                                 (missing, superlist))
563
 
 
564
 
    def assertIs(self, left, right):
565
 
        if not (left is right):
566
 
            raise AssertionError("%r is not %r." % (left, right))
567
 
 
568
 
    def assertTransportMode(self, transport, path, mode):
569
 
        """Fail if a path does not have mode mode.
570
 
        
571
 
        If modes are not supported on this transport, the assertion is ignored.
572
 
        """
573
 
        if not transport._can_roundtrip_unix_modebits():
574
 
            return
575
 
        path_stat = transport.stat(path)
576
 
        actual_mode = stat.S_IMODE(path_stat.st_mode)
577
 
        self.assertEqual(mode, actual_mode,
578
 
            'mode of %r incorrect (%o != %o)' % (path, mode, actual_mode))
579
 
 
580
 
    def assertIsInstance(self, obj, kls):
581
 
        """Fail if obj is not an instance of kls"""
582
 
        if not isinstance(obj, kls):
583
 
            self.fail("%r is an instance of %s rather than %s" % (
584
 
                obj, obj.__class__, kls))
585
 
 
586
 
    def _capture_warnings(self, a_callable, *args, **kwargs):
587
 
        """A helper for callDeprecated and applyDeprecated.
588
 
 
589
 
        :param a_callable: A callable to call.
590
 
        :param args: The positional arguments for the callable
591
 
        :param kwargs: The keyword arguments for the callable
592
 
        :return: A tuple (warnings, result). result is the result of calling
593
 
            a_callable(*args, **kwargs).
594
 
        """
595
 
        local_warnings = []
596
 
        def capture_warnings(msg, cls, stacklevel=None):
597
 
            # we've hooked into a deprecation specific callpath,
598
 
            # only deprecations should getting sent via it.
599
 
            self.assertEqual(cls, DeprecationWarning)
600
 
            local_warnings.append(msg)
601
 
        original_warning_method = symbol_versioning.warn
602
 
        symbol_versioning.set_warning_method(capture_warnings)
603
 
        try:
604
 
            result = a_callable(*args, **kwargs)
605
 
        finally:
606
 
            symbol_versioning.set_warning_method(original_warning_method)
607
 
        return (local_warnings, result)
608
 
 
609
 
    def applyDeprecated(self, deprecation_format, a_callable, *args, **kwargs):
610
 
        """Call a deprecated callable without warning the user.
611
 
 
612
 
        :param deprecation_format: The deprecation format that the callable
613
 
            should have been deprecated with. This is the same type as the 
614
 
            parameter to deprecated_method/deprecated_function. If the 
615
 
            callable is not deprecated with this format, an assertion error
616
 
            will be raised.
617
 
        :param a_callable: A callable to call. This may be a bound method or
618
 
            a regular function. It will be called with *args and **kwargs.
619
 
        :param args: The positional arguments for the callable
620
 
        :param kwargs: The keyword arguments for the callable
621
 
        :return: The result of a_callable(*args, **kwargs)
622
 
        """
623
 
        call_warnings, result = self._capture_warnings(a_callable,
624
 
            *args, **kwargs)
625
 
        expected_first_warning = symbol_versioning.deprecation_string(
626
 
            a_callable, deprecation_format)
627
 
        if len(call_warnings) == 0:
628
 
            self.fail("No assertion generated by call to %s" %
629
 
                a_callable)
630
 
        self.assertEqual(expected_first_warning, call_warnings[0])
631
 
        return result
632
 
 
633
 
    def callDeprecated(self, expected, callable, *args, **kwargs):
634
 
        """Assert that a callable is deprecated in a particular way.
635
 
 
636
 
        This is a very precise test for unusual requirements. The 
637
 
        applyDeprecated helper function is probably more suited for most tests
638
 
        as it allows you to simply specify the deprecation format being used
639
 
        and will ensure that that is issued for the function being called.
640
 
 
641
 
        :param expected: a list of the deprecation warnings expected, in order
642
 
        :param callable: The callable to call
643
 
        :param args: The positional arguments for the callable
644
 
        :param kwargs: The keyword arguments for the callable
645
 
        """
646
 
        call_warnings, result = self._capture_warnings(callable,
647
 
            *args, **kwargs)
648
 
        self.assertEqual(expected, call_warnings)
649
 
        return result
650
 
 
651
 
    def _startLogFile(self):
652
 
        """Send bzr and test log messages to a temporary file.
653
 
 
654
 
        The file is removed as the test is torn down.
655
 
        """
656
 
        fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
657
 
        self._log_file = os.fdopen(fileno, 'w+')
658
 
        self._log_nonce = bzrlib.trace.enable_test_log(self._log_file)
659
 
        self._log_file_name = name
660
 
        self.addCleanup(self._finishLogFile)
661
 
 
662
 
    def _finishLogFile(self):
663
 
        """Finished with the log file.
664
 
 
665
 
        Read contents into memory, close, and delete.
666
 
        """
667
 
        if self._log_file is None:
668
 
            return
669
 
        bzrlib.trace.disable_test_log(self._log_nonce)
670
 
        self._log_file.seek(0)
671
 
        self._log_contents = self._log_file.read()
672
 
        self._log_file.close()
673
 
        os.remove(self._log_file_name)
674
 
        self._log_file = self._log_file_name = None
675
 
 
676
 
    def addCleanup(self, callable):
677
 
        """Arrange to run a callable when this case is torn down.
678
 
 
679
 
        Callables are run in the reverse of the order they are registered, 
680
 
        ie last-in first-out.
681
 
        """
682
 
        if callable in self._cleanups:
683
 
            raise ValueError("cleanup function %r already registered on %s" 
684
 
                    % (callable, self))
685
 
        self._cleanups.append(callable)
686
 
 
687
 
    def _cleanEnvironment(self):
688
 
        new_env = {
689
 
            'HOME': os.getcwd(),
690
 
            'APPDATA': os.getcwd(),
691
 
            'BZR_EMAIL': None,
692
 
            'BZREMAIL': None, # may still be present in the environment
693
 
            'EMAIL': None,
694
 
            'BZR_PROGRESS_BAR': None,
695
 
        }
696
 
        self.__old_env = {}
697
 
        self.addCleanup(self._restoreEnvironment)
698
 
        for name, value in new_env.iteritems():
699
 
            self._captureVar(name, value)
700
 
 
701
 
    def _captureVar(self, name, newvalue):
702
 
        """Set an environment variable, and reset it when finished."""
703
 
        self.__old_env[name] = osutils.set_or_unset_env(name, newvalue)
704
 
 
705
 
    def _restoreEnvironment(self):
706
 
        for name, value in self.__old_env.iteritems():
707
 
            osutils.set_or_unset_env(name, value)
708
 
 
709
 
    def tearDown(self):
710
 
        self._runCleanups()
711
 
        unittest.TestCase.tearDown(self)
712
 
 
713
 
    def time(self, callable, *args, **kwargs):
714
 
        """Run callable and accrue the time it takes to the benchmark time.
715
 
        
716
 
        If lsprofiling is enabled (i.e. by --lsprof-time to bzr selftest) then
717
 
        this will cause lsprofile statistics to be gathered and stored in
718
 
        self._benchcalls.
719
 
        """
720
 
        if self._benchtime is None:
721
 
            self._benchtime = 0
722
 
        start = time.time()
723
 
        try:
724
 
            if not self._gather_lsprof_in_benchmarks:
725
 
                return callable(*args, **kwargs)
726
 
            else:
727
 
                # record this benchmark
728
 
                ret, stats = bzrlib.lsprof.profile(callable, *args, **kwargs)
729
 
                stats.sort()
730
 
                self._benchcalls.append(((callable, args, kwargs), stats))
731
 
                return ret
732
 
        finally:
733
 
            self._benchtime += time.time() - start
734
 
 
735
 
    def _runCleanups(self):
736
 
        """Run registered cleanup functions. 
737
 
 
738
 
        This should only be called from TestCase.tearDown.
739
 
        """
740
 
        # TODO: Perhaps this should keep running cleanups even if 
741
 
        # one of them fails?
742
 
        for cleanup_fn in reversed(self._cleanups):
743
 
            cleanup_fn()
744
 
 
745
 
    def log(self, *args):
746
 
        mutter(*args)
747
 
 
748
 
    def _get_log(self):
749
 
        """Return as a string the log for this test"""
750
 
        if self._log_file_name:
751
 
            return open(self._log_file_name).read()
752
 
        else:
753
 
            return self._log_contents
754
 
        # TODO: Delete the log after it's been read in
755
 
 
756
 
    def capture(self, cmd, retcode=0):
757
 
        """Shortcut that splits cmd into words, runs, and returns stdout"""
758
 
        return self.run_bzr_captured(cmd.split(), retcode=retcode)[0]
759
 
 
760
 
    def run_bzr_captured(self, argv, retcode=0, encoding=None, stdin=None):
761
 
        """Invoke bzr and return (stdout, stderr).
762
 
 
763
 
        Useful for code that wants to check the contents of the
764
 
        output, the way error messages are presented, etc.
765
 
 
766
 
        This should be the main method for tests that want to exercise the
767
 
        overall behavior of the bzr application (rather than a unit test
768
 
        or a functional test of the library.)
769
 
 
770
 
        Much of the old code runs bzr by forking a new copy of Python, but
771
 
        that is slower, harder to debug, and generally not necessary.
772
 
 
773
 
        This runs bzr through the interface that catches and reports
774
 
        errors, and with logging set to something approximating the
775
 
        default, so that error reporting can be checked.
776
 
 
777
 
        :param argv: arguments to invoke bzr
778
 
        :param retcode: expected return code, or None for don't-care.
779
 
        :param encoding: encoding for sys.stdout and sys.stderr
780
 
        :param stdin: A string to be used as stdin for the command.
781
 
        """
782
 
        if encoding is None:
783
 
            encoding = bzrlib.user_encoding
784
 
        if stdin is not None:
785
 
            stdin = StringIO(stdin)
786
 
        stdout = StringIOWrapper()
787
 
        stderr = StringIOWrapper()
788
 
        stdout.encoding = encoding
789
 
        stderr.encoding = encoding
790
 
 
791
 
        self.log('run bzr: %r', argv)
792
 
        # FIXME: don't call into logging here
793
 
        handler = logging.StreamHandler(stderr)
794
 
        handler.setLevel(logging.INFO)
795
 
        logger = logging.getLogger('')
796
 
        logger.addHandler(handler)
797
 
        old_ui_factory = bzrlib.ui.ui_factory
798
 
        bzrlib.ui.ui_factory = bzrlib.tests.blackbox.TestUIFactory(
799
 
            stdout=stdout,
800
 
            stderr=stderr)
801
 
        bzrlib.ui.ui_factory.stdin = stdin
802
 
        try:
803
 
            result = self.apply_redirected(stdin, stdout, stderr,
804
 
                                           bzrlib.commands.run_bzr_catch_errors,
805
 
                                           argv)
806
 
        finally:
807
 
            logger.removeHandler(handler)
808
 
            bzrlib.ui.ui_factory = old_ui_factory
809
 
 
810
 
        out = stdout.getvalue()
811
 
        err = stderr.getvalue()
812
 
        if out:
813
 
            self.log('output:\n%r', out)
814
 
        if err:
815
 
            self.log('errors:\n%r', err)
816
 
        if retcode is not None:
817
 
            self.assertEquals(retcode, result)
818
 
        return out, err
819
 
 
820
 
    def run_bzr(self, *args, **kwargs):
821
 
        """Invoke bzr, as if it were run from the command line.
822
 
 
823
 
        This should be the main method for tests that want to exercise the
824
 
        overall behavior of the bzr application (rather than a unit test
825
 
        or a functional test of the library.)
826
 
 
827
 
        This sends the stdout/stderr results into the test's log,
828
 
        where it may be useful for debugging.  See also run_captured.
829
 
 
830
 
        :param stdin: A string to be used as stdin for the command.
831
 
        """
832
 
        retcode = kwargs.pop('retcode', 0)
833
 
        encoding = kwargs.pop('encoding', None)
834
 
        stdin = kwargs.pop('stdin', None)
835
 
        return self.run_bzr_captured(args, retcode=retcode, encoding=encoding, stdin=stdin)
836
 
 
837
 
    def run_bzr_decode(self, *args, **kwargs):
838
 
        if 'encoding' in kwargs:
839
 
            encoding = kwargs['encoding']
840
 
        else:
841
 
            encoding = bzrlib.user_encoding
842
 
        return self.run_bzr(*args, **kwargs)[0].decode(encoding)
843
 
 
844
 
    def run_bzr_error(self, error_regexes, *args, **kwargs):
845
 
        """Run bzr, and check that stderr contains the supplied regexes
846
 
        
847
 
        :param error_regexes: Sequence of regular expressions which 
848
 
            must each be found in the error output. The relative ordering
849
 
            is not enforced.
850
 
        :param args: command-line arguments for bzr
851
 
        :param kwargs: Keyword arguments which are interpreted by run_bzr
852
 
            This function changes the default value of retcode to be 3,
853
 
            since in most cases this is run when you expect bzr to fail.
854
 
        :return: (out, err) The actual output of running the command (in case you
855
 
                 want to do more inspection)
856
 
 
857
 
        Examples of use:
858
 
            # Make sure that commit is failing because there is nothing to do
859
 
            self.run_bzr_error(['no changes to commit'],
860
 
                               'commit', '-m', 'my commit comment')
861
 
            # Make sure --strict is handling an unknown file, rather than
862
 
            # giving us the 'nothing to do' error
863
 
            self.build_tree(['unknown'])
864
 
            self.run_bzr_error(['Commit refused because there are unknown files'],
865
 
                               'commit', '--strict', '-m', 'my commit comment')
866
 
        """
867
 
        kwargs.setdefault('retcode', 3)
868
 
        out, err = self.run_bzr(*args, **kwargs)
869
 
        for regex in error_regexes:
870
 
            self.assertContainsRe(err, regex)
871
 
        return out, err
872
 
 
873
 
    def run_bzr_subprocess(self, *args, **kwargs):
874
 
        """Run bzr in a subprocess for testing.
875
 
 
876
 
        This starts a new Python interpreter and runs bzr in there. 
877
 
        This should only be used for tests that have a justifiable need for
878
 
        this isolation: e.g. they are testing startup time, or signal
879
 
        handling, or early startup code, etc.  Subprocess code can't be 
880
 
        profiled or debugged so easily.
881
 
 
882
 
        :param retcode: The status code that is expected.  Defaults to 0.  If
883
 
            None is supplied, the status code is not checked.
884
 
        :param env_changes: A dictionary which lists changes to environment
885
 
            variables. A value of None will unset the env variable.
886
 
            The values must be strings. The change will only occur in the
887
 
            child, so you don't need to fix the environment after running.
888
 
        :param universal_newlines: Convert CRLF => LF
889
 
        """
890
 
        env_changes = kwargs.get('env_changes', {})
891
 
        process = self.start_bzr_subprocess(args, env_changes=env_changes)
892
 
        # We distinguish between retcode=None and retcode not passed.
893
 
        supplied_retcode = kwargs.get('retcode', 0)
894
 
        return self.finish_bzr_subprocess(process, retcode=supplied_retcode,
895
 
            universal_newlines=kwargs.get('universal_newlines', False),
896
 
            process_args=args)
897
 
 
898
 
    def start_bzr_subprocess(self, process_args, env_changes=None,
899
 
                             skip_if_plan_to_signal=False):
900
 
        """Start bzr in a subprocess for testing.
901
 
 
902
 
        This starts a new Python interpreter and runs bzr in there.
903
 
        This should only be used for tests that have a justifiable need for
904
 
        this isolation: e.g. they are testing startup time, or signal
905
 
        handling, or early startup code, etc.  Subprocess code can't be
906
 
        profiled or debugged so easily.
907
 
 
908
 
        :param process_args: a list of arguments to pass to the bzr executable,
909
 
            for example `['--version']`.
910
 
        :param env_changes: A dictionary which lists changes to environment
911
 
            variables. A value of None will unset the env variable.
912
 
            The values must be strings. The change will only occur in the
913
 
            child, so you don't need to fix the environment after running.
914
 
        :param skip_if_plan_to_signal: raise TestSkipped when true and os.kill
915
 
            is not available.
916
 
 
917
 
        :returns: Popen object for the started process.
918
 
        """
919
 
        if skip_if_plan_to_signal:
920
 
            if not getattr(os, 'kill', None):
921
 
                raise TestSkipped("os.kill not available.")
922
 
 
923
 
        if env_changes is None:
924
 
            env_changes = {}
925
 
        old_env = {}
926
 
 
927
 
        def cleanup_environment():
928
 
            for env_var, value in env_changes.iteritems():
929
 
                old_env[env_var] = osutils.set_or_unset_env(env_var, value)
930
 
 
931
 
        def restore_environment():
932
 
            for env_var, value in old_env.iteritems():
933
 
                osutils.set_or_unset_env(env_var, value)
934
 
 
935
 
        bzr_path = self.get_bzr_path()
936
 
 
937
 
        try:
938
 
            # win32 subprocess doesn't support preexec_fn
939
 
            # so we will avoid using it on all platforms, just to
940
 
            # make sure the code path is used, and we don't break on win32
941
 
            cleanup_environment()
942
 
            process = Popen([sys.executable, bzr_path] + list(process_args),
943
 
                             stdin=PIPE, stdout=PIPE, stderr=PIPE)
944
 
        finally:
945
 
            restore_environment()
946
 
        return process
947
 
 
948
 
    def get_bzr_path(self):
949
 
        """Return the path of the 'bzr' executable for this test suite."""
950
 
        bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
951
 
        if not os.path.isfile(bzr_path):
952
 
            # We are probably installed. Assume sys.argv is the right file
953
 
            bzr_path = sys.argv[0]
954
 
        return bzr_path
955
 
 
956
 
    def finish_bzr_subprocess(self, process, retcode=0, send_signal=None,
957
 
                              universal_newlines=False, process_args=None):
958
 
        """Finish the execution of process.
959
 
 
960
 
        :param process: the Popen object returned from start_bzr_subprocess.
961
 
        :param retcode: The status code that is expected.  Defaults to 0.  If
962
 
            None is supplied, the status code is not checked.
963
 
        :param send_signal: an optional signal to send to the process.
964
 
        :param universal_newlines: Convert CRLF => LF
965
 
        :returns: (stdout, stderr)
966
 
        """
967
 
        if send_signal is not None:
968
 
            os.kill(process.pid, send_signal)
969
 
        out, err = process.communicate()
970
 
 
971
 
        if universal_newlines:
972
 
            out = out.replace('\r\n', '\n')
973
 
            err = err.replace('\r\n', '\n')
974
 
 
975
 
        if retcode is not None and retcode != process.returncode:
976
 
            if process_args is None:
977
 
                process_args = "(unknown args)"
978
 
            mutter('Output of bzr %s:\n%s', process_args, out)
979
 
            mutter('Error for bzr %s:\n%s', process_args, err)
980
 
            self.fail('Command bzr %s failed with retcode %s != %s'
981
 
                      % (process_args, retcode, process.returncode))
982
 
        return [out, err]
983
 
 
984
 
    def check_inventory_shape(self, inv, shape):
985
 
        """Compare an inventory to a list of expected names.
986
 
 
987
 
        Fail if they are not precisely equal.
988
 
        """
989
 
        extras = []
990
 
        shape = list(shape)             # copy
991
 
        for path, ie in inv.entries():
992
 
            name = path.replace('\\', '/')
993
 
            if ie.kind == 'dir':
994
 
                name = name + '/'
995
 
            if name in shape:
996
 
                shape.remove(name)
997
 
            else:
998
 
                extras.append(name)
999
 
        if shape:
1000
 
            self.fail("expected paths not found in inventory: %r" % shape)
1001
 
        if extras:
1002
 
            self.fail("unexpected paths found in inventory: %r" % extras)
1003
 
 
1004
 
    def apply_redirected(self, stdin=None, stdout=None, stderr=None,
1005
 
                         a_callable=None, *args, **kwargs):
1006
 
        """Call callable with redirected std io pipes.
1007
 
 
1008
 
        Returns the return code."""
1009
 
        if not callable(a_callable):
1010
 
            raise ValueError("a_callable must be callable.")
1011
 
        if stdin is None:
1012
 
            stdin = StringIO("")
1013
 
        if stdout is None:
1014
 
            if getattr(self, "_log_file", None) is not None:
1015
 
                stdout = self._log_file
1016
 
            else:
1017
 
                stdout = StringIO()
1018
 
        if stderr is None:
1019
 
            if getattr(self, "_log_file", None is not None):
1020
 
                stderr = self._log_file
1021
 
            else:
1022
 
                stderr = StringIO()
1023
 
        real_stdin = sys.stdin
1024
 
        real_stdout = sys.stdout
1025
 
        real_stderr = sys.stderr
1026
 
        try:
1027
 
            sys.stdout = stdout
1028
 
            sys.stderr = stderr
1029
 
            sys.stdin = stdin
1030
 
            return a_callable(*args, **kwargs)
1031
 
        finally:
1032
 
            sys.stdout = real_stdout
1033
 
            sys.stderr = real_stderr
1034
 
            sys.stdin = real_stdin
1035
 
 
1036
 
    @symbol_versioning.deprecated_method(symbol_versioning.zero_eleven)
1037
 
    def merge(self, branch_from, wt_to):
1038
 
        """A helper for tests to do a ui-less merge.
1039
 
 
1040
 
        This should move to the main library when someone has time to integrate
1041
 
        it in.
1042
 
        """
1043
 
        # minimal ui-less merge.
1044
 
        wt_to.branch.fetch(branch_from)
1045
 
        base_rev = common_ancestor(branch_from.last_revision(),
1046
 
                                   wt_to.branch.last_revision(),
1047
 
                                   wt_to.branch.repository)
1048
 
        merge_inner(wt_to.branch, branch_from.basis_tree(),
1049
 
                    wt_to.branch.repository.revision_tree(base_rev),
1050
 
                    this_tree=wt_to)
1051
 
        wt_to.add_parent_tree_id(branch_from.last_revision())
1052
 
 
1053
 
 
1054
 
BzrTestBase = TestCase
1055
 
 
1056
 
     
1057
 
class TestCaseInTempDir(TestCase):
1058
 
    """Derived class that runs a test within a temporary directory.
1059
 
 
1060
 
    This is useful for tests that need to create a branch, etc.
1061
 
 
1062
 
    The directory is created in a slightly complex way: for each
1063
 
    Python invocation, a new temporary top-level directory is created.
1064
 
    All test cases create their own directory within that.  If the
1065
 
    tests complete successfully, the directory is removed.
1066
 
 
1067
 
    InTempDir is an old alias for FunctionalTestCase.
1068
 
    """
1069
 
 
1070
 
    TEST_ROOT = None
1071
 
    _TEST_NAME = 'test'
1072
 
    OVERRIDE_PYTHON = 'python'
1073
 
 
1074
 
    def check_file_contents(self, filename, expect):
1075
 
        self.log("check contents of file %s" % filename)
1076
 
        contents = file(filename, 'r').read()
1077
 
        if contents != expect:
1078
 
            self.log("expected: %r" % expect)
1079
 
            self.log("actually: %r" % contents)
1080
 
            self.fail("contents of %s not as expected" % filename)
1081
 
 
1082
 
    def _make_test_root(self):
1083
 
        if TestCaseInTempDir.TEST_ROOT is not None:
1084
 
            return
1085
 
        i = 0
1086
 
        while True:
1087
 
            root = u'test%04d.tmp' % i
1088
 
            try:
1089
 
                os.mkdir(root)
1090
 
            except OSError, e:
1091
 
                if e.errno == errno.EEXIST:
1092
 
                    i += 1
1093
 
                    continue
1094
 
                else:
1095
 
                    raise
1096
 
            # successfully created
1097
 
            TestCaseInTempDir.TEST_ROOT = osutils.abspath(root)
1098
 
            break
1099
 
        # make a fake bzr directory there to prevent any tests propagating
1100
 
        # up onto the source directory's real branch
1101
 
        bzrdir.BzrDir.create_standalone_workingtree(TestCaseInTempDir.TEST_ROOT)
1102
 
 
1103
 
    def setUp(self):
1104
 
        super(TestCaseInTempDir, self).setUp()
1105
 
        self._make_test_root()
1106
 
        _currentdir = os.getcwdu()
1107
 
        # shorten the name, to avoid test failures due to path length
1108
 
        short_id = self.id().replace('bzrlib.tests.', '') \
1109
 
                   .replace('__main__.', '')[-100:]
1110
 
        # it's possible the same test class is run several times for
1111
 
        # parameterized tests, so make sure the names don't collide.  
1112
 
        i = 0
1113
 
        while True:
1114
 
            if i > 0:
1115
 
                candidate_dir = '%s/%s.%d' % (self.TEST_ROOT, short_id, i)
1116
 
            else:
1117
 
                candidate_dir = '%s/%s' % (self.TEST_ROOT, short_id)
1118
 
            if os.path.exists(candidate_dir):
1119
 
                i = i + 1
1120
 
                continue
1121
 
            else:
1122
 
                os.mkdir(candidate_dir)
1123
 
                self.test_home_dir = candidate_dir + '/home'
1124
 
                os.mkdir(self.test_home_dir)
1125
 
                self.test_dir = candidate_dir + '/work'
1126
 
                os.mkdir(self.test_dir)
1127
 
                os.chdir(self.test_dir)
1128
 
                break
1129
 
        os.environ['HOME'] = self.test_home_dir
1130
 
        os.environ['APPDATA'] = self.test_home_dir
1131
 
        def _leaveDirectory():
1132
 
            os.chdir(_currentdir)
1133
 
        self.addCleanup(_leaveDirectory)
1134
 
        
1135
 
    def build_tree(self, shape, line_endings='native', transport=None):
1136
 
        """Build a test tree according to a pattern.
1137
 
 
1138
 
        shape is a sequence of file specifications.  If the final
1139
 
        character is '/', a directory is created.
1140
 
 
1141
 
        This assumes that all the elements in the tree being built are new.
1142
 
 
1143
 
        This doesn't add anything to a branch.
1144
 
        :param line_endings: Either 'binary' or 'native'
1145
 
                             in binary mode, exact contents are written
1146
 
                             in native mode, the line endings match the
1147
 
                             default platform endings.
1148
 
 
1149
 
        :param transport: A transport to write to, for building trees on 
1150
 
                          VFS's. If the transport is readonly or None,
1151
 
                          "." is opened automatically.
1152
 
        """
1153
 
        # It's OK to just create them using forward slashes on windows.
1154
 
        if transport is None or transport.is_readonly():
1155
 
            transport = get_transport(".")
1156
 
        for name in shape:
1157
 
            self.assert_(isinstance(name, basestring))
1158
 
            if name[-1] == '/':
1159
 
                transport.mkdir(urlutils.escape(name[:-1]))
1160
 
            else:
1161
 
                if line_endings == 'binary':
1162
 
                    end = '\n'
1163
 
                elif line_endings == 'native':
1164
 
                    end = os.linesep
1165
 
                else:
1166
 
                    raise errors.BzrError('Invalid line ending request %r' % (line_endings,))
1167
 
                content = "contents of %s%s" % (name.encode('utf-8'), end)
1168
 
                # Technically 'put()' is the right command. However, put
1169
 
                # uses an AtomicFile, which requires an extra rename into place
1170
 
                # As long as the files didn't exist in the past, append() will
1171
 
                # do the same thing as put()
1172
 
                # On jam's machine, make_kernel_like_tree is:
1173
 
                #   put:    4.5-7.5s (averaging 6s)
1174
 
                #   append: 2.9-4.5s
1175
 
                #   put_non_atomic: 2.9-4.5s
1176
 
                transport.put_bytes_non_atomic(urlutils.escape(name), content)
1177
 
 
1178
 
    def build_tree_contents(self, shape):
1179
 
        build_tree_contents(shape)
1180
 
 
1181
 
    def failUnlessExists(self, path):
1182
 
        """Fail unless path, which may be abs or relative, exists."""
1183
 
        self.failUnless(osutils.lexists(path))
1184
 
 
1185
 
    def failIfExists(self, path):
1186
 
        """Fail if path, which may be abs or relative, exists."""
1187
 
        self.failIf(osutils.lexists(path))
1188
 
        
1189
 
    def assertFileEqual(self, content, path):
1190
 
        """Fail if path does not contain 'content'."""
1191
 
        self.failUnless(osutils.lexists(path))
1192
 
        # TODO: jam 20060427 Shouldn't this be 'rb'?
1193
 
        self.assertEqualDiff(content, open(path, 'r').read())
1194
 
 
1195
 
 
1196
 
class TestCaseWithTransport(TestCaseInTempDir):
1197
 
    """A test case that provides get_url and get_readonly_url facilities.
1198
 
 
1199
 
    These back onto two transport servers, one for readonly access and one for
1200
 
    read write access.
1201
 
 
1202
 
    If no explicit class is provided for readonly access, a
1203
 
    ReadonlyTransportDecorator is used instead which allows the use of non disk
1204
 
    based read write transports.
1205
 
 
1206
 
    If an explicit class is provided for readonly access, that server and the 
1207
 
    readwrite one must both define get_url() as resolving to os.getcwd().
1208
 
    """
1209
 
 
1210
 
    def __init__(self, methodName='testMethod'):
1211
 
        super(TestCaseWithTransport, self).__init__(methodName)
1212
 
        self.__readonly_server = None
1213
 
        self.__server = None
1214
 
        self.transport_server = default_transport
1215
 
        self.transport_readonly_server = None
1216
 
 
1217
 
    def get_readonly_url(self, relpath=None):
1218
 
        """Get a URL for the readonly transport.
1219
 
 
1220
 
        This will either be backed by '.' or a decorator to the transport 
1221
 
        used by self.get_url()
1222
 
        relpath provides for clients to get a path relative to the base url.
1223
 
        These should only be downwards relative, not upwards.
1224
 
        """
1225
 
        base = self.get_readonly_server().get_url()
1226
 
        if relpath is not None:
1227
 
            if not base.endswith('/'):
1228
 
                base = base + '/'
1229
 
            base = base + relpath
1230
 
        return base
1231
 
 
1232
 
    def get_readonly_server(self):
1233
 
        """Get the server instance for the readonly transport
1234
 
 
1235
 
        This is useful for some tests with specific servers to do diagnostics.
1236
 
        """
1237
 
        if self.__readonly_server is None:
1238
 
            if self.transport_readonly_server is None:
1239
 
                # readonly decorator requested
1240
 
                # bring up the server
1241
 
                self.get_url()
1242
 
                self.__readonly_server = ReadonlyServer()
1243
 
                self.__readonly_server.setUp(self.__server)
1244
 
            else:
1245
 
                self.__readonly_server = self.transport_readonly_server()
1246
 
                self.__readonly_server.setUp()
1247
 
            self.addCleanup(self.__readonly_server.tearDown)
1248
 
        return self.__readonly_server
1249
 
 
1250
 
    def get_server(self):
1251
 
        """Get the read/write server instance.
1252
 
 
1253
 
        This is useful for some tests with specific servers that need
1254
 
        diagnostics.
1255
 
        """
1256
 
        if self.__server is None:
1257
 
            self.__server = self.transport_server()
1258
 
            self.__server.setUp()
1259
 
            self.addCleanup(self.__server.tearDown)
1260
 
        return self.__server
1261
 
 
1262
 
    def get_url(self, relpath=None):
1263
 
        """Get a URL (or maybe a path) for the readwrite transport.
1264
 
 
1265
 
        This will either be backed by '.' or to an equivalent non-file based
1266
 
        facility.
1267
 
        relpath provides for clients to get a path relative to the base url.
1268
 
        These should only be downwards relative, not upwards.
1269
 
        """
1270
 
        base = self.get_server().get_url()
1271
 
        if relpath is not None and relpath != '.':
1272
 
            if not base.endswith('/'):
1273
 
                base = base + '/'
1274
 
            # XXX: Really base should be a url; we did after all call
1275
 
            # get_url()!  But sometimes it's just a path (from
1276
 
            # LocalAbspathServer), and it'd be wrong to append urlescaped data
1277
 
            # to a non-escaped local path.
1278
 
            if base.startswith('./') or base.startswith('/'):
1279
 
                base += relpath
1280
 
            else:
1281
 
                base += urlutils.escape(relpath)
1282
 
        return base
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 make_branch(self, relpath, format=None):
1301
 
        """Create a branch on the transport at relpath."""
1302
 
        repo = self.make_repository(relpath, format=format)
1303
 
        return repo.bzrdir.create_branch()
1304
 
 
1305
 
    def make_bzrdir(self, relpath, format=None):
1306
 
        try:
1307
 
            # might be a relative or absolute path
1308
 
            maybe_a_url = self.get_url(relpath)
1309
 
            segments = maybe_a_url.rsplit('/', 1)
1310
 
            t = get_transport(maybe_a_url)
1311
 
            if len(segments) > 1 and segments[-1] not in ('', '.'):
1312
 
                try:
1313
 
                    t.mkdir('.')
1314
 
                except errors.FileExists:
1315
 
                    pass
1316
 
            if format is None:
1317
 
                format = bzrlib.bzrdir.BzrDirFormat.get_default_format()
1318
 
            return format.initialize_on_transport(t)
1319
 
        except errors.UninitializableFormat:
1320
 
            raise TestSkipped("Format %s is not initializable." % format)
1321
 
 
1322
 
    def make_repository(self, relpath, shared=False, format=None):
1323
 
        """Create a repository on our default transport at relpath."""
1324
 
        made_control = self.make_bzrdir(relpath, format=format)
1325
 
        return made_control.create_repository(shared=shared)
1326
 
 
1327
 
    def make_branch_and_memory_tree(self, relpath):
1328
 
        """Create a branch on the default transport and a MemoryTree for it."""
1329
 
        b = self.make_branch(relpath)
1330
 
        return memorytree.MemoryTree.create_on_branch(b)
1331
 
 
1332
 
    def make_branch_and_tree(self, relpath, format=None):
1333
 
        """Create a branch on the transport and a tree locally.
1334
 
 
1335
 
        If the transport is not a LocalTransport, the Tree can't be created on
1336
 
        the transport.  In that case the working tree is created in the local
1337
 
        directory, and the returned tree's branch and repository will also be
1338
 
        accessed locally.
1339
 
 
1340
 
        This will fail if the original default transport for this test
1341
 
        case wasn't backed by the working directory, as the branch won't
1342
 
        be on disk for us to open it.  
1343
 
 
1344
 
        :param format: The BzrDirFormat.
1345
 
        :returns: the WorkingTree.
1346
 
        """
1347
 
        # TODO: always use the local disk path for the working tree,
1348
 
        # this obviously requires a format that supports branch references
1349
 
        # so check for that by checking bzrdir.BzrDirFormat.get_default_format()
1350
 
        # RBC 20060208
1351
 
        b = self.make_branch(relpath, format=format)
1352
 
        try:
1353
 
            return b.bzrdir.create_workingtree()
1354
 
        except errors.NotLocalUrl:
1355
 
            # We can only make working trees locally at the moment.  If the
1356
 
            # transport can't support them, then reopen the branch on a local
1357
 
            # transport, and create the working tree there.  
1358
 
            #
1359
 
            # Possibly we should instead keep
1360
 
            # the non-disk-backed branch and create a local checkout?
1361
 
            bd = bzrdir.BzrDir.open(relpath)
1362
 
            return bd.create_workingtree()
1363
 
 
1364
 
    def assertIsDirectory(self, relpath, transport):
1365
 
        """Assert that relpath within transport is a directory.
1366
 
 
1367
 
        This may not be possible on all transports; in that case it propagates
1368
 
        a TransportNotPossible.
1369
 
        """
1370
 
        try:
1371
 
            mode = transport.stat(relpath).st_mode
1372
 
        except errors.NoSuchFile:
1373
 
            self.fail("path %s is not a directory; no such file"
1374
 
                      % (relpath))
1375
 
        if not stat.S_ISDIR(mode):
1376
 
            self.fail("path %s is not a directory; has mode %#o"
1377
 
                      % (relpath, mode))
1378
 
 
1379
 
 
1380
 
class ChrootedTestCase(TestCaseWithTransport):
1381
 
    """A support class that provides readonly urls outside the local namespace.
1382
 
 
1383
 
    This is done by checking if self.transport_server is a MemoryServer. if it
1384
 
    is then we are chrooted already, if it is not then an HttpServer is used
1385
 
    for readonly urls.
1386
 
 
1387
 
    TODO RBC 20060127: make this an option to TestCaseWithTransport so it can
1388
 
                       be used without needed to redo it when a different 
1389
 
                       subclass is in use ?
1390
 
    """
1391
 
 
1392
 
    def setUp(self):
1393
 
        super(ChrootedTestCase, self).setUp()
1394
 
        if not self.transport_server == bzrlib.transport.memory.MemoryServer:
1395
 
            self.transport_readonly_server = bzrlib.transport.http.HttpServer
1396
 
 
1397
 
 
1398
 
def filter_suite_by_re(suite, pattern):
1399
 
    result = TestUtil.TestSuite()
1400
 
    filter_re = re.compile(pattern)
1401
 
    for test in iter_suite_tests(suite):
1402
 
        if filter_re.search(test.id()):
1403
 
            result.addTest(test)
1404
 
    return result
1405
 
 
1406
 
 
1407
 
def run_suite(suite, name='test', verbose=False, pattern=".*",
1408
 
              stop_on_failure=False, keep_output=False,
1409
 
              transport=None, lsprof_timed=None, bench_history=None):
1410
 
    TestCaseInTempDir._TEST_NAME = name
1411
 
    TestCase._gather_lsprof_in_benchmarks = lsprof_timed
1412
 
    if verbose:
1413
 
        verbosity = 2
1414
 
        pb = None
1415
 
    else:
1416
 
        verbosity = 1
1417
 
        pb = progress.ProgressBar()
1418
 
    runner = TextTestRunner(stream=sys.stdout,
1419
 
                            descriptions=0,
1420
 
                            verbosity=verbosity,
1421
 
                            keep_output=keep_output,
1422
 
                            pb=pb,
1423
 
                            bench_history=bench_history)
1424
 
    runner.stop_on_failure=stop_on_failure
1425
 
    if pattern != '.*':
1426
 
        suite = filter_suite_by_re(suite, pattern)
1427
 
    result = runner.run(suite)
1428
 
    return result.wasSuccessful()
1429
 
 
1430
 
 
1431
 
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
1432
 
             keep_output=False,
1433
 
             transport=None,
1434
 
             test_suite_factory=None,
1435
 
             lsprof_timed=None,
1436
 
             bench_history=None):
1437
 
    """Run the whole test suite under the enhanced runner"""
1438
 
    # XXX: Very ugly way to do this...
1439
 
    # Disable warning about old formats because we don't want it to disturb
1440
 
    # any blackbox tests.
1441
 
    from bzrlib import repository
1442
 
    repository._deprecation_warning_done = True
1443
 
 
1444
 
    global default_transport
1445
 
    if transport is None:
1446
 
        transport = default_transport
1447
 
    old_transport = default_transport
1448
 
    default_transport = transport
1449
 
    try:
1450
 
        if test_suite_factory is None:
1451
 
            suite = test_suite()
1452
 
        else:
1453
 
            suite = test_suite_factory()
1454
 
        return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
1455
 
                     stop_on_failure=stop_on_failure, keep_output=keep_output,
1456
 
                     transport=transport,
1457
 
                     lsprof_timed=lsprof_timed,
1458
 
                     bench_history=bench_history)
1459
 
    finally:
1460
 
        default_transport = old_transport
1461
 
 
1462
 
 
1463
 
def test_suite():
1464
 
    """Build and return TestSuite for the whole of bzrlib.
1465
 
    
1466
 
    This function can be replaced if you need to change the default test
1467
 
    suite on a global basis, but it is not encouraged.
1468
 
    """
1469
 
    testmod_names = [
1470
 
                   'bzrlib.tests.test_ancestry',
1471
 
                   'bzrlib.tests.test_api',
1472
 
                   'bzrlib.tests.test_atomicfile',
1473
 
                   'bzrlib.tests.test_bad_files',
1474
 
                   'bzrlib.tests.test_branch',
1475
 
                   'bzrlib.tests.test_bundle',
1476
 
                   'bzrlib.tests.test_bzrdir',
1477
 
                   'bzrlib.tests.test_cache_utf8',
1478
 
                   'bzrlib.tests.test_command',
1479
 
                   'bzrlib.tests.test_commit',
1480
 
                   'bzrlib.tests.test_commit_merge',
1481
 
                   'bzrlib.tests.test_config',
1482
 
                   'bzrlib.tests.test_conflicts',
1483
 
                   'bzrlib.tests.test_decorators',
1484
 
                   'bzrlib.tests.test_diff',
1485
 
                   'bzrlib.tests.test_doc_generate',
1486
 
                   'bzrlib.tests.test_errors',
1487
 
                   'bzrlib.tests.test_escaped_store',
1488
 
                   'bzrlib.tests.test_fetch',
1489
 
                   'bzrlib.tests.test_ftp_transport',
1490
 
                   'bzrlib.tests.test_gpg',
1491
 
                   'bzrlib.tests.test_graph',
1492
 
                   'bzrlib.tests.test_hashcache',
1493
 
                   'bzrlib.tests.test_http',
1494
 
                   'bzrlib.tests.test_http_response',
1495
 
                   'bzrlib.tests.test_identitymap',
1496
 
                   'bzrlib.tests.test_ignores',
1497
 
                   'bzrlib.tests.test_inv',
1498
 
                   'bzrlib.tests.test_knit',
1499
 
                   'bzrlib.tests.test_lazy_import',
1500
 
                   'bzrlib.tests.test_lockdir',
1501
 
                   'bzrlib.tests.test_lockable_files',
1502
 
                   'bzrlib.tests.test_log',
1503
 
                   'bzrlib.tests.test_memorytree',
1504
 
                   'bzrlib.tests.test_merge',
1505
 
                   'bzrlib.tests.test_merge3',
1506
 
                   'bzrlib.tests.test_merge_core',
1507
 
                   'bzrlib.tests.test_missing',
1508
 
                   'bzrlib.tests.test_msgeditor',
1509
 
                   'bzrlib.tests.test_nonascii',
1510
 
                   'bzrlib.tests.test_options',
1511
 
                   'bzrlib.tests.test_osutils',
1512
 
                   'bzrlib.tests.test_patch',
1513
 
                   'bzrlib.tests.test_patches',
1514
 
                   'bzrlib.tests.test_permissions',
1515
 
                   'bzrlib.tests.test_plugins',
1516
 
                   'bzrlib.tests.test_progress',
1517
 
                   'bzrlib.tests.test_reconcile',
1518
 
                   'bzrlib.tests.test_repository',
1519
 
                   'bzrlib.tests.test_revert',
1520
 
                   'bzrlib.tests.test_revision',
1521
 
                   'bzrlib.tests.test_revisionnamespaces',
1522
 
                   'bzrlib.tests.test_revisiontree',
1523
 
                   'bzrlib.tests.test_rio',
1524
 
                   'bzrlib.tests.test_sampler',
1525
 
                   'bzrlib.tests.test_selftest',
1526
 
                   'bzrlib.tests.test_setup',
1527
 
                   'bzrlib.tests.test_sftp_transport',
1528
 
                   'bzrlib.tests.test_smart_add',
1529
 
                   'bzrlib.tests.test_smart_transport',
1530
 
                   'bzrlib.tests.test_source',
1531
 
                   'bzrlib.tests.test_status',
1532
 
                   'bzrlib.tests.test_store',
1533
 
                   'bzrlib.tests.test_symbol_versioning',
1534
 
                   'bzrlib.tests.test_testament',
1535
 
                   'bzrlib.tests.test_textfile',
1536
 
                   'bzrlib.tests.test_textmerge',
1537
 
                   'bzrlib.tests.test_trace',
1538
 
                   'bzrlib.tests.test_transactions',
1539
 
                   'bzrlib.tests.test_transform',
1540
 
                   'bzrlib.tests.test_transport',
1541
 
                   'bzrlib.tests.test_tree',
1542
 
                   'bzrlib.tests.test_treebuilder',
1543
 
                   'bzrlib.tests.test_tsort',
1544
 
                   'bzrlib.tests.test_tuned_gzip',
1545
 
                   'bzrlib.tests.test_ui',
1546
 
                   'bzrlib.tests.test_upgrade',
1547
 
                   'bzrlib.tests.test_urlutils',
1548
 
                   'bzrlib.tests.test_versionedfile',
1549
 
                   'bzrlib.tests.test_version',
1550
 
                   'bzrlib.tests.test_weave',
1551
 
                   'bzrlib.tests.test_whitebox',
1552
 
                   'bzrlib.tests.test_workingtree',
1553
 
                   'bzrlib.tests.test_xml',
1554
 
                   ]
1555
 
    test_transport_implementations = [
1556
 
        'bzrlib.tests.test_transport_implementations',
1557
 
        'bzrlib.tests.test_read_bundle',
1558
 
        ]
1559
 
    suite = TestUtil.TestSuite()
1560
 
    loader = TestUtil.TestLoader()
1561
 
    suite.addTest(loader.loadTestsFromModuleNames(testmod_names))
1562
 
    from bzrlib.transport import TransportTestProviderAdapter
1563
 
    adapter = TransportTestProviderAdapter()
1564
 
    adapt_modules(test_transport_implementations, adapter, loader, suite)
1565
 
    for package in packages_to_test():
1566
 
        suite.addTest(package.test_suite())
1567
 
    for m in MODULES_TO_TEST:
1568
 
        suite.addTest(loader.loadTestsFromModule(m))
1569
 
    for m in MODULES_TO_DOCTEST:
1570
 
        suite.addTest(doctest.DocTestSuite(m))
1571
 
    for name, plugin in bzrlib.plugin.all_plugins().items():
1572
 
        if getattr(plugin, 'test_suite', None) is not None:
1573
 
            suite.addTest(plugin.test_suite())
1574
 
    return suite
1575
 
 
1576
 
 
1577
 
def adapt_modules(mods_list, adapter, loader, suite):
1578
 
    """Adapt the modules in mods_list using adapter and add to suite."""
1579
 
    for test in iter_suite_tests(loader.loadTestsFromModuleNames(mods_list)):
1580
 
        suite.addTests(adapter.adapt(test))