# -*- coding: utf-8; -*- import configparser import os 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 from wuttjamaican.app import AppHandler from wuttjamaican.testing import FileConfigTestCase class TestWuttaConfig(FileConfigTestCase): 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']) def test_get_app(self): config = conf.WuttaConfig() app = config.get_app() # make sure we get the true default handler class self.assertIsInstance(app, AppHandler) self.assertIs(type(app), AppHandler) 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 TestGetConfigPaths(FileConfigTestCase): def test_winsvc(self): myconf = self.write_file('my.conf', """ [wutta.config] winsvc.RattailFileMonitor = /path/to/other/file """) files = conf.get_config_paths(files=[myconf], winsvc='RattailFileMonitor') self.assertEqual(files, ['/path/to/other/file']) class TestMakeConfig(FileConfigTestCase): # 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 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)