/brz/remove-bazaar

# Copyright (C) 2005-2013, 2016 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."""

import gc

import doctest

from functools import reduce

from io import BytesIO, StringIO, TextIOWrapper

import os

import signal

import sys

import threading

import time

import unittest

import warnings

from testtools import (

    ExtendedToOriginalDecorator,

    MultiTestResult,

    )

from testtools.content import Content

from testtools.content_type import ContentType

from testtools.matchers import (

    DocTestMatches,

    Equals,

    )

import testtools.testresult.doubles

import breezy

from .. import (

    branchbuilder,

    controldir,

    errors,

    hooks,

    lockdir,

    memorytree,

    osutils,

    repository,

    symbol_versioning,

    tests,

    transport,

    workingtree,

    )

from ..bzr import (

    bzrdir,

    remote,

    workingtree_3,

    workingtree_4,

    )

from ..bzr import (

    groupcompress_repo,

    )

from ..git import (

    workingtree as git_workingtree,

    )

from ..symbol_versioning import (

    deprecated_function,

    deprecated_in,

    deprecated_method,

    )

from . import (

    features,

    test_lsprof,

    test_server,

    TestUtil,

    )

from ..trace import note, mutter

from ..transport import memory

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 MetaTestLog(tests.TestCase):

    def test_logging(self):

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

        self.log('a test message')

        details = self.getDetails()

        log = details['log']

        self.assertThat(log.content_type, Equals(ContentType(

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

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

        self.assertThat(self.get_log(),

                        DocTestMatches(u"...a test message\n", doctest.ELLIPSIS))

class TestTreeShape(tests.TestCaseInTempDir):

    def test_unicode_paths(self):

        self.requireFeature(features.UnicodeFilenameFeature)

        filename = u'hell\u00d8'

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

        self.assertPathExists(filename)

class TestClassesAvailable(tests.TestCase):

    """As a convenience we expose Test* classes from breezy.tests"""

    def test_test_case(self):

        from . import TestCase

    def test_test_loader(self):

        from . import TestLoader

    def test_test_suite(self):

        from . import TestSuite

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 .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 .per_transport import transport_test_permutations

        from ..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 .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"],

                                   breezy.transport.Transport))

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

                                   breezy.transport.Server))

class TestBranchScenarios(tests.TestCase):

    def test_scenarios(self):

        # check that constructor parameters are passed through to the adapted

        # test.

        from .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 .per_controldir 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 .per_repository import formats_to_scenarios

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

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

                    b'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 breezy.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(

            "breezy.tests.test_selftest.TestTestScenarioApplication."

            "test_apply_scenario(new id)",

            adapted_test1.id())

        self.assertEqual(None, adapted_test2.bzrdir_format)

        self.assertEqual(

            "breezy.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 .per_interrepository import make_scenarios

        server1 = "a"

        server2 = "b"

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

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

              'extra_setup': 'C3'}),

            ('D0,str,str',

             {'repository_format': 'D1',

              'repository_format_to': 'D2',

              'transport_readonly_server': 'b',

              'transport_server': 'a',

              'extra_setup': 'D3'})],

            scenarios)

class TestWorkingTreeScenarios(tests.TestCase):

    def test_scenarios(self):

        # check that constructor parameters are passed through to the adapted

        # test.

        from .per_workingtree import make_scenarios

        server1 = "a"

        server2 = "b"

        formats = [workingtree_4.WorkingTreeFormat4(),

                   workingtree_3.WorkingTreeFormat3(),

                   workingtree_4.WorkingTreeFormat6()]

        scenarios = make_scenarios(server1, server2, formats,

                                   remote_server='c', remote_readonly_server='d',

                                   remote_backing_server='e')

        self.assertEqual([

            ('WorkingTreeFormat4',

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

              'transport_readonly_server': 'b',

              'transport_server': 'a',

              'workingtree_format': formats[0]}),

            ('WorkingTreeFormat3',

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

              'transport_readonly_server': 'b',

              'transport_server': 'a',

              'workingtree_format': formats[1]}),

            ('WorkingTreeFormat6',

             {'bzrdir_format': formats[2]._matchingcontroldir,

              'transport_readonly_server': 'b',

              'transport_server': 'a',

              'workingtree_format': formats[2]}),

            ('WorkingTreeFormat6,remote',

             {'bzrdir_format': formats[2]._matchingcontroldir,

              'repo_is_remote': True,

              'transport_readonly_server': 'd',

              'transport_server': 'c',

              'vfs_transport_factory': 'e',

              'workingtree_format': formats[2]}),

            ], 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, one additional instance

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

        # the tests, and one more that uses the default wt format as a

        # lightweight checkout of a remote repository.  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 .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"

        smart_server = test_server.SmartTCPServer_for_testing

        smart_readonly_server = test_server.ReadonlySmartTCPServer_for_testing

        mem_server = memory.MemoryServer

        formats = [workingtree_4.WorkingTreeFormat4(),

                   workingtree_3.WorkingTreeFormat3(), ]

        scenarios = make_scenarios(server1, server2, formats)

        self.assertEqual(9, len(scenarios))

        default_wt_format = workingtree.format_registry.get_default()

        wt4_format = workingtree_4.WorkingTreeFormat4()

        wt5_format = workingtree_4.WorkingTreeFormat5()

        wt6_format = workingtree_4.WorkingTreeFormat6()

        git_wt_format = git_workingtree.GitWorkingTreeFormat()

        expected_scenarios = [

            ('WorkingTreeFormat4',

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

              'transport_readonly_server': 'b',

              'transport_server': 'a',

              'workingtree_format': formats[0],

              '_workingtree_to_test_tree': return_parameter,

              }),

            ('WorkingTreeFormat3',

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

              'transport_readonly_server': 'b',

              'transport_server': 'a',

              'workingtree_format': formats[1],

              '_workingtree_to_test_tree': return_parameter,

              }),

            ('WorkingTreeFormat6,remote',

             {'bzrdir_format': wt6_format._matchingcontroldir,

              'repo_is_remote': True,

              'transport_readonly_server': smart_readonly_server,

              'transport_server': smart_server,

              'vfs_transport_factory': mem_server,

              'workingtree_format': wt6_format,

              '_workingtree_to_test_tree': return_parameter,

              }),

            ('RevisionTree',

             {'_workingtree_to_test_tree': revision_tree_from_workingtree,

              'bzrdir_format': default_wt_format._matchingcontroldir,

              'transport_readonly_server': 'b',

              'transport_server': 'a',

              'workingtree_format': default_wt_format,

              }),

            ('GitRevisionTree',

             {'_workingtree_to_test_tree': revision_tree_from_workingtree,

              'bzrdir_format': git_wt_format._matchingcontroldir,

              'transport_readonly_server': 'b',

              'transport_server': 'a',

              'workingtree_format': git_wt_format,

              }

             ),

            ('DirStateRevisionTree,WT4',

             {'_workingtree_to_test_tree': _dirstate_tree_from_workingtree,

              'bzrdir_format': wt4_format._matchingcontroldir,

              '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._matchingcontroldir,

              'transport_readonly_server': 'b',

              'transport_server': 'a',

              'workingtree_format': wt5_format,

              }),

            ('PreviewTree',

             {'_workingtree_to_test_tree': preview_tree_pre,

              'bzrdir_format': default_wt_format._matchingcontroldir,

              '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._matchingcontroldir,

              '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 .per_tree import (

            return_parameter,

            )

        from .per_intertree import (

            make_scenarios,

            )

        from ..bzr.workingtree_3 import WorkingTreeFormat3

        from ..bzr.workingtree_4 import WorkingTreeFormat4

        input_test = TestInterTreeScenarios(

            "test_scenarios")

        server1 = "a"

        server2 = "b"

        format1 = WorkingTreeFormat4()

        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._matchingcontroldir,

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

                "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 ..bzr.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", "longname"])

        self.assertRaises(AssertionError, self.assertEqualStat,

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

    def test_assertPathExists(self):

        self.assertPathExists('.')

        self.build_tree(['foo/', 'foo/bar'])

        self.assertPathExists('foo/bar')

        self.assertPathDoesNotExist('foo/foo')

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

        """The $BRZ_HOME and $HOME environment variables should not be unicode.

        See https://bugs.launchpad.net/bzr/+bug/464174

        """

        self.assertIsInstance(os.environ['BRZ_HOME'], str)

        self.assertIsInstance(os.environ['HOME'], str)

    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.assertFalse(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 = repository.format_registry.get_default()

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

    def test_make_branch_builder_with_format(self):