/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 itertools

import logging

import os

import platform

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, 5):

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

        % (testtools.__file__, _testtools_version))

from testtools import content

import bzrlib

from bzrlib import (

    branchbuilder,

    bzrdir,

    chk_map,

    commands as _mod_commands,

    config,

    debug,

    errors,

    hooks,

    lock as _mod_lock,

    lockdir,

    memorytree,

    osutils,

    plugin as _mod_plugin,

    pyutils,

    ui,

    urlutils,

    registry,

    symbol_versioning,

    trace,

    transport as _mod_transport,

    workingtree,

    )

try:

    import bzrlib.lsprof

except ImportError:

    # lsprof not available

    pass

from bzrlib.smart import client, request

from bzrlib.transport import (

    memory,

    pathfilter,

    )

from bzrlib.tests import (

    test_server,

    TestUtil,

    treeshape,

    )

from bzrlib.ui import NullProgressView

from bzrlib.ui.text import TextUIFactory

# 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

# These are intentionally brought into this namespace. That way plugins, etc

# can just "from bzrlib.tests import TestCase, TestLoader, etc"

TestSuite = TestUtil.TestSuite

TestLoader = TestUtil.TestLoader

# Tests should run in a clean and clearly defined environment. The goal is to

# keep them isolated from the running environment as mush as possible. The test

# framework ensures the variables defined below are set (or deleted if the

# value is None) before a test is run and reset to their original value after

# the test is run. Generally if some code depends on an environment variable,

# the tests should start without this variable in the environment. There are a

# few exceptions but you shouldn't violate this rule lightly.

isolated_environ = {

    'BZR_HOME': None,

    'HOME': None,

    # bzr now uses the Win32 API and doesn't rely on APPDATA, but the

    # tests do check our impls match APPDATA

    'BZR_EDITOR': None, # test_msgeditor manipulates this variable

    'VISUAL': None,

    'EDITOR': None,

    'BZR_EMAIL': None,

    'BZREMAIL': None, # may still be present in the environment

    'EMAIL': 'jrandom@example.com', # set EMAIL as bzr does not guess

    'BZR_PROGRESS_BAR': None,

    'BZR_LOG': None,

    'BZR_PLUGIN_PATH': None,

    'BZR_DISABLE_PLUGINS': None,

    'BZR_PLUGINS_AT': None,

    'BZR_CONCURRENCY': None,

    # Make sure that any text ui tests are consistent regardless of

    # the environment the test case is run in; you may want tests that

    # test other combinations.  'dumb' is a reasonable guess for tests

    # going to a pipe or a StringIO.

    'TERM': 'dumb',

    'LINES': '25',

    'COLUMNS': '80',

    'BZR_COLUMNS': '80',

    # Disable SSH Agent

    'SSH_AUTH_SOCK': None,

    # Proxies

    'http_proxy': None,

    'HTTP_PROXY': None,

    'https_proxy': None,

    'HTTPS_PROXY': None,

    'no_proxy': None,

    'NO_PROXY': None,

    'all_proxy': None,

    'ALL_PROXY': None,

    # Nobody cares about ftp_proxy, FTP_PROXY AFAIK. So far at

    # least. If you do (care), please update this comment

    # -- vila 20080401

    'ftp_proxy': None,

    'FTP_PROXY': None,

    'BZR_REMOTE_PATH': None,

    # Generally speaking, we don't want apport reporting on crashes in

    # the test envirnoment unless we're specifically testing apport,

    # so that it doesn't leak into the real system environment.  We

    # use an env var so it propagates to subprocesses.

    'APPORT_DISABLE': '1',

    }

def override_os_environ(test, env=None):

    """Modify os.environ keeping a copy.

    :param test: A test instance

    :param env: A dict containing variable definitions to be installed

    """

    if env is None:

        env = isolated_environ

    test._original_os_environ = dict([(var, value)

                                      for var, value in os.environ.iteritems()])

    for var, value in env.iteritems():

        osutils.set_or_unset_env(var, value)

        if var not in test._original_os_environ:

            # The var is new, add it with a value of None, so

            # restore_os_environ will delete it

            test._original_os_environ[var] = None

def restore_os_environ(test):

    """Restore os.environ to its original state.

    :param test: A test instance previously passed to override_os_environ.

    """

    for var, value in test._original_os_environ.iteritems():

        # Restore the original value (or delete it if the value has been set to

        # None in override_os_environ).

        osutils.set_or_unset_env(var, value)

class ExtendedTestResult(testtools.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.

        """

        testtools.TextTestResult.__init__(self, stream)

        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

        self._first_thread_leaker_id = None

        self._tests_leaking_threads_count = 0

        self._traceback_from_test = None

    def stopTestRun(self):

        run = self.testsRun

        actionTaken = "Ran"

        stopTime = time.time()

        timeTaken = stopTime - self.startTime

        # GZ 2010-07-19: Seems testtools has no printErrors method, and though

        #                the parent class method is similar have to duplicate

        self._show_list('ERROR', self.errors)

        self._show_list('FAIL', self.failures)

        self.stream.write(self.sep2)

        self.stream.write("%s %d test%s in %.3fs\n\n" % (actionTaken,

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

        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.write(")\n")

        else:

            if self.known_failure_count:

                self.stream.write("OK (known_failures=%d)\n" %

                    self.known_failure_count)

            else:

                self.stream.write("OK\n")

        if self.skip_count > 0:

            skipped = self.skip_count

            self.stream.write('%d test%s skipped\n' %

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

        if self.unsupported:

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

                self.stream.write("Missing feature '%s' skipped %d tests.\n" %

                    (feature, count))

        if self._strict:

            ok = self.wasStrictlySuccessful()

        else:

            ok = self.wasSuccessful()

        if self._first_thread_leaker_id:

            self.stream.write(

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

                self._first_thread_leaker_id,

                self._tests_leaking_threads_count))

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

                % (len(self._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(self._delta_to_float(

            self._now() - self._start_datetime))

    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\.', '', what)

        return what

    # GZ 2010-10-04: Cloned tests may end up harmlessly calling this method

    #                multiple times in a row, because the handler is added for

    #                each test but the container list is shared between cases.

    #                See lp:498869 lp:625574 and lp:637725 for background.

    def _record_traceback_from_test(self, exc_info):

        """Store the traceback from passed exc_info tuple till"""

        self._traceback_from_test = exc_info[2]

    def startTest(self, test):

        super(ExtendedTestResult, self).startTest(test)

        if self.count == 0:

            self.startTests()

        self.count += 1

        self.report_test_start(test)

        test.number = self.count

        self._recordTestStartTime()

        # Make testtools cases give us the real traceback on failure

        addOnException = getattr(test, "addOnException", None)

        if addOnException is not None:

            addOnException(self._record_traceback_from_test)

        # Only check for thread leaks on bzrlib derived test cases

        if isinstance(test, TestCase):

            test.addCleanup(self._check_leaked_threads, test)

    def startTests(self):

        self.report_tests_starting()

        self._active_threads = threading.enumerate()

    def stopTest(self, test):

        self._traceback_from_test = None

    def _check_leaked_threads(self, test):

        """See if any threads have leaked since last call

        A sample of live threads is stored in the _active_threads attribute,

        when this method runs it compares the current live threads and any not

        in the previous sample are treated as having leaked.

        """

        now_active_threads = set(threading.enumerate())

        threads_leaked = now_active_threads.difference(self._active_threads)

        if threads_leaked:

            self._report_thread_leak(test, threads_leaked, now_active_threads)

            self._tests_leaking_threads_count += 1

            if self._first_thread_leaker_id is None:

                self._first_thread_leaker_id = test.id()

            self._active_threads = now_active_threads

    def _recordTestStartTime(self):

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

        self._start_datetime = self._now()

    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(self._traceback_from_test)

        super(ExtendedTestResult, self).addError(test, err)

        self.error_count += 1

        self.report_error(test, err)

        if self.stop_early:

            self.stop()

    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(self._traceback_from_test)

        super(ExtendedTestResult, self).addFailure(test, err)

        self.failure_count += 1

        self.report_failure(test, err)

        if self.stop_early:

            self.stop()

    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)

        super(ExtendedTestResult, self).addSuccess(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, tb=None):

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

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

            import pdb

            pdb.post_mortem(tb)

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

        """Display information before the test run begins"""

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

        """Display information on the test just about to be run"""

    def _report_thread_leak(self, test, leaked_threads, active_threads):

        """Display information on a test that leaked one or more threads"""

        # GZ 2010-09-09: A leak summary reported separately from the general

        #                thread debugging would be nice. Tests under subunit

        #                need something not using stream, perhaps adding a

        #                testtools details object would be fitting.

        if 'threads' in selftest_debug_flags:

            self.stream.write('%s is leaking, active is now %d\n' %

                (test.id(), len(active_threads)))

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

        super(TextTestResult, self).report_tests_starting()

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

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

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

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

        super(VerboseTestResult, self).report_tests_starting()

    def report_test_start(self, test):

        name = self._shortened_test_description(test)

        width = osutils.terminal_width()

        if width is not None: