From fd7e073f0ecbf9ead795ef866108a69e0825ef48 Mon Sep 17 00:00:00 2001 From: Dan Callaghan Date: May 16 2018 05:14:26 +0000 Subject: run fedmsg consumers with a Flask app context This lets us refer to current_app.config throughout the code, instead of awkwardly having this thing where a piece of code can use current_app.config if it was called through a Flask handler but otherwise has to pass around an explicit config object. The 'greenwave_cache' setting from the fedmsg config is dropped, because we can just re-use the existing 'CACHE' setting from the Greenwave config instead. --- diff --git a/fedmsg.d/config.py b/fedmsg.d/config.py index b0aa370..664262d 100644 --- a/fedmsg.d/config.py +++ b/fedmsg.d/config.py @@ -129,9 +129,4 @@ config = dict( # Greenwave API url greenwave_api_url='https://greenwave.domain.local/api/v1.0', - - # In production, these details should match the details of the frontend's - # CACHE configuration, so that the backend and frontend can manipulate the - # same shared store. - greenwave_cache={'backend': 'dogpile.cache.null'}, ) diff --git a/functional-tests/conftest.py b/functional-tests/conftest.py index 5d5c499..f7a3480 100644 --- a/functional-tests/conftest.py +++ b/functional-tests/conftest.py @@ -139,30 +139,27 @@ def distgit_server(tmpdir_factory): p.wait() -# This is only a fixture because some tests want to point the fedmsg consumers -# at the same cache that the server process is using. -# I would like to refactor those tests to send real messages to real consumers, -# so this becomes unnecessary. -@pytest.fixture(scope='session') -def greenwave_cache_config(tmpdir_factory): - cache_file = tmpdir_factory.mktemp('greenwave-cache').join('cache.dbm') - return { - 'backend': 'dogpile.cache.dbm', - 'expiration_time': 300, - 'arguments': {'filename': cache_file.strpath}, - } - - @pytest.yield_fixture(scope='session') -def greenwave_server(tmpdir_factory, resultsdb_server, waiverdb_server, greenwave_cache_config): +def greenwave_server(tmpdir_factory, resultsdb_server, waiverdb_server): if 'GREENWAVE_TEST_URL' in os.environ: yield os.environ['GREENWAVE_TEST_URL'] else: # Start Greenwave as a subprocess + cache_file = tmpdir_factory.mktemp('greenwave').join('cache.dbm') settings_file = tmpdir_factory.mktemp('greenwave').join('settings.py') settings_file.write(textwrap.dedent("""\ - CACHE = %r - """ % greenwave_cache_config)) + CACHE = { + 'backend': 'dogpile.cache.dbm', + 'expiration_time': 300, + 'arguments': {'filename': %r}, + } + """ % cache_file.strpath)) + + # We also update the config file for *this* process, as well as the server subprocess, + # because the fedmsg consumer tests actually invoke the handler code in-process. + # This way they will see the same config as the server. + os.environ['GREENWAVE_CONFIG'] = settings_file.strpath + env = dict(os.environ, PYTHONPATH='.', GREENWAVE_CONFIG=settings_file.strpath) diff --git a/functional-tests/consumers/test_resultsdb.py b/functional-tests/consumers/test_resultsdb.py index d8b4ab3..d028352 100644 --- a/functional-tests/consumers/test_resultsdb.py +++ b/functional-tests/consumers/test_resultsdb.py @@ -37,7 +37,6 @@ def test_consume_new_result( hub.config = { 'environment': 'environment', 'topic_prefix': 'topic_prefix', - 'greenwave_cache': {'backend': 'dogpile.cache.null'}, } handler = resultsdb.ResultsDBHandler(hub) assert handler.topic == ['topic_prefix.environment.taskotron.result.new'] @@ -159,7 +158,6 @@ def test_no_message_for_unchanged_decision( hub.config = { 'environment': 'environment', 'topic_prefix': 'topic_prefix', - 'greenwave_cache': {'backend': 'dogpile.cache.null'}, } handler = resultsdb.ResultsDBHandler(hub) assert handler.topic == ['topic_prefix.environment.taskotron.result.new'] @@ -199,7 +197,6 @@ def test_invalidate_new_result_with_mocked_cache( hub.config = { 'environment': 'environment', 'topic_prefix': 'topic_prefix', - 'greenwave_cache': {'backend': 'dogpile.cache.memory'}, } handler = resultsdb.ResultsDBHandler(hub) handler.cache = mock.MagicMock() @@ -218,7 +215,7 @@ def test_invalidate_new_result_with_mocked_cache( @mock.patch('greenwave.consumers.resultsdb.fedmsg.publish') def test_invalidate_new_result_with_real_cache( mock_fedmsg, load_config, requests_session, greenwave_server, - testdatabuilder, greenwave_cache_config): + testdatabuilder): load_config.return_value = {'greenwave_api_url': greenwave_server + 'api/v1.0'} nvr = testdatabuilder.unique_nvr() for testcase_name in ['dist.rpmdeplint', 'dist.upgradepath', 'dist.abicheck']: @@ -272,7 +269,6 @@ def test_invalidate_new_result_with_real_cache( hub.config = { 'environment': 'environment', 'topic_prefix': 'topic_prefix', - 'greenwave_cache': greenwave_cache_config, } handler = resultsdb.ResultsDBHandler(hub) assert handler.topic == [ @@ -324,7 +320,6 @@ def test_invalidate_new_result_with_no_preexisting_cache( hub.config = { 'environment': 'environment', 'topic_prefix': 'topic_prefix', - 'greenwave_cache': {'backend': 'dogpile.cache.memory'}, } handler = resultsdb.ResultsDBHandler(hub) handler.cache.delete = mock.MagicMock() @@ -368,7 +363,6 @@ def test_consume_compose_id_result( hub.config = { 'environment': 'environment', 'topic_prefix': 'topic_prefix', - 'greenwave_cache': {'backend': 'dogpile.cache.null'}, } handler = resultsdb.ResultsDBHandler(hub) assert handler.topic == ['topic_prefix.environment.taskotron.result.new'] @@ -441,7 +435,6 @@ def test_consume_legacy_result( hub.config = { 'environment': 'environment', 'topic_prefix': 'topic_prefix', - 'greenwave_cache': {'backend': 'dogpile.cache.null'}, } handler = resultsdb.ResultsDBHandler(hub) assert handler.topic == ['topic_prefix.environment.taskotron.result.new'] diff --git a/greenwave/consumers/resultsdb.py b/greenwave/consumers/resultsdb.py index 13b4810..5fb2610 100644 --- a/greenwave/consumers/resultsdb.py +++ b/greenwave/consumers/resultsdb.py @@ -12,13 +12,13 @@ to the message bus about the newly satisfied/unsatisfied policy. import collections import logging -import dogpile.cache +from flask import current_app import fedmsg.consumers import requests +import greenwave.app_factory import greenwave.cache import greenwave.resources -from greenwave.utils import load_config requests_session = requests.Session() @@ -53,15 +53,13 @@ class ResultsDBHandler(fedmsg.consumers.FedmsgConsumer): super(ResultsDBHandler, self).__init__(hub, *args, **kwargs) - # Initialize the cache. - self.cache = dogpile.cache.make_region( - key_mangler=dogpile.cache.util.sha1_mangle_key) - self.cache.configure(**hub.config['greenwave_cache']) + self.flask_app = greenwave.app_factory.create_app() + self.cache = self.flask_app.cache log.info('Greenwave resultsdb handler listening on: %s', self.topic) @staticmethod - def announcement_subjects(config, message): + def announcement_subjects(message): """ Yields subjects for announcement consideration from the message. Args: @@ -75,7 +73,7 @@ class ResultsDBHandler(fedmsg.consumers.FedmsgConsumer): data = message['msg']['task'] # Old format announcement_keys = [ - set(keys) for keys in config['ANNOUNCEMENT_SUBJECT_KEYS'] + set(keys) for keys in current_app.config['ANNOUNCEMENT_SUBJECT_KEYS'] ] def _decode(value): @@ -97,7 +95,6 @@ class ResultsDBHandler(fedmsg.consumers.FedmsgConsumer): """ message = message.get('body', message) log.debug('Processing message "%s"', message) - config = load_config() try: testcase = message['msg']['testcase']['name'] # New format @@ -109,17 +106,17 @@ class ResultsDBHandler(fedmsg.consumers.FedmsgConsumer): except KeyError: result_id = message['msg']['result']['id'] # Old format - for subject in self.announcement_subjects(config, message): - log.debug('Considering subject "%s"', subject) - self._invalidate_cache(subject) - self._publish_decision_changes(config, subject, result_id, testcase) + with self.flask_app.app_context(): + for subject in self.announcement_subjects(message): + log.debug('Considering subject "%s"', subject) + self._invalidate_cache(subject) + self._publish_decision_changes(subject, result_id, testcase) - def _publish_decision_changes(self, config, subject, result_id, testcase): + def _publish_decision_changes(self, subject, result_id, testcase): """ Process the given subject and publish a message if the decision is changed. Args: - config (dict): The greenwave configuration. subject (munch.Munch): A subject argument, used to query greenwave. result_id (int): A result ID to ignore for comparison. testcase (munch.Munch): The name of a testcase to consider. @@ -127,12 +124,12 @@ class ResultsDBHandler(fedmsg.consumers.FedmsgConsumer): # Build a set of all policies which might apply to this new results applicable_policies = set() - for policy in config['policies']: + for policy in current_app.config['policies']: for rule in policy.rules: if getattr(rule, 'test_case_name', None) == testcase: applicable_policies.add(policy) log.debug("messaging: found %i applicable policies of %i for testcase %r", - len(applicable_policies), len(config['policies']), testcase) + len(applicable_policies), len(current_app.config['policies']), testcase) # Given all of our applicable policies, build a map of all decision # context we know about, and which product versions they relate to. diff --git a/greenwave/consumers/waiverdb.py b/greenwave/consumers/waiverdb.py index 3c0e6ff..783ca68 100644 --- a/greenwave/consumers/waiverdb.py +++ b/greenwave/consumers/waiverdb.py @@ -10,11 +10,13 @@ to the message bus about the newly satisfied/unsatisfied policy. """ import logging -import requests import json + +# from flask import current_app import fedmsg.consumers +import requests -from greenwave.utils import load_config +import greenwave.app_factory requests_session = requests.Session() @@ -48,6 +50,8 @@ class WaiverDBHandler(fedmsg.consumers.FedmsgConsumer): self.fedmsg_config = fedmsg.config.load_config() super(WaiverDBHandler, self).__init__(hub, *args, **kwargs) + + self.flask_app = greenwave.app_factory.create_app() log.info('Greenwave waiverdb handler listening on: %s', self.topic) def consume(self, message): @@ -62,9 +66,8 @@ class WaiverDBHandler(fedmsg.consumers.FedmsgConsumer): msg = message['msg'] product_version = msg['product_version'] - config = load_config() testcase = msg['testcase'] - for policy in config['policies']: + for policy in self.flask_app.config['policies']: for rule in policy.rules: if getattr(rule, 'test_case_name', None) == testcase: data = { diff --git a/greenwave/resources.py b/greenwave/resources.py index a03cba9..9bc2731 100644 --- a/greenwave/resources.py +++ b/greenwave/resources.py @@ -107,11 +107,10 @@ def retrieve_waivers(product_version, items): # NOTE - not cached. @greenwave.utils.retry(timeout=300, interval=30, wait_on=urllib3.exceptions.NewConnectionError) def retrieve_decision(greenwave_url, data): - # TODO - get REQUESTS_TIMEOUT and REQUESTS_VERIFY here somehow. This is usually - # called from the fedmsg-hub backend which doesn't have access to the flask - # application context. We need to load the app context and config at backend - # startup to clean this up. + timeout = current_app.config['REQUESTS_TIMEOUT'] + verify = current_app.config['REQUESTS_VERIFY'] headers = {'Content-Type': 'application/json'} - response = requests_session.post(greenwave_url, headers=headers, data=json.dumps(data)) + response = requests_session.post(greenwave_url, headers=headers, data=json.dumps(data), + timeout=timeout, verify=verify) response.raise_for_status() return response.json() diff --git a/greenwave/tests/test_resultsdb_consumer.py b/greenwave/tests/test_resultsdb_consumer.py index 7a6fb65..ecfb51f 100644 --- a/greenwave/tests/test_resultsdb_consumer.py +++ b/greenwave/tests/test_resultsdb_consumer.py @@ -1,15 +1,18 @@ # SPDX-License-Identifier: GPL-2.0+ +import greenwave.app_factory import greenwave.consumers.resultsdb def test_announcement_keys_decode_with_list(): cls = greenwave.consumers.resultsdb.ResultsDBHandler - config = {'ANNOUNCEMENT_SUBJECT_KEYS': [('foo',)]} + app = greenwave.app_factory.create_app() + app.config['ANNOUNCEMENT_SUBJECT_KEYS'] = [('foo',)] message = {'msg': {'data': { u'foo'.encode('utf-8'): [u'bar'.encode('utf-8')], }}} - subjects = cls.announcement_subjects(config, message) + with app.app_context(): + subjects = list(cls.announcement_subjects(message)) - assert list(subjects) == [{u'foo': u'bar'}] + assert subjects == [{u'foo': u'bar'}]