/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: Andrew Bennetts
  • Date: 2006-09-15 01:03:27 UTC
  • mto: (1910.19.10 add get_smart_client)
  • mto: This revision was merged to the branch mainline in revision 2012.
  • Revision ID: andrew.bennetts@canonical.com-20060915010327-9a82c824fe58fc00
Use sys.stdout consistently, rather than mixed with print.

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
 
 
891
        old_env = {}
 
892
 
 
893
        def cleanup_environment():
 
894
            for env_var, value in env_changes.iteritems():
 
895
                old_env[env_var] = osutils.set_or_unset_env(env_var, value)
 
896
 
 
897
        def restore_environment():
 
898
            for env_var, value in old_env.iteritems():
 
899
                osutils.set_or_unset_env(env_var, value)
 
900
 
 
901
        bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
 
902
        if not os.path.isfile(bzr_path):
 
903
            # We are probably installed. Assume sys.argv is the right file
 
904
            bzr_path = sys.argv[0]
 
905
        args = list(args)
 
906
 
 
907
        try:
 
908
            # win32 subprocess doesn't support preexec_fn
 
909
            # so we will avoid using it on all platforms, just to
 
910
            # make sure the code path is used, and we don't break on win32
 
911
            cleanup_environment()
 
912
            process = Popen([sys.executable, bzr_path]+args,
 
913
                             stdout=PIPE, stderr=PIPE)
 
914
        finally:
 
915
            restore_environment()
 
916
            
 
917
        out = process.stdout.read()
 
918
        err = process.stderr.read()
 
919
 
 
920
        if kwargs.get('universal_newlines', False):
 
921
            out = out.replace('\r\n', '\n')
 
922
            err = err.replace('\r\n', '\n')
 
923
 
 
924
        retcode = process.wait()
 
925
        supplied_retcode = kwargs.get('retcode', 0)
 
926
        if supplied_retcode is not None:
 
927
            if supplied_retcode != retcode:
 
928
                mutter('Output of bzr %s:\n%s', args, out)
 
929
                mutter('Error for bzr %s:\n%s', args, err)
 
930
                self.fail('Command bzr %s failed with retcode %s != %s'
 
931
                          % (args, supplied_retcode, retcode))
 
932
        return [out, err]
 
933
 
 
934
    def start_bzr_subprocess(self, *args):
 
935
        """Start bzr in a subprocess for testing.
 
936
 
 
937
        This starts a new Python interpreter and runs bzr in there. 
 
938
        This should only be used for tests that have a justifiable need for
 
939
        this isolation: e.g. they are testing startup time, or signal
 
940
        handling, or early startup code, etc.  Subprocess code can't be 
 
941
        profiled or debugged so easily.
 
942
 
 
943
        :returns: Popen object for the started process.
 
944
        """
 
945
        # TODO: this ought to remove BZR_PDB when running the subprocess- the
 
946
        # user probably doesn't want to debug it, and anyhow since its files
 
947
        # are redirected they can't usefully get at it.  It just makes the
 
948
        # test suite hang.
 
949
        bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
 
950
        args = list(args)
 
951
        process = Popen([sys.executable, bzr_path]+args, stdout=PIPE, 
 
952
                         stderr=PIPE)
 
953
        return process
 
954
 
 
955
    def finish_bzr_subprocess(self, process, retcode=0, send_signal=None):
 
956
        """Finish the execution of process.
 
957
 
 
958
        :param process: the Popen object returned from start_bzr_subprocess.
 
959
        :param retcode: the expected return code of the process, if None any
 
960
            value is accepted, otherwise if there is a difference a failure will
 
961
            be raised.
 
962
        :param send_signal: an optional signal to send to the process.
 
963
        :returns: (stdout, stderr)
 
964
        """
 
965
        if send_signal is not None:
 
966
            os.kill(process.pid, send_signal)
 
967
        result = process.communicate()
 
968
        if retcode is not None:
 
969
            self.assertEqual(retcode, process.returncode)
 
970
        return result
 
971
 
 
972
    def check_inventory_shape(self, inv, shape):
 
973
        """Compare an inventory to a list of expected names.
 
974
 
 
975
        Fail if they are not precisely equal.
 
976
        """
 
977
        extras = []
 
978
        shape = list(shape)             # copy
 
979
        for path, ie in inv.entries():
 
980
            name = path.replace('\\', '/')
 
981
            if ie.kind == 'dir':
 
982
                name = name + '/'
 
983
            if name in shape:
 
984
                shape.remove(name)
 
985
            else:
 
986
                extras.append(name)
 
987
        if shape:
 
988
            self.fail("expected paths not found in inventory: %r" % shape)
 
989
        if extras:
 
990
            self.fail("unexpected paths found in inventory: %r" % extras)
 
991
 
 
992
    def apply_redirected(self, stdin=None, stdout=None, stderr=None,
 
993
                         a_callable=None, *args, **kwargs):
 
994
        """Call callable with redirected std io pipes.
 
995
 
 
996
        Returns the return code."""
 
997
        if not callable(a_callable):
 
998
            raise ValueError("a_callable must be callable.")
 
999
        if stdin is None:
 
1000
            stdin = StringIO("")
 
1001
        if stdout is None:
 
1002
            if getattr(self, "_log_file", None) is not None:
 
1003
                stdout = self._log_file
 
1004
            else:
 
1005
                stdout = StringIO()
 
1006
        if stderr is None:
 
1007
            if getattr(self, "_log_file", None is not None):
 
1008
                stderr = self._log_file
 
1009
            else:
 
1010
                stderr = StringIO()
 
1011
        real_stdin = sys.stdin
 
1012
        real_stdout = sys.stdout
 
1013
        real_stderr = sys.stderr
 
1014
        try:
 
1015
            sys.stdout = stdout
 
1016
            sys.stderr = stderr
 
1017
            sys.stdin = stdin
 
1018
            return a_callable(*args, **kwargs)
 
1019
        finally:
 
1020
            sys.stdout = real_stdout
 
1021
            sys.stderr = real_stderr
 
1022
            sys.stdin = real_stdin
 
1023
 
 
1024
    @symbol_versioning.deprecated_method(symbol_versioning.zero_eleven)
 
1025
    def merge(self, branch_from, wt_to):
 
1026
        """A helper for tests to do a ui-less merge.
 
1027
 
 
1028
        This should move to the main library when someone has time to integrate
 
1029
        it in.
 
1030
        """
 
1031
        # minimal ui-less merge.
 
1032
        wt_to.branch.fetch(branch_from)
 
1033
        base_rev = common_ancestor(branch_from.last_revision(),
 
1034
                                   wt_to.branch.last_revision(),
 
1035
                                   wt_to.branch.repository)
 
1036
        merge_inner(wt_to.branch, branch_from.basis_tree(),
 
1037
                    wt_to.branch.repository.revision_tree(base_rev),
 
1038
                    this_tree=wt_to)
 
1039
        wt_to.add_parent_tree_id(branch_from.last_revision())
 
1040
 
 
1041
 
 
1042
BzrTestBase = TestCase
 
1043
 
 
1044
     
 
1045
class TestCaseInTempDir(TestCase):
 
1046
    """Derived class that runs a test within a temporary directory.
 
1047
 
 
1048
    This is useful for tests that need to create a branch, etc.
 
1049
 
 
1050
    The directory is created in a slightly complex way: for each
 
1051
    Python invocation, a new temporary top-level directory is created.
 
1052
    All test cases create their own directory within that.  If the
 
1053
    tests complete successfully, the directory is removed.
 
1054
 
 
1055
    InTempDir is an old alias for FunctionalTestCase.
 
1056
    """
 
1057
 
 
1058
    TEST_ROOT = None
 
1059
    _TEST_NAME = 'test'
 
1060
    OVERRIDE_PYTHON = 'python'
 
1061
 
 
1062
    def check_file_contents(self, filename, expect):
 
1063
        self.log("check contents of file %s" % filename)
 
1064
        contents = file(filename, 'r').read()
 
1065
        if contents != expect:
 
1066
            self.log("expected: %r" % expect)
 
1067
            self.log("actually: %r" % contents)
 
1068
            self.fail("contents of %s not as expected" % filename)
 
1069
 
 
1070
    def _make_test_root(self):
 
1071
        if TestCaseInTempDir.TEST_ROOT is not None:
 
1072
            return
 
1073
        i = 0
 
1074
        while True:
 
1075
            root = u'test%04d.tmp' % i
 
1076
            try:
 
1077
                os.mkdir(root)
 
1078
            except OSError, e:
 
1079
                if e.errno == errno.EEXIST:
 
1080
                    i += 1
 
1081
                    continue
 
1082
                else:
 
1083
                    raise
 
1084
            # successfully created
 
1085
            TestCaseInTempDir.TEST_ROOT = osutils.abspath(root)
 
1086
            break
 
1087
        # make a fake bzr directory there to prevent any tests propagating
 
1088
        # up onto the source directory's real branch
 
1089
        bzrdir.BzrDir.create_standalone_workingtree(TestCaseInTempDir.TEST_ROOT)
 
1090
 
 
1091
    def setUp(self):
 
1092
        super(TestCaseInTempDir, self).setUp()
 
1093
        self._make_test_root()
 
1094
        _currentdir = os.getcwdu()
 
1095
        # shorten the name, to avoid test failures due to path length
 
1096
        short_id = self.id().replace('bzrlib.tests.', '') \
 
1097
                   .replace('__main__.', '')[-100:]
 
1098
        # it's possible the same test class is run several times for
 
1099
        # parameterized tests, so make sure the names don't collide.  
 
1100
        i = 0
 
1101
        while True:
 
1102
            if i > 0:
 
1103
                candidate_dir = '%s/%s.%d' % (self.TEST_ROOT, short_id, i)
 
1104
            else:
 
1105
                candidate_dir = '%s/%s' % (self.TEST_ROOT, short_id)
 
1106
            if os.path.exists(candidate_dir):
 
1107
                i = i + 1
 
1108
                continue
 
1109
            else:
 
1110
                os.mkdir(candidate_dir)
 
1111
                self.test_home_dir = candidate_dir + '/home'
 
1112
                os.mkdir(self.test_home_dir)
 
1113
                self.test_dir = candidate_dir + '/work'
 
1114
                os.mkdir(self.test_dir)
 
1115
                os.chdir(self.test_dir)
 
1116
                break
 
1117
        os.environ['HOME'] = self.test_home_dir
 
1118
        os.environ['APPDATA'] = self.test_home_dir
 
1119
        def _leaveDirectory():
 
1120
            os.chdir(_currentdir)
 
1121
        self.addCleanup(_leaveDirectory)
 
1122
        
 
1123
    def build_tree(self, shape, line_endings='native', transport=None):
 
1124
        """Build a test tree according to a pattern.
 
1125
 
 
1126
        shape is a sequence of file specifications.  If the final
 
1127
        character is '/', a directory is created.
 
1128
 
 
1129
        This assumes that all the elements in the tree being built are new.
 
1130
 
 
1131
        This doesn't add anything to a branch.
 
1132
        :param line_endings: Either 'binary' or 'native'
 
1133
                             in binary mode, exact contents are written
 
1134
                             in native mode, the line endings match the
 
1135
                             default platform endings.
 
1136
 
 
1137
        :param transport: A transport to write to, for building trees on 
 
1138
                          VFS's. If the transport is readonly or None,
 
1139
                          "." is opened automatically.
 
1140
        """
 
1141
        # It's OK to just create them using forward slashes on windows.
 
1142
        if transport is None or transport.is_readonly():
 
1143
            transport = get_transport(".")
 
1144
        for name in shape:
 
1145
            self.assert_(isinstance(name, basestring))
 
1146
            if name[-1] == '/':
 
1147
                transport.mkdir(urlutils.escape(name[:-1]))
 
1148
            else:
 
1149
                if line_endings == 'binary':
 
1150
                    end = '\n'
 
1151
                elif line_endings == 'native':
 
1152
                    end = os.linesep
 
1153
                else:
 
1154
                    raise errors.BzrError('Invalid line ending request %r' % (line_endings,))
 
1155
                content = "contents of %s%s" % (name.encode('utf-8'), end)
 
1156
                # Technically 'put()' is the right command. However, put
 
1157
                # uses an AtomicFile, which requires an extra rename into place
 
1158
                # As long as the files didn't exist in the past, append() will
 
1159
                # do the same thing as put()
 
1160
                # On jam's machine, make_kernel_like_tree is:
 
1161
                #   put:    4.5-7.5s (averaging 6s)
 
1162
                #   append: 2.9-4.5s
 
1163
                #   put_non_atomic: 2.9-4.5s
 
1164
                transport.put_bytes_non_atomic(urlutils.escape(name), content)
 
1165
 
 
1166
    def build_tree_contents(self, shape):
 
1167
        build_tree_contents(shape)
 
1168
 
 
1169
    def failUnlessExists(self, path):
 
1170
        """Fail unless path, which may be abs or relative, exists."""
 
1171
        self.failUnless(osutils.lexists(path))
 
1172
 
 
1173
    def failIfExists(self, path):
 
1174
        """Fail if path, which may be abs or relative, exists."""
 
1175
        self.failIf(osutils.lexists(path))
 
1176
        
 
1177
    def assertFileEqual(self, content, path):
 
1178
        """Fail if path does not contain 'content'."""
 
1179
        self.failUnless(osutils.lexists(path))
 
1180
        # TODO: jam 20060427 Shouldn't this be 'rb'?
 
1181
        self.assertEqualDiff(content, open(path, 'r').read())
 
1182
 
 
1183
 
 
1184
class TestCaseWithTransport(TestCaseInTempDir):
 
1185
    """A test case that provides get_url and get_readonly_url facilities.
 
1186
 
 
1187
    These back onto two transport servers, one for readonly access and one for
 
1188
    read write access.
 
1189
 
 
1190
    If no explicit class is provided for readonly access, a
 
1191
    ReadonlyTransportDecorator is used instead which allows the use of non disk
 
1192
    based read write transports.
 
1193
 
 
1194
    If an explicit class is provided for readonly access, that server and the 
 
1195
    readwrite one must both define get_url() as resolving to os.getcwd().
 
1196
    """
 
1197
 
 
1198
    def __init__(self, methodName='testMethod'):
 
1199
        super(TestCaseWithTransport, self).__init__(methodName)
 
1200
        self.__readonly_server = None
 
1201
        self.__server = None
 
1202
        self.transport_server = default_transport
 
1203
        self.transport_readonly_server = None
 
1204
 
 
1205
    def get_readonly_url(self, relpath=None):
 
1206
        """Get a URL for the readonly transport.
 
1207
 
 
1208
        This will either be backed by '.' or a decorator to the transport 
 
1209
        used by self.get_url()
 
1210
        relpath provides for clients to get a path relative to the base url.
 
1211
        These should only be downwards relative, not upwards.
 
1212
        """
 
1213
        base = self.get_readonly_server().get_url()
 
1214
        if relpath is not None:
 
1215
            if not base.endswith('/'):
 
1216
                base = base + '/'
 
1217
            base = base + relpath
 
1218
        return base
 
1219
 
 
1220
    def get_readonly_server(self):
 
1221
        """Get the server instance for the readonly transport
 
1222
 
 
1223
        This is useful for some tests with specific servers to do diagnostics.
 
1224
        """
 
1225
        if self.__readonly_server is None:
 
1226
            if self.transport_readonly_server is None:
 
1227
                # readonly decorator requested
 
1228
                # bring up the server
 
1229
                self.get_url()
 
1230
                self.__readonly_server = ReadonlyServer()
 
1231
                self.__readonly_server.setUp(self.__server)
 
1232
            else:
 
1233
                self.__readonly_server = self.transport_readonly_server()
 
1234
                self.__readonly_server.setUp()
 
1235
            self.addCleanup(self.__readonly_server.tearDown)
 
1236
        return self.__readonly_server
 
1237
 
 
1238
    def get_server(self):
 
1239
        """Get the read/write server instance.
 
1240
 
 
1241
        This is useful for some tests with specific servers that need
 
1242
        diagnostics.
 
1243
        """
 
1244
        if self.__server is None:
 
1245
            self.__server = self.transport_server()
 
1246
            self.__server.setUp()
 
1247
            self.addCleanup(self.__server.tearDown)
 
1248
        return self.__server
 
1249
 
 
1250
    def get_url(self, relpath=None):
 
1251
        """Get a URL (or maybe a path) for the readwrite transport.
 
1252
 
 
1253
        This will either be backed by '.' or to an equivalent non-file based
 
1254
        facility.
 
1255
        relpath provides for clients to get a path relative to the base url.
 
1256
        These should only be downwards relative, not upwards.
 
1257
        """
 
1258
        base = self.get_server().get_url()
 
1259
        if relpath is not None and relpath != '.':
 
1260
            if not base.endswith('/'):
 
1261
                base = base + '/'
 
1262
            # XXX: Really base should be a url; we did after all call
 
1263
            # get_url()!  But sometimes it's just a path (from
 
1264
            # LocalAbspathServer), and it'd be wrong to append urlescaped data
 
1265
            # to a non-escaped local path.
 
1266
            if base.startswith('./') or base.startswith('/'):
 
1267
                base += relpath
 
1268
            else:
 
1269
                base += urlutils.escape(relpath)
 
1270
        return base
 
1271
 
 
1272
    def get_transport(self):
 
1273
        """Return a writeable transport for the test scratch space"""
 
1274
        t = get_transport(self.get_url())
 
1275
        self.assertFalse(t.is_readonly())
 
1276
        return t
 
1277
 
 
1278
    def get_readonly_transport(self):
 
1279
        """Return a readonly transport for the test scratch space
 
1280
        
 
1281
        This can be used to test that operations which should only need
 
1282
        readonly access in fact do not try to write.
 
1283
        """
 
1284
        t = get_transport(self.get_readonly_url())
 
1285
        self.assertTrue(t.is_readonly())
 
1286
        return t
 
1287
 
 
1288
    def make_branch(self, relpath, format=None):
 
1289
        """Create a branch on the transport at relpath."""
 
1290
        repo = self.make_repository(relpath, format=format)
 
1291
        return repo.bzrdir.create_branch()
 
1292
 
 
1293
    def make_bzrdir(self, relpath, format=None):
 
1294
        try:
 
1295
            # might be a relative or absolute path
 
1296
            maybe_a_url = self.get_url(relpath)
 
1297
            segments = maybe_a_url.rsplit('/', 1)
 
1298
            t = get_transport(maybe_a_url)
 
1299
            if len(segments) > 1 and segments[-1] not in ('', '.'):
 
1300
                try:
 
1301
                    t.mkdir('.')
 
1302
                except errors.FileExists:
 
1303
                    pass
 
1304
            if format is None:
 
1305
                format = bzrlib.bzrdir.BzrDirFormat.get_default_format()
 
1306
            return format.initialize_on_transport(t)
 
1307
        except errors.UninitializableFormat:
 
1308
            raise TestSkipped("Format %s is not initializable." % format)
 
1309
 
 
1310
    def make_repository(self, relpath, shared=False, format=None):
 
1311
        """Create a repository on our default transport at relpath."""
 
1312
        made_control = self.make_bzrdir(relpath, format=format)
 
1313
        return made_control.create_repository(shared=shared)
 
1314
 
 
1315
    def make_branch_and_tree(self, relpath, format=None):
 
1316
        """Create a branch on the transport and a tree locally.
 
1317
 
 
1318
        If the transport is not a LocalTransport, the Tree can't be created on
 
1319
        the transport.  In that case the working tree is created in the local
 
1320
        directory, and the returned tree's branch and repository will also be
 
1321
        accessed locally.
 
1322
 
 
1323
        This will fail if the original default transport for this test
 
1324
        case wasn't backed by the working directory, as the branch won't
 
1325
        be on disk for us to open it.  
 
1326
 
 
1327
        :param format: The BzrDirFormat.
 
1328
        :returns: the WorkingTree.
 
1329
        """
 
1330
        # TODO: always use the local disk path for the working tree,
 
1331
        # this obviously requires a format that supports branch references
 
1332
        # so check for that by checking bzrdir.BzrDirFormat.get_default_format()
 
1333
        # RBC 20060208
 
1334
        b = self.make_branch(relpath, format=format)
 
1335
        try:
 
1336
            return b.bzrdir.create_workingtree()
 
1337
        except errors.NotLocalUrl:
 
1338
            # We can only make working trees locally at the moment.  If the
 
1339
            # transport can't support them, then reopen the branch on a local
 
1340
            # transport, and create the working tree there.  
 
1341
            #
 
1342
            # Possibly we should instead keep
 
1343
            # the non-disk-backed branch and create a local checkout?
 
1344
            bd = bzrdir.BzrDir.open(relpath)
 
1345
            return bd.create_workingtree()
 
1346
 
 
1347
    def assertIsDirectory(self, relpath, transport):
 
1348
        """Assert that relpath within transport is a directory.
 
1349
 
 
1350
        This may not be possible on all transports; in that case it propagates
 
1351
        a TransportNotPossible.
 
1352
        """
 
1353
        try:
 
1354
            mode = transport.stat(relpath).st_mode
 
1355
        except errors.NoSuchFile:
 
1356
            self.fail("path %s is not a directory; no such file"
 
1357
                      % (relpath))
 
1358
        if not stat.S_ISDIR(mode):
 
1359
            self.fail("path %s is not a directory; has mode %#o"
 
1360
                      % (relpath, mode))
 
1361
 
 
1362
 
 
1363
class ChrootedTestCase(TestCaseWithTransport):
 
1364
    """A support class that provides readonly urls outside the local namespace.
 
1365
 
 
1366
    This is done by checking if self.transport_server is a MemoryServer. if it
 
1367
    is then we are chrooted already, if it is not then an HttpServer is used
 
1368
    for readonly urls.
 
1369
 
 
1370
    TODO RBC 20060127: make this an option to TestCaseWithTransport so it can
 
1371
                       be used without needed to redo it when a different 
 
1372
                       subclass is in use ?
 
1373
    """
 
1374
 
 
1375
    def setUp(self):
 
1376
        super(ChrootedTestCase, self).setUp()
 
1377
        if not self.transport_server == bzrlib.transport.memory.MemoryServer:
 
1378
            self.transport_readonly_server = bzrlib.transport.http.HttpServer
 
1379
 
 
1380
 
 
1381
def filter_suite_by_re(suite, pattern):
 
1382
    result = TestUtil.TestSuite()
 
1383
    filter_re = re.compile(pattern)
 
1384
    for test in iter_suite_tests(suite):
 
1385
        if filter_re.search(test.id()):
 
1386
            result.addTest(test)
 
1387
    return result
 
1388
 
 
1389
 
 
1390
def run_suite(suite, name='test', verbose=False, pattern=".*",
 
1391
              stop_on_failure=False, keep_output=False,
 
1392
              transport=None, lsprof_timed=None, bench_history=None):
 
1393
    TestCaseInTempDir._TEST_NAME = name
 
1394
    TestCase._gather_lsprof_in_benchmarks = lsprof_timed
 
1395
    if verbose:
 
1396
        verbosity = 2
 
1397
        pb = None
 
1398
    else:
 
1399
        verbosity = 1
 
1400
        pb = progress.ProgressBar()
 
1401
    runner = TextTestRunner(stream=sys.stdout,
 
1402
                            descriptions=0,
 
1403
                            verbosity=verbosity,
 
1404
                            keep_output=keep_output,
 
1405
                            pb=pb,
 
1406
                            bench_history=bench_history)
 
1407
    runner.stop_on_failure=stop_on_failure
 
1408
    if pattern != '.*':
 
1409
        suite = filter_suite_by_re(suite, pattern)
 
1410
    result = runner.run(suite)
 
1411
    return result.wasSuccessful()
 
1412
 
 
1413
 
 
1414
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
 
1415
             keep_output=False,
 
1416
             transport=None,
 
1417
             test_suite_factory=None,
 
1418
             lsprof_timed=None,
 
1419
             bench_history=None):
 
1420
    """Run the whole test suite under the enhanced runner"""
 
1421
    # XXX: Very ugly way to do this...
 
1422
    # Disable warning about old formats because we don't want it to disturb
 
1423
    # any blackbox tests.
 
1424
    from bzrlib import repository
 
1425
    repository._deprecation_warning_done = True
 
1426
 
 
1427
    global default_transport
 
1428
    if transport is None:
 
1429
        transport = default_transport
 
1430
    old_transport = default_transport
 
1431
    default_transport = transport
 
1432
    try:
 
1433
        if test_suite_factory is None:
 
1434
            suite = test_suite()
 
1435
        else:
 
1436
            suite = test_suite_factory()
 
1437
        return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
 
1438
                     stop_on_failure=stop_on_failure, keep_output=keep_output,
 
1439
                     transport=transport,
 
1440
                     lsprof_timed=lsprof_timed,
 
1441
                     bench_history=bench_history)
 
1442
    finally:
 
1443
        default_transport = old_transport
 
1444
 
 
1445
 
 
1446
def test_suite():
 
1447
    """Build and return TestSuite for the whole of bzrlib.
 
1448
    
 
1449
    This function can be replaced if you need to change the default test
 
1450
    suite on a global basis, but it is not encouraged.
 
1451
    """
 
1452
    testmod_names = [
 
1453
                   'bzrlib.tests.test_ancestry',
 
1454
                   'bzrlib.tests.test_api',
 
1455
                   'bzrlib.tests.test_atomicfile',
 
1456
                   'bzrlib.tests.test_bad_files',
 
1457
                   'bzrlib.tests.test_branch',
 
1458
                   'bzrlib.tests.test_bundle',
 
1459
                   'bzrlib.tests.test_bzrdir',
 
1460
                   'bzrlib.tests.test_cache_utf8',
 
1461
                   'bzrlib.tests.test_command',
 
1462
                   'bzrlib.tests.test_commit',
 
1463
                   'bzrlib.tests.test_commit_merge',
 
1464
                   'bzrlib.tests.test_config',
 
1465
                   'bzrlib.tests.test_conflicts',
 
1466
                   'bzrlib.tests.test_decorators',
 
1467
                   'bzrlib.tests.test_diff',
 
1468
                   'bzrlib.tests.test_doc_generate',
 
1469
                   'bzrlib.tests.test_errors',
 
1470
                   'bzrlib.tests.test_escaped_store',
 
1471
                   'bzrlib.tests.test_fetch',
 
1472
                   'bzrlib.tests.test_gpg',
 
1473
                   'bzrlib.tests.test_graph',
 
1474
                   'bzrlib.tests.test_hashcache',
 
1475
                   'bzrlib.tests.test_http',
 
1476
                   'bzrlib.tests.test_http_response',
 
1477
                   'bzrlib.tests.test_identitymap',
 
1478
                   'bzrlib.tests.test_ignores',
 
1479
                   'bzrlib.tests.test_inv',
 
1480
                   'bzrlib.tests.test_knit',
 
1481
                   'bzrlib.tests.test_lazy_import',
 
1482
                   'bzrlib.tests.test_lockdir',
 
1483
                   'bzrlib.tests.test_lockable_files',
 
1484
                   'bzrlib.tests.test_log',
 
1485
                   'bzrlib.tests.test_merge',
 
1486
                   'bzrlib.tests.test_merge3',
 
1487
                   'bzrlib.tests.test_merge_core',
 
1488
                   'bzrlib.tests.test_missing',
 
1489
                   'bzrlib.tests.test_msgeditor',
 
1490
                   'bzrlib.tests.test_nonascii',
 
1491
                   'bzrlib.tests.test_options',
 
1492
                   'bzrlib.tests.test_osutils',
 
1493
                   'bzrlib.tests.test_patch',
 
1494
                   'bzrlib.tests.test_patches',
 
1495
                   'bzrlib.tests.test_permissions',
 
1496
                   'bzrlib.tests.test_plugins',
 
1497
                   'bzrlib.tests.test_progress',
 
1498
                   'bzrlib.tests.test_reconcile',
 
1499
                   'bzrlib.tests.test_repository',
 
1500
                   'bzrlib.tests.test_revert',
 
1501
                   'bzrlib.tests.test_revision',
 
1502
                   'bzrlib.tests.test_revisionnamespaces',
 
1503
                   'bzrlib.tests.test_revisiontree',
 
1504
                   'bzrlib.tests.test_rio',
 
1505
                   'bzrlib.tests.test_sampler',
 
1506
                   'bzrlib.tests.test_selftest',
 
1507
                   'bzrlib.tests.test_setup',
 
1508
                   'bzrlib.tests.test_sftp_transport',
 
1509
                   'bzrlib.tests.test_ftp_transport',
 
1510
                   'bzrlib.tests.test_smart_add',
 
1511
                   'bzrlib.tests.test_source',
 
1512
                   'bzrlib.tests.test_status',
 
1513
                   'bzrlib.tests.test_store',
 
1514
                   'bzrlib.tests.test_symbol_versioning',
 
1515
                   'bzrlib.tests.test_testament',
 
1516
                   'bzrlib.tests.test_textfile',
 
1517
                   'bzrlib.tests.test_textmerge',
 
1518
                   'bzrlib.tests.test_trace',
 
1519
                   'bzrlib.tests.test_transactions',
 
1520
                   'bzrlib.tests.test_transform',
 
1521
                   'bzrlib.tests.test_transport',
 
1522
                   'bzrlib.tests.test_tree',
 
1523
                   'bzrlib.tests.test_tsort',
 
1524
                   'bzrlib.tests.test_tuned_gzip',
 
1525
                   'bzrlib.tests.test_ui',
 
1526
                   'bzrlib.tests.test_upgrade',
 
1527
                   'bzrlib.tests.test_urlutils',
 
1528
                   'bzrlib.tests.test_versionedfile',
 
1529
                   'bzrlib.tests.test_version',
 
1530
                   'bzrlib.tests.test_weave',
 
1531
                   'bzrlib.tests.test_whitebox',
 
1532
                   'bzrlib.tests.test_workingtree',
 
1533
                   'bzrlib.tests.test_xml',
 
1534
                   ]
 
1535
    test_transport_implementations = [
 
1536
        'bzrlib.tests.test_transport_implementations',
 
1537
        'bzrlib.tests.test_read_bundle',
 
1538
        ]
 
1539
    suite = TestUtil.TestSuite()
 
1540
    loader = TestUtil.TestLoader()
 
1541
    suite.addTest(loader.loadTestsFromModuleNames(testmod_names))
 
1542
    from bzrlib.transport import TransportTestProviderAdapter
 
1543
    adapter = TransportTestProviderAdapter()
 
1544
    adapt_modules(test_transport_implementations, adapter, loader, suite)
 
1545
    for package in packages_to_test():
 
1546
        suite.addTest(package.test_suite())
 
1547
    for m in MODULES_TO_TEST:
 
1548
        suite.addTest(loader.loadTestsFromModule(m))
 
1549
    for m in MODULES_TO_DOCTEST:
 
1550
        suite.addTest(doctest.DocTestSuite(m))
 
1551
    for name, plugin in bzrlib.plugin.all_plugins().items():
 
1552
        if getattr(plugin, 'test_suite', None) is not None:
 
1553
            suite.addTest(plugin.test_suite())
 
1554
    return suite
 
1555
 
 
1556
 
 
1557
def adapt_modules(mods_list, adapter, loader, suite):
 
1558
    """Adapt the modules in mods_list using adapter and add to suite."""
 
1559
    for test in iter_suite_tests(loader.loadTestsFromModuleNames(mods_list)):
 
1560
        suite.addTests(adapter.adapt(test))