/brz/remove-bazaar

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

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

from cStringIO import StringIO

import os

import signal

import sys

import time

import unittest

import warnings

import bzrlib

from bzrlib import (

    branchbuilder,

    bzrdir,

    debug,

    errors,

    lockdir,

    memorytree,

    osutils,

    progress,

    remote,

    repository,

    symbol_versioning,

    tests,

    transport,

    workingtree,

    )

from bzrlib.repofmt import (

    groupcompress_repo,

    pack_repo,

    weaverepo,

    )

from bzrlib.symbol_versioning import (

    deprecated_function,

    deprecated_in,

    deprecated_method,

    )

from bzrlib.tests import (

    SubUnitFeature,

    test_lsprof,

    test_sftp_transport,

    TestUtil,

    )

from bzrlib.trace import note

from bzrlib.transport.memory import MemoryServer, MemoryTransport

from bzrlib.version import _get_bzr_source_tree

def _test_ids(test_suite):

    """Get the ids for the tests in a test suite."""

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

class SelftestTests(tests.TestCase):

    def test_import_tests(self):

        mod = TestUtil._load_module_by_name('bzrlib.tests.test_selftest')

        self.assertEqual(mod.SelftestTests, SelftestTests)

    def test_import_test_failure(self):

        self.assertRaises(ImportError,

                          TestUtil._load_module_by_name,

                          'bzrlib.no-name-yet')

class MetaTestLog(tests.TestCase):

    def test_logging(self):

        """Test logs are captured when a test fails."""

        self.log('a test message')

        self._log_file.flush()

        self.assertContainsRe(self._get_log(keep_log_file=True),

                              'a test message\n')

class TestUnicodeFilename(tests.TestCase):

    def test_probe_passes(self):

        """UnicodeFilename._probe passes."""

        # We can't test much more than that because the behaviour depends

        # on the platform.

        tests.UnicodeFilename._probe()

class TestTreeShape(tests.TestCaseInTempDir):

    def test_unicode_paths(self):

        self.requireFeature(tests.UnicodeFilename)

        filename = u'hell\u00d8'

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

        self.failUnlessExists(filename)

class TestTransportScenarios(tests.TestCase):

    """A group of tests that test the transport implementation adaption core.

    This is a meta test that the tests are applied to all available

    transports.

    This will be generalised in the future which is why it is in this

    test file even though it is specific to transport tests at the moment.

    """

    def test_get_transport_permutations(self):

        # this checks that get_test_permutations defined by the module is

        # called by the get_transport_test_permutations function.

        class MockModule(object):

            def get_test_permutations(self):

                return sample_permutation

        sample_permutation = [(1,2), (3,4)]

        from bzrlib.tests.per_transport import get_transport_test_permutations

        self.assertEqual(sample_permutation,

                         get_transport_test_permutations(MockModule()))

    def test_scenarios_include_all_modules(self):

        # this checks that the scenario generator returns as many permutations

        # as there are in all the registered transport modules - we assume if

        # this matches its probably doing the right thing especially in

        # combination with the tests for setting the right classes below.

        from bzrlib.tests.per_transport import transport_test_permutations

        from bzrlib.transport import _get_transport_modules

        modules = _get_transport_modules()

        permutation_count = 0

        for module in modules:

            try:

                permutation_count += len(reduce(getattr,

                    (module + ".get_test_permutations").split('.')[1:],

                     __import__(module))())

            except errors.DependencyNotPresent:

                pass

        scenarios = transport_test_permutations()

        self.assertEqual(permutation_count, len(scenarios))

    def test_scenarios_include_transport_class(self):

        # This test used to know about all the possible transports and the

        # order they were returned but that seems overly brittle (mbp

        # 20060307)

        from bzrlib.tests.per_transport import transport_test_permutations

        scenarios = transport_test_permutations()

        # there are at least that many builtin transports

        self.assertTrue(len(scenarios) > 6)

        one_scenario = scenarios[0]

        self.assertIsInstance(one_scenario[0], str)

        self.assertTrue(issubclass(one_scenario[1]["transport_class"],

                                   bzrlib.transport.Transport))

        self.assertTrue(issubclass(one_scenario[1]["transport_server"],

                                   bzrlib.transport.Server))

class TestBranchScenarios(tests.TestCase):

    def test_scenarios(self):

        # check that constructor parameters are passed through to the adapted

        # test.

        from bzrlib.tests.per_branch import make_scenarios

        server1 = "a"

        server2 = "b"

        formats = [("c", "C"), ("d", "D")]

        scenarios = make_scenarios(server1, server2, formats)

        self.assertEqual(2, len(scenarios))

        self.assertEqual([

            ('str',

             {'branch_format': 'c',

              'bzrdir_format': 'C',

              'transport_readonly_server': 'b',

              'transport_server': 'a'}),

            ('str',

             {'branch_format': 'd',

              'bzrdir_format': 'D',

              'transport_readonly_server': 'b',

              'transport_server': 'a'})],

            scenarios)

class TestBzrDirScenarios(tests.TestCase):

    def test_scenarios(self):

        # check that constructor parameters are passed through to the adapted

        # test.

        from bzrlib.tests.per_bzrdir import make_scenarios

        vfs_factory = "v"

        server1 = "a"

        server2 = "b"

        formats = ["c", "d"]

        scenarios = make_scenarios(vfs_factory, server1, server2, formats)

        self.assertEqual([

            ('str',

             {'bzrdir_format': 'c',

              'transport_readonly_server': 'b',

              'transport_server': 'a',

              'vfs_transport_factory': 'v'}),

            ('str',

             {'bzrdir_format': 'd',

              'transport_readonly_server': 'b',

              'transport_server': 'a',

              'vfs_transport_factory': 'v'})],

            scenarios)

class TestRepositoryScenarios(tests.TestCase):

    def test_formats_to_scenarios(self):

        from bzrlib.tests.per_repository import formats_to_scenarios

        formats = [("(c)", remote.RemoteRepositoryFormat()),

                   ("(d)", repository.format_registry.get(

                    'Bazaar repository format 2a (needs bzr 1.16 or later)\n'))]

        no_vfs_scenarios = formats_to_scenarios(formats, "server", "readonly",

            None)

        vfs_scenarios = formats_to_scenarios(formats, "server", "readonly",

            vfs_transport_factory="vfs")

        # no_vfs generate scenarios without vfs_transport_factory

        expected = [

            ('RemoteRepositoryFormat(c)',

             {'bzrdir_format': remote.RemoteBzrDirFormat(),

              'repository_format': remote.RemoteRepositoryFormat(),

              'transport_readonly_server': 'readonly',

              'transport_server': 'server'}),

            ('RepositoryFormat2a(d)',

             {'bzrdir_format': bzrdir.BzrDirMetaFormat1(),

              'repository_format': groupcompress_repo.RepositoryFormat2a(),

              'transport_readonly_server': 'readonly',

              'transport_server': 'server'})]

        self.assertEqual(expected, no_vfs_scenarios)

        self.assertEqual([

            ('RemoteRepositoryFormat(c)',

             {'bzrdir_format': remote.RemoteBzrDirFormat(),

              'repository_format': remote.RemoteRepositoryFormat(),

              'transport_readonly_server': 'readonly',

              'transport_server': 'server',

              'vfs_transport_factory': 'vfs'}),

            ('RepositoryFormat2a(d)',

             {'bzrdir_format': bzrdir.BzrDirMetaFormat1(),

              'repository_format': groupcompress_repo.RepositoryFormat2a(),

              'transport_readonly_server': 'readonly',

              'transport_server': 'server',

              'vfs_transport_factory': 'vfs'})],

            vfs_scenarios)

class TestTestScenarioApplication(tests.TestCase):

    """Tests for the test adaption facilities."""

    def test_apply_scenario(self):

        from bzrlib.tests import apply_scenario

        input_test = TestTestScenarioApplication("test_apply_scenario")

        # setup two adapted tests

        adapted_test1 = apply_scenario(input_test,

            ("new id",

            {"bzrdir_format":"bzr_format",

             "repository_format":"repo_fmt",

             "transport_server":"transport_server",

             "transport_readonly_server":"readonly-server"}))

        adapted_test2 = apply_scenario(input_test,

            ("new id 2", {"bzrdir_format":None}))

        # input_test should have been altered.

        self.assertRaises(AttributeError, getattr, input_test, "bzrdir_format")

        # the new tests are mutually incompatible, ensuring it has

        # made new ones, and unspecified elements in the scenario

        # should not have been altered.

        self.assertEqual("bzr_format", adapted_test1.bzrdir_format)

        self.assertEqual("repo_fmt", adapted_test1.repository_format)

        self.assertEqual("transport_server", adapted_test1.transport_server)

        self.assertEqual("readonly-server",

            adapted_test1.transport_readonly_server)

        self.assertEqual(

            "bzrlib.tests.test_selftest.TestTestScenarioApplication."

            "test_apply_scenario(new id)",

            adapted_test1.id())

        self.assertEqual(None, adapted_test2.bzrdir_format)

        self.assertEqual(

            "bzrlib.tests.test_selftest.TestTestScenarioApplication."

            "test_apply_scenario(new id 2)",

            adapted_test2.id())

class TestInterRepositoryScenarios(tests.TestCase):

    def test_scenarios(self):

        # check that constructor parameters are passed through to the adapted

        # test.

        from bzrlib.tests.per_interrepository import make_scenarios

        server1 = "a"

        server2 = "b"

        formats = [("C0", "C1", "C2"), ("D0", "D1", "D2")]

        scenarios = make_scenarios(server1, server2, formats)

        self.assertEqual([

            ('C0,str,str',

             {'repository_format': 'C1',

              'repository_format_to': 'C2',

              'transport_readonly_server': 'b',

              'transport_server': 'a'}),

            ('D0,str,str',

             {'repository_format': 'D1',

              'repository_format_to': 'D2',

              'transport_readonly_server': 'b',

              'transport_server': 'a'})],

            scenarios)

class TestWorkingTreeScenarios(tests.TestCase):

    def test_scenarios(self):

        # check that constructor parameters are passed through to the adapted

        # test.

        from bzrlib.tests.per_workingtree import make_scenarios

        server1 = "a"

        server2 = "b"

        formats = [workingtree.WorkingTreeFormat2(),

                   workingtree.WorkingTreeFormat3(),]

        scenarios = make_scenarios(server1, server2, formats)

        self.assertEqual([

            ('WorkingTreeFormat2',

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

              'transport_readonly_server': 'b',

              'transport_server': 'a',

              'workingtree_format': formats[0]}),

            ('WorkingTreeFormat3',

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

              'transport_readonly_server': 'b',

              'transport_server': 'a',

              'workingtree_format': formats[1]})],

            scenarios)

class TestTreeScenarios(tests.TestCase):

    def test_scenarios(self):

        # the tree implementation scenario generator is meant to setup one

        # instance for each working tree format, and one additional instance

        # that will use the default wt format, but create a revision tree for

        # the tests.  this means that the wt ones should have the

        # workingtree_to_test_tree attribute set to 'return_parameter' and the

        # revision one set to revision_tree_from_workingtree.

        from bzrlib.tests.per_tree import (

            _dirstate_tree_from_workingtree,

            make_scenarios,

            preview_tree_pre,

            preview_tree_post,

            return_parameter,

            revision_tree_from_workingtree

            )

        server1 = "a"

        server2 = "b"

        formats = [workingtree.WorkingTreeFormat2(),

                   workingtree.WorkingTreeFormat3(),]

        scenarios = make_scenarios(server1, server2, formats)

        self.assertEqual(7, len(scenarios))

        default_wt_format = workingtree.WorkingTreeFormat4._default_format

        wt4_format = workingtree.WorkingTreeFormat4()

        wt5_format = workingtree.WorkingTreeFormat5()

        expected_scenarios = [

            ('WorkingTreeFormat2',

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

              'transport_readonly_server': 'b',

              'transport_server': 'a',

              'workingtree_format': formats[0],

              '_workingtree_to_test_tree': return_parameter,

              }),

            ('WorkingTreeFormat3',

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

              'transport_readonly_server': 'b',

              'transport_server': 'a',

              'workingtree_format': formats[1],

              '_workingtree_to_test_tree': return_parameter,

             }),

            ('RevisionTree',

             {'_workingtree_to_test_tree': revision_tree_from_workingtree,

              'bzrdir_format': default_wt_format._matchingbzrdir,

              'transport_readonly_server': 'b',

              'transport_server': 'a',

              'workingtree_format': default_wt_format,

             }),

            ('DirStateRevisionTree,WT4',

             {'_workingtree_to_test_tree': _dirstate_tree_from_workingtree,

              'bzrdir_format': wt4_format._matchingbzrdir,

              'transport_readonly_server': 'b',

              'transport_server': 'a',

              'workingtree_format': wt4_format,

             }),

            ('DirStateRevisionTree,WT5',

             {'_workingtree_to_test_tree': _dirstate_tree_from_workingtree,

              'bzrdir_format': wt5_format._matchingbzrdir,

              'transport_readonly_server': 'b',

              'transport_server': 'a',

              'workingtree_format': wt5_format,

             }),

            ('PreviewTree',

             {'_workingtree_to_test_tree': preview_tree_pre,

              'bzrdir_format': default_wt_format._matchingbzrdir,

              'transport_readonly_server': 'b',

              'transport_server': 'a',

              'workingtree_format': default_wt_format}),

            ('PreviewTreePost',

             {'_workingtree_to_test_tree': preview_tree_post,

              'bzrdir_format': default_wt_format._matchingbzrdir,

              'transport_readonly_server': 'b',

              'transport_server': 'a',

              'workingtree_format': default_wt_format}),

             ]

        self.assertEqual(expected_scenarios, scenarios)

class TestInterTreeScenarios(tests.TestCase):

    """A group of tests that test the InterTreeTestAdapter."""

    def test_scenarios(self):

        # check that constructor parameters are passed through to the adapted

        # test.

        # for InterTree tests we want the machinery to bring up two trees in

        # each instance: the base one, and the one we are interacting with.

        # because each optimiser can be direction specific, we need to test

        # each optimiser in its chosen direction.

        # unlike the TestProviderAdapter we dont want to automatically add a

        # parameterized one for WorkingTree - the optimisers will tell us what

        # ones to add.

        from bzrlib.tests.per_tree import (

            return_parameter,

            revision_tree_from_workingtree

            )

        from bzrlib.tests.per_intertree import (

            make_scenarios,

            )

        from bzrlib.workingtree import WorkingTreeFormat2, WorkingTreeFormat3

        input_test = TestInterTreeScenarios(

            "test_scenarios")

        server1 = "a"

        server2 = "b"

        format1 = WorkingTreeFormat2()

        format2 = WorkingTreeFormat3()

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

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

        scenarios = make_scenarios(server1, server2, formats)

        self.assertEqual(2, len(scenarios))

        expected_scenarios = [

            ("1", {

                "bzrdir_format": format1._matchingbzrdir,

                "intertree_class": formats[0][1],

                "workingtree_format": formats[0][2],

                "workingtree_format_to": formats[0][3],

                "mutable_trees_to_test_trees": formats[0][4],

                "_workingtree_to_test_tree": return_parameter,

                "transport_server": server1,

                "transport_readonly_server": server2,

                }),

            ("2", {

                "bzrdir_format": format2._matchingbzrdir,

                "intertree_class": formats[1][1],

                "workingtree_format": formats[1][2],

                "workingtree_format_to": formats[1][3],

                "mutable_trees_to_test_trees": formats[1][4],

                "_workingtree_to_test_tree": return_parameter,

                "transport_server": server1,

                "transport_readonly_server": server2,

                }),

            ]

        self.assertEqual(scenarios, expected_scenarios)

class TestTestCaseInTempDir(tests.TestCaseInTempDir):

    def test_home_is_not_working(self):

        self.assertNotEqual(self.test_dir, self.test_home_dir)

        cwd = osutils.getcwd()

        self.assertIsSameRealPath(self.test_dir, cwd)

        self.assertIsSameRealPath(self.test_home_dir, os.environ['HOME'])

    def test_assertEqualStat_equal(self):

        from bzrlib.tests.test_dirstate import _FakeStat

        self.build_tree(["foo"])

        real = os.lstat("foo")

        fake = _FakeStat(real.st_size, real.st_mtime, real.st_ctime,

            real.st_dev, real.st_ino, real.st_mode)

        self.assertEqualStat(real, fake)

    def test_assertEqualStat_notequal(self):

        self.build_tree(["foo", "bar"])

        self.assertRaises(AssertionError, self.assertEqualStat,

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

class TestTestCaseWithMemoryTransport(tests.TestCaseWithMemoryTransport):

    def test_home_is_non_existant_dir_under_root(self):

        """The test_home_dir for TestCaseWithMemoryTransport is missing.

        This is because TestCaseWithMemoryTransport is for tests that do not

        need any disk resources: they should be hooked into bzrlib in such a

        way that no global settings are being changed by the test (only a

        few tests should need to do that), and having a missing dir as home is

        an effective way to ensure that this is the case.

        """

        self.assertIsSameRealPath(

            self.TEST_ROOT + "/MemoryTransportMissingHomeDir",

            self.test_home_dir)

        self.assertIsSameRealPath(self.test_home_dir, os.environ['HOME'])

    def test_cwd_is_TEST_ROOT(self):

        self.assertIsSameRealPath(self.test_dir, self.TEST_ROOT)

        cwd = osutils.getcwd()

        self.assertIsSameRealPath(self.test_dir, cwd)

    def test_make_branch_and_memory_tree(self):

        """In TestCaseWithMemoryTransport we should not make the branch on disk.

        This is hard to comprehensively robustly test, so we settle for making

        a branch and checking no directory was created at its relpath.

        """

        tree = self.make_branch_and_memory_tree('dir')

        # Guard against regression into MemoryTransport leaking

        # files to disk instead of keeping them in memory.

        self.failIf(osutils.lexists('dir'))

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

        format.repository_format = weaverepo.RepositoryFormat7()

        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.failIf(osutils.lexists('dir'))

        self.assertIsInstance(tree, memorytree.MemoryTree)

        self.assertEqual(format.repository_format.__class__,

            tree.branch.repository._format.__class__)

    def test_make_branch_builder(self):

        builder = self.make_branch_builder('dir')

        self.assertIsInstance(builder, branchbuilder.BranchBuilder)

        # Guard against regression into MemoryTransport leaking

        # files to disk instead of keeping them in memory.

        self.failIf(osutils.lexists('dir'))

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

        repo_format = weaverepo.RepositoryFormat7()

        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.failIf(osutils.lexists('dir'))

        self.assertEqual(format.repository_format.__class__,

                         the_branch.repository._format.__class__)

        self.assertEqual(repo_format.get_format_string(),

                         self.get_transport().get_bytes(

                            'dir/.bzr/repository/format'))

    def test_make_branch_builder_with_format_name(self):

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

        the_branch = builder.get_branch()

        # Guard against regression into MemoryTransport leaking

        # files to disk instead of keeping them in memory.

        self.failIf(osutils.lexists('dir'))

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

        self.assertEqual(dir_format.repository_format.__class__,

                         the_branch.repository._format.__class__)

        self.assertEqual('Bazaar-NG Knit Repository Format 1',

                         self.get_transport().get_bytes(

                            'dir/.bzr/repository/format'))

    def test_dangling_locks_cause_failures(self):

        class TestDanglingLock(tests.TestCaseWithMemoryTransport):

            def test_function(self):

                t = self.get_transport('.')

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

                l.create()

                l.attempt_lock()

        test = TestDanglingLock('test_function')

        result = test.run()

        if self._lock_check_thorough:

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

        else:

            # When _lock_check_thorough is disabled, then we don't trigger a

            # failure

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

class TestTestCaseWithTransport(tests.TestCaseWithTransport):

    """Tests for the convenience functions TestCaseWithTransport introduces."""

    def test_get_readonly_url_none(self):

        from bzrlib.transport import get_transport

        from bzrlib.transport.memory import MemoryServer

        from bzrlib.transport.readonly import ReadonlyTransportDecorator

        self.vfs_transport_factory = MemoryServer

        self.transport_readonly_server = None

        # calling get_readonly_transport() constructs a decorator on the url

        # for the server

        url = self.get_readonly_url()

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

        t = get_transport(url)

        t2 = get_transport(url2)

        self.failUnless(isinstance(t, ReadonlyTransportDecorator))

        self.failUnless(isinstance(t2, ReadonlyTransportDecorator))

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

    def test_get_readonly_url_http(self):

        from bzrlib.tests.http_server import HttpServer

        from bzrlib.transport import get_transport

        from bzrlib.transport.local import LocalURLServer

        from bzrlib.transport.http import HttpTransportBase

        self.transport_server = LocalURLServer

        self.transport_readonly_server = HttpServer

        # calling get_readonly_transport() gives us a HTTP server instance.

        url = self.get_readonly_url()

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

        # the transport returned may be any HttpTransportBase subclass

        t = get_transport(url)

        t2 = get_transport(url2)

        self.failUnless(isinstance(t, HttpTransportBase))

        self.failUnless(isinstance(t2, HttpTransportBase))

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

    def test_is_directory(self):

        """Test assertIsDirectory assertion"""

        t = self.get_transport()

        self.build_tree(['a_dir/', 'a_file'], transport=t)

        self.assertIsDirectory('a_dir', t)

        self.assertRaises(AssertionError, self.assertIsDirectory, 'a_file', t)

        self.assertRaises(AssertionError, self.assertIsDirectory, 'not_here', t)

    def test_make_branch_builder(self):

        builder = self.make_branch_builder('dir')

        rev_id = builder.build_commit()

        self.failUnlessExists('dir')