/brz/remove-bazaar

# Copyright (C) 2005-2014, 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 finding and reading the bzr config file[s]."""

from textwrap import dedent

import os

import sys

import threading

import configobj

from testtools import matchers

from .. import (

    branch,

    config,

    controldir,

    diff,

    errors,

    lock,

    osutils,

    mail_client,

    ui,

    urlutils,

    registry as _mod_registry,

    tests,

    trace,

    )

from ..bzr import (

    remote,

    )

from ..sixish import (

    BytesIO,

    text_type,

    )

from ..transport import remote as transport_remote

from . import (

    features,

    scenarios,

    test_server,

    )

def lockable_config_scenarios():

    return [

        ('global',

         {'config_class': config.GlobalConfig,

          'config_args': [],

          'config_section': 'DEFAULT'}),

        ('locations',

         {'config_class': config.LocationConfig,

          'config_args': ['.'],

          'config_section': '.'}),]

load_tests = scenarios.load_tests_apply_scenarios

# Register helpers to build stores

config.test_store_builder_registry.register(

    'configobj', lambda test: config.TransportIniFileStore(

        test.get_transport(), 'configobj.conf'))

config.test_store_builder_registry.register(

    'breezy', lambda test: config.GlobalStore())

config.test_store_builder_registry.register(

    'location', lambda test: config.LocationStore())

def build_backing_branch(test, relpath,

                         transport_class=None, server_class=None):

    """Test helper to create a backing branch only once.

    Some tests needs multiple stores/stacks to check concurrent update

    behaviours. As such, they need to build different branch *objects* even if

    they share the branch on disk.

    :param relpath: The relative path to the branch. (Note that the helper

        should always specify the same relpath).

    :param transport_class: The Transport class the test needs to use.

    :param server_class: The server associated with the ``transport_class``

        above.

    Either both or neither of ``transport_class`` and ``server_class`` should

    be specified.

    """

    if transport_class is not None and server_class is not None:

        test.transport_class = transport_class

        test.transport_server = server_class

    elif not (transport_class is None and server_class is None):

        raise AssertionError('Specify both ``transport_class`` and '

                             '``server_class`` or neither of them')

    if getattr(test, 'backing_branch', None) is None:

        # First call, let's build the branch on disk

        test.backing_branch = test.make_branch(relpath)

def build_branch_store(test):

    build_backing_branch(test, 'branch')

    b = branch.Branch.open('branch')

    return config.BranchStore(b)

config.test_store_builder_registry.register('branch', build_branch_store)

def build_control_store(test):

    build_backing_branch(test, 'branch')

    b = controldir.ControlDir.open('branch')

    return config.ControlStore(b)

config.test_store_builder_registry.register('control', build_control_store)

def build_remote_branch_store(test):

    # There is only one permutation (but we won't be able to handle more with

    # this design anyway)

    (transport_class,

     server_class) = transport_remote.get_test_permutations()[0]

    build_backing_branch(test, 'branch', transport_class, server_class)

    b = branch.Branch.open(test.get_url('branch'))

    return config.BranchStore(b)

config.test_store_builder_registry.register('remote_branch',

                                            build_remote_branch_store)

config.test_stack_builder_registry.register(

    'breezy', lambda test: config.GlobalStack())

config.test_stack_builder_registry.register(

    'location', lambda test: config.LocationStack('.'))

def build_branch_stack(test):

    build_backing_branch(test, 'branch')

    b = branch.Branch.open('branch')

    return config.BranchStack(b)

config.test_stack_builder_registry.register('branch', build_branch_stack)

def build_branch_only_stack(test):

    # There is only one permutation (but we won't be able to handle more with

    # this design anyway)

    (transport_class,

     server_class) = transport_remote.get_test_permutations()[0]

    build_backing_branch(test, 'branch', transport_class, server_class)

    b = branch.Branch.open(test.get_url('branch'))

    return config.BranchOnlyStack(b)

config.test_stack_builder_registry.register('branch_only',

                                            build_branch_only_stack)

def build_remote_control_stack(test):

    # There is only one permutation (but we won't be able to handle more with

    # this design anyway)

    (transport_class,

     server_class) = transport_remote.get_test_permutations()[0]

    # We need only a bzrdir for this, not a full branch, but it's not worth

    # creating a dedicated helper to create only the bzrdir

    build_backing_branch(test, 'branch', transport_class, server_class)

    b = branch.Branch.open(test.get_url('branch'))

    return config.RemoteControlStack(b.controldir)

config.test_stack_builder_registry.register('remote_control',

                                            build_remote_control_stack)

sample_long_alias="log -r-15..-1 --line"

sample_config_text = u"""

[DEFAULT]

email=Erik B\u00e5gfors <erik@bagfors.nu>

editor=vim

change_editor=vimdiff -of @new_path @old_path

gpg_signing_key=DD4D5088

log_format=short

validate_signatures_in_log=true

acceptable_keys=amy

user_global_option=something

bzr.mergetool.sometool=sometool {base} {this} {other} -o {result}

bzr.mergetool.funkytool=funkytool "arg with spaces" {this_temp}

bzr.mergetool.newtool='"newtool with spaces" {this_temp}'

bzr.default_mergetool=sometool

[ALIASES]

h=help

ll=""".encode('utf-8') + sample_long_alias.encode('utf-8') + b"\n"

sample_always_signatures = b"""

[DEFAULT]

check_signatures=ignore

create_signatures=always

"""

sample_ignore_signatures = b"""

[DEFAULT]

check_signatures=require

create_signatures=never

"""

sample_maybe_signatures = b"""

[DEFAULT]

check_signatures=ignore

create_signatures=when-required

"""

sample_branches_text = b"""

[http://www.example.com]

# Top level policy

email=Robert Collins <robertc@example.org>

normal_option = normal

appendpath_option = append

appendpath_option:policy = appendpath

norecurse_option = norecurse

norecurse_option:policy = norecurse

[http://www.example.com/ignoreparent]

# different project: ignore parent dir config

ignore_parents=true

[http://www.example.com/norecurse]

# configuration items that only apply to this dir

recurse=false

normal_option = norecurse

[http://www.example.com/dir]

appendpath_option = normal

[/b/]

check_signatures=require

# test trailing / matching with no children

[/a/]

check_signatures=check-available

gpg_signing_key=default

user_local_option=local

# test trailing / matching

[/a/*]

#subdirs will match but not the parent

[/a/c]

check_signatures=ignore

post_commit=breezy.tests.test_config.post_commit

#testing explicit beats globs

"""

def create_configs(test):

    """Create configuration files for a given test.

    This requires creating a tree (and populate the ``test.tree`` attribute)

    and its associated branch and will populate the following attributes:

    - branch_config: A BranchConfig for the associated branch.

    - locations_config : A LocationConfig for the associated branch

    - breezy_config: A GlobalConfig.

    The tree and branch are created in a 'tree' subdirectory so the tests can

    still use the test directory to stay outside of the branch.

    """

    tree = test.make_branch_and_tree('tree')

    test.tree = tree

    test.branch_config = config.BranchConfig(tree.branch)

    test.locations_config = config.LocationConfig(tree.basedir)

    test.breezy_config = config.GlobalConfig()

def create_configs_with_file_option(test):

    """Create configuration files with a ``file`` option set in each.

    This builds on ``create_configs`` and add one ``file`` option in each

    configuration with a value which allows identifying the configuration file.

    """

    create_configs(test)

    test.breezy_config.set_user_option('file', 'breezy')

    test.locations_config.set_user_option('file', 'locations')

    test.branch_config.set_user_option('file', 'branch')

class TestOptionsMixin:

    def assertOptions(self, expected, conf):

        # We don't care about the parser (as it will make tests hard to write

        # and error-prone anyway)

        self.assertThat([opt[:4] for opt in conf._get_options()],

                        matchers.Equals(expected))

class InstrumentedConfigObj(object):

    """A config obj look-enough-alike to record calls made to it."""

    def __contains__(self, thing):

        self._calls.append(('__contains__', thing))

        return False

    def __getitem__(self, key):

        self._calls.append(('__getitem__', key))

        return self

    def __init__(self, input, encoding=None):

        self._calls = [('__init__', input, encoding)]

    def __setitem__(self, key, value):

        self._calls.append(('__setitem__', key, value))

    def __delitem__(self, key):

        self._calls.append(('__delitem__', key))

    def keys(self):

        self._calls.append(('keys',))

        return []

    def reload(self):

        self._calls.append(('reload',))

    def write(self, arg):

        self._calls.append(('write',))

    def as_bool(self, value):

        self._calls.append(('as_bool', value))

        return False

    def get_value(self, section, name):

        self._calls.append(('get_value', section, name))

        return None

class FakeBranch(object):

    def __init__(self, base=None):

        if base is None:

            self.base = "http://example.com/branches/demo"

        else:

            self.base = base

        self._transport = self.control_files = \

            FakeControlFilesAndTransport()

    def _get_config(self):

        return config.TransportConfig(self._transport, 'branch.conf')

    def lock_write(self):

        return lock.LogicalLockResult(self.unlock)

    def unlock(self):

        pass

class FakeControlFilesAndTransport(object):

    def __init__(self):

        self.files = {}

        self._transport = self

    def get(self, filename):

        # from Transport

        try:

            return BytesIO(self.files[filename])

        except KeyError:

            raise errors.NoSuchFile(filename)

    def get_bytes(self, filename):

        # from Transport

        try:

            return self.files[filename]

        except KeyError:

            raise errors.NoSuchFile(filename)

    def put(self, filename, fileobj):

        self.files[filename] = fileobj.read()

    def put_file(self, filename, fileobj):

        return self.put(filename, fileobj)

class InstrumentedConfig(config.Config):

    """An instrumented config that supplies stubs for template methods."""

    def __init__(self):

        super(InstrumentedConfig, self).__init__()

        self._calls = []

        self._signatures = config.CHECK_NEVER

    def _get_user_id(self):

        self._calls.append('_get_user_id')

        return "Robert Collins <robert.collins@example.org>"

    def _get_signature_checking(self):

        self._calls.append('_get_signature_checking')

        return self._signatures

    def _get_change_editor(self):

        self._calls.append('_get_change_editor')

        return 'vimdiff -fo @new_path @old_path'

bool_config = b"""[DEFAULT]

active = true

inactive = false

[UPPERCASE]

active = True

nonactive = False

"""

class TestConfigObj(tests.TestCase):

    def test_get_bool(self):

        co = config.ConfigObj(BytesIO(bool_config))

        self.assertIs(co.get_bool('DEFAULT', 'active'), True)

        self.assertIs(co.get_bool('DEFAULT', 'inactive'), False)

        self.assertIs(co.get_bool('UPPERCASE', 'active'), True)

        self.assertIs(co.get_bool('UPPERCASE', 'nonactive'), False)

    def test_hash_sign_in_value(self):

        """

        Before 4.5.0, ConfigObj did not quote # signs in values, so they'd be

        treated as comments when read in again. (#86838)

        """

        co = config.ConfigObj()

        co['test'] = 'foo#bar'

        outfile = BytesIO()

        co.write(outfile=outfile)

        lines = outfile.getvalue().splitlines()

        self.assertEqual(lines, [b'test = "foo#bar"'])

        co2 = config.ConfigObj(lines)

        self.assertEqual(co2['test'], 'foo#bar')

    def test_triple_quotes(self):

        # Bug #710410: if the value string has triple quotes

        # then ConfigObj versions up to 4.7.2 will quote them wrong

        # and won't able to read them back

        triple_quotes_value = '''spam

""" that's my spam """

eggs'''

        co = config.ConfigObj()

        co['test'] = triple_quotes_value

        # While writing this test another bug in ConfigObj has been found:

        # method co.write() without arguments produces list of lines

        # one option per line, and multiline values are not split

        # across multiple lines,

        # and that breaks the parsing these lines back by ConfigObj.

        # This issue only affects test, but it's better to avoid

        # `co.write()` construct at all.

        # [bialix 20110222] bug report sent to ConfigObj's author

        outfile = BytesIO()

        co.write(outfile=outfile)

        output = outfile.getvalue()

        # now we're trying to read it back

        co2 = config.ConfigObj(BytesIO(output))

        self.assertEqual(triple_quotes_value, co2['test'])

erroneous_config = b"""[section] # line 1

good=good # line 2

[section] # line 3

whocares=notme # line 4

"""

class TestConfigObjErrors(tests.TestCase):

    def test_duplicate_section_name_error_line(self):

        try:

            co = configobj.ConfigObj(BytesIO(erroneous_config),

                                     raise_errors=True)

        except config.configobj.DuplicateError as e:

            self.assertEqual(3, e.line_number)

        else:

            self.fail('Error in config file not detected')

class TestConfig(tests.TestCase):

    def test_constructs(self):

        config.Config()

    def test_user_email(self):

        my_config = InstrumentedConfig()

        self.assertEqual('robert.collins@example.org', my_config.user_email())

        self.assertEqual(['_get_user_id'], my_config._calls)

    def test_username(self):

        my_config = InstrumentedConfig()

        self.assertEqual('Robert Collins <robert.collins@example.org>',

                         my_config.username())

        self.assertEqual(['_get_user_id'], my_config._calls)

    def test_get_user_option_default(self):

        my_config = config.Config()

        self.assertEqual(None, my_config.get_user_option('no_option'))

    def test_validate_signatures_in_log_default(self):

        my_config = config.Config()

        self.assertEqual(False, my_config.validate_signatures_in_log())

    def test_get_change_editor(self):

        my_config = InstrumentedConfig()

        change_editor = my_config.get_change_editor('old_tree', 'new_tree')

        self.assertEqual(['_get_change_editor'], my_config._calls)

        self.assertIs(diff.DiffFromTool, change_editor.__class__)

        self.assertEqual(['vimdiff', '-fo', '@new_path', '@old_path'],

                         change_editor.command_template)

class TestConfigPath(tests.TestCase):

    def setUp(self):

        super(TestConfigPath, self).setUp()

        self.overrideEnv('HOME', '/home/bogus')

        self.overrideEnv('XDG_CACHE_HOME', '')

        if sys.platform == 'win32':

            self.overrideEnv(

                'BRZ_HOME',

                r'C:\Documents and Settings\bogus\Application Data')

            self.brz_home = \

                'C:/Documents and Settings/bogus/Application Data/breezy'

        else:

            self.brz_home = '/home/bogus/.config/breezy'

    def test_config_dir(self):

        self.assertEqual(config.config_dir(), self.brz_home)

    def test_config_dir_is_unicode(self):

        self.assertIsInstance(config.config_dir(), text_type)

    def test_config_filename(self):

        self.assertEqual(config.config_filename(),

                         self.brz_home + '/breezy.conf')

    def test_locations_config_filename(self):

        self.assertEqual(config.locations_config_filename(),

                         self.brz_home + '/locations.conf')

    def test_authentication_config_filename(self):

        self.assertEqual(config.authentication_config_filename(),

                         self.brz_home + '/authentication.conf')

    def test_xdg_cache_dir(self):

        self.assertEqual(config.xdg_cache_dir(),

            '/home/bogus/.cache')

class TestConfigPathFallback(tests.TestCaseInTempDir):

    def setUp(self):

        super(TestConfigPathFallback, self).setUp()

        self.overrideEnv('HOME', self.test_dir)

        self.overrideEnv('XDG_CACHE_HOME', '')

        self.bzr_home = os.path.join(self.test_dir, '.bazaar')

        os.mkdir(self.bzr_home)

    def test_config_dir(self):

        self.assertEqual(config.config_dir(), self.bzr_home)

    def test_config_dir_is_unicode(self):

        self.assertIsInstance(config.config_dir(), text_type)

    def test_config_filename(self):

        self.assertEqual(config.config_filename(),

                         self.bzr_home + '/bazaar.conf')

    def test_locations_config_filename(self):

        self.assertEqual(config.locations_config_filename(),

                         self.bzr_home + '/locations.conf')

    def test_authentication_config_filename(self):

        self.assertEqual(config.authentication_config_filename(),

                         self.bzr_home + '/authentication.conf')

    def test_xdg_cache_dir(self):

        self.assertEqual(config.xdg_cache_dir(),

            os.path.join(self.test_dir, '.cache'))

class TestXDGConfigDir(tests.TestCaseInTempDir):

    # must be in temp dir because config tests for the existence of the bazaar

    # subdirectory of $XDG_CONFIG_HOME

    def setUp(self):

        if sys.platform == 'win32':

            raise tests.TestNotApplicable(

                'XDG config dir not used on this platform')

        super(TestXDGConfigDir, self).setUp()

        self.overrideEnv('HOME', self.test_home_dir)

        # BRZ_HOME overrides everything we want to test so unset it.

        self.overrideEnv('BRZ_HOME', None)

    def test_xdg_config_dir_exists(self):

        """When ~/.config/bazaar exists, use it as the config dir."""

        newdir = osutils.pathjoin(self.test_home_dir, '.config', 'bazaar')

        os.makedirs(newdir)

        self.assertEqual(config.config_dir(), newdir)

    def test_xdg_config_home(self):

        """When XDG_CONFIG_HOME is set, use it."""

        xdgconfigdir = osutils.pathjoin(self.test_home_dir, 'xdgconfig')

        self.overrideEnv('XDG_CONFIG_HOME', xdgconfigdir)

        newdir = osutils.pathjoin(xdgconfigdir, 'bazaar')

        os.makedirs(newdir)

        self.assertEqual(config.config_dir(), newdir)

class TestIniConfig(tests.TestCaseInTempDir):

    def make_config_parser(self, s):

        conf = config.IniBasedConfig.from_string(s)

        return conf, conf._get_parser()

class TestIniConfigBuilding(TestIniConfig):

    def test_contructs(self):

        config.IniBasedConfig()

    def test_from_fp(self):

        my_config = config.IniBasedConfig.from_string(sample_config_text)

        self.assertIsInstance(my_config._get_parser(), configobj.ConfigObj)

    def test_cached(self):

        my_config = config.IniBasedConfig.from_string(sample_config_text)

        parser = my_config._get_parser()

        self.assertTrue(my_config._get_parser() is parser)

    def _dummy_chown(self, path, uid, gid):

        self.path, self.uid, self.gid = path, uid, gid

    def test_ini_config_ownership(self):

        """Ensure that chown is happening during _write_config_file"""

        self.requireFeature(features.chown_feature)

        self.overrideAttr(os, 'chown', self._dummy_chown)

        self.path = self.uid = self.gid = None

        conf = config.IniBasedConfig(file_name='./foo.conf')

        conf._write_config_file()

        self.assertEqual(self.path, './foo.conf')

        self.assertTrue(isinstance(self.uid, int))

        self.assertTrue(isinstance(self.gid, int))

class TestIniConfigSaving(tests.TestCaseInTempDir):

    def test_cant_save_without_a_file_name(self):

        conf = config.IniBasedConfig()

        self.assertRaises(AssertionError, conf._write_config_file)

    def test_saved_with_content(self):

        content = b'foo = bar\n'