~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2005, 2006, 2007 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

from pprint import pformat

import re

import shlex

import stat

from subprocess import Popen, PIPE

import sys

import tempfile

import unittest

import time

from bzrlib import (

    bzrdir,

    debug,

    errors,

    memorytree,

    osutils,

    progress,

    urlutils,

    )

import bzrlib.branch

import bzrlib.commands

import bzrlib.bundle.serializer

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.plugin

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 LocalURLServer

from bzrlib.transport.memory import MemoryServer

from bzrlib.transport.readonly import ReadonlyServer

from bzrlib.trace import mutter, note

from bzrlib.tests import TestUtil

from bzrlib.tests.HttpServer import HttpServer

from bzrlib.tests.TestUtil import (

                          TestSuite,

                          TestLoader,

                          )

from bzrlib.tests.treeshape import build_tree_contents

from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2

default_transport = LocalURLServer

MODULES_TO_TEST = []

MODULES_TO_DOCTEST = [

                      bzrlib.bundle.serializer,

                      bzrlib.errors,

                      bzrlib.export,

                      bzrlib.inventory,

                      bzrlib.iterablefile,

                      bzrlib.lockdir,

                      bzrlib.merge3,

                      bzrlib.option,

                      bzrlib.store,

                      ]

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 ExtendedTestResult(unittest._TextTestResult):

    """Accepts, reports and accumulates the results of running tests.

    Compared to this unittest version this class adds support for profiling,

    benchmarking, stopping as soon as a test fails,  and skipping tests.

    There are further-specialized subclasses for different types of display.

    """

    stop_early = False

    def __init__(self, stream, descriptions, verbosity,

                 bench_history=None,

                 num_tests=None,

                 ):

        """Construct new TestResult.

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

            benchmark results.

        """

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

        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

        self.ui = bzrlib.ui.ui_factory

        self.num_tests = num_tests

        self.error_count = 0

        self.failure_count = 0

        self.skip_count = 0

        self.count = 0

        self._overall_start_time = time.time()

    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."""

        # some benchmarks can take thousands of seconds to run, so we need 8

        # places

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

    def _shortened_test_description(self, test):

        what = test.id()

        what = re.sub(r'^bzrlib\.(tests|benchmarks)\.', '', what)

        return what

    def startTest(self, test):

        unittest.TestResult.startTest(self, test)

        self.report_test_start(test)

        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)

        self.report_error(test, err)

        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)

        self.report_failure(test, err)

        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()))

        self.report_success(test)

        unittest.TestResult.addSuccess(self, test)

    def addSkipped(self, test, skip_excinfo):

        self.extractBenchmarkTime(test)

        self.report_skip(test, skip_excinfo)

        # 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)

    def finished(self):

        pass

    def report_cleaning_up(self):

        pass

    def report_success(self, test):

        pass

class TextTestResult(ExtendedTestResult):

    """Displays progress and results of tests in text form"""

    def __init__(self, *args, **kw):

        ExtendedTestResult.__init__(self, *args, **kw)

        self.pb = self.ui.nested_progress_bar()

        self.pb.show_pct = False

        self.pb.show_spinner = False

        self.pb.show_eta = False, 

        self.pb.show_count = False

        self.pb.show_bar = False

    def report_starting(self):

        self.pb.update('[test 0/%d] starting...' % (self.num_tests))

    def _progress_prefix_text(self):

        a = '[%d' % self.count

        if self.num_tests is not None:

            a +='/%d' % self.num_tests

        a += ' in %ds' % (time.time() - self._overall_start_time)

        if self.error_count:

            a += ', %d errors' % self.error_count

        if self.failure_count:

            a += ', %d failed' % self.failure_count

        if self.skip_count:

            a += ', %d skipped' % self.skip_count

        a += ']'

        return a

    def report_test_start(self, test):

        self.count += 1

        self.pb.update(

                self._progress_prefix_text()

                + ' ' 

                + self._shortened_test_description(test))

    def report_error(self, test, err):

        self.error_count += 1

        self.pb.note('ERROR: %s\n    %s\n', 

            self._shortened_test_description(test),

            err[1],

            )

    def report_failure(self, test, err):

        self.failure_count += 1

        self.pb.note('FAIL: %s\n    %s\n', 

            self._shortened_test_description(test),

            err[1],

            )

    def report_skip(self, test, skip_excinfo):

        self.skip_count += 1

        if False:

            # at the moment these are mostly not things we can fix

            # and so they just produce stipple; use the verbose reporter

            # to see them.

            if False:

                # show test and reason for skip

                self.pb.note('SKIP: %s\n    %s\n', 

                    self._shortened_test_description(test),

                    skip_excinfo[1])

            else:

                # since the class name was left behind in the still-visible

                # progress bar...

                self.pb.note('SKIP: %s', skip_excinfo[1])

    def report_cleaning_up(self):

        self.pb.update('cleaning up...')

    def finished(self):

        self.pb.finished()

class VerboseTestResult(ExtendedTestResult):

    """Produce long output, with one line per test run plus times"""

    def _ellipsize_to_right(self, a_string, final_width):

        """Truncate and pad a string, keeping the right hand side"""

        if len(a_string) > final_width:

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

        else:

            result = a_string

        return result.ljust(final_width)

    def report_starting(self):

        self.stream.write('running %d tests...\n' % self.num_tests)

    def report_test_start(self, test):

        self.count += 1

        name = self._shortened_test_description(test)

        # width needs space for 6 char status, plus 1 for slash, plus 2 10-char

        # numbers, plus a trailing blank

        self.stream.write(self._ellipsize_to_right(name,

                            osutils.terminal_width()-30))

        self.stream.flush()

    def report_error(self, test, err):

        self.error_count += 1

        self.stream.writeln('ERROR %s\n    %s' 

                % (self._testTimeString(), err[1]))

    def report_failure(self, test, err):

        self.failure_count += 1

        self.stream.writeln(' FAIL %s\n    %s'

                % (self._testTimeString(), err[1]))

    def report_success(self, test):

        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)

        self.stream.flush()

    def report_skip(self, test, skip_excinfo):

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

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

class TextTestRunner(object):

    stop_on_failure = False

    def __init__(self,

                 stream=sys.stderr,

                 descriptions=0,

                 verbosity=1,

                 keep_output=False,

                 bench_history=None):

        self.stream = unittest._WritelnDecorator(stream)

        self.descriptions = descriptions

        self.verbosity = verbosity

        self.keep_output = keep_output

        self._bench_history = bench_history

    def run(self, test):

        "Run the given test case or test suite."

        startTime = time.time()

        if self.verbosity == 1:

            result_class = TextTestResult

        elif self.verbosity >= 2:

            result_class = VerboseTestResult

        result = result_class(self.stream,

                              self.descriptions,

                              self.verbosity,

                              bench_history=self._bench_history,

                              num_tests=test.countTestCases(),

                              )

        result.stop_early = self.stop_on_failure

        result.report_starting()

        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")

        result.report_cleaning_up()

        # 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 = TestCaseWithMemoryTransport.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())

                try:

                    osutils.rmtree(test_root)

                except OSError, e:

                    if sys.platform == 'win32' and e.errno == errno.EACCES:

                        print >>sys.stderr, ('Permission denied: '

                                             'unable to remove testing dir '

                                             '%s' % os.path.basename(test_root))

                    else:

                        raise

        else:

            note("Failed tests working directories are in '%s'\n", test_root)

        TestCaseWithMemoryTransport.TEST_ROOT = None

        result.finished()

        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._silenceUI()

        self._startLogFile()

        self._benchcalls = []

        self._benchtime = None

        # prevent hooks affecting tests

        self._preserved_hooks = bzrlib.branch.Branch.hooks

        self.addCleanup(self._restoreHooks)

        # this list of hooks must be kept in sync with the defaults

        # in branch.py

        bzrlib.branch.Branch.hooks = bzrlib.branch.BranchHooks()

    def _silenceUI(self):

        """Turn off UI for duration of test"""

        # by default the UI is off; tests can turn it on if they want it.

        saved = bzrlib.ui.ui_factory

        def _restore():

            bzrlib.ui.ui_factory = saved

        bzrlib.ui.ui_factory = bzrlib.ui.SilentUIFactory()

        self.addCleanup(_restore)

    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 assertEqual(self, a, b, message=''):

        if a == b:

            return

        if message:

            message += '\n'

        raise AssertionError("%snot equal:\na = %s\nb = %s\n"

            % (message,

               pformat(a, indent=4), pformat(b, indent=4)))

    assertEquals = assertEqual

    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 +