~bzr-pqm/bzr/bzr.dev

#

# 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

"""Tests for the test framework."""

 

from cStringIO import StringIO

import os

import signal

import sys

import unittest

import warnings

 

from testtools.content import Content

from testtools.content_type import ContentType

from testtools.matchers import (

    DocTestMatches,

    Equals,

    )

 

import bzrlib

from bzrlib import (

    branchbuilder,

    bzrdir,

    errors,

    lockdir,

    memorytree,

    osutils,

    tests,

    transport,

    workingtree,

    )

from bzrlib.repofmt import (

    groupcompress_repo,

    )

from bzrlib.symbol_versioning import (

    deprecated_function,

    features,

    test_lsprof,

    test_server,

    TestUtil,

    )

from bzrlib.trace import note, mutter

from bzrlib.transport import memory

 

 

def _test_ids(test_suite):

    return [t.id() for t in tests.iter_suite_tests(test_suite)]

 

 

class MetaTestLog(tests.TestCase):

 

    def test_logging(self):

            "text", "plain", {"charset": "utf8"})))

        self.assertThat(u"".join(log.iter_text()), Equals(self.get_log()))

        self.assertThat(self.get_log(),

 

 

class TestTreeShape(tests.TestCaseInTempDir):

 

    def test_unicode_paths(self):

 

        filename = u'hell\u00d8'

        self.build_tree_contents([(filename, 'contents of hello')])

 

 

class TestClassesAvailable(tests.TestCase):

        from bzrlib.tests.per_workingtree import make_scenarios

        server1 = "a"

        server2 = "b"

        scenarios = make_scenarios(server1, server2, formats)

        self.assertEqual([

             {'bzrdir_format': formats[0]._matchingbzrdir,

              'transport_readonly_server': 'b',

              'transport_server': 'a',

            )

        server1 = "a"

        server2 = "b"

        scenarios = make_scenarios(server1, server2, formats)

        self.assertEqual(7, len(scenarios))

        expected_scenarios = [

             {'bzrdir_format': formats[0]._matchingbzrdir,

              'transport_readonly_server': 'b',

              'transport_server': 'a',

        # ones to add.

        from bzrlib.tests.per_tree import (

            return_parameter,

            )

        from bzrlib.tests.per_intertree import (

            make_scenarios,

            )

        input_test = TestInterTreeScenarios(

            "test_scenarios")

        server1 = "a"

        server2 = "b"

        format2 = WorkingTreeFormat3()

        formats = [("1", str, format1, format2, "converter1"),

            ("2", int, format2, format1, "converter2")]

        self.assertRaises(AssertionError, self.assertEqualStat,

            os.lstat("foo"), os.lstat("longname"))

 

 

class TestTestCaseWithMemoryTransport(tests.TestCaseWithMemoryTransport):

 

        tree = self.make_branch_and_memory_tree('dir')

        # Guard against regression into MemoryTransport leaking

        # files to disk instead of keeping them in memory.

        self.assertIsInstance(tree, memorytree.MemoryTree)

 

    def test_make_branch_and_memory_tree_with_format(self):

        """make_branch_and_memory_tree should accept a format option."""

        format = bzrdir.BzrDirMetaFormat1()

        tree = self.make_branch_and_memory_tree('dir', format=format)

        # Guard against regression into MemoryTransport leaking

        # files to disk instead of keeping them in memory.

        self.assertIsInstance(tree, memorytree.MemoryTree)

        self.assertEqual(format.repository_format.__class__,

            tree.branch.repository._format.__class__)

        self.assertIsInstance(builder, branchbuilder.BranchBuilder)

        # Guard against regression into MemoryTransport leaking

        # files to disk instead of keeping them in memory.

 

    def test_make_branch_builder_with_format(self):

        # Use a repo layout that doesn't conform to a 'named' layout, to ensure

        # that the format objects are used.

        format = bzrdir.BzrDirMetaFormat1()

        format.repository_format = repo_format

        builder = self.make_branch_builder('dir', format=format)

        the_branch = builder.get_branch()

        # Guard against regression into MemoryTransport leaking

        # files to disk instead of keeping them in memory.

        self.assertEqual(format.repository_format.__class__,

                         the_branch.repository._format.__class__)

        self.assertEqual(repo_format.get_format_string(),

        the_branch = builder.get_branch()

        # Guard against regression into MemoryTransport leaking

        # files to disk instead of keeping them in memory.

        dir_format = bzrdir.format_registry.make_bzrdir('knit')

        self.assertEqual(dir_format.repository_format.__class__,

                         the_branch.repository._format.__class__)

    def test_dangling_locks_cause_failures(self):

        class TestDanglingLock(tests.TestCaseWithMemoryTransport):

            def test_function(self):

                l = lockdir.LockDir(t, 'lock')

                l.create()

                l.attempt_lock()

        # for the server

        url = self.get_readonly_url()

        url2 = self.get_readonly_url('foo/bar')

        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))

 

    def test_get_readonly_url_http(self):

        url = self.get_readonly_url()

        url2 = self.get_readonly_url('foo/bar')

        # the transport returned may be any HttpTransportBase subclass

        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))

 

    def test_is_directory(self):

    def test_make_branch_builder(self):

        builder = self.make_branch_builder('dir')

        rev_id = builder.build_commit()

        a_dir = bzrdir.BzrDir.open('dir')

        self.assertRaises(errors.NoWorkingTree, a_dir.open_workingtree)

        a_branch = a_dir.open_branch()

        self.assertIsInstance(result_bzrdir.transport,

                              memory.MemoryTransport)

        # should not be on disk, should only be in memory

 

 

class TestChrootedTest(tests.ChrootedTestCase):

 

    def test_root_is_root(self):

        url = t.base

        self.assertEqual(url, t.clone('..').base)

 

class TestProfileResult(tests.TestCase):

 

    def test_profiles_tests(self):

        result = tests.ProfileResult(terminal)

        class Sample(tests.TestCase):

            def a(self):

                descriptions=0,

                verbosity=1,

                )

        test_case.run(MultiTestResult(result, capture))

        run_case = capture._events[0][1]

        timed_string = result._testTimeString(run_case)

        self.check_timing(ShortDelayTestCase('test_short_delay'),

                          r"^ +[0-9]+ms$")

 

    def _time_hello_world_encoding(self):

        """Profile two sleep calls

 

 

    def test_lsprofiling(self):

        """Verbose test result prints lsprof statistics from test cases."""

        result_stream = StringIO()

        result = bzrlib.tests.VerboseTestResult(

            result_stream,

        self.assertContainsRe(output,

            r"LSProf output for <type 'unicode'>\(\('world',\), {'errors': 'replace'}\)\n")

 

    def test_known_failure(self):

        class InstrumentedTestResult(tests.ExtendedTestResult):

            def stopTestRun(self): pass

            def report_tests_starting(self): pass

        result = InstrumentedTestResult(None, None, None, None)

        class Test(tests.TestCase):

            def test_function(self):

        test = Test("test_function")

        test.run(result)

        # it should invoke 'report_known_failure'.

            descriptions=0,

            verbosity=2,

            )

 

    def get_passing_test(self):

        """Return a test object that can't be run usefully."""

                self._call = test, feature

        result = InstrumentedTestResult(None, None, None, None)

        test = SampleTestCase('_test_pass')

        result.startTest(test)

        result.addNotSupported(test, feature)

        # it should invoke 'report_unsupported'.

            verbosity=2,

            )

        test = self.get_passing_test()

        result.startTest(test)

        prefix = len(result_stream.getvalue())

        result.report_unsupported(test, feature)

            def addNotSupported(self, test, feature):

                self._call = test, feature

        result = InstrumentedTestResult(None, None, None, None)

        class Test(tests.TestCase):

            def test_function(self):

                raise tests.UnavailableFeature(feature)

    def test_strict_with_known_failure(self):

        result = bzrlib.tests.TextTestResult(self._log_file, descriptions=0,

                                             verbosity=1)

        self.assertFalse(result.wasStrictlySuccessful())

        self.assertEqual(None, result._extractBenchmarkTime(test))

 

        self.assertEquals(2, result.count)

 

 

class TestRunner(tests.TestCase):

 

    def dummy_test(self):

        test = unittest.TestSuite()

        test.addTest(Test("known_failure_test"))

        def failing_test():

        test.addTest(unittest.FunctionTestCase(failing_test))

        stream = StringIO()

        runner = tests.TextTestRunner(stream=stream)

            '^----------------------------------------------------------------------\n'

            'Traceback \\(most recent call last\\):\n'

            '  .*' # File .*, line .*, in failing_test' - but maybe not from .pyc

            '.*'

            '^----------------------------------------------------------------------\n'

            '.*'

        # the final output.

        class Test(tests.TestCase):

            def known_failure_test(self):

        test = Test("known_failure_test")

        stream = StringIO()

        runner = tests.TextTestRunner(stream=stream)

            '\n'

            'OK \\(known_failures=1\\)\n')

 

    def test_result_decorator(self):

        # decorate results

        calls = []

            def startTest(self, test):

                calls.append('start')

        test = unittest.FunctionTestCase(lambda:None)

        stream = StringIO()

 

    def test_unsupported_features_listed(self):

        """When unsupported features are encountered they are detailed."""

            def _probe(self): return False

            def _probe(self): return False

        # create sample tests

        test1 = SampleTestCase('_test_pass')

            ],

            lines[-3:])

 

    def test_verbose_test_count(self):

        """A verbose test run reports the right test count at the start"""

        suite = TestUtil.TestSuite([

    def test_startTestRun(self):

        """run should call result.startTestRun()"""

        calls = []

            def startTestRun(self):

                calls.append('startTestRun')

        test = unittest.FunctionTestCase(lambda:None)

        stream = StringIO()

    def test_stopTestRun(self):

        """run should call result.stopTestRun()"""

        calls = []

            def stopTestRun(self):

                calls.append('stopTestRun')

        test = unittest.FunctionTestCase(lambda:None)

        stream = StringIO()

        result = self.run_test_runner(runner, test)

        self.assertLength(1, calls)

 

 

class SampleTestCase(tests.TestCase):

 

        # Note this test won't fail with hooks that the core library doesn't

        # use - but it trigger with a plugin that adds hooks, so its still a

        # useful warning in that case.

            bzrlib.smart.server.SmartTCPServer.hooks)

 

    def test__gather_lsprof_in_benchmarks(self):

        """When _gather_lsprof_in_benchmarks is on, accumulate profile data.

 

        Each self.time() call is individually and separately profiled.

        """

        # overrides the class member with an instance member so no cleanup

        # needed.

        self._gather_lsprof_in_benchmarks = True

        transport_server = memory.MemoryServer()

        transport_server.start_server()

        self.addCleanup(transport_server.stop_server)

        bzrdir.BzrDir.create(t.base)

        self.assertRaises(errors.BzrError,

            bzrdir.BzrDir.open_from_transport, t)

 

    def test_requireFeature_available(self):

        """self.requireFeature(available) is a no-op."""

            def _probe(self):return True

        feature = Available()

        self.requireFeature(feature)

 

    def test_requireFeature_unavailable(self):

        """self.requireFeature(unavailable) raises UnavailableFeature."""

            def _probe(self):return False

        feature = Unavailable()

        self.assertRaises(tests.UnavailableFeature,

        test.run(unittest.TestResult())

        self.assertEqual('original', obj.test_attr)

 

    def _probe(self):

        return False

missing_feature = _MissingFeature()

        result = self._run_test('test_fail')

        self.assertEqual(1, len(result.failures))

        result_content = result.failures[0][1]

        self.assertContainsRe(result_content, 'this was a failing test')

 

    def test_error_has_log(self):

        result = self._run_test('test_error')

        self.assertEqual(1, len(result.errors))

        result_content = result.errors[0][1]

        self.assertContainsRe(result_content, 'this test errored')

 

    def test_skip_has_no_log(self):

        result = self._run_test('test_xfail')

        self.assertEqual(1, len(result.expectedFailures))

        result_content = result.expectedFailures[0][1]

        self.assertNotContainsRe(result_content, 'test with expected failure')

 

    def test_unexpected_success_has_log(self):

    def test_make_branch_and_tree_with_format(self):

        # we should be able to supply a format to make_branch_and_tree

        self.make_branch_and_tree('a', format=bzrlib.bzrdir.BzrDirMetaFormat1())

        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('a')._format,

                              bzrlib.bzrdir.BzrDirMetaFormat1)

 

    def test_make_branch_and_memory_tree(self):

        # we should be able to get a new branch and a mutable tree from

        self.assertLength(2, output.readlines())

 

    def test_lsprof_tests(self):

        class Test(object):

            def __call__(test, result):

                test.run(result)

            def run(test, result):

            def countTestCases(self):

                return 1

        self.run_selftest(test_suite_factory=Test, lsprof_tests=True)

 

    def test_random(self):

        # test randomising by listing a number of tests.

        self.assertNotContainsRe(content, 'test with expected failure')

        self.assertEqual(1, len(result.expectedFailures))

        result_content = result.expectedFailures[0][1]

        self.assertNotContainsRe(result_content, 'test with expected failure')

 

    def test_unexpected_success_has_log(self):

        content, result = self.run_subunit_stream('test_unexpected_success')

        self.assertContainsRe(content, '(?m)^log$')

        self.assertContainsRe(content, 'test with unexpected success')

        self.assertEqual(1, len(result.unexpectedSuccesses))

        test = result.unexpectedSuccesses[0]

        # RemotedTestCase doesn't preserve the "details"

        # stdout and stderr of the invoked run_bzr

        current_factory = bzrlib.ui.ui_factory

        self.run_bzr(['foo'])

        self.assertNotEqual(sys.stdout, self.factory.stdout)

        self.assertNotEqual(sys.stderr, self.factory.stderr)

        self.assertEqual('foo\n', self.factory.stdout.getvalue())

 

 

class TestStartBzrSubProcess(tests.TestCase):

 

 

    def _popen(self, *args, **kwargs):

        self.check_popen_state()

        self._popen_args = args

        self._popen_kwargs = kwargs

        raise _DontSpawnProcess()

 

    def test_run_bzr_subprocess_no_plugins(self):

        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [])

        command = self._popen_args[0]

 

    def test_allow_plugins(self):

        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],

        command = self._popen_args[0]

        self.assertEqual([], command[2:])

 

    def test_set_env(self):

        # set in the child

        def check_environment():

            self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])

        self.check_popen_state = check_environment

        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],

        # not set in theparent

        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)

 

    def test_run_bzr_subprocess_env_del(self):

        """run_bzr_subprocess can remove environment variables too."""

        def check_environment():

            self.assertFalse('EXISTANT_ENV_VAR' in os.environ)

        os.environ['EXISTANT_ENV_VAR'] = 'set variable'

        self.check_popen_state = check_environment

        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],

        # Still set in parent

        self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])

        del os.environ['EXISTANT_ENV_VAR']

 

    def test_env_del_missing(self):

        def check_environment():

            self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)

        self.check_popen_state = check_environment

        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],

 

    def test_working_dir(self):

        """Test that we can specify the working dir for the child"""

        chdirs = []

        def chdir(path):

            chdirs.append(path)

        self.assertEqual(['foo', 'current'], chdirs)

 

    def test_get_bzr_path_with_cwd_bzrlib(self):

        self.assertEqual('bzr: interrupted\n', result[1])