# Its configuration is manually coded.
entry_points="""\
[paste.app_factory]
- main = pgwui_server:main
+ main = pgwui_server.pgwui_server:main
"""
# },
)
-# 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
-# <http://www.gnu.org/licenses/>.
-#
-
-# Karl O. Pinc <kop@karlpinc.com>
-
-'''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()
--- /dev/null
+# 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
+# <http://www.gnu.org/licenses/>.
+#
+
+# Karl O. Pinc <kop@karlpinc.com>
+
+'''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()
+++ /dev/null
-# 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
-# <http://www.gnu.org/licenses/>.
-#
-
-# Karl O. Pinc <kop@karlpinc.com>
-
-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)
--- /dev/null
+# 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
+# <http://www.gnu.org/licenses/>.
+#
+
+# Karl O. Pinc <kop@karlpinc.com>
+
+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)
projects / pgwui_server / commitdiff