~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/__init__.py

  • Committer: mbp at sourcefrog
  • Date: 2005-03-09 04:08:15 UTC
  • Revision ID: mbp@sourcefrog.net-20050309040815-13242001617e4a06
import from baz patch-364

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 not an instance of %s" % (obj, kls))
345
 
 
346
 
    def _startLogFile(self):
347
 
        """Send bzr and test log messages to a temporary file.
348
 
 
349
 
        The file is removed as the test is torn down.
350
 
        """
351
 
        fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
352
 
        encoder, decoder, stream_reader, stream_writer = codecs.lookup('UTF-8')
353
 
        self._log_file = stream_writer(os.fdopen(fileno, 'w+'))
354
 
        self._log_nonce = bzrlib.trace.enable_test_log(self._log_file)
355
 
        self._log_file_name = name
356
 
        self.addCleanup(self._finishLogFile)
357
 
 
358
 
    def _finishLogFile(self):
359
 
        """Finished with the log file.
360
 
 
361
 
        Read contents into memory, close, and delete.
362
 
        """
363
 
        bzrlib.trace.disable_test_log(self._log_nonce)
364
 
        self._log_file.seek(0)
365
 
        self._log_contents = self._log_file.read()
366
 
        self._log_file.close()
367
 
        os.remove(self._log_file_name)
368
 
        self._log_file = self._log_file_name = None
369
 
 
370
 
    def addCleanup(self, callable):
371
 
        """Arrange to run a callable when this case is torn down.
372
 
 
373
 
        Callables are run in the reverse of the order they are registered, 
374
 
        ie last-in first-out.
375
 
        """
376
 
        if callable in self._cleanups:
377
 
            raise ValueError("cleanup function %r already registered on %s" 
378
 
                    % (callable, self))
379
 
        self._cleanups.append(callable)
380
 
 
381
 
    def _cleanEnvironment(self):
382
 
        new_env = {
383
 
            'HOME': os.getcwd(),
384
 
            'APPDATA': os.getcwd(),
385
 
            'BZREMAIL': None,
386
 
            'EMAIL': None,
387
 
        }
388
 
        self.__old_env = {}
389
 
        self.addCleanup(self._restoreEnvironment)
390
 
        for name, value in new_env.iteritems():
391
 
            self._captureVar(name, value)
392
 
 
393
 
 
394
 
    def _captureVar(self, name, newvalue):
395
 
        """Set an environment variable, preparing it to be reset when finished."""
396
 
        self.__old_env[name] = os.environ.get(name, None)
397
 
        if newvalue is None:
398
 
            if name in os.environ:
399
 
                del os.environ[name]
400
 
        else:
401
 
            os.environ[name] = newvalue
402
 
 
403
 
    @staticmethod
404
 
    def _restoreVar(name, value):
405
 
        if value is None:
406
 
            if name in os.environ:
407
 
                del os.environ[name]
408
 
        else:
409
 
            os.environ[name] = value
410
 
 
411
 
    def _restoreEnvironment(self):
412
 
        for name, value in self.__old_env.iteritems():
413
 
            self._restoreVar(name, value)
414
 
 
415
 
    def tearDown(self):
416
 
        self._runCleanups()
417
 
        unittest.TestCase.tearDown(self)
418
 
 
419
 
    def _runCleanups(self):
420
 
        """Run registered cleanup functions. 
421
 
 
422
 
        This should only be called from TestCase.tearDown.
423
 
        """
424
 
        # TODO: Perhaps this should keep running cleanups even if 
425
 
        # one of them fails?
426
 
        for cleanup_fn in reversed(self._cleanups):
427
 
            cleanup_fn()
428
 
 
429
 
    def log(self, *args):
430
 
        mutter(*args)
431
 
 
432
 
    def _get_log(self):
433
 
        """Return as a string the log for this test"""
434
 
        if self._log_file_name:
435
 
            return open(self._log_file_name).read()
436
 
        else:
437
 
            return self._log_contents
438
 
        # TODO: Delete the log after it's been read in
439
 
 
440
 
    def capture(self, cmd, retcode=0):
441
 
        """Shortcut that splits cmd into words, runs, and returns stdout"""
442
 
        return self.run_bzr_captured(cmd.split(), retcode=retcode)[0]
443
 
 
444
 
    def run_bzr_captured(self, argv, retcode=0):
445
 
        """Invoke bzr and return (stdout, stderr).
446
 
 
447
 
        Useful for code that wants to check the contents of the
448
 
        output, the way error messages are presented, etc.
449
 
 
450
 
        This should be the main method for tests that want to exercise the
451
 
        overall behavior of the bzr application (rather than a unit test
452
 
        or a functional test of the library.)
453
 
 
454
 
        Much of the old code runs bzr by forking a new copy of Python, but
455
 
        that is slower, harder to debug, and generally not necessary.
456
 
 
457
 
        This runs bzr through the interface that catches and reports
458
 
        errors, and with logging set to something approximating the
459
 
        default, so that error reporting can be checked.
460
 
 
461
 
        argv -- arguments to invoke bzr
462
 
        retcode -- expected return code, or None for don't-care.
463
 
        """
464
 
        stdout = StringIO()
465
 
        stderr = StringIO()
466
 
        self.log('run bzr: %s', ' '.join(argv))
467
 
        # FIXME: don't call into logging here
468
 
        handler = logging.StreamHandler(stderr)
469
 
        handler.setFormatter(bzrlib.trace.QuietFormatter())
470
 
        handler.setLevel(logging.INFO)
471
 
        logger = logging.getLogger('')
472
 
        logger.addHandler(handler)
473
 
        try:
474
 
            result = self.apply_redirected(None, stdout, stderr,
475
 
                                           bzrlib.commands.run_bzr_catch_errors,
476
 
                                           argv)
477
 
        finally:
478
 
            logger.removeHandler(handler)
479
 
        out = stdout.getvalue()
480
 
        err = stderr.getvalue()
481
 
        if out:
482
 
            self.log('output:\n%s', out)
483
 
        if err:
484
 
            self.log('errors:\n%s', err)
485
 
        if retcode is not None:
486
 
            self.assertEquals(result, retcode)
487
 
        return out, err
488
 
 
489
 
    def run_bzr(self, *args, **kwargs):
490
 
        """Invoke bzr, as if it were run from the command line.
491
 
 
492
 
        This should be the main method for tests that want to exercise the
493
 
        overall behavior of the bzr application (rather than a unit test
494
 
        or a functional test of the library.)
495
 
 
496
 
        This sends the stdout/stderr results into the test's log,
497
 
        where it may be useful for debugging.  See also run_captured.
498
 
        """
499
 
        retcode = kwargs.pop('retcode', 0)
500
 
        return self.run_bzr_captured(args, retcode)
501
 
 
502
 
    def check_inventory_shape(self, inv, shape):
503
 
        """Compare an inventory to a list of expected names.
504
 
 
505
 
        Fail if they are not precisely equal.
506
 
        """
507
 
        extras = []
508
 
        shape = list(shape)             # copy
509
 
        for path, ie in inv.entries():
510
 
            name = path.replace('\\', '/')
511
 
            if ie.kind == 'dir':
512
 
                name = name + '/'
513
 
            if name in shape:
514
 
                shape.remove(name)
515
 
            else:
516
 
                extras.append(name)
517
 
        if shape:
518
 
            self.fail("expected paths not found in inventory: %r" % shape)
519
 
        if extras:
520
 
            self.fail("unexpected paths found in inventory: %r" % extras)
521
 
 
522
 
    def apply_redirected(self, stdin=None, stdout=None, stderr=None,
523
 
                         a_callable=None, *args, **kwargs):
524
 
        """Call callable with redirected std io pipes.
525
 
 
526
 
        Returns the return code."""
527
 
        if not callable(a_callable):
528
 
            raise ValueError("a_callable must be callable.")
529
 
        if stdin is None:
530
 
            stdin = StringIO("")
531
 
        if stdout is None:
532
 
            if getattr(self, "_log_file", None) is not None:
533
 
                stdout = self._log_file
534
 
            else:
535
 
                stdout = StringIO()
536
 
        if stderr is None:
537
 
            if getattr(self, "_log_file", None is not None):
538
 
                stderr = self._log_file
539
 
            else:
540
 
                stderr = StringIO()
541
 
        real_stdin = sys.stdin
542
 
        real_stdout = sys.stdout
543
 
        real_stderr = sys.stderr
544
 
        try:
545
 
            sys.stdout = stdout
546
 
            sys.stderr = stderr
547
 
            sys.stdin = stdin
548
 
            return a_callable(*args, **kwargs)
549
 
        finally:
550
 
            sys.stdout = real_stdout
551
 
            sys.stderr = real_stderr
552
 
            sys.stdin = real_stdin
553
 
 
554
 
    def merge(self, branch_from, wt_to):
555
 
        """A helper for tests to do a ui-less merge.
556
 
 
557
 
        This should move to the main library when someone has time to integrate
558
 
        it in.
559
 
        """
560
 
        # minimal ui-less merge.
561
 
        wt_to.branch.fetch(branch_from)
562
 
        base_rev = common_ancestor(branch_from.last_revision(),
563
 
                                   wt_to.branch.last_revision(),
564
 
                                   wt_to.branch.repository)
565
 
        merge_inner(wt_to.branch, branch_from.basis_tree(), 
566
 
                    wt_to.branch.repository.revision_tree(base_rev),
567
 
                    this_tree=wt_to)
568
 
        wt_to.add_pending_merge(branch_from.last_revision())
569
 
 
570
 
 
571
 
BzrTestBase = TestCase
572
 
 
573
 
     
574
 
class TestCaseInTempDir(TestCase):
575
 
    """Derived class that runs a test within a temporary directory.
576
 
 
577
 
    This is useful for tests that need to create a branch, etc.
578
 
 
579
 
    The directory is created in a slightly complex way: for each
580
 
    Python invocation, a new temporary top-level directory is created.
581
 
    All test cases create their own directory within that.  If the
582
 
    tests complete successfully, the directory is removed.
583
 
 
584
 
    InTempDir is an old alias for FunctionalTestCase.
585
 
    """
586
 
 
587
 
    TEST_ROOT = None
588
 
    _TEST_NAME = 'test'
589
 
    OVERRIDE_PYTHON = 'python'
590
 
 
591
 
    def check_file_contents(self, filename, expect):
592
 
        self.log("check contents of file %s" % filename)
593
 
        contents = file(filename, 'r').read()
594
 
        if contents != expect:
595
 
            self.log("expected: %r" % expect)
596
 
            self.log("actually: %r" % contents)
597
 
            self.fail("contents of %s not as expected" % filename)
598
 
 
599
 
    def _make_test_root(self):
600
 
        if TestCaseInTempDir.TEST_ROOT is not None:
601
 
            return
602
 
        i = 0
603
 
        while True:
604
 
            root = u'test%04d.tmp' % i
605
 
            try:
606
 
                os.mkdir(root)
607
 
            except OSError, e:
608
 
                if e.errno == errno.EEXIST:
609
 
                    i += 1
610
 
                    continue
611
 
                else:
612
 
                    raise
613
 
            # successfully created
614
 
            TestCaseInTempDir.TEST_ROOT = osutils.abspath(root)
615
 
            break
616
 
        # make a fake bzr directory there to prevent any tests propagating
617
 
        # up onto the source directory's real branch
618
 
        bzrdir.BzrDir.create_standalone_workingtree(TestCaseInTempDir.TEST_ROOT)
619
 
 
620
 
    def setUp(self):
621
 
        super(TestCaseInTempDir, self).setUp()
622
 
        self._make_test_root()
623
 
        _currentdir = os.getcwdu()
624
 
        # shorten the name, to avoid test failures due to path length
625
 
        short_id = self.id().replace('bzrlib.tests.', '') \
626
 
                   .replace('__main__.', '')[-100:]
627
 
        # it's possible the same test class is run several times for
628
 
        # parameterized tests, so make sure the names don't collide.  
629
 
        i = 0
630
 
        while True:
631
 
            if i > 0:
632
 
                candidate_dir = '%s/%s.%d' % (self.TEST_ROOT, short_id, i)
633
 
            else:
634
 
                candidate_dir = '%s/%s' % (self.TEST_ROOT, short_id)
635
 
            if os.path.exists(candidate_dir):
636
 
                i = i + 1
637
 
                continue
638
 
            else:
639
 
                self.test_dir = candidate_dir
640
 
                os.mkdir(self.test_dir)
641
 
                os.chdir(self.test_dir)
642
 
                break
643
 
        os.environ['HOME'] = self.test_dir
644
 
        os.environ['APPDATA'] = self.test_dir
645
 
        def _leaveDirectory():
646
 
            os.chdir(_currentdir)
647
 
        self.addCleanup(_leaveDirectory)
648
 
        
649
 
    def build_tree(self, shape, line_endings='native', transport=None):
650
 
        """Build a test tree according to a pattern.
651
 
 
652
 
        shape is a sequence of file specifications.  If the final
653
 
        character is '/', a directory is created.
654
 
 
655
 
        This doesn't add anything to a branch.
656
 
        :param line_endings: Either 'binary' or 'native'
657
 
                             in binary mode, exact contents are written
658
 
                             in native mode, the line endings match the
659
 
                             default platform endings.
660
 
 
661
 
        :param transport: A transport to write to, for building trees on 
662
 
                          VFS's. If the transport is readonly or None,
663
 
                          "." is opened automatically.
664
 
        """
665
 
        # XXX: It's OK to just create them using forward slashes on windows?
666
 
        if transport is None or transport.is_readonly():
667
 
            transport = get_transport(".")
668
 
        for name in shape:
669
 
            self.assert_(isinstance(name, basestring))
670
 
            if name[-1] == '/':
671
 
                transport.mkdir(urlescape(name[:-1]))
672
 
            else:
673
 
                if line_endings == 'binary':
674
 
                    end = '\n'
675
 
                elif line_endings == 'native':
676
 
                    end = os.linesep
677
 
                else:
678
 
                    raise errors.BzrError('Invalid line ending request %r' % (line_endings,))
679
 
                content = "contents of %s%s" % (name, end)
680
 
                transport.put(urlescape(name), StringIO(content))
681
 
 
682
 
    def build_tree_contents(self, shape):
683
 
        build_tree_contents(shape)
684
 
 
685
 
    def failUnlessExists(self, path):
686
 
        """Fail unless path, which may be abs or relative, exists."""
687
 
        self.failUnless(osutils.lexists(path))
688
 
 
689
 
    def failIfExists(self, path):
690
 
        """Fail if path, which may be abs or relative, exists."""
691
 
        self.failIf(osutils.lexists(path))
692
 
        
693
 
    def assertFileEqual(self, content, path):
694
 
        """Fail if path does not contain 'content'."""
695
 
        self.failUnless(osutils.lexists(path))
696
 
        self.assertEqualDiff(content, open(path, 'r').read())
697
 
 
698
 
 
699
 
class TestCaseWithTransport(TestCaseInTempDir):
700
 
    """A test case that provides get_url and get_readonly_url facilities.
701
 
 
702
 
    These back onto two transport servers, one for readonly access and one for
703
 
    read write access.
704
 
 
705
 
    If no explicit class is provided for readonly access, a
706
 
    ReadonlyTransportDecorator is used instead which allows the use of non disk
707
 
    based read write transports.
708
 
 
709
 
    If an explicit class is provided for readonly access, that server and the 
710
 
    readwrite one must both define get_url() as resolving to os.getcwd().
711
 
    """
712
 
 
713
 
    def __init__(self, methodName='testMethod'):
714
 
        super(TestCaseWithTransport, self).__init__(methodName)
715
 
        self.__readonly_server = None
716
 
        self.__server = None
717
 
        self.transport_server = default_transport
718
 
        self.transport_readonly_server = None
719
 
 
720
 
    def get_readonly_url(self, relpath=None):
721
 
        """Get a URL for the readonly transport.
722
 
 
723
 
        This will either be backed by '.' or a decorator to the transport 
724
 
        used by self.get_url()
725
 
        relpath provides for clients to get a path relative to the base url.
726
 
        These should only be downwards relative, not upwards.
727
 
        """
728
 
        base = self.get_readonly_server().get_url()
729
 
        if relpath is not None:
730
 
            if not base.endswith('/'):
731
 
                base = base + '/'
732
 
            base = base + relpath
733
 
        return base
734
 
 
735
 
    def get_readonly_server(self):
736
 
        """Get the server instance for the readonly transport
737
 
 
738
 
        This is useful for some tests with specific servers to do diagnostics.
739
 
        """
740
 
        if self.__readonly_server is None:
741
 
            if self.transport_readonly_server is None:
742
 
                # readonly decorator requested
743
 
                # bring up the server
744
 
                self.get_url()
745
 
                self.__readonly_server = ReadonlyServer()
746
 
                self.__readonly_server.setUp(self.__server)
747
 
            else:
748
 
                self.__readonly_server = self.transport_readonly_server()
749
 
                self.__readonly_server.setUp()
750
 
            self.addCleanup(self.__readonly_server.tearDown)
751
 
        return self.__readonly_server
752
 
 
753
 
    def get_server(self):
754
 
        """Get the read/write server instance.
755
 
 
756
 
        This is useful for some tests with specific servers that need
757
 
        diagnostics.
758
 
        """
759
 
        if self.__server is None:
760
 
            self.__server = self.transport_server()
761
 
            self.__server.setUp()
762
 
            self.addCleanup(self.__server.tearDown)
763
 
        return self.__server
764
 
 
765
 
    def get_url(self, relpath=None):
766
 
        """Get a URL for the readwrite transport.
767
 
 
768
 
        This will either be backed by '.' or to an equivalent non-file based
769
 
        facility.
770
 
        relpath provides for clients to get a path relative to the base url.
771
 
        These should only be downwards relative, not upwards.
772
 
        """
773
 
        base = self.get_server().get_url()
774
 
        if relpath is not None and relpath != '.':
775
 
            if not base.endswith('/'):
776
 
                base = base + '/'
777
 
            base = base + relpath
778
 
        return base
779
 
 
780
 
    def get_transport(self):
781
 
        """Return a writeable transport for the test scratch space"""
782
 
        t = get_transport(self.get_url())
783
 
        self.assertFalse(t.is_readonly())
784
 
        return t
785
 
 
786
 
    def get_readonly_transport(self):
787
 
        """Return a readonly transport for the test scratch space
788
 
        
789
 
        This can be used to test that operations which should only need
790
 
        readonly access in fact do not try to write.
791
 
        """
792
 
        t = get_transport(self.get_readonly_url())
793
 
        self.assertTrue(t.is_readonly())
794
 
        return t
795
 
 
796
 
    def make_branch(self, relpath, format=None):
797
 
        """Create a branch on the transport at relpath."""
798
 
        repo = self.make_repository(relpath, format=format)
799
 
        return repo.bzrdir.create_branch()
800
 
 
801
 
    def make_bzrdir(self, relpath, format=None):
802
 
        try:
803
 
            url = self.get_url(relpath)
804
 
            segments = relpath.split('/')
805
 
            if segments and segments[-1] not in ('', '.'):
806
 
                parent = self.get_url('/'.join(segments[:-1]))
807
 
                t = get_transport(parent)
808
 
                try:
809
 
                    t.mkdir(segments[-1])
810
 
                except errors.FileExists:
811
 
                    pass
812
 
            if format is None:
813
 
                format=bzrlib.bzrdir.BzrDirFormat.get_default_format()
814
 
            # FIXME: make this use a single transport someday. RBC 20060418
815
 
            return format.initialize_on_transport(get_transport(relpath))
816
 
        except errors.UninitializableFormat:
817
 
            raise TestSkipped("Format %s is not initializable.")
818
 
 
819
 
    def make_repository(self, relpath, shared=False, format=None):
820
 
        """Create a repository on our default transport at relpath."""
821
 
        made_control = self.make_bzrdir(relpath, format=format)
822
 
        return made_control.create_repository(shared=shared)
823
 
 
824
 
    def make_branch_and_tree(self, relpath, format=None):
825
 
        """Create a branch on the transport and a tree locally.
826
 
 
827
 
        Returns the tree.
828
 
        """
829
 
        # TODO: always use the local disk path for the working tree,
830
 
        # this obviously requires a format that supports branch references
831
 
        # so check for that by checking bzrdir.BzrDirFormat.get_default_format()
832
 
        # RBC 20060208
833
 
        b = self.make_branch(relpath, format=format)
834
 
        try:
835
 
            return b.bzrdir.create_workingtree()
836
 
        except errors.NotLocalUrl:
837
 
            # new formats - catch No tree error and create
838
 
            # a branch reference and a checkout.
839
 
            # old formats at that point - raise TestSkipped.
840
 
            # TODO: rbc 20060208
841
 
            return WorkingTreeFormat2().initialize(bzrdir.BzrDir.open(relpath))
842
 
 
843
 
    def assertIsDirectory(self, relpath, transport):
844
 
        """Assert that relpath within transport is a directory.
845
 
 
846
 
        This may not be possible on all transports; in that case it propagates
847
 
        a TransportNotPossible.
848
 
        """
849
 
        try:
850
 
            mode = transport.stat(relpath).st_mode
851
 
        except errors.NoSuchFile:
852
 
            self.fail("path %s is not a directory; no such file"
853
 
                      % (relpath))
854
 
        if not stat.S_ISDIR(mode):
855
 
            self.fail("path %s is not a directory; has mode %#o"
856
 
                      % (relpath, mode))
857
 
 
858
 
 
859
 
class ChrootedTestCase(TestCaseWithTransport):
860
 
    """A support class that provides readonly urls outside the local namespace.
861
 
 
862
 
    This is done by checking if self.transport_server is a MemoryServer. if it
863
 
    is then we are chrooted already, if it is not then an HttpServer is used
864
 
    for readonly urls.
865
 
 
866
 
    TODO RBC 20060127: make this an option to TestCaseWithTransport so it can
867
 
                       be used without needed to redo it when a different 
868
 
                       subclass is in use ?
869
 
    """
870
 
 
871
 
    def setUp(self):
872
 
        super(ChrootedTestCase, self).setUp()
873
 
        if not self.transport_server == bzrlib.transport.memory.MemoryServer:
874
 
            self.transport_readonly_server = bzrlib.transport.http.HttpServer
875
 
 
876
 
 
877
 
def filter_suite_by_re(suite, pattern):
878
 
    result = TestSuite()
879
 
    filter_re = re.compile(pattern)
880
 
    for test in iter_suite_tests(suite):
881
 
        if filter_re.search(test.id()):
882
 
            result.addTest(test)
883
 
    return result
884
 
 
885
 
 
886
 
def run_suite(suite, name='test', verbose=False, pattern=".*",
887
 
              stop_on_failure=False, keep_output=False,
888
 
              transport=None):
889
 
    TestCaseInTempDir._TEST_NAME = name
890
 
    if verbose:
891
 
        verbosity = 2
892
 
    else:
893
 
        verbosity = 1
894
 
    runner = TextTestRunner(stream=sys.stdout,
895
 
                            descriptions=0,
896
 
                            verbosity=verbosity)
897
 
    runner.stop_on_failure=stop_on_failure
898
 
    if pattern != '.*':
899
 
        suite = filter_suite_by_re(suite, pattern)
900
 
    result = runner.run(suite)
901
 
    # This is still a little bogus, 
902
 
    # but only a little. Folk not using our testrunner will
903
 
    # have to delete their temp directories themselves.
904
 
    test_root = TestCaseInTempDir.TEST_ROOT
905
 
    if result.wasSuccessful() or not keep_output:
906
 
        if test_root is not None:
907
 
            print 'Deleting test root %s...' % test_root
908
 
            try:
909
 
                shutil.rmtree(test_root)
910
 
            finally:
911
 
                print
912
 
    else:
913
 
        print "Failed tests working directories are in '%s'\n" % TestCaseInTempDir.TEST_ROOT
914
 
    return result.wasSuccessful()
915
 
 
916
 
 
917
 
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
918
 
             keep_output=False,
919
 
             transport=None):
920
 
    """Run the whole test suite under the enhanced runner"""
921
 
    global default_transport
922
 
    if transport is None:
923
 
        transport = default_transport
924
 
    old_transport = default_transport
925
 
    default_transport = transport
926
 
    suite = test_suite()
927
 
    try:
928
 
        return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
929
 
                     stop_on_failure=stop_on_failure, keep_output=keep_output,
930
 
                     transport=transport)
931
 
    finally:
932
 
        default_transport = old_transport
933
 
 
934
 
 
935
 
 
936
 
def test_suite():
937
 
    """Build and return TestSuite for the whole program."""
938
 
    from doctest import DocTestSuite
939
 
 
940
 
    global MODULES_TO_DOCTEST
941
 
 
942
 
    testmod_names = [ \
943
 
                   'bzrlib.tests.test_ancestry',
944
 
                   'bzrlib.tests.test_annotate',
945
 
                   'bzrlib.tests.test_api',
946
 
                   'bzrlib.tests.test_bad_files',
947
 
                   'bzrlib.tests.test_branch',
948
 
                   'bzrlib.tests.test_bzrdir',
949
 
                   'bzrlib.tests.test_command',
950
 
                   'bzrlib.tests.test_commit',
951
 
                   'bzrlib.tests.test_commit_merge',
952
 
                   'bzrlib.tests.test_config',
953
 
                   'bzrlib.tests.test_conflicts',
954
 
                   'bzrlib.tests.test_decorators',
955
 
                   'bzrlib.tests.test_diff',
956
 
                   'bzrlib.tests.test_doc_generate',
957
 
                   'bzrlib.tests.test_errors',
958
 
                   'bzrlib.tests.test_escaped_store',
959
 
                   'bzrlib.tests.test_fetch',
960
 
                   'bzrlib.tests.test_gpg',
961
 
                   'bzrlib.tests.test_graph',
962
 
                   'bzrlib.tests.test_hashcache',
963
 
                   'bzrlib.tests.test_http',
964
 
                   'bzrlib.tests.test_identitymap',
965
 
                   'bzrlib.tests.test_inv',
966
 
                   'bzrlib.tests.test_knit',
967
 
                   'bzrlib.tests.test_lockdir',
968
 
                   'bzrlib.tests.test_lockable_files',
969
 
                   'bzrlib.tests.test_log',
970
 
                   'bzrlib.tests.test_merge',
971
 
                   'bzrlib.tests.test_merge3',
972
 
                   'bzrlib.tests.test_merge_core',
973
 
                   'bzrlib.tests.test_missing',
974
 
                   'bzrlib.tests.test_msgeditor',
975
 
                   'bzrlib.tests.test_nonascii',
976
 
                   'bzrlib.tests.test_options',
977
 
                   'bzrlib.tests.test_osutils',
978
 
                   'bzrlib.tests.test_permissions',
979
 
                   'bzrlib.tests.test_plugins',
980
 
                   'bzrlib.tests.test_progress',
981
 
                   'bzrlib.tests.test_reconcile',
982
 
                   'bzrlib.tests.test_repository',
983
 
                   'bzrlib.tests.test_revision',
984
 
                   'bzrlib.tests.test_revisionnamespaces',
985
 
                   'bzrlib.tests.test_revprops',
986
 
                   'bzrlib.tests.test_rio',
987
 
                   'bzrlib.tests.test_sampler',
988
 
                   'bzrlib.tests.test_selftest',
989
 
                   'bzrlib.tests.test_setup',
990
 
                   'bzrlib.tests.test_sftp_transport',
991
 
                   'bzrlib.tests.test_smart_add',
992
 
                   'bzrlib.tests.test_source',
993
 
                   'bzrlib.tests.test_store',
994
 
                   'bzrlib.tests.test_symbol_versioning',
995
 
                   'bzrlib.tests.test_testament',
996
 
                   'bzrlib.tests.test_textmerge',
997
 
                   'bzrlib.tests.test_trace',
998
 
                   'bzrlib.tests.test_transactions',
999
 
                   'bzrlib.tests.test_transform',
1000
 
                   'bzrlib.tests.test_transport',
1001
 
                   'bzrlib.tests.test_tsort',
1002
 
                   'bzrlib.tests.test_tuned_gzip',
1003
 
                   'bzrlib.tests.test_ui',
1004
 
                   'bzrlib.tests.test_upgrade',
1005
 
                   'bzrlib.tests.test_versionedfile',
1006
 
                   'bzrlib.tests.test_weave',
1007
 
                   'bzrlib.tests.test_whitebox',
1008
 
                   'bzrlib.tests.test_workingtree',
1009
 
                   'bzrlib.tests.test_xml',
1010
 
                   ]
1011
 
    test_transport_implementations = [
1012
 
        'bzrlib.tests.test_transport_implementations']
1013
 
 
1014
 
    TestCase.BZRPATH = osutils.pathjoin(
1015
 
            osutils.realpath(osutils.dirname(bzrlib.__path__[0])), 'bzr')
1016
 
    print '%10s: %s' % ('bzr', osutils.realpath(sys.argv[0]))
1017
 
    print '%10s: %s' % ('bzrlib', bzrlib.__path__[0])
1018
 
    print
1019
 
    suite = TestSuite()
1020
 
    # python2.4's TestLoader.loadTestsFromNames gives very poor 
1021
 
    # errors if it fails to load a named module - no indication of what's
1022
 
    # actually wrong, just "no such module".  We should probably override that
1023
 
    # class, but for the moment just load them ourselves. (mbp 20051202)
1024
 
    loader = TestLoader()
1025
 
    from bzrlib.transport import TransportTestProviderAdapter
1026
 
    adapter = TransportTestProviderAdapter()
1027
 
    adapt_modules(test_transport_implementations, adapter, loader, suite)
1028
 
    for mod_name in testmod_names:
1029
 
        mod = _load_module_by_name(mod_name)
1030
 
        suite.addTest(loader.loadTestsFromModule(mod))
1031
 
    for package in packages_to_test():
1032
 
        suite.addTest(package.test_suite())
1033
 
    for m in MODULES_TO_TEST:
1034
 
        suite.addTest(loader.loadTestsFromModule(m))
1035
 
    for m in (MODULES_TO_DOCTEST):
1036
 
        suite.addTest(DocTestSuite(m))
1037
 
    for name, plugin in bzrlib.plugin.all_plugins().items():
1038
 
        if getattr(plugin, 'test_suite', None) is not None:
1039
 
            suite.addTest(plugin.test_suite())
1040
 
    return suite
1041
 
 
1042
 
 
1043
 
def adapt_modules(mods_list, adapter, loader, suite):
1044
 
    """Adapt the modules in mods_list using adapter and add to suite."""
1045
 
    for mod_name in mods_list:
1046
 
        mod = _load_module_by_name(mod_name)
1047
 
        for test in iter_suite_tests(loader.loadTestsFromModule(mod)):
1048
 
            suite.addTests(adapter.adapt(test))
1049
 
 
1050
 
 
1051
 
def _load_module_by_name(mod_name):
1052
 
    parts = mod_name.split('.')
1053
 
    module = __import__(mod_name)
1054
 
    del parts[0]
1055
 
    # for historical reasons python returns the top-level module even though
1056
 
    # it loads the submodule; we need to walk down to get the one we want.
1057
 
    while parts:
1058
 
        module = getattr(module, parts.pop(0))
1059
 
    return module