2
0
Fork 0
wuttjamaican/tests/test_conf.py
Lance Edgar b458272207 Add make_engine_from_config() method for AppHandler
and other misc. tweaks needed to get this incorporated into Rattail
2023-11-19 20:36:51 -06:00

626 lines
24 KiB
Python

# -*- 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)