/brz/remove-bazaar

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

« back to all changes in this revision

Viewing changes to bzrlib/tests/__init__.py

  • Committer: Aaron Bentley
  • Date: 2006-09-19 16:17:31 UTC
  • mto: This revision was merged to the branch mainline in revision 2162.
  • Revision ID: abentley@panoramicfeedback.com-20060919161731-4a099268251f858c
Implement specific file id and dangling id handling

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