/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: Robert Collins
  • Date: 2006-05-04 10:36:13 UTC
  • mto: (1697.1.1 integration)
  • mto: This revision was merged to the branch mainline in revision 1701.
  • Revision ID: robertc@robertcollins.net-20060504103613-bf4d13f925aa8520
Extend LockableFiles to support break_lock() calls.

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 errno
 
33
import logging
 
34
import os
 
35
import re
 
36
import shutil
 
37
import stat
 
38
import sys
 
39
import tempfile
 
40
import unittest
 
41
import time
 
42
 
 
43
 
 
44
import bzrlib.branch
 
45
import bzrlib.bzrdir as bzrdir
 
46
import bzrlib.commands
 
47
import bzrlib.errors as errors
 
48
import bzrlib.inventory
 
49
import bzrlib.iterablefile
 
50
import bzrlib.lockdir
 
51
from bzrlib.merge import merge_inner
 
52
import bzrlib.merge3
 
53
import bzrlib.osutils
 
54
import bzrlib.osutils as osutils
 
55
import bzrlib.plugin
 
56
from bzrlib.revision import common_ancestor
 
57
import bzrlib.store
 
58
import bzrlib.trace
 
59
from bzrlib.transport import urlescape, get_transport
 
60
import bzrlib.transport
 
61
from bzrlib.transport.local import LocalRelpathServer
 
62
from bzrlib.transport.readonly import ReadonlyServer
 
63
from bzrlib.trace import mutter
 
64
from bzrlib.tests.TestUtil import TestLoader, TestSuite
 
65
from bzrlib.tests.treeshape import build_tree_contents
 
66
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2
 
67
 
 
68
default_transport = LocalRelpathServer
 
69
 
 
70
MODULES_TO_TEST = []
 
71
MODULES_TO_DOCTEST = [
 
72
                      bzrlib.branch,
 
73
                      bzrlib.commands,
 
74
                      bzrlib.errors,
 
75
                      bzrlib.inventory,
 
76
                      bzrlib.iterablefile,
 
77
                      bzrlib.lockdir,
 
78
                      bzrlib.merge3,
 
79
                      bzrlib.option,
 
80
                      bzrlib.osutils,
 
81
                      bzrlib.store
 
82
                      ]
 
83
def packages_to_test():
 
84
    """Return a list of packages to test.
 
85
 
 
86
    The packages are not globally imported so that import failures are
 
87
    triggered when running selftest, not when importing the command.
 
88
    """
 
89
    import bzrlib.doc
 
90
    import bzrlib.tests.blackbox
 
91
    import bzrlib.tests.branch_implementations
 
92
    import bzrlib.tests.bzrdir_implementations
 
93
    import bzrlib.tests.interrepository_implementations
 
94
    import bzrlib.tests.interversionedfile_implementations
 
95
    import bzrlib.tests.repository_implementations
 
96
    import bzrlib.tests.revisionstore_implementations
 
97
    import bzrlib.tests.workingtree_implementations
 
98
    return [
 
99
            bzrlib.doc,
 
100
            bzrlib.tests.blackbox,
 
101
            bzrlib.tests.branch_implementations,
 
102
            bzrlib.tests.bzrdir_implementations,
 
103
            bzrlib.tests.interrepository_implementations,
 
104
            bzrlib.tests.interversionedfile_implementations,
 
105
            bzrlib.tests.repository_implementations,
 
106
            bzrlib.tests.revisionstore_implementations,
 
107
            bzrlib.tests.workingtree_implementations,
 
108
            ]
 
109
 
 
110
 
 
111
class _MyResult(unittest._TextTestResult):
 
112
    """Custom TestResult.
 
113
 
 
114
    Shows output in a different format, including displaying runtime for tests.
 
115
    """
 
116
    stop_early = False
 
117
 
 
118
    def _elapsedTime(self):
 
119
        return "%5dms" % (1000 * (time.time() - self._start_time))
 
120
 
 
121
    def startTest(self, test):
 
122
        unittest.TestResult.startTest(self, test)
 
123
        # In a short description, the important words are in
 
124
        # the beginning, but in an id, the important words are
 
125
        # at the end
 
126
        SHOW_DESCRIPTIONS = False
 
127
        if self.showAll:
 
128
            width = osutils.terminal_width()
 
129
            name_width = width - 15
 
130
            what = None
 
131
            if SHOW_DESCRIPTIONS:
 
132
                what = test.shortDescription()
 
133
                if what:
 
134
                    if len(what) > name_width:
 
135
                        what = what[:name_width-3] + '...'
 
136
            if what is None:
 
137
                what = test.id()
 
138
                if what.startswith('bzrlib.tests.'):
 
139
                    what = what[13:]
 
140
                if len(what) > name_width:
 
141
                    what = '...' + what[3-name_width:]
 
142
            what = what.ljust(name_width)
 
143
            self.stream.write(what)
 
144
        self.stream.flush()
 
145
        self._start_time = time.time()
 
146
 
 
147
    def addError(self, test, err):
 
148
        if isinstance(err[1], TestSkipped):
 
149
            return self.addSkipped(test, err)    
 
150
        unittest.TestResult.addError(self, test, err)
 
151
        if self.showAll:
 
152
            self.stream.writeln("ERROR %s" % self._elapsedTime())
 
153
        elif self.dots:
 
154
            self.stream.write('E')
 
155
        self.stream.flush()
 
156
        if self.stop_early:
 
157
            self.stop()
 
158
 
 
159
    def addFailure(self, test, err):
 
160
        unittest.TestResult.addFailure(self, test, err)
 
161
        if self.showAll:
 
162
            self.stream.writeln(" FAIL %s" % self._elapsedTime())
 
163
        elif self.dots:
 
164
            self.stream.write('F')
 
165
        self.stream.flush()
 
166
        if self.stop_early:
 
167
            self.stop()
 
168
 
 
169
    def addSuccess(self, test):
 
170
        if self.showAll:
 
171
            self.stream.writeln('   OK %s' % self._elapsedTime())
 
172
        elif self.dots:
 
173
            self.stream.write('~')
 
174
        self.stream.flush()
 
175
        unittest.TestResult.addSuccess(self, test)
 
176
 
 
177
    def addSkipped(self, test, skip_excinfo):
 
178
        if self.showAll:
 
179
            print >>self.stream, ' SKIP %s' % self._elapsedTime()
 
180
            print >>self.stream, '     %s' % skip_excinfo[1]
 
181
        elif self.dots:
 
182
            self.stream.write('S')
 
183
        self.stream.flush()
 
184
        # seems best to treat this as success from point-of-view of unittest
 
185
        # -- it actually does nothing so it barely matters :)
 
186
        unittest.TestResult.addSuccess(self, test)
 
187
 
 
188
    def printErrorList(self, flavour, errors):
 
189
        for test, err in errors:
 
190
            self.stream.writeln(self.separator1)
 
191
            self.stream.writeln("%s: %s" % (flavour, self.getDescription(test)))
 
192
            if getattr(test, '_get_log', None) is not None:
 
193
                print >>self.stream
 
194
                print >>self.stream, \
 
195
                        ('vvvv[log from %s]' % test.id()).ljust(78,'-')
 
196
                print >>self.stream, test._get_log()
 
197
                print >>self.stream, \
 
198
                        ('^^^^[log from %s]' % test.id()).ljust(78,'-')
 
199
            self.stream.writeln(self.separator2)
 
200
            self.stream.writeln("%s" % err)
 
201
 
 
202
 
 
203
class TextTestRunner(unittest.TextTestRunner):
 
204
    stop_on_failure = False
 
205
 
 
206
    def _makeResult(self):
 
207
        result = _MyResult(self.stream, self.descriptions, self.verbosity)
 
208
        result.stop_early = self.stop_on_failure
 
209
        return result
 
210
 
 
211
 
 
212
def iter_suite_tests(suite):
 
213
    """Return all tests in a suite, recursing through nested suites"""
 
214
    for item in suite._tests:
 
215
        if isinstance(item, unittest.TestCase):
 
216
            yield item
 
217
        elif isinstance(item, unittest.TestSuite):
 
218
            for r in iter_suite_tests(item):
 
219
                yield r
 
220
        else:
 
221
            raise Exception('unknown object %r inside test suite %r'
 
222
                            % (item, suite))
 
223
 
 
224
 
 
225
class TestSkipped(Exception):
 
226
    """Indicates that a test was intentionally skipped, rather than failing."""
 
227
    # XXX: Not used yet
 
228
 
 
229
 
 
230
class CommandFailed(Exception):
 
231
    pass
 
232
 
 
233
class TestCase(unittest.TestCase):
 
234
    """Base class for bzr unit tests.
 
235
    
 
236
    Tests that need access to disk resources should subclass 
 
237
    TestCaseInTempDir not TestCase.
 
238
 
 
239
    Error and debug log messages are redirected from their usual
 
240
    location into a temporary file, the contents of which can be
 
241
    retrieved by _get_log().  We use a real OS file, not an in-memory object,
 
242
    so that it can also capture file IO.  When the test completes this file
 
243
    is read into memory and removed from disk.
 
244
       
 
245
    There are also convenience functions to invoke bzr's command-line
 
246
    routine, and to build and check bzr trees.
 
247
   
 
248
    In addition to the usual method of overriding tearDown(), this class also
 
249
    allows subclasses to register functions into the _cleanups list, which is
 
250
    run in order as the object is torn down.  It's less likely this will be
 
251
    accidentally overlooked.
 
252
    """
 
253
 
 
254
    BZRPATH = 'bzr'
 
255
    _log_file_name = None
 
256
    _log_contents = ''
 
257
 
 
258
    def __init__(self, methodName='testMethod'):
 
259
        super(TestCase, self).__init__(methodName)
 
260
        self._cleanups = []
 
261
 
 
262
    def setUp(self):
 
263
        unittest.TestCase.setUp(self)
 
264
        self._cleanEnvironment()
 
265
        bzrlib.trace.disable_default_logging()
 
266
        self._startLogFile()
 
267
 
 
268
    def _ndiff_strings(self, a, b):
 
269
        """Return ndiff between two strings containing lines.
 
270
        
 
271
        A trailing newline is added if missing to make the strings
 
272
        print properly."""
 
273
        if b and b[-1] != '\n':
 
274
            b += '\n'
 
275
        if a and a[-1] != '\n':
 
276
            a += '\n'
 
277
        difflines = difflib.ndiff(a.splitlines(True),
 
278
                                  b.splitlines(True),
 
279
                                  linejunk=lambda x: False,
 
280
                                  charjunk=lambda x: False)
 
281
        return ''.join(difflines)
 
282
 
 
283
    def assertEqualDiff(self, a, b, message=None):
 
284
        """Assert two texts are equal, if not raise an exception.
 
285
        
 
286
        This is intended for use with multi-line strings where it can 
 
287
        be hard to find the differences by eye.
 
288
        """
 
289
        # TODO: perhaps override assertEquals to call this for strings?
 
290
        if a == b:
 
291
            return
 
292
        if message is None:
 
293
            message = "texts not equal:\n"
 
294
        raise AssertionError(message + 
 
295
                             self._ndiff_strings(a, b))      
 
296
        
 
297
    def assertEqualMode(self, mode, mode_test):
 
298
        self.assertEqual(mode, mode_test,
 
299
                         'mode mismatch %o != %o' % (mode, mode_test))
 
300
 
 
301
    def assertStartsWith(self, s, prefix):
 
302
        if not s.startswith(prefix):
 
303
            raise AssertionError('string %r does not start with %r' % (s, prefix))
 
304
 
 
305
    def assertEndsWith(self, s, suffix):
 
306
        if not s.endswith(prefix):
 
307
            raise AssertionError('string %r does not end with %r' % (s, suffix))
 
308
 
 
309
    def assertContainsRe(self, haystack, needle_re):
 
310
        """Assert that a contains something matching a regular expression."""
 
311
        if not re.search(needle_re, haystack):
 
312
            raise AssertionError('pattern "%s" not found in "%s"'
 
313
                    % (needle_re, haystack))
 
314
 
 
315
    def assertSubset(self, sublist, superlist):
 
316
        """Assert that every entry in sublist is present in superlist."""
 
317
        missing = []
 
318
        for entry in sublist:
 
319
            if entry not in superlist:
 
320
                missing.append(entry)
 
321
        if len(missing) > 0:
 
322
            raise AssertionError("value(s) %r not present in container %r" % 
 
323
                                 (missing, superlist))
 
324
 
 
325
    def assertIs(self, left, right):
 
326
        if not (left is right):
 
327
            raise AssertionError("%r is not %r." % (left, right))
 
328
 
 
329
    def assertTransportMode(self, transport, path, mode):
 
330
        """Fail if a path does not have mode mode.
 
331
        
 
332
        If modes are not supported on this transport, the assertion is ignored.
 
333
        """
 
334
        if not transport._can_roundtrip_unix_modebits():
 
335
            return
 
336
        path_stat = transport.stat(path)
 
337
        actual_mode = stat.S_IMODE(path_stat.st_mode)
 
338
        self.assertEqual(mode, actual_mode,
 
339
            'mode of %r incorrect (%o != %o)' % (path, mode, actual_mode))
 
340
 
 
341
    def assertIsInstance(self, obj, kls):
 
342
        """Fail if obj is not an instance of kls"""
 
343
        if not isinstance(obj, kls):
 
344
            self.fail("%r is an instance of %s rather than %s" % (
 
345
                obj, obj.__class__, kls))
 
346
 
 
347
    def _startLogFile(self):
 
348
        """Send bzr and test log messages to a temporary file.
 
349
 
 
350
        The file is removed as the test is torn down.
 
351
        """
 
352
        fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
 
353
        encoder, decoder, stream_reader, stream_writer = codecs.lookup('UTF-8')
 
354
        self._log_file = stream_writer(os.fdopen(fileno, 'w+'))
 
355
        self._log_nonce = bzrlib.trace.enable_test_log(self._log_file)
 
356
        self._log_file_name = name
 
357
        self.addCleanup(self._finishLogFile)
 
358
 
 
359
    def _finishLogFile(self):
 
360
        """Finished with the log file.
 
361
 
 
362
        Read contents into memory, close, and delete.
 
363
        """
 
364
        bzrlib.trace.disable_test_log(self._log_nonce)
 
365
        self._log_file.seek(0)
 
366
        self._log_contents = self._log_file.read()
 
367
        self._log_file.close()
 
368
        os.remove(self._log_file_name)
 
369
        self._log_file = self._log_file_name = None
 
370
 
 
371
    def addCleanup(self, callable):
 
372
        """Arrange to run a callable when this case is torn down.
 
373
 
 
374
        Callables are run in the reverse of the order they are registered, 
 
375
        ie last-in first-out.
 
376
        """
 
377
        if callable in self._cleanups:
 
378
            raise ValueError("cleanup function %r already registered on %s" 
 
379
                    % (callable, self))
 
380
        self._cleanups.append(callable)
 
381
 
 
382
    def _cleanEnvironment(self):
 
383
        new_env = {
 
384
            'HOME': os.getcwd(),
 
385
            'APPDATA': os.getcwd(),
 
386
            'BZREMAIL': None,
 
387
            'EMAIL': None,
 
388
        }
 
389
        self.__old_env = {}
 
390
        self.addCleanup(self._restoreEnvironment)
 
391
        for name, value in new_env.iteritems():
 
392
            self._captureVar(name, value)
 
393
 
 
394
 
 
395
    def _captureVar(self, name, newvalue):
 
396
        """Set an environment variable, preparing it to be reset when finished."""
 
397
        self.__old_env[name] = os.environ.get(name, None)
 
398
        if newvalue is None:
 
399
            if name in os.environ:
 
400
                del os.environ[name]
 
401
        else:
 
402
            os.environ[name] = newvalue
 
403
 
 
404
    @staticmethod
 
405
    def _restoreVar(name, value):
 
406
        if value is None:
 
407
            if name in os.environ:
 
408
                del os.environ[name]
 
409
        else:
 
410
            os.environ[name] = value
 
411
 
 
412
    def _restoreEnvironment(self):
 
413
        for name, value in self.__old_env.iteritems():
 
414
            self._restoreVar(name, value)
 
415
 
 
416
    def tearDown(self):
 
417
        self._runCleanups()
 
418
        unittest.TestCase.tearDown(self)
 
419
 
 
420
    def _runCleanups(self):
 
421
        """Run registered cleanup functions. 
 
422
 
 
423
        This should only be called from TestCase.tearDown.
 
424
        """
 
425
        # TODO: Perhaps this should keep running cleanups even if 
 
426
        # one of them fails?
 
427
        for cleanup_fn in reversed(self._cleanups):
 
428
            cleanup_fn()
 
429
 
 
430
    def log(self, *args):
 
431
        mutter(*args)
 
432
 
 
433
    def _get_log(self):
 
434
        """Return as a string the log for this test"""
 
435
        if self._log_file_name:
 
436
            return open(self._log_file_name).read()
 
437
        else:
 
438
            return self._log_contents
 
439
        # TODO: Delete the log after it's been read in
 
440
 
 
441
    def capture(self, cmd, retcode=0):
 
442
        """Shortcut that splits cmd into words, runs, and returns stdout"""
 
443
        return self.run_bzr_captured(cmd.split(), retcode=retcode)[0]
 
444
 
 
445
    def run_bzr_captured(self, argv, retcode=0, stdin=None):
 
446
        """Invoke bzr and return (stdout, stderr).
 
447
 
 
448
        Useful for code that wants to check the contents of the
 
449
        output, the way error messages are presented, etc.
 
450
 
 
451
        This should be the main method for tests that want to exercise the
 
452
        overall behavior of the bzr application (rather than a unit test
 
453
        or a functional test of the library.)
 
454
 
 
455
        Much of the old code runs bzr by forking a new copy of Python, but
 
456
        that is slower, harder to debug, and generally not necessary.
 
457
 
 
458
        This runs bzr through the interface that catches and reports
 
459
        errors, and with logging set to something approximating the
 
460
        default, so that error reporting can be checked.
 
461
 
 
462
        argv -- arguments to invoke bzr
 
463
        retcode -- expected return code, or None for don't-care.
 
464
        :param stdin: A string to be used as stdin for the command.
 
465
        """
 
466
        if stdin is not None:
 
467
            stdin = StringIO(stdin)
 
468
        stdout = StringIO()
 
469
        stderr = StringIO()
 
470
        self.log('run bzr: %s', ' '.join(argv))
 
471
        # FIXME: don't call into logging here
 
472
        handler = logging.StreamHandler(stderr)
 
473
        handler.setFormatter(bzrlib.trace.QuietFormatter())
 
474
        handler.setLevel(logging.INFO)
 
475
        logger = logging.getLogger('')
 
476
        logger.addHandler(handler)
 
477
        try:
 
478
            result = self.apply_redirected(stdin, stdout, stderr,
 
479
                                           bzrlib.commands.run_bzr_catch_errors,
 
480
                                           argv)
 
481
        finally:
 
482
            logger.removeHandler(handler)
 
483
        out = stdout.getvalue()
 
484
        err = stderr.getvalue()
 
485
        if out:
 
486
            self.log('output:\n%s', out)
 
487
        if err:
 
488
            self.log('errors:\n%s', err)
 
489
        if retcode is not None:
 
490
            self.assertEquals(result, retcode)
 
491
        return out, err
 
492
 
 
493
    def run_bzr(self, *args, **kwargs):
 
494
        """Invoke bzr, as if it were run from the command line.
 
495
 
 
496
        This should be the main method for tests that want to exercise the
 
497
        overall behavior of the bzr application (rather than a unit test
 
498
        or a functional test of the library.)
 
499
 
 
500
        This sends the stdout/stderr results into the test's log,
 
501
        where it may be useful for debugging.  See also run_captured.
 
502
 
 
503
        :param stdin: A string to be used as stdin for the command.
 
504
        """
 
505
        retcode = kwargs.pop('retcode', 0)
 
506
        stdin = kwargs.pop('stdin', None)
 
507
        return self.run_bzr_captured(args, retcode, stdin)
 
508
 
 
509
    def check_inventory_shape(self, inv, shape):
 
510
        """Compare an inventory to a list of expected names.
 
511
 
 
512
        Fail if they are not precisely equal.
 
513
        """
 
514
        extras = []
 
515
        shape = list(shape)             # copy
 
516
        for path, ie in inv.entries():
 
517
            name = path.replace('\\', '/')
 
518
            if ie.kind == 'dir':
 
519
                name = name + '/'
 
520
            if name in shape:
 
521
                shape.remove(name)
 
522
            else:
 
523
                extras.append(name)
 
524
        if shape:
 
525
            self.fail("expected paths not found in inventory: %r" % shape)
 
526
        if extras:
 
527
            self.fail("unexpected paths found in inventory: %r" % extras)
 
528
 
 
529
    def apply_redirected(self, stdin=None, stdout=None, stderr=None,
 
530
                         a_callable=None, *args, **kwargs):
 
531
        """Call callable with redirected std io pipes.
 
532
 
 
533
        Returns the return code."""
 
534
        if not callable(a_callable):
 
535
            raise ValueError("a_callable must be callable.")
 
536
        if stdin is None:
 
537
            stdin = StringIO("")
 
538
        if stdout is None:
 
539
            if getattr(self, "_log_file", None) is not None:
 
540
                stdout = self._log_file
 
541
            else:
 
542
                stdout = StringIO()
 
543
        if stderr is None:
 
544
            if getattr(self, "_log_file", None is not None):
 
545
                stderr = self._log_file
 
546
            else:
 
547
                stderr = StringIO()
 
548
        real_stdin = sys.stdin
 
549
        real_stdout = sys.stdout
 
550
        real_stderr = sys.stderr
 
551
        try:
 
552
            sys.stdout = stdout
 
553
            sys.stderr = stderr
 
554
            sys.stdin = stdin
 
555
            return a_callable(*args, **kwargs)
 
556
        finally:
 
557
            sys.stdout = real_stdout
 
558
            sys.stderr = real_stderr
 
559
            sys.stdin = real_stdin
 
560
 
 
561
    def merge(self, branch_from, wt_to):
 
562
        """A helper for tests to do a ui-less merge.
 
563
 
 
564
        This should move to the main library when someone has time to integrate
 
565
        it in.
 
566
        """
 
567
        # minimal ui-less merge.
 
568
        wt_to.branch.fetch(branch_from)
 
569
        base_rev = common_ancestor(branch_from.last_revision(),
 
570
                                   wt_to.branch.last_revision(),
 
571
                                   wt_to.branch.repository)
 
572
        merge_inner(wt_to.branch, branch_from.basis_tree(), 
 
573
                    wt_to.branch.repository.revision_tree(base_rev),
 
574
                    this_tree=wt_to)
 
575
        wt_to.add_pending_merge(branch_from.last_revision())
 
576
 
 
577
 
 
578
BzrTestBase = TestCase
 
579
 
 
580
     
 
581
class TestCaseInTempDir(TestCase):
 
582
    """Derived class that runs a test within a temporary directory.
 
583
 
 
584
    This is useful for tests that need to create a branch, etc.
 
585
 
 
586
    The directory is created in a slightly complex way: for each
 
587
    Python invocation, a new temporary top-level directory is created.
 
588
    All test cases create their own directory within that.  If the
 
589
    tests complete successfully, the directory is removed.
 
590
 
 
591
    InTempDir is an old alias for FunctionalTestCase.
 
592
    """
 
593
 
 
594
    TEST_ROOT = None
 
595
    _TEST_NAME = 'test'
 
596
    OVERRIDE_PYTHON = 'python'
 
597
 
 
598
    def check_file_contents(self, filename, expect):
 
599
        self.log("check contents of file %s" % filename)
 
600
        contents = file(filename, 'r').read()
 
601
        if contents != expect:
 
602
            self.log("expected: %r" % expect)
 
603
            self.log("actually: %r" % contents)
 
604
            self.fail("contents of %s not as expected" % filename)
 
605
 
 
606
    def _make_test_root(self):
 
607
        if TestCaseInTempDir.TEST_ROOT is not None:
 
608
            return
 
609
        i = 0
 
610
        while True:
 
611
            root = u'test%04d.tmp' % i
 
612
            try:
 
613
                os.mkdir(root)
 
614
            except OSError, e:
 
615
                if e.errno == errno.EEXIST:
 
616
                    i += 1
 
617
                    continue
 
618
                else:
 
619
                    raise
 
620
            # successfully created
 
621
            TestCaseInTempDir.TEST_ROOT = osutils.abspath(root)
 
622
            break
 
623
        # make a fake bzr directory there to prevent any tests propagating
 
624
        # up onto the source directory's real branch
 
625
        bzrdir.BzrDir.create_standalone_workingtree(TestCaseInTempDir.TEST_ROOT)
 
626
 
 
627
    def setUp(self):
 
628
        super(TestCaseInTempDir, self).setUp()
 
629
        self._make_test_root()
 
630
        _currentdir = os.getcwdu()
 
631
        # shorten the name, to avoid test failures due to path length
 
632
        short_id = self.id().replace('bzrlib.tests.', '') \
 
633
                   .replace('__main__.', '')[-100:]
 
634
        # it's possible the same test class is run several times for
 
635
        # parameterized tests, so make sure the names don't collide.  
 
636
        i = 0
 
637
        while True:
 
638
            if i > 0:
 
639
                candidate_dir = '%s/%s.%d' % (self.TEST_ROOT, short_id, i)
 
640
            else:
 
641
                candidate_dir = '%s/%s' % (self.TEST_ROOT, short_id)
 
642
            if os.path.exists(candidate_dir):
 
643
                i = i + 1
 
644
                continue
 
645
            else:
 
646
                self.test_dir = candidate_dir
 
647
                os.mkdir(self.test_dir)
 
648
                os.chdir(self.test_dir)
 
649
                break
 
650
        os.environ['HOME'] = self.test_dir
 
651
        os.environ['APPDATA'] = self.test_dir
 
652
        def _leaveDirectory():
 
653
            os.chdir(_currentdir)
 
654
        self.addCleanup(_leaveDirectory)
 
655
        
 
656
    def build_tree(self, shape, line_endings='native', transport=None):
 
657
        """Build a test tree according to a pattern.
 
658
 
 
659
        shape is a sequence of file specifications.  If the final
 
660
        character is '/', a directory is created.
 
661
 
 
662
        This doesn't add anything to a branch.
 
663
        :param line_endings: Either 'binary' or 'native'
 
664
                             in binary mode, exact contents are written
 
665
                             in native mode, the line endings match the
 
666
                             default platform endings.
 
667
 
 
668
        :param transport: A transport to write to, for building trees on 
 
669
                          VFS's. If the transport is readonly or None,
 
670
                          "." is opened automatically.
 
671
        """
 
672
        # XXX: It's OK to just create them using forward slashes on windows?
 
673
        if transport is None or transport.is_readonly():
 
674
            transport = get_transport(".")
 
675
        for name in shape:
 
676
            self.assert_(isinstance(name, basestring))
 
677
            if name[-1] == '/':
 
678
                transport.mkdir(urlescape(name[:-1]))
 
679
            else:
 
680
                if line_endings == 'binary':
 
681
                    end = '\n'
 
682
                elif line_endings == 'native':
 
683
                    end = os.linesep
 
684
                else:
 
685
                    raise errors.BzrError('Invalid line ending request %r' % (line_endings,))
 
686
                content = "contents of %s%s" % (name, end)
 
687
                transport.put(urlescape(name), StringIO(content))
 
688
 
 
689
    def build_tree_contents(self, shape):
 
690
        build_tree_contents(shape)
 
691
 
 
692
    def failUnlessExists(self, path):
 
693
        """Fail unless path, which may be abs or relative, exists."""
 
694
        self.failUnless(osutils.lexists(path))
 
695
 
 
696
    def failIfExists(self, path):
 
697
        """Fail if path, which may be abs or relative, exists."""
 
698
        self.failIf(osutils.lexists(path))
 
699
        
 
700
    def assertFileEqual(self, content, path):
 
701
        """Fail if path does not contain 'content'."""
 
702
        self.failUnless(osutils.lexists(path))
 
703
        self.assertEqualDiff(content, open(path, 'r').read())
 
704
 
 
705
 
 
706
class TestCaseWithTransport(TestCaseInTempDir):
 
707
    """A test case that provides get_url and get_readonly_url facilities.
 
708
 
 
709
    These back onto two transport servers, one for readonly access and one for
 
710
    read write access.
 
711
 
 
712
    If no explicit class is provided for readonly access, a
 
713
    ReadonlyTransportDecorator is used instead which allows the use of non disk
 
714
    based read write transports.
 
715
 
 
716
    If an explicit class is provided for readonly access, that server and the 
 
717
    readwrite one must both define get_url() as resolving to os.getcwd().
 
718
    """
 
719
 
 
720
    def __init__(self, methodName='testMethod'):
 
721
        super(TestCaseWithTransport, self).__init__(methodName)
 
722
        self.__readonly_server = None
 
723
        self.__server = None
 
724
        self.transport_server = default_transport
 
725
        self.transport_readonly_server = None
 
726
 
 
727
    def get_readonly_url(self, relpath=None):
 
728
        """Get a URL for the readonly transport.
 
729
 
 
730
        This will either be backed by '.' or a decorator to the transport 
 
731
        used by self.get_url()
 
732
        relpath provides for clients to get a path relative to the base url.
 
733
        These should only be downwards relative, not upwards.
 
734
        """
 
735
        base = self.get_readonly_server().get_url()
 
736
        if relpath is not None:
 
737
            if not base.endswith('/'):
 
738
                base = base + '/'
 
739
            base = base + relpath
 
740
        return base
 
741
 
 
742
    def get_readonly_server(self):
 
743
        """Get the server instance for the readonly transport
 
744
 
 
745
        This is useful for some tests with specific servers to do diagnostics.
 
746
        """
 
747
        if self.__readonly_server is None:
 
748
            if self.transport_readonly_server is None:
 
749
                # readonly decorator requested
 
750
                # bring up the server
 
751
                self.get_url()
 
752
                self.__readonly_server = ReadonlyServer()
 
753
                self.__readonly_server.setUp(self.__server)
 
754
            else:
 
755
                self.__readonly_server = self.transport_readonly_server()
 
756
                self.__readonly_server.setUp()
 
757
            self.addCleanup(self.__readonly_server.tearDown)
 
758
        return self.__readonly_server
 
759
 
 
760
    def get_server(self):
 
761
        """Get the read/write server instance.
 
762
 
 
763
        This is useful for some tests with specific servers that need
 
764
        diagnostics.
 
765
        """
 
766
        if self.__server is None:
 
767
            self.__server = self.transport_server()
 
768
            self.__server.setUp()
 
769
            self.addCleanup(self.__server.tearDown)
 
770
        return self.__server
 
771
 
 
772
    def get_url(self, relpath=None):
 
773
        """Get a URL for the readwrite transport.
 
774
 
 
775
        This will either be backed by '.' or to an equivalent non-file based
 
776
        facility.
 
777
        relpath provides for clients to get a path relative to the base url.
 
778
        These should only be downwards relative, not upwards.
 
779
        """
 
780
        base = self.get_server().get_url()
 
781
        if relpath is not None and relpath != '.':
 
782
            if not base.endswith('/'):
 
783
                base = base + '/'
 
784
            base = base + relpath
 
785
        return base
 
786
 
 
787
    def get_transport(self):
 
788
        """Return a writeable transport for the test scratch space"""
 
789
        t = get_transport(self.get_url())
 
790
        self.assertFalse(t.is_readonly())
 
791
        return t
 
792
 
 
793
    def get_readonly_transport(self):
 
794
        """Return a readonly transport for the test scratch space
 
795
        
 
796
        This can be used to test that operations which should only need
 
797
        readonly access in fact do not try to write.
 
798
        """
 
799
        t = get_transport(self.get_readonly_url())
 
800
        self.assertTrue(t.is_readonly())
 
801
        return t
 
802
 
 
803
    def make_branch(self, relpath, format=None):
 
804
        """Create a branch on the transport at relpath."""
 
805
        repo = self.make_repository(relpath, format=format)
 
806
        return repo.bzrdir.create_branch()
 
807
 
 
808
    def make_bzrdir(self, relpath, format=None):
 
809
        try:
 
810
            url = self.get_url(relpath)
 
811
            segments = relpath.split('/')
 
812
            if segments and segments[-1] not in ('', '.'):
 
813
                parent = self.get_url('/'.join(segments[:-1]))
 
814
                t = get_transport(parent)
 
815
                try:
 
816
                    t.mkdir(segments[-1])
 
817
                except errors.FileExists:
 
818
                    pass
 
819
            if format is None:
 
820
                format=bzrlib.bzrdir.BzrDirFormat.get_default_format()
 
821
            # FIXME: make this use a single transport someday. RBC 20060418
 
822
            return format.initialize_on_transport(get_transport(relpath))
 
823
        except errors.UninitializableFormat:
 
824
            raise TestSkipped("Format %s is not initializable.")
 
825
 
 
826
    def make_repository(self, relpath, shared=False, format=None):
 
827
        """Create a repository on our default transport at relpath."""
 
828
        made_control = self.make_bzrdir(relpath, format=format)
 
829
        return made_control.create_repository(shared=shared)
 
830
 
 
831
    def make_branch_and_tree(self, relpath, format=None):
 
832
        """Create a branch on the transport and a tree locally.
 
833
 
 
834
        Returns the tree.
 
835
        """
 
836
        # TODO: always use the local disk path for the working tree,
 
837
        # this obviously requires a format that supports branch references
 
838
        # so check for that by checking bzrdir.BzrDirFormat.get_default_format()
 
839
        # RBC 20060208
 
840
        b = self.make_branch(relpath, format=format)
 
841
        try:
 
842
            return b.bzrdir.create_workingtree()
 
843
        except errors.NotLocalUrl:
 
844
            # new formats - catch No tree error and create
 
845
            # a branch reference and a checkout.
 
846
            # old formats at that point - raise TestSkipped.
 
847
            # TODO: rbc 20060208
 
848
            return WorkingTreeFormat2().initialize(bzrdir.BzrDir.open(relpath))
 
849
 
 
850
    def assertIsDirectory(self, relpath, transport):
 
851
        """Assert that relpath within transport is a directory.
 
852
 
 
853
        This may not be possible on all transports; in that case it propagates
 
854
        a TransportNotPossible.
 
855
        """
 
856
        try:
 
857
            mode = transport.stat(relpath).st_mode
 
858
        except errors.NoSuchFile:
 
859
            self.fail("path %s is not a directory; no such file"
 
860
                      % (relpath))
 
861
        if not stat.S_ISDIR(mode):
 
862
            self.fail("path %s is not a directory; has mode %#o"
 
863
                      % (relpath, mode))
 
864
 
 
865
 
 
866
class ChrootedTestCase(TestCaseWithTransport):
 
867
    """A support class that provides readonly urls outside the local namespace.
 
868
 
 
869
    This is done by checking if self.transport_server is a MemoryServer. if it
 
870
    is then we are chrooted already, if it is not then an HttpServer is used
 
871
    for readonly urls.
 
872
 
 
873
    TODO RBC 20060127: make this an option to TestCaseWithTransport so it can
 
874
                       be used without needed to redo it when a different 
 
875
                       subclass is in use ?
 
876
    """
 
877
 
 
878
    def setUp(self):
 
879
        super(ChrootedTestCase, self).setUp()
 
880
        if not self.transport_server == bzrlib.transport.memory.MemoryServer:
 
881
            self.transport_readonly_server = bzrlib.transport.http.HttpServer
 
882
 
 
883
 
 
884
def filter_suite_by_re(suite, pattern):
 
885
    result = TestSuite()
 
886
    filter_re = re.compile(pattern)
 
887
    for test in iter_suite_tests(suite):
 
888
        if filter_re.search(test.id()):
 
889
            result.addTest(test)
 
890
    return result
 
891
 
 
892
 
 
893
def run_suite(suite, name='test', verbose=False, pattern=".*",
 
894
              stop_on_failure=False, keep_output=False,
 
895
              transport=None):
 
896
    TestCaseInTempDir._TEST_NAME = name
 
897
    if verbose:
 
898
        verbosity = 2
 
899
    else:
 
900
        verbosity = 1
 
901
    runner = TextTestRunner(stream=sys.stdout,
 
902
                            descriptions=0,
 
903
                            verbosity=verbosity)
 
904
    runner.stop_on_failure=stop_on_failure
 
905
    if pattern != '.*':
 
906
        suite = filter_suite_by_re(suite, pattern)
 
907
    result = runner.run(suite)
 
908
    # This is still a little bogus, 
 
909
    # but only a little. Folk not using our testrunner will
 
910
    # have to delete their temp directories themselves.
 
911
    test_root = TestCaseInTempDir.TEST_ROOT
 
912
    if result.wasSuccessful() or not keep_output:
 
913
        if test_root is not None:
 
914
            print 'Deleting test root %s...' % test_root
 
915
            try:
 
916
                shutil.rmtree(test_root)
 
917
            finally:
 
918
                print
 
919
    else:
 
920
        print "Failed tests working directories are in '%s'\n" % TestCaseInTempDir.TEST_ROOT
 
921
    return result.wasSuccessful()
 
922
 
 
923
 
 
924
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
 
925
             keep_output=False,
 
926
             transport=None):
 
927
    """Run the whole test suite under the enhanced runner"""
 
928
    global default_transport
 
929
    if transport is None:
 
930
        transport = default_transport
 
931
    old_transport = default_transport
 
932
    default_transport = transport
 
933
    suite = test_suite()
 
934
    try:
 
935
        return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
 
936
                     stop_on_failure=stop_on_failure, keep_output=keep_output,
 
937
                     transport=transport)
 
938
    finally:
 
939
        default_transport = old_transport
 
940
 
 
941
 
 
942
 
 
943
def test_suite():
 
944
    """Build and return TestSuite for the whole program."""
 
945
    from doctest import DocTestSuite
 
946
 
 
947
    global MODULES_TO_DOCTEST
 
948
 
 
949
    testmod_names = [ \
 
950
                   'bzrlib.tests.test_ancestry',
 
951
                   'bzrlib.tests.test_annotate',
 
952
                   'bzrlib.tests.test_api',
 
953
                   'bzrlib.tests.test_bad_files',
 
954
                   'bzrlib.tests.test_branch',
 
955
                   'bzrlib.tests.test_bzrdir',
 
956
                   'bzrlib.tests.test_command',
 
957
                   'bzrlib.tests.test_commit',
 
958
                   'bzrlib.tests.test_commit_merge',
 
959
                   'bzrlib.tests.test_config',
 
960
                   'bzrlib.tests.test_conflicts',
 
961
                   'bzrlib.tests.test_decorators',
 
962
                   'bzrlib.tests.test_diff',
 
963
                   'bzrlib.tests.test_doc_generate',
 
964
                   'bzrlib.tests.test_errors',
 
965
                   'bzrlib.tests.test_escaped_store',
 
966
                   'bzrlib.tests.test_fetch',
 
967
                   'bzrlib.tests.test_gpg',
 
968
                   'bzrlib.tests.test_graph',
 
969
                   'bzrlib.tests.test_hashcache',
 
970
                   'bzrlib.tests.test_http',
 
971
                   'bzrlib.tests.test_identitymap',
 
972
                   'bzrlib.tests.test_inv',
 
973
                   'bzrlib.tests.test_knit',
 
974
                   'bzrlib.tests.test_lockdir',
 
975
                   'bzrlib.tests.test_lockable_files',
 
976
                   'bzrlib.tests.test_log',
 
977
                   'bzrlib.tests.test_merge',
 
978
                   'bzrlib.tests.test_merge3',
 
979
                   'bzrlib.tests.test_merge_core',
 
980
                   'bzrlib.tests.test_missing',
 
981
                   'bzrlib.tests.test_msgeditor',
 
982
                   'bzrlib.tests.test_nonascii',
 
983
                   'bzrlib.tests.test_options',
 
984
                   'bzrlib.tests.test_osutils',
 
985
                   'bzrlib.tests.test_patch',
 
986
                   'bzrlib.tests.test_permissions',
 
987
                   'bzrlib.tests.test_plugins',
 
988
                   'bzrlib.tests.test_progress',
 
989
                   'bzrlib.tests.test_reconcile',
 
990
                   'bzrlib.tests.test_repository',
 
991
                   'bzrlib.tests.test_revision',
 
992
                   'bzrlib.tests.test_revisionnamespaces',
 
993
                   'bzrlib.tests.test_revprops',
 
994
                   'bzrlib.tests.test_rio',
 
995
                   'bzrlib.tests.test_sampler',
 
996
                   'bzrlib.tests.test_selftest',
 
997
                   'bzrlib.tests.test_setup',
 
998
                   'bzrlib.tests.test_sftp_transport',
 
999
                   'bzrlib.tests.test_smart_add',
 
1000
                   'bzrlib.tests.test_source',
 
1001
                   'bzrlib.tests.test_store',
 
1002
                   'bzrlib.tests.test_symbol_versioning',
 
1003
                   'bzrlib.tests.test_testament',
 
1004
                   'bzrlib.tests.test_textfile',
 
1005
                   'bzrlib.tests.test_textmerge',
 
1006
                   'bzrlib.tests.test_trace',
 
1007
                   'bzrlib.tests.test_transactions',
 
1008
                   'bzrlib.tests.test_transform',
 
1009
                   'bzrlib.tests.test_transport',
 
1010
                   'bzrlib.tests.test_tsort',
 
1011
                   'bzrlib.tests.test_tuned_gzip',
 
1012
                   'bzrlib.tests.test_ui',
 
1013
                   'bzrlib.tests.test_upgrade',
 
1014
                   'bzrlib.tests.test_versionedfile',
 
1015
                   'bzrlib.tests.test_weave',
 
1016
                   'bzrlib.tests.test_whitebox',
 
1017
                   'bzrlib.tests.test_workingtree',
 
1018
                   'bzrlib.tests.test_xml',
 
1019
                   ]
 
1020
    test_transport_implementations = [
 
1021
        'bzrlib.tests.test_transport_implementations']
 
1022
 
 
1023
    TestCase.BZRPATH = osutils.pathjoin(
 
1024
            osutils.realpath(osutils.dirname(bzrlib.__path__[0])), 'bzr')
 
1025
    print '%10s: %s' % ('bzr', osutils.realpath(sys.argv[0]))
 
1026
    print '%10s: %s' % ('bzrlib', bzrlib.__path__[0])
 
1027
    print
 
1028
    suite = TestSuite()
 
1029
    # python2.4's TestLoader.loadTestsFromNames gives very poor 
 
1030
    # errors if it fails to load a named module - no indication of what's
 
1031
    # actually wrong, just "no such module".  We should probably override that
 
1032
    # class, but for the moment just load them ourselves. (mbp 20051202)
 
1033
    loader = TestLoader()
 
1034
    from bzrlib.transport import TransportTestProviderAdapter
 
1035
    adapter = TransportTestProviderAdapter()
 
1036
    adapt_modules(test_transport_implementations, adapter, loader, suite)
 
1037
    for mod_name in testmod_names:
 
1038
        mod = _load_module_by_name(mod_name)
 
1039
        suite.addTest(loader.loadTestsFromModule(mod))
 
1040
    for package in packages_to_test():
 
1041
        suite.addTest(package.test_suite())
 
1042
    for m in MODULES_TO_TEST:
 
1043
        suite.addTest(loader.loadTestsFromModule(m))
 
1044
    for m in (MODULES_TO_DOCTEST):
 
1045
        suite.addTest(DocTestSuite(m))
 
1046
    for name, plugin in bzrlib.plugin.all_plugins().items():
 
1047
        if getattr(plugin, 'test_suite', None) is not None:
 
1048
            suite.addTest(plugin.test_suite())
 
1049
    return suite
 
1050
 
 
1051
 
 
1052
def adapt_modules(mods_list, adapter, loader, suite):
 
1053
    """Adapt the modules in mods_list using adapter and add to suite."""
 
1054
    for mod_name in mods_list:
 
1055
        mod = _load_module_by_name(mod_name)
 
1056
        for test in iter_suite_tests(loader.loadTestsFromModule(mod)):
 
1057
            suite.addTests(adapter.adapt(test))
 
1058
 
 
1059
 
 
1060
def _load_module_by_name(mod_name):
 
1061
    parts = mod_name.split('.')
 
1062
    module = __import__(mod_name)
 
1063
    del parts[0]
 
1064
    # for historical reasons python returns the top-level module even though
 
1065
    # it loads the submodule; we need to walk down to get the one we want.
 
1066
    while parts:
 
1067
        module = getattr(module, parts.pop(0))
 
1068
    return module