wuttjamaican/tests/test_conf.py

# -*- coding: utf-8; -*-

import configparser
import os
import shutil
import tempfile
from unittest import TestCase
from unittest.mock import patch, MagicMock

import sqlalchemy as sa

from wuttjamaican import conf
from wuttjamaican.exc import ConfigurationError
from wuttjamaican.db import Session


class TestWuttaConfig(TestCase):

    def setUp(self):
        self.tempdir = tempfile.mkdtemp()

    def tearDown(self):
        shutil.rmtree(self.tempdir)

    def write_file(self, filename, content):
        path = os.path.join(self.tempdir, filename)
        with open(path, 'wt') as f:
            f.write(content)
        return path

    def test_contstructor_basic(self):
        config = conf.WuttaConfig()
        self.assertEqual(config.appname, 'wutta')
        self.assertEqual(config.files_read, [])

    def test_constructor_valid_files(self):
        myfile = self.write_file('my.conf', '')
        config = conf.WuttaConfig(files=[myfile])
        self.assertEqual(len(config.files_read), 1)
        self.assertEqual(config.files_read[0], myfile)

    def test_constructor_missing_files(self):
        invalid = os.path.join(self.tempdir, 'invalid.conf')
        self.assertRaises(FileNotFoundError, conf.WuttaConfig, files=[invalid])

    def test_constructor_required_files_are_present(self):
        first = self.write_file('first.conf', """\
[foo]
bar = 1
baz = A
""")

        second = self.write_file('second.conf', """\
[wutta.config]
require = %(here)s/first.conf

[foo]
baz = B
""")

        config = conf.WuttaConfig(files=[second])
        self.assertEqual(len(config.files_read), 2)
        # nb. files_read listing is in order of "priority" which is
        # same the as order in which files were initially read
        self.assertEqual(config.files_read[0], second)
        self.assertEqual(config.files_read[1], first)
        self.assertEqual(config.get('foo.bar'), '1')
        self.assertEqual(config.get('foo.baz'), 'B')

    def test_constructor_required_files_are_missing(self):
        second = self.write_file('second.conf', """\
[wutta.config]
require = %(here)s/first.conf

[foo]
baz = B
""")

        self.assertRaises(FileNotFoundError, conf.WuttaConfig, files=[second])

    def test_constructor_included_files_are_present(self):
        first = self.write_file('first.conf', """\
[foo]
bar = 1
baz = A
""")

        second = self.write_file('second.conf', """\
[wutta.config]
include = %(here)s/first.conf

[foo]
baz = B
""")

        config = conf.WuttaConfig(files=[second])
        self.assertEqual(len(config.files_read), 2)
        # nb. files_read listing is in order of "priority" which is
        # same the as order in which files were initially read
        self.assertEqual(config.files_read[0], second)
        self.assertEqual(config.files_read[1], first)
        self.assertEqual(config.get('foo.bar'), '1')
        self.assertEqual(config.get('foo.baz'), 'B')

    def test_constructor_included_files_are_missing(self):
        second = self.write_file('second.conf', """\
[wutta.config]
include = %(here)s/first.conf

[foo]
baz = B
""")

        config = conf.WuttaConfig(files=[second])
        self.assertEqual(len(config.files_read), 1)
        self.assertEqual(config.files_read[0], second)
        self.assertIsNone(config.get('foo.bar'))
        self.assertEqual(config.get('foo.baz'), 'B')

    def test_prioritized_files(self):
        first = self.write_file('first.conf', """\
[foo]
bar = 1
""")

        second = self.write_file('second.conf', """\
[wutta.config]
require = %(here)s/first.conf
""")

        config = conf.WuttaConfig(files=[second])
        files = config.get_prioritized_files()
        self.assertEqual(len(files), 2)
        self.assertEqual(files[0], second)
        self.assertEqual(files[1], first)

    def test_constructor_defaults(self):
        config = conf.WuttaConfig()
        self.assertEqual(config.defaults, {})
        self.assertIsNone(config.get('foo'))

        config = conf.WuttaConfig(defaults={'foo': 'bar'})
        self.assertEqual(config.defaults, {'foo': 'bar'})
        self.assertEqual(config.get('foo'), 'bar')

    def test_constructor_db_flags(self):
        myfile = self.write_file('my.conf', """\
[wutta.config]
usedb = true
preferdb = true
""")

        # flags are off by default
        config = conf.WuttaConfig()
        self.assertFalse(config.usedb)
        self.assertFalse(config.preferdb)

        # but may override via constructor
        config = conf.WuttaConfig(usedb=True, preferdb=True)
        self.assertTrue(config.usedb)
        self.assertTrue(config.preferdb)

        # and also may override via config file
        config = conf.WuttaConfig(files=[myfile])
        self.assertTrue(config.usedb)
        self.assertTrue(config.preferdb)

    def test_constructor_db_not_supported(self):

        # flags are off by default
        config = conf.WuttaConfig()
        self.assertFalse(config.usedb)
        self.assertFalse(config.preferdb)

        # but caller may enable the flags (if sqlalchemy available)
        config = conf.WuttaConfig(usedb=True, preferdb=True)
        self.assertTrue(config.usedb)
        self.assertTrue(config.preferdb)

        # but db flags are force-disabled if sqlalchemy not available,
        # regardless of flag values caller provides...

        orig_import = __import__

        def mock_import(name, *args, **kwargs):
            if name == 'db':
                raise ImportError
            return orig_import(name, *args, **kwargs)

        with patch('builtins.__import__', side_effect=mock_import):
            config = conf.WuttaConfig(usedb=True, preferdb=True)
            self.assertFalse(config.usedb)
            self.assertFalse(config.preferdb)

    def test_constructor_may_configure_logging(self):
        myfile = self.write_file('my.conf', """\
[wutta.config]
configure_logging = true
""")

        with patch.object(conf.WuttaConfig, '_configure_logging') as method:

            # no logging config by default
            config = conf.WuttaConfig()
            method.assert_not_called()

            # but may override via constructor
            method.reset_mock()
            config = conf.WuttaConfig(configure_logging=True)
            method.assert_called_once()

            # and also may override via config file
            method.reset_mock()
            config = conf.WuttaConfig(files=[myfile])
            method.assert_called_once()

    def test_constructor_configures_logging(self):
        myfile = self.write_file('my.conf', """\
[wutta]
timezone.default = America/Chicago

[wutta.config]
configure_logging = true
""")

        with patch('wuttjamaican.conf.logging') as logging:

            # basic constructor attempts logging config
            config = conf.WuttaConfig(configure_logging=True)
            logging.config.fileConfig.assert_called_once()

            # if logging config fails, error is *not* raised
            logging.config.fileConfig.reset_mock()
            logging.config.fileConfig.side_effect = configparser.NoSectionError('logging')
            config = conf.WuttaConfig(configure_logging=True)
            logging.config.fileConfig.assert_called_once()

            # and it works if we specify config file
            logging.config.fileConfig.reset_mock()
            config = conf.WuttaConfig(files=[myfile])
            logging.config.fileConfig.assert_called_once()

    def test_setdefault(self):
        config = conf.WuttaConfig()

        # value is empty by default
        self.assertIsNone(config.get('foo'))

        # but we can change that by setting default
        config.setdefault('foo', 'bar')
        self.assertEqual(config.get('foo'), 'bar')

        # also, value is returned when we set default
        self.assertIsNone(config.get('baz'))
        self.assertEqual(config.setdefault('baz', 'blarg'), 'blarg')

    def test_get_require_with_default(self):
        config = conf.WuttaConfig()
        self.assertRaises(ValueError, config.get, 'foo', require=True, default='bar')

    def test_get_require_missing(self):
        config = conf.WuttaConfig()
        self.assertRaises(ConfigurationError, config.get, 'foo', require=True)

    def test_get_with_default(self):
        config = conf.WuttaConfig()
        # nb. returns None if no default specified
        self.assertIsNone(config.get('foo'))
        self.assertEqual(config.get('foo', default='bar'), 'bar')

    def test_get_from_db(self):
        # minimal config, but at least it needs db cxn info
        config = conf.WuttaConfig(defaults={'wutta.db.default.url': 'sqlite://'})

        session = Session()

        # setup table for testing
        session.execute(sa.text("""
        create table setting (
                name varchar(255) primary key,
                value text
        );
        """))
        session.commit()

        # setting not yet defined
        self.assertIsNone(config.get_from_db('foo'))

        # insert setting value to db
        session.execute(sa.text("insert into setting values ('foo', 'bar')"))
        session.commit()

        # now setting returns a value
        self.assertEqual(config.get_from_db('foo'), 'bar')

        # also works if we provide the session
        self.assertEqual(config.get_from_db('foo', session=session), 'bar')

        session.close()

    def test_get_default(self):
        config = conf.WuttaConfig()
        self.assertIsNone(config.get('foo'))
        self.assertEqual(config.get('foo', default='bar'), 'bar')

    def test_get_require(self):
        config = conf.WuttaConfig()
        self.assertIsNone(config.get('foo'))
        self.assertRaises(ConfigurationError, config.get, 'foo', require=True)

    def test_get_require_message(self):
        config = conf.WuttaConfig()
        self.assertIsNone(config.get('foo'))
        try:
            config.get('foo', require=True, message="makin stuff up")
        except ConfigurationError as error:
            self.assertIn("makin stuff up", str(error))

    def test_get_preferdb(self):

        # start out with a default value
        config = conf.WuttaConfig(defaults={'wutta.db.default.url': 'sqlite://',
                                            'foo': 'bar'})
        self.assertEqual(config.get('foo'), 'bar')

        session = Session()

        # setup table for testing
        session.execute(sa.text("""
        create table setting (
                name varchar(255) primary key,
                value text
        );
        """))
        session.execute(sa.text("insert into setting values ('foo', 'baz')"))
        session.commit()

        # we did not specify usedb=True, so original default is still returned
        self.assertFalse(config.usedb)
        self.assertEqual(config.get('foo'), 'bar')

        # usedb but no preferdb means original default is still returned
        self.assertEqual(config.get('foo', usedb=True), 'bar')

        # but preferdb should mean newer db value is returned
        self.assertEqual(config.get('foo', usedb=True, preferdb=True), 'baz')

        # try a different key to ensure db fallback works if no default present
        session.execute(sa.text("insert into setting values ('blarg', 'blitz')"))
        session.commit()
        self.assertIsNone(config.get('blarg'))
        self.assertEqual(config.get('blarg', usedb=True), 'blitz')

        session.close()

    def test_get_ambiguous(self):
        config = conf.WuttaConfig()

        # value is returned if key is not ambiguous
        config.setdefault('foo', 'bar')
        self.assertEqual(config.get('foo'), 'bar')

        # but None is returned if key is ambiguous
        config.setdefault('foo.bar', 'baz')
        self.assertIsNone(config.get('foo'))

    def test_require(self):
        config = conf.WuttaConfig()
        self.assertRaises(ConfigurationError, config.require, 'foo')

    def test_get_bool(self):
        config = conf.WuttaConfig()
        self.assertFalse(config.get_bool('foo.bar'))
        config.setdefault('foo.bar', 'true')
        self.assertTrue(config.get_bool('foo.bar'))

    def test_get_int(self):
        config = conf.WuttaConfig()
        self.assertIsNone(config.get_int('foo.bar'))
        config.setdefault('foo.bar', '42')
        self.assertEqual(config.get_int('foo.bar'), 42)

    def test_get_list(self):
        config = conf.WuttaConfig()
        self.assertEqual(config.get_list('foo.bar'), [])
        config.setdefault('foo.bar', 'hello world')
        self.assertEqual(config.get_list('foo.bar'), ['hello', 'world'])


class TestGenericDefaultFiles(TestCase):

    def test_linux(self):
        files = conf.generic_default_files('wuttatest')
        self.assertIsInstance(files, list)
        self.assertTrue(len(files) > 1)
        self.assertIn('/etc/wuttatest.conf', files)

    def test_win32(self):
        win32com = MagicMock()
        win32com.shell.SHGetSpecialFolderPath.return_value = r'C:' + os.sep
        with patch.dict('sys.modules', **{'win32com.shell': win32com}):
            with patch('wuttjamaican.conf.sys', platform='win32'):

                files = conf.generic_default_files('wuttatest')
                self.assertIsInstance(files, list)
                self.assertTrue(len(files) > 1)
                self.assertIn(os.path.join('C:', 'wuttatest.conf'), files)

    def test_win32_broken(self):
        orig_import = __import__

        def mock_import(name, *args, **kwargs):
            if name == 'win32com.shell':
                raise ImportError
            return orig_import(name, *args, **kwargs)

        with patch('builtins.__import__', side_effect=mock_import):
            with patch('wuttjamaican.conf.sys', platform='win32'):

                files = conf.generic_default_files('wuttatest')
                self.assertIsInstance(files, list)
                self.assertEqual(len(files), 0)


class TestMakeConfig(TestCase):

    # nb. we use appname='wuttatest' in this suite to avoid any
    # "valid" default config files, env vars etc. which may be present
    # on the dev machine

    def setUp(self):
        self.tempdir = tempfile.mkdtemp()

    def tearDown(self):
        shutil.rmtree(self.tempdir)

    def write_file(self, filename, content):
        path = os.path.join(self.tempdir, filename)
        with open(path, 'wt') as f:
            f.write(content)
        return path

    def test_generic_default_files(self):
        generic = self.write_file('generic.conf', '')

        with patch('wuttjamaican.conf.generic_default_files') as generic_default_files:
            with patch('wuttjamaican.conf.WuttaConfig') as WuttaConfig:

                # generic files are used if nothing is specified
                generic_default_files.return_value = [generic]
                config = conf.make_config(appname='wuttatest')
                generic_default_files.assert_called_once_with('wuttatest')
                WuttaConfig.assert_called_once_with([generic], appname='wuttatest',
                                                    usedb=None, preferdb=None)

                # make sure empty defaults works too
                generic_default_files.reset_mock()
                generic_default_files.return_value = []
                WuttaConfig.reset_mock()
                config = conf.make_config(appname='wuttatest')
                generic_default_files.assert_called_once_with('wuttatest')
                WuttaConfig.assert_called_once_with([], appname='wuttatest',
                                                    usedb=None, preferdb=None)

    def test_specify_default_files(self):
        generic = self.write_file('generic.conf', '')
        myfile = self.write_file('my.conf', '')

        with patch('wuttjamaican.conf.generic_default_files') as generic_default_files:
            with patch('wuttjamaican.conf.WuttaConfig') as WuttaConfig:

                # generic defaults are used if nothing specified
                generic_default_files.return_value = [generic]
                config = conf.make_config(appname='wuttatest')
                generic_default_files.assert_called_once_with('wuttatest')
                WuttaConfig.assert_called_once_with([generic], appname='wuttatest',
                                                    usedb=None, preferdb=None)

                # can specify single default file
                generic_default_files.reset_mock()
                WuttaConfig.reset_mock()
                config = conf.make_config(appname='wuttatest', default_files=myfile)
                generic_default_files.assert_not_called()
                WuttaConfig.assert_called_once_with([myfile], appname='wuttatest',
                                                    usedb=None, preferdb=None)

                # can specify default files as list
                generic_default_files.reset_mock()
                WuttaConfig.reset_mock()
                config = conf.make_config(appname='wuttatest', default_files=[myfile])
                generic_default_files.assert_not_called()
                WuttaConfig.assert_called_once_with([myfile], appname='wuttatest',
                                                    usedb=None, preferdb=None)

                # can specify default files as callable
                generic_default_files.reset_mock()
                WuttaConfig.reset_mock()
                config = conf.make_config(appname='wuttatest', default_files=lambda appname: [myfile])
                generic_default_files.assert_not_called()
                WuttaConfig.assert_called_once_with([myfile], appname='wuttatest',
                                                    usedb=None, preferdb=None)

    def test_specify_plus_files(self):
        generic = self.write_file('generic.conf', '')
        myfile = self.write_file('my.conf', '')

        with patch('wuttjamaican.conf.generic_default_files') as generic_default_files:
            with patch('wuttjamaican.conf.WuttaConfig') as WuttaConfig:

                generic_default_files.return_value = [generic]

                # no plus files by default
                config = conf.make_config(appname='wuttatest')
                generic_default_files.assert_called_once_with('wuttatest')
                WuttaConfig.assert_called_once_with([generic], appname='wuttatest',
                                                    usedb=None, preferdb=None)

                # can specify single plus file
                generic_default_files.reset_mock()
                WuttaConfig.reset_mock()
                config = conf.make_config(appname='wuttatest', plus_files=myfile)
                generic_default_files.assert_called_once_with('wuttatest')
                WuttaConfig.assert_called_once_with([generic, myfile], appname='wuttatest',
                                                    usedb=None, preferdb=None)

                # can specify plus files as list
                generic_default_files.reset_mock()
                WuttaConfig.reset_mock()
                config = conf.make_config(appname='wuttatest', plus_files=[myfile])
                generic_default_files.assert_called_once_with('wuttatest')
                WuttaConfig.assert_called_once_with([generic, myfile], appname='wuttatest',
                                                    usedb=None, preferdb=None)

                # can specify plus files via env
                generic_default_files.reset_mock()
                WuttaConfig.reset_mock()
                config = conf.make_config(appname='wuttatest',
                                          env={'WUTTATEST_CONFIG_PLUS_FILES': myfile})
                generic_default_files.assert_called_once_with('wuttatest')
                WuttaConfig.assert_called_once_with([generic, myfile], appname='wuttatest',
                                                    usedb=None, preferdb=None)

    def test_specify_primary_files(self):
        generic = self.write_file('generic.conf', '')
        myfile = self.write_file('my.conf', '')

        with patch('wuttjamaican.conf.generic_default_files') as generic_default_files:
            with patch('wuttjamaican.conf.WuttaConfig') as WuttaConfig:

                generic_default_files.return_value = [generic]

                # generic files by default
                config = conf.make_config(appname='wuttatest')
                generic_default_files.assert_called_once_with('wuttatest')
                WuttaConfig.assert_called_once_with([generic], appname='wuttatest',
                                                    usedb=None, preferdb=None)

                # can specify single primary file (nb. no default files)
                generic_default_files.reset_mock()
                WuttaConfig.reset_mock()
                config = conf.make_config(myfile, appname='wuttatest')
                generic_default_files.assert_not_called()
                WuttaConfig.assert_called_once_with([myfile], appname='wuttatest',
                                                    usedb=None, preferdb=None)

                # can specify primary files as list
                generic_default_files.reset_mock()
                WuttaConfig.reset_mock()
                config = conf.make_config([myfile], appname='wuttatest')
                generic_default_files.assert_not_called()
                WuttaConfig.assert_called_once_with([myfile], appname='wuttatest',
                                                    usedb=None, preferdb=None)

                # can specify primary files via env
                generic_default_files.reset_mock()
                WuttaConfig.reset_mock()
                config = conf.make_config(appname='wuttatest',
                                          env={'WUTTATEST_CONFIG_FILES': myfile})
                generic_default_files.assert_not_called()
                WuttaConfig.assert_called_once_with([myfile], appname='wuttatest',
                                                    usedb=None, preferdb=None)

    def test_extensions(self):
        generic = self.write_file('generic.conf', '')
        myfile = self.write_file('my.conf', '')

        with patch('wuttjamaican.conf.WuttaConfig') as WuttaConfig:
            with patch('wuttjamaican.conf.load_entry_points') as load_entry_points:

                # no entry points loaded if extend=False
                config = conf.make_config(appname='wuttatest', extend=False)
                WuttaConfig.assert_called_once_with([], appname='wuttatest',
                                                    usedb=None, preferdb=None)
                load_entry_points.assert_not_called()

                # confirm entry points for default appname
                load_entry_points.reset_mock()
                WuttaConfig.reset_mock()
                config = conf.make_config(appname='wutta')
                WuttaConfig.assert_called_once_with([], appname='wutta',
                                                    usedb=None, preferdb=None)
                load_entry_points.assert_called_once_with('wutta.config.extensions')

                # confirm entry points for custom appname
                load_entry_points.reset_mock()
                WuttaConfig.reset_mock()
                config = conf.make_config(appname='wuttatest')
                WuttaConfig.assert_called_once_with([], appname='wuttatest',
                                                    usedb=None, preferdb=None)
                load_entry_points.assert_called_once_with('wuttatest.config.extensions')

                # confirm extensions are invoked
                load_entry_points.reset_mock()
                foo_obj = MagicMock()
                foo_cls = MagicMock(return_value=foo_obj)
                load_entry_points.return_value = {'foo': foo_cls}
                WuttaConfig.reset_mock()
                testconfig = MagicMock()
                WuttaConfig.return_value = testconfig
                config = conf.make_config(appname='wuttatest')
                WuttaConfig.assert_called_once_with([], appname='wuttatest',
                                                    usedb=None, preferdb=None)
                load_entry_points.assert_called_once_with('wuttatest.config.extensions')
                foo_cls.assert_called_once_with()
                foo_obj.configure.assert_called_once_with(testconfig)