/brz/remove-bazaar

# Copyright (C) 2005-2010 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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

"""Testing framework extensions"""

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

import codecs

import copy

from cStringIO import StringIO

import difflib

import doctest

import errno

import logging

import math

import os

import pprint

import random

import re

import shlex

import stat

import subprocess

import sys

import tempfile

import threading

import time

import traceback

import unittest

import warnings

import testtools

# nb: check this before importing anything else from within it

_testtools_version = getattr(testtools, '__version__', ())

if _testtools_version < (0, 9, 2):

    raise ImportError("need at least testtools 0.9.2: %s is %r"

        % (testtools.__file__, _testtools_version))

from testtools import content

from bzrlib import (

    branchbuilder,

    bzrdir,

    chk_map,

    config,

    debug,

    errors,

    hooks,

    lock as _mod_lock,

    memorytree,

    osutils,

    progress,

    ui,

    urlutils,

    registry,

    transport as _mod_transport,

    workingtree,

    )

import bzrlib.branch

import bzrlib.commands

import bzrlib.timestamp

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

from bzrlib.smart import client, request, server

import bzrlib.store

from bzrlib import symbol_versioning

from bzrlib.symbol_versioning import (

    DEPRECATED_PARAMETER,

    deprecated_function,

    deprecated_in,

    deprecated_method,

    deprecated_passed,

    )

import bzrlib.trace

from bzrlib.transport import (

    memory,

    pathfilter,

    )

from bzrlib.trace import mutter, note

from bzrlib.tests import (

    test_server,

    TestUtil,

    treeshape,

    )

from bzrlib.tests.http_server import HttpServer

from bzrlib.tests.TestUtil import (

                          TestSuite,

                          TestLoader,

                          )

from bzrlib.ui import NullProgressView

from bzrlib.ui.text import TextUIFactory

import bzrlib.version_info_formats.format_custom

from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2

# Mark this python module as being part of the implementation

# of unittest: this gives us better tracebacks where the last

# shown frame is the test code, not our assertXYZ.

__unittest = 1

default_transport = test_server.LocalURLServer

_unitialized_attr = object()

"""A sentinel needed to act as a default value in a method signature."""

# Subunit result codes, defined here to prevent a hard dependency on subunit.

SUBUNIT_SEEK_SET = 0

SUBUNIT_SEEK_CUR = 1

class ExtendedTestResult(unittest._TextTestResult):

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

    Compared to the 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.

    When a test finishes, in whatever way, it calls one of the addSuccess,

    addFailure or addError classes.  These in turn may redirect to a more

    specific case for the special test results supported by our extended

    tests.

    Note that just one of these objects is fed the results from many tests.

    """

    stop_early = False

    def __init__(self, stream, descriptions, verbosity,

                 bench_history=None,

                 strict=False,

                 ):

        """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 = ui.ui_factory

        self.num_tests = 0

        self.error_count = 0

        self.failure_count = 0

        self.known_failure_count = 0

        self.skip_count = 0

        self.not_applicable_count = 0

        self.unsupported = {}

        self.count = 0

        self._overall_start_time = time.time()

        self._strict = strict

    def stopTestRun(self):

        run = self.testsRun

        actionTaken = "Ran"

        stopTime = time.time()

        timeTaken = stopTime - self.startTime

        self.printErrors()

        self.stream.writeln(self.separator2)

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

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

        self.stream.writeln()

        if not self.wasSuccessful():

            self.stream.write("FAILED (")

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

            if failed:

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

            if errored:

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

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

            if self.known_failure_count:

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

                self.stream.write("known_failure_count=%d" %

                    self.known_failure_count)

            self.stream.writeln(")")

        else:

            if self.known_failure_count:

                self.stream.writeln("OK (known_failures=%d)" %

                    self.known_failure_count)

            else:

                self.stream.writeln("OK")

        if self.skip_count > 0:

            skipped = self.skip_count

            self.stream.writeln('%d test%s skipped' %

                                (skipped, skipped != 1 and "s" or ""))

        if self.unsupported:

            for feature, count in sorted(self.unsupported.items()):

                self.stream.writeln("Missing feature '%s' skipped %d tests." %

                    (feature, count))

        if self._strict:

            ok = self.wasStrictlySuccessful()

        else:

            ok = self.wasSuccessful()

        if TestCase._first_thread_leaker_id:

            self.stream.write(

                '%s is leaking threads among %d leaking tests.\n' % (

                TestCase._first_thread_leaker_id,

                TestCase._leaking_threads_tests))

            # We don't report the main thread as an active one.

            self.stream.write(

                '%d non-main threads were left active in the end.\n'

                % (TestCase._active_threads - 1))

    def getDescription(self, test):

        return test.id()

    def _extractBenchmarkTime(self, testCase, details=None):

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

        if details and 'benchtime' in details:

            return float(''.join(details['benchtime'].iter_bytes()))

        return 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, testCase):

        benchmark_time = self._extractBenchmarkTime(testCase)

        if benchmark_time is not None:

            return self._formatTime(benchmark_time) + "*"

        else:

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

        if self.count == 0:

            self.startTests()

        self.report_test_start(test)

        test.number = self.count

        self._recordTestStartTime()

    def startTests(self):

        import platform

        if getattr(sys, 'frozen', None) is None:

            bzr_path = osutils.realpath(sys.argv[0])

        else:

            bzr_path = sys.executable

        self.stream.write(

            'bzr selftest: %s\n' % (bzr_path,))

        self.stream.write(

            '   %s\n' % (

                    bzrlib.__path__[0],))

        self.stream.write(

            '   bzr-%s python-%s %s\n' % (

                    bzrlib.version_string,

                    bzrlib._format_version_tuple(sys.version_info),

                    platform.platform(aliased=1),

                    ))

        self.stream.write('\n')

    def _recordTestStartTime(self):

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

        self._start_time = time.time()

    def _cleanupLogFile(self, test):

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

    def addError(self, test, err):

        """Tell result that test finished with an error.

        Called from the TestCase run() method when the test

        fails with an unexpected error.

        """

        self._post_mortem()

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

        self.error_count += 1

        self.report_error(test, err)

        if self.stop_early:

            self.stop()

        self._cleanupLogFile(test)

    def addFailure(self, test, err):

        """Tell result that test failed.

        Called from the TestCase run() method when the test

        fails because e.g. an assert() method failed.

        """

        self._post_mortem()

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

        self.failure_count += 1

        self.report_failure(test, err)

        if self.stop_early:

            self.stop()

        self._cleanupLogFile(test)

    def addSuccess(self, test, details=None):

        """Tell result that test completed successfully.

        Called from the TestCase run()

        """

        if self._bench_history is not None:

            benchmark_time = self._extractBenchmarkTime(test, details)

            if benchmark_time is not None:

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

                    self._formatTime(benchmark_time),

                    test.id()))

        self.report_success(test)

        self._cleanupLogFile(test)

        unittest.TestResult.addSuccess(self, test)

        test._log_contents = ''

    def addExpectedFailure(self, test, err):

        self.known_failure_count += 1

        self.report_known_failure(test, err)

    def addNotSupported(self, test, feature):

        """The test will not be run because of a missing feature.

        """

        # this can be called in two different ways: it may be that the

        # test started running, and then raised (through requireFeature)

        # UnavailableFeature.  Alternatively this method can be called

        # while probing for features before running the test code proper; in

        # that case we will see startTest and stopTest, but the test will

        # never actually run.

        self.unsupported.setdefault(str(feature), 0)

        self.unsupported[str(feature)] += 1

        self.report_unsupported(test, feature)

    def addSkip(self, test, reason):

        """A test has not run for 'reason'."""

        self.skip_count += 1

        self.report_skip(test, reason)

    def addNotApplicable(self, test, reason):

        self.not_applicable_count += 1

        self.report_not_applicable(test, reason)

    def _post_mortem(self):

        """Start a PDB post mortem session."""

        if os.environ.get('BZR_TEST_PDB', None):

            import pdb;pdb.post_mortem()

    def progress(self, offset, whence):

        """The test is adjusting the count of tests to run."""

        if whence == SUBUNIT_SEEK_SET:

            self.num_tests = offset

        elif whence == SUBUNIT_SEEK_CUR:

            self.num_tests += offset

        else:

            raise errors.BzrError("Unknown whence %r" % whence)

    def report_cleaning_up(self):

        pass

    def startTestRun(self):

        self.startTime = time.time()

    def report_success(self, test):

        pass

    def wasStrictlySuccessful(self):

        if self.unsupported or self.known_failure_count:

            return False

        return self.wasSuccessful()

class TextTestResult(ExtendedTestResult):

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

    def __init__(self, stream, descriptions, verbosity,

                 bench_history=None,

                 pb=None,

                 strict=None,

                 ):

        ExtendedTestResult.__init__(self, stream, descriptions, verbosity,

            bench_history, strict)

        # We no longer pass them around, but just rely on the UIFactory stack

        # for state

        if pb is not None:

            warnings.warn("Passing pb to TextTestResult is deprecated")

        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

        self.pb.update_latency = 0

        self.pb.show_transport_activity = False

    def stopTestRun(self):

        # called when the tests that are going to run have run

        self.pb.clear()

        self.pb.finished()

        super(TextTestResult, self).stopTestRun()

    def startTestRun(self):

        super(TextTestResult, self).startTestRun()

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

    def printErrors(self):

        # clear the pb to make room for the error listing

        self.pb.clear()

        super(TextTestResult, self).printErrors()

    def _progress_prefix_text(self):

        # the longer this text, the less space we have to show the test

        # name...

        a = '[%d' % self.count              # total that have been run

        # tests skipped as known not to be relevant are not important enough

        # to show here

        ## if self.skip_count:

        ##     a += ', %d skip' % self.skip_count

        ## if self.known_failure_count:

        ##     a += '+%dX' % self.known_failure_count

        if self.num_tests:

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

        a += ' in '

        runtime = time.time() - self._overall_start_time

        if runtime >= 60:

            a += '%dm%ds' % (runtime / 60, runtime % 60)

        else:

            a += '%ds' % runtime

        total_fail_count = self.error_count + self.failure_count

        if total_fail_count:

            a += ', %d failed' % total_fail_count

        # if self.unsupported:

        #     a += ', %d missing' % len(self.unsupported)

        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 _test_description(self, test):

        return self._shortened_test_description(test)

    def report_error(self, test, err):

        self.stream.write('ERROR: %s\n    %s\n' % (

            self._test_description(test),

            err[1],

            ))

    def report_failure(self, test, err):

        self.stream.write('FAIL: %s\n    %s\n' % (

            self._test_description(test),

            err[1],

            ))

    def report_known_failure(self, test, err):

        pass

    def report_skip(self, test, reason):

        pass

    def report_not_applicable(self, test, reason):

        pass

    def report_unsupported(self, test, feature):

        """test cannot be run because feature is missing."""

    def report_cleaning_up(self):

        self.pb.update('Cleaning up')

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 startTestRun(self):

        super(VerboseTestResult, self).startTestRun()

        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 = osutils.terminal_width()

        if width is not None:

            # width needs space for 6 char status, plus 1 for slash, plus an

            # 11-char time string, plus a trailing blank

            # when NUMBERED_DIRS: plus 5 chars on test number, plus 1 char on

            # space

            self.stream.write(self._ellipsize_to_right(name, width-18))

        else:

            self.stream.write(name)

        self.stream.flush()

    def _error_summary(self, err):

        indent = ' ' * 4

        return '%s%s' % (indent, err[1])

    def report_error(self, test, err):

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

                % (self._testTimeString(test),

                   self._error_summary(err)))

    def report_failure(self, test, err):

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

                % (self._testTimeString(test),

                   self._error_summary(err)))

    def report_known_failure(self, test, err):

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

                % (self._testTimeString(test),

                   self._error_summary(err)))

    def report_success(self, test):

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

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

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

            stats.pprint(file=self.stream)

        # flush the stream so that we get smooth output. This verbose mode is

        # used to show the output in PQM.

        self.stream.flush()

    def report_skip(self, test, reason):

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

                % (self._testTimeString(test), reason))

    def report_not_applicable(self, test, reason):

        self.stream.writeln('  N/A %s\n    %s'

                % (self._testTimeString(test), reason))

    def report_unsupported(self, test, feature):

        """test cannot be run because feature is missing."""

        self.stream.writeln("NODEP %s\n    The feature '%s' is not available."

                %(self._testTimeString(test), feature))

class TextTestRunner(object):

    stop_on_failure = False

    def __init__(self,

                 stream=sys.stderr,

                 descriptions=0,

                 verbosity=1,

                 bench_history=None,

                 strict=False,

                 result_decorators=None,

                 ):

        """Create a TextTestRunner.

        :param result_decorators: An optional list of decorators to apply

            to the result object being used by the runner. Decorators are

            applied left to right - the first element in the list is the 

            innermost decorator.

        """

        # stream may know claim to know to write unicode strings, but in older

        # pythons this goes sufficiently wrong that it is a bad idea. (

        # specifically a built in file with encoding 'UTF-8' will still try

        # to encode using ascii.

        new_encoding = osutils.get_terminal_encoding()

        codec = codecs.lookup(new_encoding)

        if type(codec) is tuple:

            # Python 2.4

            encode = codec[0]

        else:

            encode = codec.encode

        stream = osutils.UnicodeOrBytesToBytesWriter(encode, stream)

        stream.encoding = new_encoding