~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2005, 2006 by Canonical Ltd

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

# TODO: Perhaps there should be an API to find out if bzr running under the

# test suite -- some plugins might want to avoid making intrusive changes if

# this is the case.  However, we want behaviour under to test to diverge as

# little as possible, so this should be used rarely if it's added at all.

# (Suggestion from j-a-meinel, 2005-11-24)

# NOTE: Some classes in here use camelCaseNaming() rather than

# underscore_naming().  That's for consistency with unittest; it's not the

# general style of bzrlib.  Please continue that consistency when adding e.g.

# new assertFoo() methods.

import codecs

from cStringIO import StringIO

import difflib

import doctest

import errno

import logging

import os

import re

import shlex

import stat

from subprocess import Popen, PIPE

import sys

import tempfile

import unittest

import time

from bzrlib import memorytree

import bzrlib.branch

import bzrlib.bzrdir as bzrdir

import bzrlib.commands

import bzrlib.bundle.serializer

import bzrlib.errors as errors

import bzrlib.export

import bzrlib.inventory

import bzrlib.iterablefile

import bzrlib.lockdir

try:

    import bzrlib.lsprof

except ImportError:

    # lsprof not available

    pass

from bzrlib.merge import merge_inner

import bzrlib.merge3

import bzrlib.osutils

import bzrlib.osutils as osutils

import bzrlib.plugin

import bzrlib.progress as progress

from bzrlib.revision import common_ancestor

import bzrlib.store

from bzrlib import symbol_versioning

import bzrlib.trace

from bzrlib.transport import get_transport

import bzrlib.transport

from bzrlib.transport.local import LocalRelpathServer

from bzrlib.transport.readonly import ReadonlyServer

from bzrlib.trace import mutter

from bzrlib.tests import TestUtil

from bzrlib.tests.TestUtil import (

                          TestSuite,

                          TestLoader,

                          )

from bzrlib.tests.treeshape import build_tree_contents

import bzrlib.urlutils as urlutils

from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2

default_transport = LocalRelpathServer

MODULES_TO_TEST = []

MODULES_TO_DOCTEST = [

                      bzrlib.branch,

                      bzrlib.bundle.serializer,

                      bzrlib.commands,

                      bzrlib.errors,

                      bzrlib.export,

                      bzrlib.inventory,

                      bzrlib.iterablefile,

                      bzrlib.lockdir,

                      bzrlib.merge3,

                      bzrlib.option,

                      bzrlib.osutils,

                      bzrlib.store,

                      bzrlib.transport,

                      ]

def packages_to_test():

    """Return a list of packages to test.

    The packages are not globally imported so that import failures are

    triggered when running selftest, not when importing the command.

    """

    import bzrlib.doc

    import bzrlib.tests.blackbox

    import bzrlib.tests.branch_implementations

    import bzrlib.tests.bzrdir_implementations

    import bzrlib.tests.interrepository_implementations

    import bzrlib.tests.interversionedfile_implementations

    import bzrlib.tests.intertree_implementations

    import bzrlib.tests.repository_implementations

    import bzrlib.tests.revisionstore_implementations

    import bzrlib.tests.tree_implementations

    import bzrlib.tests.workingtree_implementations

    return [

            bzrlib.doc,

            bzrlib.tests.blackbox,

            bzrlib.tests.branch_implementations,

            bzrlib.tests.bzrdir_implementations,

            bzrlib.tests.interrepository_implementations,

            bzrlib.tests.interversionedfile_implementations,

            bzrlib.tests.intertree_implementations,

            bzrlib.tests.repository_implementations,

            bzrlib.tests.revisionstore_implementations,

            bzrlib.tests.tree_implementations,

            bzrlib.tests.workingtree_implementations,

            ]

class _MyResult(unittest._TextTestResult):

    """Custom TestResult.

    Shows output in a different format, including displaying runtime for tests.

    """

    stop_early = False

    def __init__(self, stream, descriptions, verbosity, pb=None,

                 bench_history=None):

        """Construct new TestResult.

        :param bench_history: Optionally, a writable file object to accumulate

            benchmark results.

        """

        unittest._TextTestResult.__init__(self, stream, descriptions, verbosity)

        self.pb = pb

        if bench_history is not None:

            from bzrlib.version import _get_bzr_source_tree

            src_tree = _get_bzr_source_tree()

            if src_tree:

                try:

                    revision_id = src_tree.get_parent_ids()[0]

                except IndexError:

                    # XXX: if this is a brand new tree, do the same as if there

                    # is no branch.

                    revision_id = ''

            else:

                # XXX: If there's no branch, what should we do?

                revision_id = ''

            bench_history.write("--date %s %s\n" % (time.time(), revision_id))

        self._bench_history = bench_history

    def extractBenchmarkTime(self, testCase):

        """Add a benchmark time for the current test case."""

        self._benchmarkTime = getattr(testCase, "_benchtime", None)

    def _elapsedTestTimeString(self):

        """Return a time string for the overall time the current test has taken."""

        return self._formatTime(time.time() - self._start_time)

    def _testTimeString(self):

        if self._benchmarkTime is not None:

            return "%s/%s" % (

                self._formatTime(self._benchmarkTime),

                self._elapsedTestTimeString())

        else:

            return "      %s" % self._elapsedTestTimeString()

    def _formatTime(self, seconds):

        """Format seconds as milliseconds with leading spaces."""

        return "%5dms" % (1000 * seconds)

    def _ellipsise_unimportant_words(self, a_string, final_width,

                                   keep_start=False):

        """Add ellipses (sp?) for overly long strings.

        :param keep_start: If true preserve the start of a_string rather

                           than the end of it.

        """

        if keep_start:

            if len(a_string) > final_width:

                result = a_string[:final_width-3] + '...'

            else:

                result = a_string

        else:

            if len(a_string) > final_width:

                result = '...' + a_string[3-final_width:]

            else:

                result = a_string

        return result.ljust(final_width)

    def startTest(self, test):

        unittest.TestResult.startTest(self, test)

        # In a short description, the important words are in

        # the beginning, but in an id, the important words are

        # at the end

        SHOW_DESCRIPTIONS = False

        if not self.showAll and self.dots and self.pb is not None:

            final_width = 13

        else:

            final_width = osutils.terminal_width()

            final_width = final_width - 15 - 8

        what = None

        if SHOW_DESCRIPTIONS:

            what = test.shortDescription()

            if what:

                what = self._ellipsise_unimportant_words(what, final_width, keep_start=True)

        if what is None:

            what = test.id()

            if what.startswith('bzrlib.tests.'):

                what = what[13:]

            what = self._ellipsise_unimportant_words(what, final_width)

        if self.showAll:

            self.stream.write(what)

        elif self.dots and self.pb is not None:

            self.pb.update(what, self.testsRun - 1, None)

        self.stream.flush()

        self._recordTestStartTime()

    def _recordTestStartTime(self):

        """Record that a test has started."""

        self._start_time = time.time()

    def addError(self, test, err):

        if isinstance(err[1], TestSkipped):

            return self.addSkipped(test, err)    

        unittest.TestResult.addError(self, test, err)

        # We can only do this if we have one of our TestCases, not if

        # we have a doctest.

        setKeepLogFile = getattr(test, 'setKeepLogFile', None)

        if setKeepLogFile is not None:

            setKeepLogfile()

        self.extractBenchmarkTime(test)

        if self.showAll:

            self.stream.writeln("ERROR %s" % self._testTimeString())

        elif self.dots and self.pb is None:

            self.stream.write('E')

        elif self.dots:

            self.pb.update(self._ellipsise_unimportant_words('ERROR', 13), self.testsRun, None)

            self.pb.note(self._ellipsise_unimportant_words(

                            test.id() + ': ERROR',

                            osutils.terminal_width()))

        self.stream.flush()

        if self.stop_early:

            self.stop()

    def addFailure(self, test, err):

        unittest.TestResult.addFailure(self, test, err)

        # We can only do this if we have one of our TestCases, not if

        # we have a doctest.

        setKeepLogFile = getattr(test, 'setKeepLogFile', None)

        if setKeepLogFile is not None:

            setKeepLogfile()

        self.extractBenchmarkTime(test)

        if self.showAll:

            self.stream.writeln(" FAIL %s" % self._testTimeString())

        elif self.dots and self.pb is None:

            self.stream.write('F')

        elif self.dots:

            self.pb.update(self._ellipsise_unimportant_words('FAIL', 13), self.testsRun, None)

            self.pb.note(self._ellipsise_unimportant_words(

                            test.id() + ': FAIL',

                            osutils.terminal_width()))

        self.stream.flush()

        if self.stop_early:

            self.stop()

    def addSuccess(self, test):

        self.extractBenchmarkTime(test)

        if self._bench_history is not None:

            if self._benchmarkTime is not None:

                self._bench_history.write("%s %s\n" % (

                    self._formatTime(self._benchmarkTime),

                    test.id()))

        if self.showAll:

            self.stream.writeln('   OK %s' % self._testTimeString())

            for bench_called, stats in getattr(test, '_benchcalls', []):

                self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)

                stats.pprint(file=self.stream)

        elif self.dots and self.pb is None:

            self.stream.write('~')

        elif self.dots:

            self.pb.update(self._ellipsise_unimportant_words('OK', 13), self.testsRun, None)

        self.stream.flush()

        unittest.TestResult.addSuccess(self, test)

    def addSkipped(self, test, skip_excinfo):

        self.extractBenchmarkTime(test)

        if self.showAll:

            print >>self.stream, ' SKIP %s' % self._testTimeString()

            print >>self.stream, '     %s' % skip_excinfo[1]

        elif self.dots and self.pb is None:

            self.stream.write('S')

        elif self.dots:

            self.pb.update(self._ellipsise_unimportant_words('SKIP', 13), self.testsRun, None)

        self.stream.flush()

        # seems best to treat this as success from point-of-view of unittest

        # -- it actually does nothing so it barely matters :)

        try:

            test.tearDown()

        except KeyboardInterrupt:

            raise

        except:

            self.addError(test, test.__exc_info())

        else:

            unittest.TestResult.addSuccess(self, test)

    def printErrorList(self, flavour, errors):

        for test, err in errors:

            self.stream.writeln(self.separator1)

            self.stream.writeln("%s: %s" % (flavour, self.getDescription(test)))

            if getattr(test, '_get_log', None) is not None:

                print >>self.stream

                print >>self.stream, \

                        ('vvvv[log from %s]' % test.id()).ljust(78,'-')

                print >>self.stream, test._get_log()

                print >>self.stream, \

                        ('^^^^[log from %s]' % test.id()).ljust(78,'-')

            self.stream.writeln(self.separator2)

            self.stream.writeln("%s" % err)

class TextTestRunner(object):

    stop_on_failure = False

    def __init__(self,

                 stream=sys.stderr,

                 descriptions=0,

                 verbosity=1,

                 keep_output=False,

                 pb=None,

                 bench_history=None):

        self.stream = unittest._WritelnDecorator(stream)

        self.descriptions = descriptions

        self.verbosity = verbosity

        self.keep_output = keep_output

        self.pb = pb

        self._bench_history = bench_history

    def _makeResult(self):

        result = _MyResult(self.stream,

                           self.descriptions,

                           self.verbosity,

                           pb=self.pb,

                           bench_history=self._bench_history)

        result.stop_early = self.stop_on_failure

        return result

    def run(self, test):

        "Run the given test case or test suite."

        result = self._makeResult()

        startTime = time.time()

        if self.pb is not None:

            self.pb.update('Running tests', 0, test.countTestCases())

        test.run(result)

        stopTime = time.time()

        timeTaken = stopTime - startTime

        result.printErrors()

        self.stream.writeln(result.separator2)

        run = result.testsRun

        self.stream.writeln("Ran %d test%s in %.3fs" %

                            (run, run != 1 and "s" or "", timeTaken))

        self.stream.writeln()

        if not result.wasSuccessful():

            self.stream.write("FAILED (")

            failed, errored = map(len, (result.failures, result.errors))

            if failed:

                self.stream.write("failures=%d" % failed)

            if errored:

                if failed: self.stream.write(", ")

                self.stream.write("errors=%d" % errored)

            self.stream.writeln(")")

        else:

            self.stream.writeln("OK")

        if self.pb is not None:

            self.pb.update('Cleaning up', 0, 1)

        # This is still a little bogus, 

        # but only a little. Folk not using our testrunner will

        # have to delete their temp directories themselves.

        test_root = TestCaseInTempDir.TEST_ROOT

        if result.wasSuccessful() or not self.keep_output:

            if test_root is not None:

                # If LANG=C we probably have created some bogus paths

                # which rmtree(unicode) will fail to delete

                # so make sure we are using rmtree(str) to delete everything

                # except on win32, where rmtree(str) will fail

                # since it doesn't have the property of byte-stream paths

                # (they are either ascii or mbcs)

                if sys.platform == 'win32':

                    # make sure we are using the unicode win32 api

                    test_root = unicode(test_root)

                else:

                    test_root = test_root.encode(

                        sys.getfilesystemencoding())

                osutils.rmtree(test_root)

        else:

            if self.pb is not None:

                self.pb.note("Failed tests working directories are in '%s'\n",

                             test_root)

            else:

                self.stream.writeln(

                    "Failed tests working directories are in '%s'\n" %

                    test_root)

        TestCaseInTempDir.TEST_ROOT = None

        if self.pb is not None:

            self.pb.clear()

        return result

def iter_suite_tests(suite):

    """Return all tests in a suite, recursing through nested suites"""

    for item in suite._tests:

        if isinstance(item, unittest.TestCase):

            yield item

        elif isinstance(item, unittest.TestSuite):

            for r in iter_suite_tests(item):

                yield r

        else:

            raise Exception('unknown object %r inside test suite %r'

                            % (item, suite))

class TestSkipped(Exception):

    """Indicates that a test was intentionally skipped, rather than failing."""

class CommandFailed(Exception):

    pass

class StringIOWrapper(object):

    """A wrapper around cStringIO which just adds an encoding attribute.

    Internally we can check sys.stdout to see what the output encoding

    should be. However, cStringIO has no encoding attribute that we can

    set. So we wrap it instead.

    """

    encoding='ascii'

    _cstring = None

    def __init__(self, s=None):

        if s is not None:

            self.__dict__['_cstring'] = StringIO(s)

        else:

            self.__dict__['_cstring'] = StringIO()

    def __getattr__(self, name, getattr=getattr):

        return getattr(self.__dict__['_cstring'], name)

    def __setattr__(self, name, val):

        if name == 'encoding':

            self.__dict__['encoding'] = val

        else:

            return setattr(self._cstring, name, val)

class TestCase(unittest.TestCase):

    """Base class for bzr unit tests.

    Tests that need access to disk resources should subclass 

    TestCaseInTempDir not TestCase.

    Error and debug log messages are redirected from their usual

    location into a temporary file, the contents of which can be

    retrieved by _get_log().  We use a real OS file, not an in-memory object,

    so that it can also capture file IO.  When the test completes this file

    is read into memory and removed from disk.

    There are also convenience functions to invoke bzr's command-line

    routine, and to build and check bzr trees.

    In addition to the usual method of overriding tearDown(), this class also

    allows subclasses to register functions into the _cleanups list, which is

    run in order as the object is torn down.  It's less likely this will be

    accidentally overlooked.

    """

    _log_file_name = None

    _log_contents = ''

    _keep_log_file = False

    # record lsprof data when performing benchmark calls.

    _gather_lsprof_in_benchmarks = False

    def __init__(self, methodName='testMethod'):

        super(TestCase, self).__init__(methodName)

        self._cleanups = []

    def setUp(self):

        unittest.TestCase.setUp(self)

        self._cleanEnvironment()

        bzrlib.trace.disable_default_logging()

        self._startLogFile()

        self._benchcalls = []

        self._benchtime = None

    def _ndiff_strings(self, a, b):

        """Return ndiff between two strings containing lines.

        A trailing newline is added if missing to make the strings

        print properly."""

        if b and b[-1] != '\n':

            b += '\n'

        if a and a[-1] != '\n':

            a += '\n'

        difflines = difflib.ndiff(a.splitlines(True),

                                  b.splitlines(True),

                                  linejunk=lambda x: False,

                                  charjunk=lambda x: False)

        return ''.join(difflines)

    def assertEqualDiff(self, a, b, message=None):

        """Assert two texts are equal, if not raise an exception.

        This is intended for use with multi-line strings where it can 

        be hard to find the differences by eye.

        """

        # TODO: perhaps override assertEquals to call this for strings?

        if a == b:

            return

        if message is None:

            message = "texts not equal:\n"

        raise AssertionError(message + 

                             self._ndiff_strings(a, b))      

    def assertEqualMode(self, mode, mode_test):

        self.assertEqual(mode, mode_test,

                         'mode mismatch %o != %o' % (mode, mode_test))

    def assertStartsWith(self, s, prefix):

        if not s.startswith(prefix):

            raise AssertionError('string %r does not start with %r' % (s, prefix))

    def assertEndsWith(self, s, suffix):

        """Asserts that s ends with suffix."""

        if not s.endswith(suffix):

            raise AssertionError('string %r does not end with %r' % (s, suffix))

    def assertContainsRe(self, haystack, needle_re):

        """Assert that a contains something matching a regular expression."""

        if not re.search(needle_re, haystack):

            raise AssertionError('pattern "%s" not found in "%s"'

                    % (needle_re, haystack))

    def assertNotContainsRe(self, haystack, needle_re):

        """Assert that a does not match a regular expression"""

        if re.search(needle_re, haystack):

            raise AssertionError('pattern "%s" found in "%s"'

                    % (needle_re, haystack))

    def assertSubset(self, sublist, superlist):

        """Assert that every entry in sublist is present in superlist."""

        missing = []

        for entry in sublist:

            if entry not in superlist:

                missing.append(entry)

        if len(missing) > 0:

            raise AssertionError("value(s) %r not present in container %r" % 

                                 (missing, superlist))

    def assertIs(self, left, right):

        if not (left is right):

            raise AssertionError("%r is not %r." % (left, right))

    def assertTransportMode(self, transport, path, mode):

        """Fail if a path does not have mode mode.

        If modes are not supported on this transport, the assertion is ignored.

        """

        if not transport._can_roundtrip_unix_modebits():

            return

        path_stat = transport.stat(path)

        actual_mode = stat.S_IMODE(path_stat.st_mode)

        self.assertEqual(mode, actual_mode,

            'mode of %r incorrect (%o != %o)' % (path, mode, actual_mode))

    def assertIsInstance(self, obj, kls):

        """Fail if obj is not an instance of kls"""

        if not isinstance(obj, kls):

            self.fail("%r is an instance of %s rather than %s" % (

                obj, obj.__class__, kls))

    def _capture_warnings(self, a_callable, *args, **kwargs):

        """A helper for callDeprecated and applyDeprecated.

        :param a_callable: A callable to call.

        :param args: The positional arguments for the callable

        :param kwargs: The keyword arguments for the callable

        :return: A tuple (warnings, result). result is the result of calling

            a_callable(*args, **kwargs).

        """

        local_warnings = []

        def capture_warnings(msg, cls, stacklevel=None):

            # we've hooked into a deprecation specific callpath,

            # only deprecations should getting sent via it.

            self.assertEqual(cls, DeprecationWarning)

            local_warnings.append(msg)

        original_warning_method = symbol_versioning.warn

        symbol_versioning.set_warning_method(capture_warnings)

        try:

            result = a_callable(*args, **kwargs)