~didrocks/ubuntuone-client/use_result_var

#
# Author: Guillermo Gonzalez <guillermo.gonzalez@canonical.com>
#
# Copyright 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 version 3, as published
# by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, 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, see <http://www.gnu.org/licenses/>.
"""tests for the syncdaemon config module """
from __future__ import with_statement

import logging
import os
from ConfigParser import ConfigParser
from xdg.BaseDirectory import (
    xdg_data_home,
    xdg_cache_home,
)

from ubuntuone.syncdaemon import config
from contrib.testing.testcase import (
    BaseTwistedTestCase,
    environ,
)


class TestConfigBasic(BaseTwistedTestCase):
    """Basic _Config object tests"""

    def setUp(self):
        BaseTwistedTestCase.setUp(self)
        self.test_root = self.mktemp()

    def assertThrottlingSection(self, expected, current, on, read, write):
        """Assert for equality two ConfigParser and against the on, read and
        write args
        """
        self.assertEquals(expected.getboolean(config.THROTTLING, 'on'), on)
        self.assertEquals(expected.getint(config.THROTTLING, 'read_limit'),
                          read)
        self.assertEquals(expected.getint(config.THROTTLING, 'write_limit'),
                          write)
        self.assertEquals(expected.getboolean(config.THROTTLING, 'on'),
                          current.get_throttling())
        self.assertEquals(expected.getint(config.THROTTLING, 'read_limit'),
                          current.get_throttling_read_limit())
        self.assertEquals(expected.getint(config.THROTTLING, 'write_limit'),
                          current.get_throttling_write_limit())

    def test_load_empty(self):
        """test loading the a non-existent config file"""
        conf_file = os.path.join(self.test_root, 'test_missing_config.conf')
        # create the config object with an empty config file
        conf = config._Config(conf_file)
        self.assertEquals(False, conf.get_throttling())
        self.assertEquals(2097152, conf.get_throttling_read_limit())
        self.assertEquals(2097152, conf.get_throttling_write_limit())

    def test_load_basic(self):
        """test loading the config file containing only the throttling values"""
        conf_file = os.path.join(self.test_root, 'test_load_config.conf')
        # write some throttling values to the config file
        with open(conf_file, 'w') as fp:
            fp.write('[bandwidth_throttling]\n')
            fp.write('on = True\n')
            fp.write('read_limit = 1000\n')
            fp.write('write_limit = 200\n')
        conf = config._Config(conf_file)
        self.assertEquals(True, conf.get_throttling())
        self.assertEquals(1000, conf.get_throttling_read_limit())
        self.assertEquals(200, conf.get_throttling_write_limit())

    def test_load_extra_data(self):
        """test loading the a config file with other sections too"""
        conf_file = os.path.join(self.test_root, 'test_load_extra_config.conf')
        # write some throttling values to the config file
        with open(conf_file, 'w') as fp:
            fp.write('[__main__]\n')
            fp.write('log_level = INFO\n')
            fp.write('disable_ssl_verify = True\n')
            fp.write('\n')
            fp.write('[bandwidth_throttling]\n')
            fp.write('on = True\n')
            fp.write('read_limit = 1000\n')
            fp.write('write_limit = 200\n')
        conf = config._Config(conf_file)
        self.assertEquals(True, conf.get_throttling())
        self.assertEquals(1000, conf.get_throttling_read_limit())
        self.assertEquals(200, conf.get_throttling_write_limit())

    def test_write_new(self):
        """test writing the throttling section to a new config file"""
        conf_file = os.path.join(self.test_root, 'test_write_new_config.conf')
        self.assertFalse(os.path.exists(conf_file))
        conf = config._Config(conf_file)
        conf.set_throttling(True)
        conf.set_throttling_read_limit(1000)
        conf.set_throttling_write_limit(100)
        conf.save()
        # load the config in a barebone ConfigParser and check
        conf_1 = ConfigParser()
        conf_1.read(conf_file)
        self.assertThrottlingSection(conf_1, conf, True, 1000, 100)

    def test_write_existing(self):
        """test writing the throttling section to a existing config file"""
        conf_file = os.path.join(self.test_root,
                                 'test_write_existing_config.conf')
        # write some throttling values to the config file
        with open(conf_file, 'w') as fp:
            fp.write('[bandwidth_throttling]\n')
            fp.write('on = False\n')
            fp.write('read_limit = 1000\n')
            fp.write('write_limit = 100\n')
        self.assertTrue(os.path.exists(conf_file))
        conf = config._Config(conf_file)
        conf.set_throttling(True)
        conf.set_throttling_read_limit(2000)
        conf.set_throttling_write_limit(200)
        conf.save()
        # load the config in a barebone ConfigParser and check
        conf_1 = ConfigParser()
        conf_1.read(conf_file)
        self.assertThrottlingSection(conf_1, conf, True, 2000, 200)

    def test_write_extra(self):
        """test writing the throttling section back to the config file,
        including extra sections
        """
        conf_file = os.path.join(self.test_root, 'test_write_extra_config.conf')
        # write some throttling values to the config file
        with open(conf_file, 'w') as fp:
            fp.write('[__main__]\n')
            fp.write('log_level = INFO\n')
            fp.write('disable_ssl_verify = True\n')
            fp.write('\n')
            fp.write('[bandwidth_throttling]\n')
            fp.write('on = False\n')
            fp.write('read_limit = 2000\n')
            fp.write('write_limit = 200\n')
        self.assertTrue(os.path.exists(conf_file))
        conf = config._Config(conf_file)
        conf.set_throttling(True)
        conf.set_throttling_read_limit(3000)
        conf.set_throttling_write_limit(300)
        conf.save()
        # load the config in a barebone ConfigParser and check
        conf_1 = ConfigParser()
        conf_1.read(conf_file)
        self.assertThrottlingSection(conf_1, conf, True, 3000, 300)
        self.assertEquals(conf_1.get('__main__', 'log_level'),
                          conf.get('__main__', 'log_level'))
        self.assertEquals(conf_1.getboolean('__main__', 'disable_ssl_verify'),
                          conf.getboolean('__main__', 'disable_ssl_verify'))

    def test_write_existing_partial(self):
        """test writing a partially updated throttling section
        to a existing config file
        """
        conf_file = os.path.join(self.test_root,
                                 'test_write_existing_config.conf')
        # write some throttling values to the config file
        with open(conf_file, 'w') as fp:
            fp.write('[bandwidth_throttling]\n')
            fp.write('on = True\n')
            fp.write('read_limit = 1000\n')
            fp.write('write_limit = 100\n')
        self.assertTrue(os.path.exists(conf_file))
        conf = config._Config(conf_file)
        conf.set_throttling(False)
        conf.save()
        # load the config in a barebone ConfigParser and check
        conf_1 = ConfigParser()
        conf_1.read(conf_file)
        self.assertThrottlingSection(conf_1, conf, False, 1000, 100)

    def test_load_negative_limits(self):
        """test loading the config file with negative read/write limits"""
        conf_file = os.path.join(self.test_root, 'test_load_config.conf')
        # write some throttling values to the config file
        with open(conf_file, 'w') as fp:
            fp.write('[bandwidth_throttling]\n')
            fp.write('on = True\n')
            fp.write('read_limit = -1\n')
            fp.write('write_limit = -1\n')
        conf = config._Config(conf_file)
        self.assertEquals(True, conf.get_throttling())
        self.assertEquals(None, conf.get_throttling_read_limit())
        self.assertEquals(None, conf.get_throttling_write_limit())


    def test_load_partial_config(self):
        """test loading a partial config file and fallback to defaults"""
        conf_file = os.path.join(self.test_root, 'test_load_config.conf')
        # write some throttling values to the config file
        with open(conf_file, 'w') as fp:
            fp.write('[bandwidth_throttling]\n')
            fp.write('on = True\n')
            fp.write('read_limit = 1\n')
        conf = config._Config(conf_file)
        self.assertEquals(True, conf.get_throttling())
        self.assertEquals(1, conf.get_throttling_read_limit())
        self.assertEquals(2097152, conf.get_throttling_write_limit())

    def test_override(self):
        """test loading the config file containing only the throttling values"""
        conf_file = os.path.join(self.test_root, 'test_load_config.conf')
        # write some throttling values to the config file
        with open(conf_file, 'w') as fp:
            fp.write('[bandwidth_throttling]\n')
            fp.write('on = True\n')
            fp.write('read_limit = 1000\n')
            fp.write('write_limit = 200\n')
        conf = config._Config(conf_file)
        conf_orig = config._Config(conf_file)
        overridden_opts = [('bandwidth_throttling', 'on', False)]
        conf.override_options(overridden_opts)
        self.assertEquals(False, conf.get_throttling())
        self.assertFalse(conf.get_throttling() == conf_orig.get_throttling())
        self.assertEquals(1000, conf.get_throttling_read_limit())
        self.assertEquals(200, conf.get_throttling_write_limit())
        conf.save()
        # load the config in a barebone ConfigParser and check
        conf_1 = ConfigParser()
        conf_1.read(conf_file)
        self.assertThrottlingSection(conf_1, conf_orig, True, 1000, 200)

    def test_load_udf_autosubscribe(self):
        """Test load/set/override of udf_autosubscribe config value."""
        conf_file = os.path.join(self.test_root,
                                 'test_udf_autosubscribe_config.conf')
        # write some throttling values to the config file
        with open(conf_file, 'w') as fp:
            fp.write('[__main__]\n')
            fp.write('log_level = INFO\n')
            fp.write('disable_ssl_verify = True\n')
            fp.write('udf_autosubscribe = True\n')
            fp.write('\n')
            fp.write('[bandwidth_throttling]\n')
            fp.write('on = True\n')
            fp.write('read_limit = 1000\n')
            fp.write('write_limit = 200\n')

        # keep a original around
        conf_orig = config._Config(conf_file)

        # load the config
        conf = config._Config(conf_file)
        self.assertTrue(conf.get_udf_autosubscribe())
        # change it to False
        conf.set_udf_autosubscribe(False)
        self.assertFalse(conf.get_udf_autosubscribe())
        # save, load and check
        conf.save()
        conf_1 = config._Config(conf_file)
        self.assertFalse(conf_1.get_udf_autosubscribe())
        # change it to True
        conf.set_udf_autosubscribe(True)
        self.assertTrue(conf.get_udf_autosubscribe())
        # save, load and check
        conf.save()
        conf_1 = config._Config(conf_file)
        self.assertTrue(conf_1.get_udf_autosubscribe())

        # load the config, check the override of the value
        conf = config._Config(conf_file)
        self.assertTrue(conf.get_udf_autosubscribe())
        overridden_opts = [('__main__', 'udf_autosubscribe', False)]
        conf.override_options(overridden_opts)
        self.assertFalse(conf.get_udf_autosubscribe())
        self.assertNotEqual(conf.get_udf_autosubscribe(),
                            conf_orig.get_udf_autosubscribe())
        conf.save()
        conf_1 = config._Config(conf_file)
        self.assertEquals(True, conf_1.get_udf_autosubscribe())

    def test_load_share_autosubscribe(self):
        """Test load/set/override of share_autosubscribe config value."""
        conf_file = os.path.join(self.test_root,
                                 'test_share_autosubscribe_config.conf')
        # write some throttling values to the config file
        with open(conf_file, 'w') as fp:
            fp.write('[__main__]\n')
            fp.write('share_autosubscribe = True\n')

        # keep a original around
        conf_orig = config._Config(conf_file)

        # load the config
        conf = config._Config(conf_file)
        self.assertTrue(conf.get_share_autosubscribe())
        # change it to False
        conf.set_share_autosubscribe(False)
        self.assertFalse(conf.get_share_autosubscribe())
        # save, load and check
        conf.save()
        conf_1 = config._Config(conf_file)
        self.assertFalse(conf_1.get_share_autosubscribe())
        # change it to True
        conf.set_share_autosubscribe(True)
        self.assertTrue(conf.get_share_autosubscribe())
        # save, load and check
        conf.save()
        conf_1 = config._Config(conf_file)
        self.assertTrue(conf_1.get_share_autosubscribe())

        # load the config, check the override of the value
        conf = config._Config(conf_file)
        self.assertTrue(conf.get_share_autosubscribe())
        overridden_opts = [('__main__', 'share_autosubscribe', False)]
        conf.override_options(overridden_opts)
        self.assertFalse(conf.get_share_autosubscribe())
        self.assertNotEqual(conf.get_share_autosubscribe(),
                            conf_orig.get_share_autosubscribe())
        conf.save()
        conf_1 = config._Config(conf_file)
        self.assertEquals(True, conf_1.get_share_autosubscribe())

    def test_load_autoconnect(self):
        """Test load/set/override of autoconnect config value."""
        conf_file = os.path.join(self.test_root,
                                 'test_autoconnect_config.conf')
        # ensure that autoconnect is True
        with open(conf_file, 'w') as fp:
            fp.write('[__main__]\n')
            fp.write('autoconnect = True\n')

        # keep a original around
        conf_orig = config._Config(conf_file)

        # assert default is correct
        self.assertTrue(conf_orig.get_autoconnect(),
                        'autoconnect is True by default.')

        # load the config
        conf = config._Config(conf_file)
        self.assertTrue(conf.get_autoconnect())

        # change it to False
        conf.set_autoconnect(False)
        self.assertFalse(conf.get_autoconnect())

        # save, load and check
        conf.save()
        conf_1 = config._Config(conf_file)
        self.assertFalse(conf_1.get_autoconnect())
        # change it to True
        conf.set_autoconnect(True)
        self.assertTrue(conf.get_autoconnect())
        # save, load and check
        conf.save()
        conf_1 = config._Config(conf_file)
        self.assertTrue(conf_1.get_autoconnect())

        # load the config, check the override of the value
        conf = config._Config(conf_file)
        self.assertTrue(conf.get_autoconnect())
        overridden_opts = [('__main__', 'autoconnect', False)]
        conf.override_options(overridden_opts)
        self.assertFalse(conf.get_autoconnect())
        self.assertNotEqual(conf.get_autoconnect(),
                            conf_orig.get_autoconnect())
        conf.save()
        conf_1 = config._Config(conf_file)
        self.assertEquals(True, conf_1.get_autoconnect())


class ConfigglueParsersTests(BaseTwistedTestCase):
    """Tests for our custom configglue parsers"""

    def test_throttling_limit_parser(self):
        """Test throttling_limit_parser"""
        good_value = '20480'
        unset_value = '-1'
        bad_value = 'hola'
        invalid_value = None
        zero_value = '0'
        parser = config.throttling_limit_parser
        self.assertEquals(20480, parser(good_value))
        self.assertEquals(None, parser(unset_value))
        self.assertRaises(ValueError, parser, bad_value)
        self.assertRaises(TypeError, parser, invalid_value)
        self.assertEquals(None, parser(zero_value))

    def test_log_level_parser(self):
        """Test log_level_parser"""
        good_value = 'INFO'
        bad_value = 'hola'
        invalid_value = None
        parser = config.log_level_parser
        self.assertEqual(logging.INFO, parser(good_value))
        self.assertEqual(logging.DEBUG, parser(bad_value))
        self.assertEqual(logging.DEBUG, parser(invalid_value))

    def test_home_dir_parser(self):
        """Test home_dir_parser"""
        good_value = '~/hola'
        bad_value = 'hola'
        invalid_value = None
        parser = config.home_dir_parser
        with environ('HOME', '/home/fake'):
            self.assertEquals('/home/fake/hola', parser(good_value))
        self.assertEquals('hola', parser(bad_value))
        self.assertRaises(AttributeError, parser, invalid_value)

    def test_xdg_cache_dir_parser(self):
        """Test xdg_cache_dir_parser"""
        good_value = 'hola'
        bad_value = '/hola'
        invalid_value = None
        parser = config.xdg_cache_dir_parser
        self.assertEquals(os.path.join(xdg_cache_home, 'hola'),
                          parser(good_value))
        self.assertEquals('/hola', parser(bad_value))
        self.assertRaises(AttributeError, parser, invalid_value)

    def test_xdg_data_dir_parser(self):
        """Test xdg_data_dir_parser"""
        good_value = 'hola'
        bad_value = '/hola'
        invalid_value = None
        parser = config.xdg_data_dir_parser
        self.assertEquals(os.path.join(xdg_data_home, 'hola'),
                          parser(good_value))
        self.assertEquals('/hola', parser(bad_value))
        self.assertRaises(AttributeError, parser, invalid_value)


class SyncDaemonConfigParserTests(BaseTwistedTestCase):
    """Tests for SyncDaemonConfigParser"""

    def setUp(self):
        BaseTwistedTestCase.setUp(self)
        self.test_root = self.mktemp()
        self.default_config = os.path.join(os.environ['ROOTDIR'], 'data',
                                           'syncdaemon.conf')
        self.logging_config = os.path.join(os.environ['ROOTDIR'], 'data',
                                           'logging.conf')
        self.cp = config.SyncDaemonConfigParser()
        self.cp.readfp(file(self.default_config))
        self.cp.readfp(file(self.logging_config))

    def test_log_level_old_config(self):
        """Test log_level upgrade hook."""
        conf_file = os.path.join(self.test_root, 'test_old_config.conf')
        # write some throttling values to the config file
        with open(conf_file, 'w') as fp:
            fp.write('[__main__]\n')
            fp.write('log_level = DEBUG\n')
        self.assertTrue(os.path.exists(conf_file))
        self.cp.read([conf_file])
        self.cp.parse_all()
        self.assertEquals(self.cp.get('logging', 'level').value, 10)

    def test_log_level_new_config(self):
        """Test log_level upgrade hook with new config"""
        conf_file = os.path.join(self.test_root, 'test_new_config.conf')
        # write some throttling values to the config file
        with open(conf_file, 'w') as fp:
            fp.write('[logging]\n')
            fp.write('level = DEBUG\n')
        self.assertTrue(os.path.exists(conf_file))
        self.cp.read([conf_file])
        self.cp.parse_all()
        self.assertEquals(self.cp.get('logging', 'level').value, 10)

    def test_log_level_old_and_new_config(self):
        """Test log_level upgrade hook with a mixed config."""
        conf_file = os.path.join(self.test_root,
                                 'test_old_and_new_config.conf')
        # write some throttling values to the config file (not default ones)
        with open(conf_file, 'w') as fp:
            fp.write('[__main__]\n')
            fp.write('log_level = NOTE\n')
            fp.write('[logging]\n')
            fp.write('level = ERROR\n')
        self.assertTrue(os.path.exists(conf_file))
        self.cp.read([conf_file])
        self.cp.parse_all()
        self.assertEquals(self.cp.get('logging', 'level').value, logging.ERROR)

    def test_old_default_config(self):
        """Test log_level upgrade hook with an old default config"""
        self.cp.read(config.get_config_files()[0])
        # fake an old config
        value = self.cp.get('logging', 'level.default')
        help = self.cp.get('logging', 'level.help')
        parser = self.cp.get('logging', 'level.parser')
        self.cp.set('__main__', 'log_level.default', value)
        self.cp.set('__main__', 'log_level.help', help)
        self.cp.set('__main__', 'log_level.parser', parser)
        self.cp.remove_option('logging', 'level.default')
        self.cp.remove_option('logging', 'level.help')
        self.cp.remove_option('logging', 'level.parser')
        # parse it
        self.cp.parse_all()
        new_value = self.cp.get('logging', 'level')
        self.assertEquals(new_value.value, new_value.parser(value))

    def test_add_upgrade_hook(self):
        """Test add_upgrade_hook method"""
        self.cp.add_upgrade_hook('foo', 'bar', lambda x: None)
        self.assertIn(('foo', 'bar'), self.cp.upgrade_hooks)
        # try to add the same upgrade_hook
        self.assertRaises(ValueError, self.cp.add_upgrade_hook, 'foo', 'bar',
                          lambda x: None)

    def test_ignore_one(self):
        """Test ignore files config, one regex."""
        conf_file = os.path.join(self.test_root, 'test_new_config.conf')
        with open(conf_file, 'w') as fp:
            fp.write('[__main__]\n')
            fp.write('ignore = .*\\.pyc\n')  # all .pyc files
        self.assertTrue(os.path.exists(conf_file))
        self.cp.read([conf_file])
        self.cp.parse_all()
        self.assertEquals(self.cp.get('__main__', 'ignore').value,
                          [r'.*\.pyc'])

    def test_ignore_two(self):
        """Test ignore files config, two regexes."""
        conf_file = os.path.join(self.test_root, 'test_new_config.conf')
        with open(conf_file, 'w') as fp:
            fp.write('[__main__]\n')
            fp.write('ignore = .*\\.pyc\n')  # all .pyc files
            fp.write('         .*\\.sw[opnx]\n')  # all gvim temp files
        self.assertTrue(os.path.exists(conf_file))
        self.cp.read([conf_file])
        self.cp.parse_all()
        self.assertEquals(self.cp.get('__main__', 'ignore').value,
                          ['.*\\.pyc', '.*\\.sw[opnx]'])