/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

from io import BytesIO

import os

import sys

import threading

import configobj

from testtools import matchers

from .. import (

    branch,

    config,

    bedding,

    controldir,

    diff,

    errors,

    lock,

    osutils,

    mail_client,

    ui,

    urlutils,

    registry as _mod_registry,

    tests,

    trace,

    )

from ..bzr import (

    remote,

    )

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

        self._change_editor = 'vimdiff -fo {new_path} {old_path}'

    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 self._change_editor

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)

    def test_get_change_editor_implicit_args(self):

        # If there are no substitution variables, then assume the

        # old and new path are the last arguments.

        my_config = InstrumentedConfig()

        my_config._change_editor = 'vimdiff -o'

        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', '-o', '{old_path}', '{new_path}'],

                         change_editor.command_template)

    def test_get_change_editor_old_style(self):

        # Test the old style format for the change_editor setting.

        my_config = InstrumentedConfig()

        my_config._change_editor = 'vimdiff -o @old_path @new_path'

        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', '-o', '{old_path}', '{new_path}'],

                         change_editor.command_template)

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'

        config.IniBasedConfig.from_string(content, file_name='./test.conf',

                                          save=True)

        self.assertFileEqual(content, 'test.conf')

class TestIniConfigOptionExpansion(tests.TestCase):

    """Test option expansion from the IniConfig level.

    What we really want here is to test the Config level, but the class being

    abstract as far as storing values is concerned, this can't be done

    properly (yet).

    """

    # FIXME: This should be rewritten when all configs share a storage

    # implementation -- vila 2011-02-18

    def get_config(self, string=None):

        if string is None:

            string = b''

        c = config.IniBasedConfig.from_string(string)

        return c

    def assertExpansion(self, expected, conf, string, env=None):

        self.assertEqual(expected, conf.expand_options(string, env))

    def test_no_expansion(self):

        c = self.get_config('')

        self.assertExpansion('foo', c, 'foo')

    def test_env_adding_options(self):

        c = self.get_config('')

        self.assertExpansion('bar', c, '{foo}', {'foo': 'bar'})

    def test_env_overriding_options(self):

        c = self.get_config('foo=baz')

        self.assertExpansion('bar', c, '{foo}', {'foo': 'bar'})

    def test_simple_ref(self):

        c = self.get_config('foo=xxx')

        self.assertExpansion('xxx', c, '{foo}')

    def test_unknown_ref(self):

        c = self.get_config('')

        self.assertRaises(config.ExpandingUnknownOption,

                          c.expand_options, '{foo}')

    def test_indirect_ref(self):

        c = self.get_config('''

foo=xxx

bar={foo}

''')

        self.assertExpansion('xxx', c, '{bar}')

    def test_embedded_ref(self):

        c = self.get_config('''

foo=xxx

bar=foo

''')

        self.assertExpansion('xxx', c, '{{bar}}')

    def test_simple_loop(self):

        c = self.get_config('foo={foo}')

        self.assertRaises(config.OptionExpansionLoop, c.expand_options,

                          '{foo}')

    def test_indirect_loop(self):

        c = self.get_config('''

foo={bar}

bar={baz}

baz={foo}''')

        e = self.assertRaises(config.OptionExpansionLoop,

                              c.expand_options, '{foo}')

        self.assertEqual('foo->bar->baz', e.refs)

        self.assertEqual('{foo}', e.string)

    def test_list(self):

        conf = self.get_config('''

foo=start

bar=middle