import gc import os import platform import tempfile import time import sys import functools import pytest from testing.testifycompat import ( assert_equal, assert_raises, mock, ) from staticconf import ( config, errors, proxy, schema, testing, validation, ) import staticconf class TestRemoveByKeys: def test_empty_dict(self): keys = range(3) assert_equal([], config.remove_by_keys({}, keys)) def test_no_keys(self): keys = [] testMap = dict(enumerate(range(3))) assert_equal(list(testMap.items()), config.remove_by_keys(testMap, keys)) def test_overlap(self): keys = [1, 3, 5, 7] testMap = dict(enumerate(range(8))) expected = [(0, 0), (2, 2), (4, 4), (6, 6)] assert_equal(expected, config.remove_by_keys(testMap, keys)) class TestConfigMap: @pytest.fixture(autouse=True) def setup_config_map(self): self.config_map = config.ConfigMap(one=1, three=3, seven=7) def test_no_iteritems(self): assert not hasattr(self.config_map, 'iteritems') def test_getitem(self): assert_equal(self.config_map['one'], 1) assert_equal(self.config_map['seven'], 7) def test_get(self): assert_equal(self.config_map.get('three'), 3) assert_equal(self.config_map.get('four', 0), 0) def test_contains(self): assert 'one' in self.config_map assert 'two' not in self.config_map def test_len(self): assert_equal(len(self.config_map), 3) class TestConfigurationNamespace: @pytest.fixture(autouse=True) def setup_namespace(self): self.name = 'the_name' self.namespace = config.ConfigNamespace(self.name) self.config_data = dict(enumerate(['one', 'two', 'three'], 1)) def test_register_get_value_proxies(self): proxies = [mock.Mock(), mock.Mock()] for mock_proxy in proxies: self.namespace.register_proxy(mock_proxy) assert_equal(self.namespace.get_value_proxies(), proxies) @pytest.mark.skipif( 'PyPy' in platform.python_implementation(), reason="Fails on PyPy", ) def test_get_value_proxies_does_not_contain_out_of_scope_proxies(self): assert not self.namespace.get_value_proxies() def a_scope(): mock_proxy = mock.create_autospec(proxy.ValueProxy) self.namespace.register_proxy(mock_proxy) a_scope() a_scope() gc.collect() assert_equal(len(self.namespace.get_value_proxies()), 0) def test_update_values(self): values = dict(one=1, two=2) self.namespace.update_values(values) assert 'one' in self.namespace assert 'two' in self.namespace def test_get_config_values(self): self.namespace['stars'] = 'foo' values = self.namespace.get_config_values() assert_equal(values, {'stars': 'foo'}) def test_get_config_dict(self): self.namespace['one.two.three.four'] = 5 self.namespace['one.two.three.five'] = 'six' self.namespace['one.b.cats'] = [1, 2, 3] self.namespace['a.two'] = 'c' self.namespace['first'] = True d = self.namespace.get_config_dict() assert_equal(d, { 'one': { 'b': { 'cats': [1, 2, 3], }, 'two': { 'three': { 'four': 5, 'five': 'six', }, }, }, 'a': { 'two': 'c', }, 'first': True, }) def test_get_known_keys(self): proxies = [mock.Mock(), mock.Mock()] for mock_proxy in proxies: self.namespace.register_proxy(mock_proxy) expected = {mock_proxy.config_key for mock_proxy in proxies} assert_equal(self.namespace.get_known_keys(), expected) def test_validate_keys_no_unknown_keys(self): proxies = [mock.Mock(config_key=i) for i in self.config_data] for mock_proxy in proxies: self.namespace.register_proxy(mock_proxy) with mock.patch('staticconf.config.log') as mock_log: self.namespace.validate_keys(self.config_data, True) self.namespace.validate_keys(self.config_data, False) assert not mock_log.warn.mock_calls def test_validate_keys_unknown_log(self): with mock.patch('staticconf.config.log') as mock_log: self.namespace.validate_keys(self.config_data, False) assert_equal(len(mock_log.info.mock_calls), 1) def test_validate_keys_unknown_log_keys_only(self): with mock.patch('staticconf.config.log') as mock_log: self.namespace.validate_keys( self.config_data, False, log_keys_only=True, ) assert_equal(len(mock_log.info.mock_calls), 1) log_msg = mock_log.info.call_args[0][0] unknown = config.remove_by_keys( self.config_data, self.namespace.get_known_keys(), ) for k, v in unknown: # Have to cast to strings here, since log_msg is a string key_string, val_string = str(k), str(v) assert key_string in log_msg assert val_string not in log_msg def test_validate_keys_unknown_raise(self): assert_raises(errors.ConfigurationError, self.namespace.validate_keys, self.config_data, True) def test_clear(self): self.namespace.apply_config_data(self.config_data, False, False) assert self.namespace.get_config_values() self.namespace.clear() assert_equal(self.namespace.get_config_values(), {}) class TestGetNamespace: @pytest.fixture(autouse=True) def mock_namespaces(self): with mock.patch.dict(config.configuration_namespaces): yield def test_get_namespace_new(self): name = 'some_unlikely_name' assert name not in config.configuration_namespaces config.get_namespace(name) assert name in config.configuration_namespaces def test_get_namespace_existing(self): name = 'the_common_name' namespace = config.get_namespace(name) assert_equal(namespace, config.get_namespace(name)) class TestReload: @pytest.fixture(autouse=True) def mock_namespaces(self): with mock.patch.dict(config.configuration_namespaces): yield def test_reload_default(self): staticconf.DictConfiguration(dict(one='three', seven='nine')) one, seven = staticconf.get('one'), staticconf.get('seven') staticconf.DictConfiguration(dict(one='ten', seven='el')) staticconf.reload() assert_equal(one, 'ten') assert_equal(seven, 'el') def test_reload_all(self): name = 'another_one' staticconf.DictConfiguration(dict(one='three')) staticconf.DictConfiguration(dict(two='three'), namespace=name) one, two = staticconf.get('one'), staticconf.get('two', namespace=name) # access the values to set the value_proxy cache one.value, two.value staticconf.DictConfiguration(dict(one='four')) staticconf.DictConfiguration(dict(two='five'), namespace=name) staticconf.reload(all_names=True) assert_equal(one, 'four') assert_equal(two, 'five') def test_reload_single(self): name = 'another_one' staticconf.DictConfiguration(dict(one='three')) staticconf.DictConfiguration(dict(two='three'), namespace=name) one, two = staticconf.get('one'), staticconf.get('two', namespace=name) # access the values to set the value_proxy cache one.value, two.value staticconf.DictConfiguration(dict(one='four')) staticconf.DictConfiguration(dict(two='five'), namespace=name) staticconf.reload() assert_equal(one, 'four') assert_equal(two, 'three') class TestValidateConfig: @pytest.fixture(autouse=True) def patch_config(self): with mock.patch.dict(config.configuration_namespaces, clear=True): with testing.MockConfiguration(): yield def test_validate_single_passes(self): staticconf.DictConfiguration({}) config.validate() _ = staticconf.get_string('one.two') # noqa: F841 staticconf.DictConfiguration({'one.two': 'nice'}) config.validate() def test_validate_single_fails(self): _ = staticconf.get_int('one.two') # noqa: F841 assert_raises(errors.ConfigurationError, config.validate) def test_validate_all_passes(self): name = 'yan' staticconf.DictConfiguration({}, namespace=name) staticconf.DictConfiguration({}) config.validate(all_names=True) staticconf.get_string('one.two') staticconf.get_string('foo', namespace=name) staticconf.DictConfiguration({'one.two': 'nice'}) staticconf.DictConfiguration({'foo': 'nice'}, namespace=name) config.validate(all_names=True) def test_validate_all_fails(self): name = 'yan' _ = staticconf.get_string('foo', namespace=name) # noqa: F841 assert_raises(errors.ConfigurationError, config.validate, all_names=True) def test_validate_value_token(self): class ExampleSchema(schema.Schema): namespace = 'DEFAULT' thing = schema.int() assert_raises(errors.ConfigurationError, config.validate, all_names=True) class TestConfigHelp: @pytest.fixture(autouse=True) def setup_config_help(self): self.config_help = config.ConfigHelp() self.config_help.add('one', validation.validate_any, None, 'DEFAULT', "the one") self.config_help.add('when', validation.validate_time, 'NOW', 'DEFAULT', "The time") self.config_help.add('you sure', validation.validate_bool, 'No', 'DEFAULT', "Are you?") self.config_help.add('one', validation.validate_any, None, 'Beta', "the one") self.config_help.add('one', validation.validate_any, None, 'Alpha', "the one") self.config_help.add('two', validation.validate_any, None, 'Alpha', "the two") self.lines = self.config_help.view_help().split('\n') def test_view_help_format(self): line, help = self.lines[4:6] assert_equal(help, 'The time') assert_equal(line, 'when (Type: time, Default: NOW)') def test_view_help_format_namespace(self): namespace, one, _, two, _, blank = self.lines[9:15] assert_equal(namespace, 'Namespace: Alpha') assert one.startswith('one') assert two.startswith('two') assert_equal(blank, '') def test_view_help_namespace_sort(self): lines = list(filter(lambda l: l.startswith('Namespace'), self.lines)) expected = ['Namespace: DEFAULT', 'Namespace: Alpha', 'Namespace: Beta'] assert_equal(lines, expected) class TestHasDuplicateKeys: @pytest.fixture(autouse=True) def setup_base_conf(self): self.base_conf = {'fear': 'is_the', 'mind': 'killer'} def test_has_dupliacte_keys_false(self): config_data = dict(unique_keys=123) assert not config.has_duplicate_keys(config_data, self.base_conf, True) assert not config.has_duplicate_keys(config_data, self.base_conf, False) def test_has_duplicate_keys_raises(self): config_data = dict(fear=123) assert_raises( errors.ConfigurationError, config.has_duplicate_keys, config_data, self.base_conf, True) def test_has_duplicate_keys_no_raise(self): config_data = dict(mind=123) assert config.has_duplicate_keys(config_data, self.base_conf, False) class TestConfigurationWatcher: @pytest.fixture(autouse=True) def setup_mocks_and_config_watcher(self): self.loader = mock.Mock() with mock.patch('staticconf.config.time') as self.mock_time: with mock.patch('staticconf.config.os.stat') as self.mock_stat: with tempfile.NamedTemporaryFile() as file: with mock.patch( 'staticconf.config.os.path.getmtime', ) as self.mock_getmtime: file.flush() self.mtime = 234 self.mock_getmtime.return_value = self.mtime self.mock_stat.return_value.st_ino = 1 self.mock_stat.return_value.st_dev = 2 self.filename = file.name self.watcher = config.ConfigurationWatcher( self.loader, self.filename) yield def test_get_filename_list_from_string(self): with mock.patch('staticconf.config.os.path.abspath') as mock_path_abspath: mock_path_abspath.side_effect = lambda p: p filename = 'thefilename.yaml' filenames = self.watcher.get_filename_list(filename) assert_equal(filenames, [filename]) def test_get_filename_list_from_list(self): with mock.patch('staticconf.config.os.path.abspath') as mock_path_abspath: mock_path_abspath.side_effect = lambda p: p filenames = ['b', 'g', 'z', 'a'] expected = ['a', 'b', 'g', 'z'] assert_equal(self.watcher.get_filename_list(filenames), expected) def test_should_check(self): self.watcher.last_check = 123456789 self.mock_time.time.return_value = 123456789 # Still current, but no min_interval assert self.watcher.should_check # With max interval self.watcher.min_interval = 3 assert not self.watcher.should_check # Time has passed self.mock_time.time.return_value = 123456794 assert self.watcher.should_check def test_file_modified_not_modified(self): self.mock_time.time.return_value = 123460 assert not self.watcher.file_modified() assert_equal(self.watcher.last_check, self.mock_time.time.return_value) def test_file_modified(self): self.watcher.comparators[0].last_max_mtime = 123456 self.mock_getmtime.return_value = 123460 assert self.watcher.file_modified() assert_equal(self.watcher.last_check, self.mock_time.time.return_value) def test_reload_default(self): self.watcher.reload() self.loader.assert_called_with() def test_reload_custom(self): reloader = mock.Mock() watcher = config.ConfigurationWatcher( self.loader, self.filename, reloader=reloader) watcher.reload() reloader.assert_called_with() class TestInodeComparator: def test_get_inodes_empty(self): comparator = config.InodeComparator([]) assert comparator.get_inodes() == [] @mock.patch('staticconf.config.os.stat', autospec=True) def test_get_inodes(self, mock_stat): comparator = config.InodeComparator(['./one.file']) inodes = comparator.get_inodes() expected = [(mock_stat.return_value.st_dev, mock_stat.return_value.st_ino)] assert_equal(inodes, expected) class TestMTimeComparator: @mock.patch('staticconf.config.os.path.getmtime', autospec=True, return_value=1) def test_no_change(self, mock_mtime): comparator = config.MTimeComparator(['./one.file']) assert not comparator.has_changed() assert not comparator.has_changed() @mock.patch( 'staticconf.config.os.path.getmtime', autospec=True, side_effect=[0, 1, 1, 2], ) def test_changes(self, mock_mtime): comparator = config.MTimeComparator(['./one.file']) assert comparator.has_changed() assert not comparator.has_changed() assert comparator.has_changed() @mock.patch( 'staticconf.config.os.path.getmtime', autospec=True, side_effect=[1, 2, 1], ) def test_change_when_newer_time_before_older_time(self, mock_mtime): comparator = config.MTimeComparator(['./one.file']) # 1 -> 2 assert comparator.has_changed() # 2 -> 1 (can happen as a result of a revert) assert comparator.has_changed() class TestMTimeComparatorWithCompareFunc: @pytest.fixture(autouse=True) def setup_comparator(self): self._LoggingMTimeComparator = functools.partial( config.MTimeComparator, compare_func=config.build_compare_func(self._err_logger)) @pytest.fixture(autouse=True) def _reset_err_logger(self): self._err_filename = None self._exc_info = (None, None, None) def _err_logger(self, filename): self._err_filename = filename self._exc_info = sys.exc_info() def test_logs_error(self): self._LoggingMTimeComparator(['./not.a.file']) assert self._err_filename == "./not.a.file" assert all(x is not None for x in self._exc_info) def test_get_most_recent_empty(self): self._LoggingMTimeComparator([]) assert self._err_filename is None assert all(x is None for x in self._exc_info) @mock.patch('staticconf.config.os.path.getmtime', autospec=True, return_value=1) def test_no_change(self, mock_mtime): comparator = self._LoggingMTimeComparator(['./one.file']) assert not comparator.has_changed() assert not comparator.has_changed() assert self._err_filename is None assert all(x is None for x in self._exc_info) @mock.patch( 'staticconf.config.os.path.getmtime', autospec=True, side_effect=[0, 1, 1, 2], ) def test_changes(self, mock_mtime): comparator = self._LoggingMTimeComparator(['./one.file']) assert comparator.has_changed() assert not comparator.has_changed() assert comparator.has_changed() assert self._err_filename is None assert all(x is None for x in self._exc_info) class TestMD5Comparator: @pytest.fixture() def comparator(self): self.original_contents = b"abcdefghijkabcd" with tempfile.NamedTemporaryFile() as self.file: self.write_contents(self.original_contents) yield config.MD5Comparator([self.file.name]) def write_contents(self, contents): self.file.seek(0) self.file.write(contents) self.file.flush() def test_get_hashes_empty(self): comparator = config.MD5Comparator([]) assert comparator.get_hashes() == [] def test_has_changed_no_changes(self, comparator): assert not comparator.has_changed() self.write_contents(self.original_contents) assert not comparator.has_changed() def test_has_changed_with_changes(self, comparator): assert not comparator.has_changed() self.write_contents(b"this is the new content") assert comparator.has_changed() class TestReloadCallbackChain: @pytest.fixture(autouse=True) def setup_callback_chain(self): self.callbacks = list(enumerate([mock.Mock(), mock.Mock()])) self.callback_chain = config.ReloadCallbackChain(callbacks=self.callbacks) def test_init_with_callbacks(self): assert_equal(self.callback_chain.callbacks, dict(self.callbacks)) def test_add_remove(self): callback = mock.Mock() self.callback_chain.add('one', callback) assert_equal(self.callback_chain.callbacks['one'], callback) self.callback_chain.remove('one') assert 'one' not in self.callback_chain.callbacks def test_call(self): self.callback_chain.namespace = 'the_namespace' with mock.patch('staticconf.config.reload') as mock_reload: self.callback_chain() for _, callback in self.callbacks: callback.assert_called_with() mock_reload.assert_called_with(name='the_namespace', all_names=False) class TestConfigFacade: @pytest.fixture(autouse=True) def setup_facade(self): self.mock_watcher = mock.create_autospec(config.ConfigurationWatcher) self.mock_watcher.get_reloader.return_value = mock.create_autospec( config.ReloadCallbackChain) self.facade = config.ConfigFacade(self.mock_watcher) def test_load(self): filename, namespace = "filename", "namespace" loader = mock.Mock() with mock.patch( 'staticconf.config.ConfigurationWatcher', autospec=True) as mock_watcher_class: facade = config.ConfigFacade.load(filename, namespace, loader) facade.watcher.load_config.assert_called_with() assert_equal(facade.watcher, mock_watcher_class.return_value) reloader = facade.callback_chain assert_equal(reloader, facade.watcher.get_reloader()) def test_load_passes_comparators_to_configuration_watcher(self): filename, namespace = "filename", "namespace" loader = mock.Mock() comparator = mock.Mock(name='MockComparator') with mock.patch( 'staticconf.config.ConfigurationWatcher', autospec=True ) as mock_watcher_class: config.ConfigFacade.load( filename, namespace, loader, comparators=[comparator], ) mock_watcher_class.assert_called_with( mock.ANY, filename, min_interval=mock.ANY, reloader=mock.ANY, comparators=[comparator], ) def test_add_callback(self): name, func = 'name', mock.Mock() self.facade.add_callback(name, func) self.facade.callback_chain.add.assert_called_with(name, func) def test_reload_if_changed(self): self.facade.reload_if_changed() self.mock_watcher.reload_if_changed.assert_called_with(force=False) @pytest.mark.acceptance class TestConfigFacadeAcceptance: @pytest.fixture(autouse=True) def setup_env(self): self.file = tempfile.NamedTemporaryFile() self.write(b"one: A") def write(self, content, mtime_seconds=0): time.sleep(0.03) self.file.file.seek(0) self.file.write(content) self.file.flush() tstamp = time.time() + mtime_seconds os.utime(self.file.name, (tstamp, tstamp)) @pytest.fixture(autouse=True) def patch_namespace(self): self.namespace = 'testing_namespace' with testing.MockConfiguration(namespace=self.namespace): yield def test_load_end_to_end(self): loader = staticconf.YamlConfiguration callback = mock.Mock() facade = staticconf.ConfigFacade.load(self.file.name, self.namespace, loader) facade.add_callback('one', callback) assert_equal(staticconf.get('one', namespace=self.namespace), "A") self.write(b"one: B", 10) facade.reload_if_changed() assert_equal(staticconf.get('one', namespace=self.namespace), "B") callback.assert_called_with() def test_reload_end_to_end(self): loader = mock.Mock() facade = staticconf.ConfigFacade.load( self.file.name, self.namespace, loader) assert_equal(loader.call_count, 1) time.sleep(1) facade.reload_if_changed() assert_equal(loader.call_count, 1) os.utime(self.file.name, None) facade.reload_if_changed() assert_equal(loader.call_count, 2) class TestBuildLoaderCallable: @pytest.fixture(autouse=True) def patch_namespace(self): self.namespace = 'the_namespace' patcher = mock.patch('staticconf.config.get_namespace', autospec=True) with patcher as self.mock_get_namespace: yield def test_build_loader_callable(self): load_func, filename = mock.Mock(), mock.Mock() loader_callable = config.build_loader_callable( load_func, filename, self.namespace) result = loader_callable() self.mock_get_namespace.assert_called_with(self.namespace) mock_namespace = self.mock_get_namespace.return_value mock_namespace.clear.assert_called_with() load_func.assert_called_with(filename, namespace=self.namespace) assert_equal(result, load_func.return_value)