~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2005-2011 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]."""

# import system imports here

from cStringIO import StringIO

import os

import sys

import threading

from testtools import matchers

#import bzrlib specific imports here

from bzrlib import (

    branch,

    bzrdir,

    config,

    diff,

    errors,

    osutils,

    mail_client,

    ui,

    urlutils,

    tests,

    trace,

    transport,

    )

from bzrlib.tests import (

    features,

    scenarios,

    )

from bzrlib.util.configobj import configobj

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

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_command=gnome-gpg

log_format=short

user_global_option=something

[ALIASES]

h=help

ll=""" + sample_long_alias + "\n"

sample_always_signatures = """

[DEFAULT]

check_signatures=ignore

create_signatures=always

"""

sample_ignore_signatures = """

[DEFAULT]

check_signatures=require

create_signatures=never

"""

sample_maybe_signatures = """

[DEFAULT]

check_signatures=ignore

create_signatures=when-required

"""

sample_branches_text = """

[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_command=false

user_local_option=local

# test trailing / matching

[/a/*]

#subdirs will match but not the parent

[/a/c]

check_signatures=ignore

post_commit=bzrlib.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

    - bazaar_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.bazaar_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.bazaar_config.set_user_option('file', 'bazaar')

    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, user_id=None):

        if base is None:

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

        else:

            self.base = base

        self._transport = self.control_files = \

            FakeControlFilesAndTransport(user_id=user_id)

    def _get_config(self):

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

    def lock_write(self):

        pass

    def unlock(self):

        pass

class FakeControlFilesAndTransport(object):

    def __init__(self, user_id=None):

        self.files = {}

        if user_id:

            self.files['email'] = user_id

        self._transport = self

    def get_utf8(self, filename):

        # from LockableFiles

        raise AssertionError("get_utf8 should no longer be used")

    def get(self, filename):

        # from Transport

        try:

            return StringIO(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 = """[DEFAULT]

active = true

inactive = false

[UPPERCASE]

active = True

nonactive = False

"""

class TestConfigObj(tests.TestCase):

    def test_get_bool(self):

        co = config.ConfigObj(StringIO(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'

        lines = co.write()

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

        co2 = config.ConfigObj(lines)

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

erroneous_config = """[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(StringIO(erroneous_config),

                                     raise_errors=True)

        except config.configobj.DuplicateError, 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_no_default_editor(self):

        self.assertRaises(NotImplementedError, config.Config().get_editor)

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

        my_config = config.Config()

        self.assertFalse(my_config.signature_needed())

        self.assertEqual(config.CHECK_IF_POSSIBLE,

                         my_config.signature_checking())

        self.assertEqual(config.SIGN_WHEN_REQUIRED,

                         my_config.signing_policy())

    def test_signatures_template_method(self):

        my_config = InstrumentedConfig()

        self.assertEqual(config.CHECK_NEVER, my_config.signature_checking())

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

    def test_signatures_template_method_none(self):

        my_config = InstrumentedConfig()

        my_config._signatures = None

        self.assertEqual(config.CHECK_IF_POSSIBLE,

                         my_config.signature_checking())

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

    def test_gpg_signing_command_default(self):

        my_config = config.Config()

        self.assertEqual('gpg', my_config.gpg_signing_command())

    def test_get_user_option_default(self):

        my_config = config.Config()

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

    def test_post_commit_default(self):

        my_config = config.Config()

        self.assertEqual(None, my_config.post_commit())

    def test_log_format_default(self):

        my_config = config.Config()

        self.assertEqual('long', my_config.log_format())

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

        if sys.platform == 'win32':

            self.overrideEnv(

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

            self.bzr_home = \

                'C:/Documents and Settings/bogus/Application Data/bazaar/2.0'

        else:

            self.bzr_home = '/home/bogus/.bazaar'

    def test_config_dir(self):

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

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

            '/home/bogus/.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 in ('darwin', 'win32'):

            raise tests.TestNotApplicable(

                'XDG config dir not used on this platform')

        super(TestXDGConfigDir, self).setUp()

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

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

        self.overrideEnv('BZR_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):

        my_config = 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.failUnless(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.assertEquals(self.path, './foo.conf')

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

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

    def test_get_filename_parameter_is_deprecated_(self):

        conf = self.callDeprecated([

            'IniBasedConfig.__init__(get_filename) was deprecated in 2.3.'

            ' Use file_name instead.'],

            config.IniBasedConfig, lambda: 'ini.conf')

        self.assertEqual('ini.conf', conf.file_name)

    def test_get_parser_file_parameter_is_deprecated_(self):

        config_file = StringIO(sample_config_text.encode('utf-8'))

        conf = config.IniBasedConfig.from_string(sample_config_text)

        conf = self.callDeprecated([

            'IniBasedConfig._get_parser(file=xxx) was deprecated in 2.3.'

            ' Use IniBasedConfig(_content=xxx) instead.'],

            conf._get_parser, file=config_file)

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 = 'foo = bar\n'

        conf = config.IniBasedConfig.from_string(

            content, file_name='./test.conf', save=True)

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

class TestIniBaseConfigOnDisk(tests.TestCaseInTempDir):

    def test_cannot_reload_without_name(self):

        conf = config.IniBasedConfig.from_string(sample_config_text)

        self.assertRaises(AssertionError, conf.reload)

    def test_reload_see_new_value(self):

        c1 = config.IniBasedConfig.from_string('editor=vim\n',

                                               file_name='./test/conf')

        c1._write_config_file()

        c2 = config.IniBasedConfig.from_string('editor=emacs\n',

                                               file_name='./test/conf')

        c2._write_config_file()

        self.assertEqual('vim', c1.get_user_option('editor'))

        self.assertEqual('emacs', c2.get_user_option('editor'))

        # Make sure we get the Right value

        c1.reload()

        self.assertEqual('emacs', c1.get_user_option('editor'))

class TestLockableConfig(tests.TestCaseInTempDir):

    scenarios = lockable_config_scenarios()

    # Set by load_tests

    config_class = None

    config_args = None

    config_section = None

    def setUp(self):

        super(TestLockableConfig, self).setUp()

        self._content = '[%s]\none=1\ntwo=2\n' % (self.config_section,)

        self.config = self.create_config(self._content)

    def get_existing_config(self):

        return self.config_class(*self.config_args)

    def create_config(self, content):

        kwargs = dict(save=True)

        c = self.config_class.from_string(content, *self.config_args, **kwargs)

        return c

    def test_simple_read_access(self):

        self.assertEquals('1', self.config.get_user_option('one'))

    def test_simple_write_access(self):

        self.config.set_user_option('one', 'one')

        self.assertEquals('one', self.config.get_user_option('one'))

    def test_listen_to_the_last_speaker(self):

        c1 = self.config

        c2 = self.get_existing_config()

        c1.set_user_option('one', 'ONE')

        c2.set_user_option('two', 'TWO')

        self.assertEquals('ONE', c1.get_user_option('one'))

        self.assertEquals('TWO', c2.get_user_option('two'))

        # The second update respect the first one

        self.assertEquals('ONE', c2.get_user_option('one'))

    def test_last_speaker_wins(self):

        # If the same config is not shared, the same variable modified twice

        # can only see a single result.

        c1 = self.config

        c2 = self.get_existing_config()

        c1.set_user_option('one', 'c1')

        c2.set_user_option('one', 'c2')

        self.assertEquals('c2', c2._get_user_option('one'))

        # The first modification is still available until another refresh

        # occur

        self.assertEquals('c1', c1._get_user_option('one'))

        c1.set_user_option('two', 'done')

        self.assertEquals('c2', c1._get_user_option('one'))

    def test_writes_are_serialized(self):

        c1 = self.config

        c2 = self.get_existing_config()

        # We spawn a thread that will pause *during* the write

        before_writing = threading.Event()

        after_writing = threading.Event()

        writing_done = threading.Event()

        c1_orig = c1._write_config_file

        def c1_write_config_file():

            before_writing.set()

            c1_orig()

            # The lock is held we wait for the main thread to decide when to

            # continue

            after_writing.wait()

        c1._write_config_file = c1_write_config_file

        def c1_set_option():

            c1.set_user_option('one', 'c1')

            writing_done.set()

        t1 = threading.Thread(target=c1_set_option)

        # Collect the thread after the test

        self.addCleanup(t1.join)

        # Be ready to unblock the thread if the test goes wrong

        self.addCleanup(after_writing.set)

        t1.start()

        before_writing.wait()

        self.assertTrue(c1._lock.is_held)

        self.assertRaises(errors.LockContention,

                          c2.set_user_option, 'one', 'c2')

        self.assertEquals('c1', c1.get_user_option('one'))

        # Let the lock be released

        after_writing.set()

        writing_done.wait()

        c2.set_user_option('one', 'c2')

        self.assertEquals('c2', c2.get_user_option('one'))

    def test_read_while_writing(self):

       c1 = self.config

       # We spawn a thread that will pause *during* the write

       ready_to_write = threading.Event()

       do_writing = threading.Event()

       writing_done = threading.Event()

       c1_orig = c1._write_config_file

       def c1_write_config_file():

           ready_to_write.set()

           # The lock is held we wait for the main thread to decide when to

           # continue

           do_writing.wait()

           c1_orig()

           writing_done.set()

       c1._write_config_file = c1_write_config_file

       def c1_set_option():

           c1.set_user_option('one', 'c1')

       t1 = threading.Thread(target=c1_set_option)

       # Collect the thread after the test

       self.addCleanup(t1.join)

       # Be ready to unblock the thread if the test goes wrong

       self.addCleanup(do_writing.set)

       t1.start()

       # Ensure the thread is ready to write

       ready_to_write.wait()

       self.assertTrue(c1._lock.is_held)

       self.assertEquals('c1', c1.get_user_option('one'))

       # If we read during the write, we get the old value

       c2 = self.get_existing_config()

       self.assertEquals('1', c2.get_user_option('one'))

       # Let the writing occur and ensure it occurred

       do_writing.set()