From: Karl O. Pinc Date: Tue, 17 Nov 2020 20:01:10 +0000 (-0600) Subject: Move code out of __init__.py X-Git-Url: https://papio.biology.duke.edu/gitweb/?a=commitdiff_plain;h=a86b6416e792d604904a9911749d681969bbddcc;p=pgwui_server Move code out of __init__.py --- diff --git a/setup.py b/setup.py index ebda3db..a8ab059 100644 --- a/setup.py +++ b/setup.py @@ -181,7 +181,7 @@ setup( # Its configuration is manually coded. entry_points="""\ [paste.app_factory] - main = pgwui_server:main + main = pgwui_server.pgwui_server:main """ # }, ) diff --git a/src/pgwui_server/__init__.py b/src/pgwui_server/__init__.py index eb3b7c3..e69de29 100644 --- a/src/pgwui_server/__init__.py +++ b/src/pgwui_server/__init__.py @@ -1,201 +0,0 @@ -# Copyright (C) 2018, 2019, 2020 The Meme Factory, Inc. -# http://www.karlpinc.com/ - -# This file is part of PGWUI_Server. -# -# This program is free software: you can redistribute it and/or -# modify it under the terms of the GNU Affero General Public License -# as published by the Free Software Foundation, either version 3 of -# the License, or (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, but -# WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU -# Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public -# License along with this program. If not, see -# . -# - -# Karl O. Pinc - -'''Provide a way to configure PGWUI. -''' - -from pyramid.config import Configurator -import logging -import sys - -from . import exceptions as server_ex -from . import checkset -from pgwui_common import exceptions as common_ex -from pgwui_common import plugin - -# Constants - -# All the settings recognized by PGWUI -SETTINGS = set( - ['pg_host', - 'pg_port', - 'default_db', - 'dry_run', - 'route_prefix', - 'routes', - 'validate_hmac', - 'autoconfigure', - ]) - - -# Logging -log = logging.getLogger(__name__) - - -# Functions - -def dot_to_component_settings(settings, key, component): - '''Put a component's settings into its own dict, - adding to what's already there - ''' - comp_settings = settings['pgwui'].setdefault(component, dict()) - comp_settings.update(settings[key]) - del settings[key] - - -def component_setting_into_dict( - errors, component_checkers, key, settings, component): - '''Put a component's settings in its own dict and validate them - ''' - comp_settings = dot_to_component_settings(settings, key, component) - if component in component_checkers: - errors.extend( - component_checkers[component](comp_settings)) - - -def dot_to_dict(settings, key, new_key): - settings['pgwui'][new_key] = settings[key] - del settings[key] - - -def parse_assignments(lines): - '''Return a list of key/value tuples from the lines of a setting - ''' - result = [] - if isinstance(lines, str): - for line in lines.splitlines(): - if '=' in line: - key, val = line.split('=', 1) - result.append((key.rstrip(), val.lstrip())) - else: - for key, val in lines.items(): - result.append((key, val)) - return result - - -def setting_into_dict( - errors, components, component_checkers, key, settings): - '''Separate a pgwui setting into a dict on '.' chars; validate - component settings. - ''' - if key[:6] == 'pgwui.': - new_key = key[6:] - if new_key in components: - settings[key] = dict(parse_assignments(settings[key])) - component_setting_into_dict( - errors, component_checkers, key, settings, new_key) - else: - if new_key in SETTINGS: - dot_to_dict(settings, key, new_key) - else: - errors.append(common_ex.UnknownSettingKeyError(key)) - - -def dictify_settings(errors, settings, components): - '''Convert . in the pgwui settings to dict mappings, and validate - the result. - ''' - component_checkers = plugin.find_pgwui_check_settings() - settings.setdefault('pgwui', dict()) - for key in list(settings.keys()): - setting_into_dict( - errors, components, component_checkers, key, settings) - checkset.validate_setting_values(errors, settings) - checkset.validate_hmac(errors, settings) - - -def exit_reporting_errors(errors): - '''Report errors and exit - ''' - tagged = [(logging.ERROR, error) for error in errors] - tagged.append((logging.CRITICAL, server_ex.BadSettingsAbort())) - - for (level, error) in tagged: - log.log(level, error) - - for (level, error) in (tagged[0], tagged[-1]): - print(error, file=sys.stderr) # in case logging is broken - - sys.exit(1) - - -def exit_on_invalid_settings(settings, components): - '''Exit when settings don't validate - ''' - errors = [] - dictify_settings(errors, settings, components) - if errors: - exit_reporting_errors(errors) - - -def add_routes(config, settings): - '''Add routes found in pgwui.routes setting - ''' - pgwui_settings = settings['pgwui'] - if 'routes' in pgwui_settings: - routes = parse_assignments(pgwui_settings['routes']) - for name, route in routes: - config.add_route(name, route) - - -def autoconfigurable_components(settings, components): - '''Automatic pgwui component discovery - ''' - autoconfig = settings['pgwui'].get('autoconfigure', True) - if not autoconfig: - return [] - - if 'pyramid.include' in settings: - log.info(server_ex.AutoconfigureConflict()) - - return components - - -def apply_component_defaults(settings, components): - '''Apply component default settings to existing settings - ''' - components_to_config = autoconfigurable_components(settings, components) - - rp = settings['pgwui'].get('route_prefix') - with Configurator(settings=settings, route_prefix=rp) as config: - config.include('pgwui_common') - for component in components_to_config: - log.debug('Autoconfiguring PGWUI component: {}'.format(component)) - config.include(component) - add_routes(config, settings) - log.debug('Done autoconfiguring PGWUI components') - return config - - -def pgwui_server_config(settings): - '''Configure pyramid - ''' - components = plugin.find_pgwui_components() - exit_on_invalid_settings(settings, components) - return apply_component_defaults(settings, components) - - -def main(global_config, **settings): - '''Return a Pyramid WSGI application - ''' - config = pgwui_server_config(settings) - return config.make_wsgi_app() diff --git a/src/pgwui_server/pgwui_server.py b/src/pgwui_server/pgwui_server.py new file mode 100644 index 0000000..eb3b7c3 --- /dev/null +++ b/src/pgwui_server/pgwui_server.py @@ -0,0 +1,201 @@ +# Copyright (C) 2018, 2019, 2020 The Meme Factory, Inc. +# http://www.karlpinc.com/ + +# This file is part of PGWUI_Server. +# +# This program is free software: you can redistribute it and/or +# modify it under the terms of the GNU Affero General Public License +# as published by the Free Software Foundation, either version 3 of +# the License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, but +# WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public +# License along with this program. If not, see +# . +# + +# Karl O. Pinc + +'''Provide a way to configure PGWUI. +''' + +from pyramid.config import Configurator +import logging +import sys + +from . import exceptions as server_ex +from . import checkset +from pgwui_common import exceptions as common_ex +from pgwui_common import plugin + +# Constants + +# All the settings recognized by PGWUI +SETTINGS = set( + ['pg_host', + 'pg_port', + 'default_db', + 'dry_run', + 'route_prefix', + 'routes', + 'validate_hmac', + 'autoconfigure', + ]) + + +# Logging +log = logging.getLogger(__name__) + + +# Functions + +def dot_to_component_settings(settings, key, component): + '''Put a component's settings into its own dict, + adding to what's already there + ''' + comp_settings = settings['pgwui'].setdefault(component, dict()) + comp_settings.update(settings[key]) + del settings[key] + + +def component_setting_into_dict( + errors, component_checkers, key, settings, component): + '''Put a component's settings in its own dict and validate them + ''' + comp_settings = dot_to_component_settings(settings, key, component) + if component in component_checkers: + errors.extend( + component_checkers[component](comp_settings)) + + +def dot_to_dict(settings, key, new_key): + settings['pgwui'][new_key] = settings[key] + del settings[key] + + +def parse_assignments(lines): + '''Return a list of key/value tuples from the lines of a setting + ''' + result = [] + if isinstance(lines, str): + for line in lines.splitlines(): + if '=' in line: + key, val = line.split('=', 1) + result.append((key.rstrip(), val.lstrip())) + else: + for key, val in lines.items(): + result.append((key, val)) + return result + + +def setting_into_dict( + errors, components, component_checkers, key, settings): + '''Separate a pgwui setting into a dict on '.' chars; validate + component settings. + ''' + if key[:6] == 'pgwui.': + new_key = key[6:] + if new_key in components: + settings[key] = dict(parse_assignments(settings[key])) + component_setting_into_dict( + errors, component_checkers, key, settings, new_key) + else: + if new_key in SETTINGS: + dot_to_dict(settings, key, new_key) + else: + errors.append(common_ex.UnknownSettingKeyError(key)) + + +def dictify_settings(errors, settings, components): + '''Convert . in the pgwui settings to dict mappings, and validate + the result. + ''' + component_checkers = plugin.find_pgwui_check_settings() + settings.setdefault('pgwui', dict()) + for key in list(settings.keys()): + setting_into_dict( + errors, components, component_checkers, key, settings) + checkset.validate_setting_values(errors, settings) + checkset.validate_hmac(errors, settings) + + +def exit_reporting_errors(errors): + '''Report errors and exit + ''' + tagged = [(logging.ERROR, error) for error in errors] + tagged.append((logging.CRITICAL, server_ex.BadSettingsAbort())) + + for (level, error) in tagged: + log.log(level, error) + + for (level, error) in (tagged[0], tagged[-1]): + print(error, file=sys.stderr) # in case logging is broken + + sys.exit(1) + + +def exit_on_invalid_settings(settings, components): + '''Exit when settings don't validate + ''' + errors = [] + dictify_settings(errors, settings, components) + if errors: + exit_reporting_errors(errors) + + +def add_routes(config, settings): + '''Add routes found in pgwui.routes setting + ''' + pgwui_settings = settings['pgwui'] + if 'routes' in pgwui_settings: + routes = parse_assignments(pgwui_settings['routes']) + for name, route in routes: + config.add_route(name, route) + + +def autoconfigurable_components(settings, components): + '''Automatic pgwui component discovery + ''' + autoconfig = settings['pgwui'].get('autoconfigure', True) + if not autoconfig: + return [] + + if 'pyramid.include' in settings: + log.info(server_ex.AutoconfigureConflict()) + + return components + + +def apply_component_defaults(settings, components): + '''Apply component default settings to existing settings + ''' + components_to_config = autoconfigurable_components(settings, components) + + rp = settings['pgwui'].get('route_prefix') + with Configurator(settings=settings, route_prefix=rp) as config: + config.include('pgwui_common') + for component in components_to_config: + log.debug('Autoconfiguring PGWUI component: {}'.format(component)) + config.include(component) + add_routes(config, settings) + log.debug('Done autoconfiguring PGWUI components') + return config + + +def pgwui_server_config(settings): + '''Configure pyramid + ''' + components = plugin.find_pgwui_components() + exit_on_invalid_settings(settings, components) + return apply_component_defaults(settings, components) + + +def main(global_config, **settings): + '''Return a Pyramid WSGI application + ''' + config = pgwui_server_config(settings) + return config.make_wsgi_app() diff --git a/tests/test___init__.py b/tests/test___init__.py deleted file mode 100644 index 9b86c39..0000000 --- a/tests/test___init__.py +++ /dev/null @@ -1,523 +0,0 @@ -# Copyright (C) 2018, 2019, 2020 The Meme Factory, Inc. -# http://www.karlpinc.com/ - -# This file is part of PGWUI_Server. -# -# This program is free software: you can redistribute it and/or -# modify it under the terms of the GNU Affero General Public License -# as published by the Free Software Foundation, either version 3 of -# the License, or (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, but -# WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU -# Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public -# License along with this program. If not, see -# . -# - -# Karl O. Pinc - -import copy -import logging -import pytest -import sys - -import pyramid.testing - -import pgwui_common.exceptions as common_ex -import pgwui_common.plugin -# Use as a regular module, not a plugin, so lint checks work -from pgwui_testing import testing - -import pgwui_server.checkset -import pgwui_server.__init__ as pgwui_server_init - - -# Constants - -TEST_SETTINGS = { - 'pgwui.validate_hmac': 'False', - 'pgwui.dry_run': 'False', -} - - -# Use contextlib.AbstractContextManager for Python >= 3.6 -# (Or, better, use the magic mock maker that's not yet integrated.) -class MockConfigurator(): - def __init__(self, **kwargs): - pass - - def __enter__(self): - return self - - def __exit__(self, *args): - pass - - def make_wsgi_app(self): - return 'wsgi_app' - - def include(self, *args): - pass - - -mock_add_route = testing.instance_method_mock_fixture('add_route') - -mock_find_pgwui_components = testing.make_mock_fixture( - pgwui_common.plugin, 'find_pgwui_components') -mock_find_pgwui_check_settings = testing.make_mock_fixture( - pgwui_common.plugin, 'find_pgwui_check_settings') - -mock_validate_setting_values = testing.make_mock_fixture( - pgwui_server.checkset, 'validate_setting_values') -mock_validate_hmac = testing.make_mock_fixture( - pgwui_server.checkset, 'validate_hmac') - - -# Unit tests - -# dot_to_component_settings() - -def test_dot_to_component_settings_new(): - '''Adds a new dict and puts the settings in it - ''' - comp_settings = {'foo': 'foo', 'bar': 'bar'} - component = 'pgwui_component' - key = 'pgwui.' + component - settings = {'pgwui': {}, - key: comp_settings} - expected = {'pgwui': {component: comp_settings}} - - pgwui_server_init.dot_to_component_settings( - settings, key, 'pgwui_component') - - assert settings == expected - - -def test_dot_to_component_settings_old(): - '''Extends an existing dict in the settings - ''' - comp_settings = {'foo': 'foo', 'bar': 'bar'} - component = 'pgwui_component' - key = 'pgwui.' + component - settings = {'pgwui': {component: {'foo': 'bar', 'baz': 'baz'}}, - key: comp_settings} - expected = {'pgwui': - {component: {'foo': 'foo', 'bar': 'bar', 'baz': 'baz'}}} - - pgwui_server_init.dot_to_component_settings( - settings, key, 'pgwui_component') - - assert settings == expected - - -mock_dot_to_component_setting = testing.make_mock_fixture( - pgwui_server_init, 'dot_to_component_settings') - - -# component_setting_into_dict() - -def test_component_setting_into_dict_no_checker( - mock_dot_to_component_setting): - '''When there's no checker nothing is done - ''' - errors = [] - - pgwui_server_init.component_setting_into_dict( - errors, {}, 'pgwui.pgwui_component', None, 'pgwui_component') - - assert errors == [] - - -def test_component_setting_into_dict_checker( - mock_dot_to_component_setting): - '''When there's a checker its result is appended to the errors - ''' - errors = ['someerror'] - new_errors = ['new1', 'new2'] - expected = copy.deepcopy(errors) - expected.extend(new_errors) - - pgwui_server_init.component_setting_into_dict( - errors, {'pgwui_component': lambda settings: new_errors}, - 'pgwui.pgwui_component', None, 'pgwui_component') - - assert errors == expected - - -mock_component_setting_into_dict = testing.make_mock_fixture( - pgwui_server_init, 'component_setting_into_dict') - - -# dot_to_dict() - -def test_dot_to_dict(): - '''Removes pgwui.* settings, replaces them with a dict entry - ''' - settings = {'foo': 1, - 'pgwui': {}, - 'pgwui.abc': 'abc', - 'pgwui.def': 'def'} - expected = {'foo': 1, - 'pgwui': {'abc': 'abc'}, - 'pgwui.def': 'def'} - - pgwui_server_init.dot_to_dict(settings, 'pgwui.abc', 'abc') - - assert settings == expected - - -mock_dot_to_dict = testing.make_mock_fixture( - pgwui_server_init, 'dot_to_dict') - - -# parse_assignments() - -def test_parse_assignments_str(): - '''Returns key/value string tuples and ignores lines without an "="''' - lines = ('key1 = value1\n' # whitespace around = is ignored - '\n' - 'ignored\n' - 'key2=value2\n' # missing whitespace is fine - 'key3= value3=withequals\n' - ) - result = pgwui_server_init.parse_assignments(lines) - assert set(result) == set([('key1', 'value1'), - ('key2', 'value2'), - ('key3', 'value3=withequals')]) - - -def test_parse_assignments_dict(): - '''Returns key value tuples. - ''' - lines = {'key1': 'value1', - 'key2': 'value2', - } - result = pgwui_server_init.parse_assignments(lines) - assert set(result) == set([('key1', 'value1'), - ('key2', 'value2'), - ]) - - -mock_parse_assignments = testing.make_mock_fixture( - pgwui_server_init, 'parse_assignments') - - -# setting_into_dict() - -def test_setting_into_dict_unknown( - mock_parse_assignments, - mock_component_setting_into_dict, - mock_dot_to_dict): - '''No new errors when there's a non-pgwui setting''' - errors = [] - pgwui_server_init.setting_into_dict(errors, [], {}, 'foo', {}) - - assert errors == [] - - -def test_setting_into_dict_bad( - mock_parse_assignments, - mock_component_setting_into_dict, - mock_dot_to_dict): - '''Delivers an error on a bad pgwui setting''' - errors = [] - - pgwui_server_init.setting_into_dict( - errors, [], {}, 'pgwui.foo', {}) - - assert errors - assert isinstance(errors[0], common_ex.UnknownSettingKeyError) - - -def test_setting_into_dict_good( - mock_parse_assignments, - mock_component_setting_into_dict, - mock_dot_to_dict): - '''Calls dot_to_dict when a known pgwui setting is supplied''' - errors = [] - - pgwui_server_init.setting_into_dict( - errors, [], {}, 'pgwui.pg_host', {}) - - mock_dot_to_dict.assert_called_once() - assert errors == [] - - -def test_setting_into_dict_plugin_component( - mock_parse_assignments, - mock_component_setting_into_dict, - mock_dot_to_dict): - '''When a setting is for a component the setting is parsed and - moved into a dict - ''' - key = 'pgwui.pgwui_component' - settings = {key: None} - errors = [] - mock_parse_assignments.return_value = {} - - pgwui_server_init.setting_into_dict( - errors, ['pgwui_component'], {}, key, settings) - - mock_parse_assignments.assert_called_once() - mock_component_setting_into_dict.assert_called_once() - assert errors == [] - - -mock_setting_into_dict = testing.make_mock_fixture( - pgwui_server_init, 'setting_into_dict') - - -# dictify_settings() - -def test_dictify_settings(mock_find_pgwui_check_settings, - mock_setting_into_dict, - mock_validate_setting_values, - mock_validate_hmac): - '''Calls setting_into_dict() for each key in setting, - with the proper list of plugin components - ''' - settings = {'key1': 'value1', - 'key2': 'value2'} - components = ['pgwui_server'] - - errors = [] - pgwui_server_init.dictify_settings(errors, settings, components) - - assert mock_validate_setting_values.called - assert mock_validate_hmac.called - - assert mock_setting_into_dict.call_count == len(settings) - assert mock_setting_into_dict.call_args[0][1] == components - - -mock_dictify_settings = testing.make_mock_fixture( - pgwui_server_init, 'dictify_settings') - - -# exit_reporting_errors() - -@pytest.fixture -def assert_exit1(): - def run(): - - exit1_called = False - - def mock_exit(status): - nonlocal exit1_called - exit1_called = status == 1 - - return mock_exit - - assert exit1_called - - return run - - -def test_exit_reporting_errors_logged( - assert_exit1, monkeypatch, caplog, capsys): - '''All errors are logged at ERROR, and a extra one at CRITICAL - ''' - monkeypatch.setattr(sys, 'exit', assert_exit1()) - caplog.set_level(logging.INFO) - errors = ['one', 'two', 'three'] - pgwui_server_init.exit_reporting_errors(errors) - - logs = caplog.record_tuples - - assert len(logs) == 4 - - levels = [log[1] for log in logs] - for level in levels[:-1]: - assert level == logging.ERROR - assert levels[-1] == logging.CRITICAL - - -def test_exit_reporting_errors_printed( - assert_exit1, monkeypatch, capsys): - '''First and last (the extra) errors are printed on stderr - ''' - monkeypatch.setattr(sys, 'exit', assert_exit1()) - errors = ['one', 'two', 'three'] - pgwui_server_init.exit_reporting_errors(errors) - - (out, err) = capsys.readouterr() - errlines = err.split('\n')[:-1] - - assert out == '' - assert len(errlines) == 2 - assert errlines[0] == 'one' - assert errlines[1] != 'two' - assert errlines[1] != 'three' - - -mock_exit_reporting_errors = testing.make_mock_fixture( - pgwui_server_init, 'exit_reporting_errors') - - -# exit_on_invalid_settings() - -def test_exit_on_invalid_settings_invalid(monkeypatch, - mock_exit_reporting_errors): - '''Calls dictify_settings and exit_reporting_errors() when - setting is invalid - ''' - def mock_dictify_settings(errors, settings, components): - errors.append('error1') - - monkeypatch.setattr(pgwui_server_init, 'dictify_settings', - mock_dictify_settings) - - pgwui_server_init.exit_on_invalid_settings({}, []) - - assert mock_exit_reporting_errors.called - - -def test_exit_on_invalid_settings_valid(mock_dictify_settings): - '''Returns, without exiting, when all settings are valid - ''' - pgwui_server_init.exit_on_invalid_settings({}, []) - - assert True - - -mock_exit_on_invalid_settings = testing.make_mock_fixture( - pgwui_server_init, 'exit_on_invalid_settings') - - -# autoconfigurable_components() - -def test_autoconfiguable_components_no_autoconfig(): - '''When the settings have no pgwui.autoconfigure return an empty list - ''' - test_components = ['some', 'components'] - - result = pgwui_server_init.autoconfigurable_components( - {'pgwui': {'autoconfigure': False}}, test_components) - - assert result == [] - - -def test_autoconfigurable_components_log_info(caplog): - '''When pyramid.include is in the settings an INFO message is logged - ''' - caplog.set_level(logging.INFO) - - pgwui_server_init.autoconfigurable_components( - {'pgwui': {'autoconfigure': True}, - 'pyramid.include': None}, - []) - - logs = caplog.record_tuples - - assert len(logs) == 1 - - level = logs[0][1] - assert level == logging.INFO - - -def test_autoconfigurable_components_components_returned(): - '''The suppiled components are returned when autoconfigure is True - ''' - test_components = ['some', 'components'] - - result = pgwui_server_init.autoconfigurable_components( - {'pgwui': {'pgwui.autoconfigure': True}}, test_components) - - assert result == test_components - - -mock_autoconfigurable_components = testing.make_mock_fixture( - pgwui_server_init, 'autoconfigurable_components') - - -# add_routes() - -def test_add_routes_empty(mock_add_route): - '''When there is no pgwui.routes setting nothing gets added''' - with pyramid.testing.testConfig() as config: - mocked_add_route = mock_add_route(config) - pgwui_server_init.add_routes(config, {'pgwui': {}}) - - assert not mocked_add_route.called - - -def test_add_routes_notempty(mock_add_route, mock_parse_assignments): - '''When there is a pgwui.routes setting config.add_route() is called - for each route''' - test_routes = [('name1', 'route1'), - ('name2', 'route2')] - mock_parse_assignments.return_value = test_routes - with pyramid.testing.testConfig() as config: - mocked_add_route = mock_add_route(config) - pgwui_server_init.add_routes(config, {'pgwui': {'routes': ''}}) - - assert mocked_add_route.call_count == len(test_routes) - - -mock_add_routes = testing.make_mock_fixture( - pgwui_server_init, 'add_routes') - - -# apply_component_defaults() - - -def test_apply_component_defaults(monkeypatch, caplog, - mock_autoconfigurable_components, - mock_add_routes): - '''A configurator is returned, a debug log entry is made for - each autoconfigurable component - ''' - caplog.set_level(logging.DEBUG) - - mock_autoconfigurable_components.return_value = \ - ['pgwui_mock_component_name'] - monkeypatch.setattr(pgwui_server_init, 'Configurator', - MockConfigurator) - - result = pgwui_server_init.apply_component_defaults({'pgwui': {}}, []) - assert isinstance(result, MockConfigurator) - - logs = caplog.record_tuples - - assert len(logs) == 2 # One for the single autoconfig, one for finishing - - for log in logs: - level = log[1] - assert level == logging.DEBUG - - -mock_apply_component_defaults = testing.make_mock_fixture( - pgwui_server_init, 'apply_component_defaults') - - -# pgwui_server_config() - -def test_pgwui_server_config( - mock_find_pgwui_components, - mock_apply_component_defaults, mock_exit_on_invalid_settings): - '''Returns a configuration''' - test_configurator = 'test configurator' - mock_apply_component_defaults.return_value = test_configurator - - result = pgwui_server_init.pgwui_server_config({}) - - assert result == test_configurator - - -# main() -def test_main(monkeypatch): - '''Returns a wsgi app''' - monkeypatch.setattr(pgwui_server_init, 'pgwui_server_config', - lambda *args: MockConfigurator()) - - result = pgwui_server_init.main({}) - assert result == 'wsgi_app' - - -# Integration tests -def test_main_integrated(): - '''Does not raise errors or warnings''' - pgwui_server_init.main({}, **TEST_SETTINGS) diff --git a/tests/test_pgwui_server.py b/tests/test_pgwui_server.py new file mode 100644 index 0000000..2ba90a1 --- /dev/null +++ b/tests/test_pgwui_server.py @@ -0,0 +1,523 @@ +# Copyright (C) 2018, 2019, 2020 The Meme Factory, Inc. +# http://www.karlpinc.com/ + +# This file is part of PGWUI_Server. +# +# This program is free software: you can redistribute it and/or +# modify it under the terms of the GNU Affero General Public License +# as published by the Free Software Foundation, either version 3 of +# the License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, but +# WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public +# License along with this program. If not, see +# . +# + +# Karl O. Pinc + +import copy +import logging +import pytest +import sys + +import pyramid.testing + +import pgwui_common.exceptions as common_ex +import pgwui_common.plugin +# Use as a regular module, not a plugin, so lint checks work +from pgwui_testing import testing + +import pgwui_server.checkset +import pgwui_server.pgwui_server as pgwui_server + + +# Constants + +TEST_SETTINGS = { + 'pgwui.validate_hmac': 'False', + 'pgwui.dry_run': 'False', +} + + +# Use contextlib.AbstractContextManager for Python >= 3.6 +# (Or, better, use the magic mock maker that's not yet integrated.) +class MockConfigurator(): + def __init__(self, **kwargs): + pass + + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + def make_wsgi_app(self): + return 'wsgi_app' + + def include(self, *args): + pass + + +mock_add_route = testing.instance_method_mock_fixture('add_route') + +mock_find_pgwui_components = testing.make_mock_fixture( + pgwui_common.plugin, 'find_pgwui_components') +mock_find_pgwui_check_settings = testing.make_mock_fixture( + pgwui_common.plugin, 'find_pgwui_check_settings') + +mock_validate_setting_values = testing.make_mock_fixture( + pgwui_server.checkset, 'validate_setting_values') +mock_validate_hmac = testing.make_mock_fixture( + pgwui_server.checkset, 'validate_hmac') + + +# Unit tests + +# dot_to_component_settings() + +def test_dot_to_component_settings_new(): + '''Adds a new dict and puts the settings in it + ''' + comp_settings = {'foo': 'foo', 'bar': 'bar'} + component = 'pgwui_component' + key = 'pgwui.' + component + settings = {'pgwui': {}, + key: comp_settings} + expected = {'pgwui': {component: comp_settings}} + + pgwui_server.dot_to_component_settings( + settings, key, 'pgwui_component') + + assert settings == expected + + +def test_dot_to_component_settings_old(): + '''Extends an existing dict in the settings + ''' + comp_settings = {'foo': 'foo', 'bar': 'bar'} + component = 'pgwui_component' + key = 'pgwui.' + component + settings = {'pgwui': {component: {'foo': 'bar', 'baz': 'baz'}}, + key: comp_settings} + expected = {'pgwui': + {component: {'foo': 'foo', 'bar': 'bar', 'baz': 'baz'}}} + + pgwui_server.dot_to_component_settings( + settings, key, 'pgwui_component') + + assert settings == expected + + +mock_dot_to_component_setting = testing.make_mock_fixture( + pgwui_server, 'dot_to_component_settings') + + +# component_setting_into_dict() + +def test_component_setting_into_dict_no_checker( + mock_dot_to_component_setting): + '''When there's no checker nothing is done + ''' + errors = [] + + pgwui_server.component_setting_into_dict( + errors, {}, 'pgwui.pgwui_component', None, 'pgwui_component') + + assert errors == [] + + +def test_component_setting_into_dict_checker( + mock_dot_to_component_setting): + '''When there's a checker its result is appended to the errors + ''' + errors = ['someerror'] + new_errors = ['new1', 'new2'] + expected = copy.deepcopy(errors) + expected.extend(new_errors) + + pgwui_server.component_setting_into_dict( + errors, {'pgwui_component': lambda settings: new_errors}, + 'pgwui.pgwui_component', None, 'pgwui_component') + + assert errors == expected + + +mock_component_setting_into_dict = testing.make_mock_fixture( + pgwui_server, 'component_setting_into_dict') + + +# dot_to_dict() + +def test_dot_to_dict(): + '''Removes pgwui.* settings, replaces them with a dict entry + ''' + settings = {'foo': 1, + 'pgwui': {}, + 'pgwui.abc': 'abc', + 'pgwui.def': 'def'} + expected = {'foo': 1, + 'pgwui': {'abc': 'abc'}, + 'pgwui.def': 'def'} + + pgwui_server.dot_to_dict(settings, 'pgwui.abc', 'abc') + + assert settings == expected + + +mock_dot_to_dict = testing.make_mock_fixture( + pgwui_server, 'dot_to_dict') + + +# parse_assignments() + +def test_parse_assignments_str(): + '''Returns key/value string tuples and ignores lines without an "="''' + lines = ('key1 = value1\n' # whitespace around = is ignored + '\n' + 'ignored\n' + 'key2=value2\n' # missing whitespace is fine + 'key3= value3=withequals\n' + ) + result = pgwui_server.parse_assignments(lines) + assert set(result) == set([('key1', 'value1'), + ('key2', 'value2'), + ('key3', 'value3=withequals')]) + + +def test_parse_assignments_dict(): + '''Returns key value tuples. + ''' + lines = {'key1': 'value1', + 'key2': 'value2', + } + result = pgwui_server.parse_assignments(lines) + assert set(result) == set([('key1', 'value1'), + ('key2', 'value2'), + ]) + + +mock_parse_assignments = testing.make_mock_fixture( + pgwui_server, 'parse_assignments') + + +# setting_into_dict() + +def test_setting_into_dict_unknown( + mock_parse_assignments, + mock_component_setting_into_dict, + mock_dot_to_dict): + '''No new errors when there's a non-pgwui setting''' + errors = [] + pgwui_server.setting_into_dict(errors, [], {}, 'foo', {}) + + assert errors == [] + + +def test_setting_into_dict_bad( + mock_parse_assignments, + mock_component_setting_into_dict, + mock_dot_to_dict): + '''Delivers an error on a bad pgwui setting''' + errors = [] + + pgwui_server.setting_into_dict( + errors, [], {}, 'pgwui.foo', {}) + + assert errors + assert isinstance(errors[0], common_ex.UnknownSettingKeyError) + + +def test_setting_into_dict_good( + mock_parse_assignments, + mock_component_setting_into_dict, + mock_dot_to_dict): + '''Calls dot_to_dict when a known pgwui setting is supplied''' + errors = [] + + pgwui_server.setting_into_dict( + errors, [], {}, 'pgwui.pg_host', {}) + + mock_dot_to_dict.assert_called_once() + assert errors == [] + + +def test_setting_into_dict_plugin_component( + mock_parse_assignments, + mock_component_setting_into_dict, + mock_dot_to_dict): + '''When a setting is for a component the setting is parsed and + moved into a dict + ''' + key = 'pgwui.pgwui_component' + settings = {key: None} + errors = [] + mock_parse_assignments.return_value = {} + + pgwui_server.setting_into_dict( + errors, ['pgwui_component'], {}, key, settings) + + mock_parse_assignments.assert_called_once() + mock_component_setting_into_dict.assert_called_once() + assert errors == [] + + +mock_setting_into_dict = testing.make_mock_fixture( + pgwui_server, 'setting_into_dict') + + +# dictify_settings() + +def test_dictify_settings(mock_find_pgwui_check_settings, + mock_setting_into_dict, + mock_validate_setting_values, + mock_validate_hmac): + '''Calls setting_into_dict() for each key in setting, + with the proper list of plugin components + ''' + settings = {'key1': 'value1', + 'key2': 'value2'} + components = ['pgwui_server'] + + errors = [] + pgwui_server.dictify_settings(errors, settings, components) + + assert mock_validate_setting_values.called + assert mock_validate_hmac.called + + assert mock_setting_into_dict.call_count == len(settings) + assert mock_setting_into_dict.call_args[0][1] == components + + +mock_dictify_settings = testing.make_mock_fixture( + pgwui_server, 'dictify_settings') + + +# exit_reporting_errors() + +@pytest.fixture +def assert_exit1(): + def run(): + + exit1_called = False + + def mock_exit(status): + nonlocal exit1_called + exit1_called = status == 1 + + return mock_exit + + assert exit1_called + + return run + + +def test_exit_reporting_errors_logged( + assert_exit1, monkeypatch, caplog, capsys): + '''All errors are logged at ERROR, and a extra one at CRITICAL + ''' + monkeypatch.setattr(sys, 'exit', assert_exit1()) + caplog.set_level(logging.INFO) + errors = ['one', 'two', 'three'] + pgwui_server.exit_reporting_errors(errors) + + logs = caplog.record_tuples + + assert len(logs) == 4 + + levels = [log[1] for log in logs] + for level in levels[:-1]: + assert level == logging.ERROR + assert levels[-1] == logging.CRITICAL + + +def test_exit_reporting_errors_printed( + assert_exit1, monkeypatch, capsys): + '''First and last (the extra) errors are printed on stderr + ''' + monkeypatch.setattr(sys, 'exit', assert_exit1()) + errors = ['one', 'two', 'three'] + pgwui_server.exit_reporting_errors(errors) + + (out, err) = capsys.readouterr() + errlines = err.split('\n')[:-1] + + assert out == '' + assert len(errlines) == 2 + assert errlines[0] == 'one' + assert errlines[1] != 'two' + assert errlines[1] != 'three' + + +mock_exit_reporting_errors = testing.make_mock_fixture( + pgwui_server, 'exit_reporting_errors') + + +# exit_on_invalid_settings() + +def test_exit_on_invalid_settings_invalid(monkeypatch, + mock_exit_reporting_errors): + '''Calls dictify_settings and exit_reporting_errors() when + setting is invalid + ''' + def mock_dictify_settings(errors, settings, components): + errors.append('error1') + + monkeypatch.setattr(pgwui_server, 'dictify_settings', + mock_dictify_settings) + + pgwui_server.exit_on_invalid_settings({}, []) + + assert mock_exit_reporting_errors.called + + +def test_exit_on_invalid_settings_valid(mock_dictify_settings): + '''Returns, without exiting, when all settings are valid + ''' + pgwui_server.exit_on_invalid_settings({}, []) + + assert True + + +mock_exit_on_invalid_settings = testing.make_mock_fixture( + pgwui_server, 'exit_on_invalid_settings') + + +# autoconfigurable_components() + +def test_autoconfiguable_components_no_autoconfig(): + '''When the settings have no pgwui.autoconfigure return an empty list + ''' + test_components = ['some', 'components'] + + result = pgwui_server.autoconfigurable_components( + {'pgwui': {'autoconfigure': False}}, test_components) + + assert result == [] + + +def test_autoconfigurable_components_log_info(caplog): + '''When pyramid.include is in the settings an INFO message is logged + ''' + caplog.set_level(logging.INFO) + + pgwui_server.autoconfigurable_components( + {'pgwui': {'autoconfigure': True}, + 'pyramid.include': None}, + []) + + logs = caplog.record_tuples + + assert len(logs) == 1 + + level = logs[0][1] + assert level == logging.INFO + + +def test_autoconfigurable_components_components_returned(): + '''The suppiled components are returned when autoconfigure is True + ''' + test_components = ['some', 'components'] + + result = pgwui_server.autoconfigurable_components( + {'pgwui': {'pgwui.autoconfigure': True}}, test_components) + + assert result == test_components + + +mock_autoconfigurable_components = testing.make_mock_fixture( + pgwui_server, 'autoconfigurable_components') + + +# add_routes() + +def test_add_routes_empty(mock_add_route): + '''When there is no pgwui.routes setting nothing gets added''' + with pyramid.testing.testConfig() as config: + mocked_add_route = mock_add_route(config) + pgwui_server.add_routes(config, {'pgwui': {}}) + + assert not mocked_add_route.called + + +def test_add_routes_notempty(mock_add_route, mock_parse_assignments): + '''When there is a pgwui.routes setting config.add_route() is called + for each route''' + test_routes = [('name1', 'route1'), + ('name2', 'route2')] + mock_parse_assignments.return_value = test_routes + with pyramid.testing.testConfig() as config: + mocked_add_route = mock_add_route(config) + pgwui_server.add_routes(config, {'pgwui': {'routes': ''}}) + + assert mocked_add_route.call_count == len(test_routes) + + +mock_add_routes = testing.make_mock_fixture( + pgwui_server, 'add_routes') + + +# apply_component_defaults() + + +def test_apply_component_defaults(monkeypatch, caplog, + mock_autoconfigurable_components, + mock_add_routes): + '''A configurator is returned, a debug log entry is made for + each autoconfigurable component + ''' + caplog.set_level(logging.DEBUG) + + mock_autoconfigurable_components.return_value = \ + ['pgwui_mock_component_name'] + monkeypatch.setattr(pgwui_server, 'Configurator', + MockConfigurator) + + result = pgwui_server.apply_component_defaults({'pgwui': {}}, []) + assert isinstance(result, MockConfigurator) + + logs = caplog.record_tuples + + assert len(logs) == 2 # One for the single autoconfig, one for finishing + + for log in logs: + level = log[1] + assert level == logging.DEBUG + + +mock_apply_component_defaults = testing.make_mock_fixture( + pgwui_server, 'apply_component_defaults') + + +# pgwui_server_config() + +def test_pgwui_server_config( + mock_find_pgwui_components, + mock_apply_component_defaults, mock_exit_on_invalid_settings): + '''Returns a configuration''' + test_configurator = 'test configurator' + mock_apply_component_defaults.return_value = test_configurator + + result = pgwui_server.pgwui_server_config({}) + + assert result == test_configurator + + +# main() +def test_main(monkeypatch): + '''Returns a wsgi app''' + monkeypatch.setattr(pgwui_server, 'pgwui_server_config', + lambda *args: MockConfigurator()) + + result = pgwui_server.main({}) + assert result == 'wsgi_app' + + +# Integration tests +def test_main_integrated(): + '''Does not raise errors or warnings''' + pgwui_server.main({}, **TEST_SETTINGS)