~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2005, 2006, 2007 Canonical Ltd

from pprint import pformat

import random

    ui,

    workingtree,

import bzrlib.timestamp

# 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

 

                      bzrlib.timestamp,

    import bzrlib.tests.per_lock

            bzrlib.tests.per_lock,

                 use_numbered_dirs=False,

        self.ui = ui.ui_factory

        self.known_failure_count = 0

        self.unsupported = {}

        self.use_numbered_dirs = use_numbered_dirs

        test.number = self.count

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

 

        self.extractBenchmarkTime(test)

        self._cleanupLogFile(test)

            return self.addSkipped(test, err)

        elif isinstance(err[1], UnavailableFeature):

            return self.addNotSupported(test, err[1].args[0])

        self.error_count += 1

        self._cleanupLogFile(test)

        self.extractBenchmarkTime(test)

        if isinstance(err[1], KnownFailure):

            return self.addKnownFailure(test, err)

        self.failure_count += 1

    def addKnownFailure(self, test, err):

        self.known_failure_count += 1

        self.report_known_failure(test, err)

 

    def addNotSupported(self, test, feature):

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

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

        self.report_unsupported(test, feature)

 

            self.stream.write("%s: " % flavour)

            if self.use_numbered_dirs:

                self.stream.write('#%d ' % test.number)

            self.stream.writeln(self.getDescription(test))

    def __init__(self, stream, descriptions, verbosity,

                 bench_history=None,

                 num_tests=None,

                 pb=None,

                 use_numbered_dirs=False,

                 ):

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

            bench_history, num_tests, use_numbered_dirs)

        if pb is None:

            self.pb = self.ui.nested_progress_bar()

            self._supplied_pb = False

        else:

            self.pb = pb

            self._supplied_pb = True

        self.pb.show_eta = False,

        if self.known_failure_count:

            a += ', %d known failures' % self.known_failure_count

        if self.unsupported:

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

    def _test_description(self, test):

        if self.use_numbered_dirs:

            return '#%d %s' % (self.count,

                               self._shortened_test_description(test))

        else:

            return self._shortened_test_description(test)

 

            self._test_description(test),

            self._test_description(test),

    def report_known_failure(self, test, err):

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

            self._test_description(test), err[1])

 

    def report_unsupported(self, test, feature):

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

                  

        if not self._supplied_pb:

            self.pb.finished()

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

        if self.use_numbered_dirs:

            self.stream.write('%5d ' % self.count)

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

                                osutils.terminal_width()-36))

        else:

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

                                osutils.terminal_width()-30))

    def _error_summary(self, err):

        indent = ' ' * 4

        if self.use_numbered_dirs:

            indent += ' ' * 6

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

 

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

                % (self._testTimeString(),

                   self._error_summary(err)))

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

                % (self._testTimeString(),

                   self._error_summary(err)))

 

    def report_known_failure(self, test, err):

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

                % (self._testTimeString(),

                   self._error_summary(err)))

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

        # used to show the output in PQM.

        self.skip_count += 1

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

                % (self._testTimeString(),

                   self._error_summary(skip_excinfo)))

 

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

                  

                 bench_history=None,

                 use_numbered_dirs=False,

                 list_only=False

                 ):

        self.use_numbered_dirs = use_numbered_dirs

        self.list_only = list_only

                              use_numbered_dirs=self.use_numbered_dirs,

        if self.list_only:

            if self.verbosity >= 2:

                self.stream.writeln("Listing tests only ...\n")

            run = 0

            for t in iter_suite_tests(test):

                self.stream.writeln("%s" % (t.id()))

                run += 1

            actionTaken = "Listed"

        else: 

            test.run(result)

            run = result.testsRun

            actionTaken = "Ran"

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

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

            if result.known_failure_count:

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

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

                    result.known_failure_count)

            if result.known_failure_count:

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

                    result.known_failure_count)

            else:

                self.stream.writeln("OK")

        if result.skip_count > 0:

            skipped = result.skip_count

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

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

        if result.unsupported:

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

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

                    (feature, count))

                _rmtree_temp_dir(test_root)

class KnownFailure(AssertionError):

    """Indicates that a test failed in a precisely expected manner.

 

    Such failures dont block the whole test suite from passing because they are

    indicators of partially completed code or of future work. We have an

    explicit error for them so that we can ensure that they are always visible:

    KnownFailures are always shown in the output of bzr selftest.

    """

 

 

class UnavailableFeature(Exception):

    """A feature required for this test was not available.

 

    The feature should be used to construct the exception.

    """

 

 

class TestUIFactory(ui.CLIUIFactory):

    """A UI Factory for testing.

 

    Hide the progress bar but emit note()s.

    Redirect stdin.

    Allows get_password to be tested without real tty attached.

    """

 

    def __init__(self,

                 stdout=None,

                 stderr=None,

                 stdin=None):

        super(TestUIFactory, self).__init__()

        if stdin is not None:

            # We use a StringIOWrapper to be able to test various

            # encodings, but the user is still responsible to

            # encode the string and to set the encoding attribute

            # of StringIOWrapper.

            self.stdin = StringIOWrapper(stdin)

        if stdout is None:

            self.stdout = sys.stdout

        else:

            self.stdout = stdout

        if stderr is None:

            self.stderr = sys.stderr

        else:

            self.stderr = stderr

 

    def clear(self):

        """See progress.ProgressBar.clear()."""

 

    def clear_term(self):

        """See progress.ProgressBar.clear_term()."""

 

    def clear_term(self):

        """See progress.ProgressBar.clear_term()."""

 

    def finished(self):

        """See progress.ProgressBar.finished()."""

 

    def note(self, fmt_string, *args, **kwargs):

        """See progress.ProgressBar.note()."""

        self.stdout.write((fmt_string + "\n") % args)

 

    def progress_bar(self):

        return self

 

    def nested_progress_bar(self):

        return self

 

    def update(self, message, count=None, total=None):

        """See progress.ProgressBar.update()."""

 

    def get_non_echoed_password(self, prompt):

        """Get password from stdin without trying to handle the echo mode"""

        if prompt:

            self.stdout.write(prompt)

        password = self.stdin.readline()

        if not password:

            raise EOFError

        if password[-1] == '\n':

            password = password[:-1]

        return password

 

 

        self._clear_hooks()

 

    def _clear_hooks(self):

        # prevent hooks affecting tests

        import bzrlib.branch

        import bzrlib.smart.server

        self._preserved_hooks = {

            bzrlib.branch.Branch: bzrlib.branch.Branch.hooks,

            bzrlib.smart.server.SmartTCPServer: bzrlib.smart.server.SmartTCPServer.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()

        bzrlib.smart.server.SmartTCPServer.hooks = \

            bzrlib.smart.server.SmartServerHooks()

        saved = ui.ui_factory

            ui.ui_factory = saved

        ui.ui_factory = ui.SilentUIFactory()

    def assertEqual(self, a, b, message=''):

        try:

            if a == b:

                return

        except UnicodeError, e:

            # If we can't compare without getting a UnicodeError, then

            # obviously they are different

            mutter('UnicodeError: %s', e)

        if message:

            message += '\n'

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

            % (message,

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

 

    assertEquals = assertEqual

 

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

    def assertRaises(self, excClass, func, *args, **kwargs):

        """Assert that a callable raises a particular exception.

 

        :param excClass: As for the except statement, this may be either an

        exception class, or a tuple of classes.

 

        Returns the exception so that you can examine it.

        """

        try:

            func(*args, **kwargs)

        except excClass, e:

            return e

        else:

            if getattr(excClass,'__name__', None) is not None:

                excName = excClass.__name__

            else:

                # probably a tuple

                excName = str(excClass)

            raise self.failureException, "%s not raised" % excName

 

    def assertIs(self, left, right, message=None):

            if message is not None:

                raise AssertionError(message)

            else:

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

 

    def assertIsNot(self, left, right, message=None):

        if (left is right):

            if message is not None:

                raise AssertionError(message)

            else:

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

    def expectFailure(self, reason, assertion, *args, **kwargs):

        """Invoke a test, expecting it to fail for the given reason.

 

        This is for assertions that ought to succeed, but currently fail.

        (The failure is *expected* but not *wanted*.)  Please be very precise

        about the failure you're expecting.  If a new bug is introduced,

        AssertionError should be raised, not KnownFailure.

 

        Frequently, expectFailure should be followed by an opposite assertion.

        See example below.

 

        Intended to be used with a callable that raises AssertionError as the

        'assertion' parameter.  args and kwargs are passed to the 'assertion'.

 

        Raises KnownFailure if the test fails.  Raises AssertionError if the

        test succeeds.

 

        example usage::

 

          self.expectFailure('Math is broken', self.assertNotEqual, 54,

                             dynamic_val)

          self.assertEqual(42, dynamic_val)

 

          This means that a dynamic_val of 54 will cause the test to raise

          a KnownFailure.  Once math is fixed and the expectFailure is removed,

          only a dynamic_val of 42 will allow the test to pass.  Anything other

          than 54 or 42 will cause an AssertionError.

        """

        try:

            assertion(*args, **kwargs)

        except AssertionError:

            raise KnownFailure(reason)

        else:

            self.fail('Unexpected success.  Should have failed: %s' % reason)

 

            self.fail("No deprecation warning generated by call to %s" %

            'APPDATA': None,  # bzr now use Win32 API and don't rely on APPDATA

    def _restoreHooks(self):

        for klass, hooks in self._preserved_hooks.items():

            setattr(klass, 'hooks', hooks)

 

    def knownFailure(self, reason):

        """This test has failed for some known reason."""

        raise KnownFailure(reason)

 

    def run(self, result=None):

        if result is None: result = self.defaultTestResult()

        for feature in getattr(self, '_test_needs_features', []):

            if not feature.available():

                result.startTest(self)

                if getattr(result, 'addNotSupported', None):

                    result.addNotSupported(self, feature)

                else:

                    result.addSuccess(self)

                result.stopTest(self)

                return

        return unittest.TestCase.run(self, result)

 

 

        # Actually pop the cleanups from the list so tearDown running

        # twice is safe (this happens for skipped tests).

        while self._cleanups:

            self._cleanups.pop()()

                try:

                    os.remove(self._log_file_name)

                except OSError, e:

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

                        print >>sys.stderr, ('Unable to delete log file '

                                             ' %r' % self._log_file_name)

                    else:

                        raise

    def requireFeature(self, feature):

        """This test requires a specific feature is available.

 

        :raises UnavailableFeature: When feature is not available.

        """

        if not feature.available():

            raise UnavailableFeature(feature)

 

        old_ui_factory = ui.ui_factory

        ui.ui_factory = TestUIFactory(stdin=stdin, stdout=stdout, stderr=stderr)

                result = self.apply_redirected(ui.ui_factory.stdin,

                                               stdout, stderr,

            ui.ui_factory = old_ui_factory

            self.assertEquals(retcode, result,

                              message='Unexpected return code')

        :param retcode: The status code the command should return

        :param working_dir: The directory to run the command in

        error_regexes = kwargs.pop('error_regexes', [])

 

        out, err = self.run_bzr_captured(args, retcode=retcode,

            encoding=encoding, stdin=stdin, working_dir=working_dir)

 

        for regex in error_regexes:

            self.assertContainsRe(err, regex)

        return out, err

 

        out, err = self.run_bzr(error_regexes=error_regexes, *args, **kwargs)

    def reduceLockdirTimeout(self):

        """Reduce the default lock timeout for the duration of the test, so that

        if LockContention occurs during a test, it does so quickly.

 

        Tests that expect to provoke LockContention errors should call this.

        """

        orig_timeout = bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS

        def resetTimeout():

            bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = orig_timeout

        self.addCleanup(resetTimeout)

        bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = 0

        self.vfs_transport_factory = default_transport

        self.transport_server = None

        self.__vfs_server = None

 

                self.__readonly_server.setUp(self.get_vfs_only_server())

                self.__readonly_server.setUp(self.get_vfs_only_server())

    def get_vfs_only_server(self):

        """Get the vfs only read/write server instance.

        if self.__vfs_server is None:

            self.__vfs_server = MemoryServer()

            self.__vfs_server.setUp()

            self.addCleanup(self.__vfs_server.tearDown)

        return self.__vfs_server

 

    def get_server(self):

        """Get the read/write server instance.

 

        This is useful for some tests with specific servers that need

        diagnostics.

 

        This is built from the self.transport_server factory. If that is None,

        then the self.get_vfs_server is returned.

        """

            if self.transport_server is None or self.transport_server is self.vfs_transport_factory:

                return self.get_vfs_only_server()

            else:

                # bring up a decorated means of access to the vfs only server.

                self.__server = self.transport_server()

                try:

                    self.__server.setUp(self.get_vfs_only_server())

                except TypeError, e:

                    # This should never happen; the try:Except here is to assist

                    # developers having to update code rather than seeing an

                    # uninformative TypeError.

                    raise Exception, "Old server API in use: %s, %s" % (self.__server, e)

    def _adjust_url(self, base, relpath):

    def get_url(self, relpath=None):

        """Get a URL (or maybe a path) for the readwrite transport.

 

        This will either be backed by '.' or to an equivalent non-file based

        facility.

        relpath provides for clients to get a path relative to the base url.

        These should only be downwards relative, not upwards.

        """

        base = self.get_server().get_url()

        return self._adjust_url(base, relpath)

 

    def get_vfs_only_url(self, relpath=None):

        """Get a URL (or maybe a path for the plain old vfs transport.

 

        This will never be a smart protocol.

        :param relpath: provides for clients to get a path relative to the base

            url.  These should only be downwards relative, not upwards.

        """

        base = self.get_vfs_only_server().get_url()

        return self._adjust_url(base, relpath)

 

                format = 'default'

            if isinstance(format, basestring):

                format = bzrdir.format_registry.make_bzrdir(format)

        os.environ['BZR_HOME'] = self.test_home_dir

        self.reduceLockdirTimeout()

    use_numbered_dirs = False

        if self.use_numbered_dirs:  # strongly recommended on Windows

                                    # due the path length limitation (260 ch.)

            candidate_dir = '%s/%dK/%05d' % (self.TEST_ROOT,

                                             int(self.number/1000),

                                             self.number)

            os.makedirs(candidate_dir)

            self.test_home_dir = candidate_dir + '/home'

            os.mkdir(self.test_home_dir)

            self.test_dir = candidate_dir + '/work'

            os.mkdir(self.test_dir)

            os.chdir(self.test_dir)

            # put name of test inside

            f = file(candidate_dir + '/name', 'w')

            f.write(self.id())

            f.close()

            return

        # Else NAMED DIRS

                    raise errors.BzrError(

                        'Invalid line ending request %r' % line_endings)

        self.failUnlessExists(path)

        f = file(path, 'r')

        try:

            s = f.read()

        finally:

            f.close()

        self.assertEqualDiff(content, s)

 

    def failUnlessExists(self, path):

        """Fail unless path or paths, which may be abs or relative, exist."""

        if not isinstance(path, basestring):

            for p in path:

                self.failUnlessExists(p)

        else:

            self.failUnless(osutils.lexists(path),path+" does not exist")

 

    def failIfExists(self, path):

        """Fail if path or paths, which may be abs or relative, exist."""

        if not isinstance(path, basestring):

            for p in path:

                self.failIfExists(p)

        else:

            self.failIf(osutils.lexists(path),path+" exists")

 

    def assertInWorkingTree(self,path,root_path='.',tree=None):

        """Assert whether path or paths are in the WorkingTree"""

        if tree is None:

            tree = workingtree.WorkingTree.open(root_path)

        if not isinstance(path, basestring):

            for p in path:

                self.assertInWorkingTree(p,tree=tree)

        else:

            self.assertIsNot(tree.path2id(path), None,

                path+' not in working tree.')

 

    def assertNotInWorkingTree(self,path,root_path='.',tree=None):

        """Assert whether path or paths are not in the WorkingTree"""

        if tree is None:

            tree = workingtree.WorkingTree.open(root_path)

        if not isinstance(path, basestring):

            for p in path:

                self.assertNotInWorkingTree(p,tree=tree)

        else:

            self.assertIs(tree.path2id(path), None, path+' in working tree.')

    def get_vfs_only_server(self):

        if self.__vfs_server is None:

            self.__vfs_server = self.vfs_transport_factory()

            self.__vfs_server.setUp()

            self.addCleanup(self.__vfs_server.tearDown)

        return self.__vfs_server

        the transport.  In that case if the vfs_transport_factory is

        LocalURLServer the working tree is created in the local

        directory backing the transport, and the returned tree's branch and

        repository will also be accessed locally. Otherwise a lightweight

        checkout is created and returned.

            # transport can't support them, then we keep the non-disk-backed

            # branch and create a local checkout.

            if self.vfs_transport_factory is LocalURLServer:

                # the branch is colocated on disk, we cannot create a checkout.

                # hopefully callers will expect this.

                local_controldir= bzrdir.BzrDir.open(self.get_vfs_only_url(relpath))

                return local_controldir.create_workingtree()

            else:

                return b.create_checkout(relpath, lightweight=True)

    def assertTreesEqual(self, left, right):

        """Check that left and right have the same content and properties."""

        # we use a tree delta to check for equality of the content, and we

        # manually check for equality of other things such as the parents list.

        self.assertEqual(left.get_parent_ids(), right.get_parent_ids())

        differences = left.changes_from(right)

        self.assertFalse(differences.has_changed(),

            "Trees %r and %r are different: %r" % (left, right, differences))

 

        self.__vfs_server = None

        if not self.vfs_transport_factory == MemoryServer:

def filter_suite_by_re(suite, pattern, exclude_pattern=None,

                       random_order=False):

    """Create a test suite by filtering another one.

    

    :param suite:           the source suite

    :param pattern:         pattern that names must match

    :param exclude_pattern: pattern that names must not match, if any

    :param random_order:    if True, tests in the new suite will be put in

                            random order

    :returns: the newly created suite

    """ 

    return sort_suite_by_re(suite, pattern, exclude_pattern,

        random_order, False)

 

 

def sort_suite_by_re(suite, pattern, exclude_pattern=None,

                     random_order=False, append_rest=True):

    """Create a test suite by sorting another one.

    

    :param suite:           the source suite

    :param pattern:         pattern that names must match in order to go

                            first in the new suite

    :param exclude_pattern: pattern that names must not match, if any

    :param random_order:    if True, tests in the new suite will be put in

                            random order

    :param append_rest:     if False, pattern is a strict filter and not

                            just an ordering directive

    :returns: the newly created suite

    """ 

    first = []

    second = []

    if exclude_pattern is not None:

        exclude_re = re.compile(exclude_pattern)

        test_id = test.id()

        if exclude_pattern is None or not exclude_re.search(test_id):

            if filter_re.search(test_id):

                first.append(test)

            elif append_rest:

                second.append(test)

    if random_order:

        random.shuffle(first)

        random.shuffle(second)

    return TestUtil.TestSuite(first + second)

              transport=None, lsprof_timed=None, bench_history=None,

              matching_tests_first=None,

              numbered_dirs=None,

              list_only=False,

              random_seed=None,

              exclude_pattern=None,

              ):

    use_numbered_dirs = bool(numbered_dirs)

 

    if numbered_dirs is not None:

        TestCaseInTempDir.use_numbered_dirs = use_numbered_dirs

                            bench_history=bench_history,

                            use_numbered_dirs=use_numbered_dirs,

                            list_only=list_only,

                            )

    # Initialise the random number generator and display the seed used.

    # We convert the seed to a long to make it reuseable across invocations.

    random_order = False

    if random_seed is not None:

        random_order = True

        if random_seed == "now":

            random_seed = long(time.time())

        else:

            # Convert the seed to a long if we can

            try:

                random_seed = long(random_seed)

            except:

                pass

        runner.stream.writeln("Randomizing test order using seed %s\n" %

            (random_seed))

        random.seed(random_seed)

    # Customise the list of tests if requested

    if pattern != '.*' or exclude_pattern is not None or random_order:

        if matching_tests_first:

            suite = sort_suite_by_re(suite, pattern, exclude_pattern,

                random_order)

        else:

            suite = filter_suite_by_re(suite, pattern, exclude_pattern,

                random_order)

             bench_history=None,

             matching_tests_first=None,

             numbered_dirs=None,

             list_only=False,

             random_seed=None,

             exclude_pattern=None):

                     bench_history=bench_history,

                     matching_tests_first=matching_tests_first,

                     numbered_dirs=numbered_dirs,

                     list_only=list_only,

                     random_seed=random_seed,