From 590f4563f129a8c9610e46a477c99f83900044b8 Mon Sep 17 00:00:00 2001 From: Lukas Holecek Date: Jul 18 2019 05:27:31 +0000 Subject: Refactor consumers Move common consumer code to Consumer base class. Signed-off-by: Lukas Holecek --- diff --git a/functional-tests/consumers/test_resultsdb.py b/functional-tests/consumers/test_resultsdb.py index fd74041..17724af 100644 --- a/functional-tests/consumers/test_resultsdb.py +++ b/functional-tests/consumers/test_resultsdb.py @@ -19,7 +19,7 @@ def create_resultdb_handler(greenwave_server, cache_config=None): cache_config) -@mock.patch('greenwave.consumers.resultsdb.fedmsg.publish') +@mock.patch('greenwave.consumers.consumer.fedmsg.publish') def test_consume_new_result( mock_fedmsg, requests_session, greenwave_server, testdatabuilder): @@ -159,7 +159,7 @@ def test_consume_new_result( } -@mock.patch('greenwave.consumers.resultsdb.fedmsg.publish') +@mock.patch('greenwave.consumers.consumer.fedmsg.publish') def test_consume_unchanged_result( mock_fedmsg, requests_session, greenwave_server, testdatabuilder): @@ -193,7 +193,7 @@ def test_consume_unchanged_result( assert len(mock_fedmsg.mock_calls) == 0 -@mock.patch('greenwave.consumers.resultsdb.fedmsg.publish') +@mock.patch('greenwave.consumers.consumer.fedmsg.publish') def test_consume_compose_id_result( mock_fedmsg, requests_session, greenwave_server, testdatabuilder): @@ -262,7 +262,7 @@ def test_consume_compose_id_result( mock_fedmsg.assert_called_once_with(topic='decision.update', msg=msg) -@mock.patch('greenwave.consumers.resultsdb.fedmsg.publish') +@mock.patch('greenwave.consumers.consumer.fedmsg.publish') def test_consume_legacy_result( mock_fedmsg, requests_session, greenwave_server, testdatabuilder): @@ -394,7 +394,7 @@ def test_consume_legacy_result( mock_fedmsg.assert_any_call(topic='decision.update', msg=second_msg) -@mock.patch('greenwave.consumers.resultsdb.fedmsg.publish') +@mock.patch('greenwave.consumers.consumer.fedmsg.publish') def test_no_message_for_nonapplicable_policies( mock_fedmsg, requests_session, greenwave_server, testdatabuilder): @@ -432,7 +432,7 @@ def test_no_message_for_nonapplicable_policies( mock_fedmsg.assert_not_called() -@mock.patch('greenwave.consumers.resultsdb.fedmsg.publish') +@mock.patch('greenwave.consumers.consumer.fedmsg.publish') def test_consume_new_result_container_image( mock_fedmsg, requests_session, greenwave_server, testdatabuilder): diff --git a/functional-tests/consumers/test_waiverdb.py b/functional-tests/consumers/test_waiverdb.py index 3964453..32b7e4e 100644 --- a/functional-tests/consumers/test_waiverdb.py +++ b/functional-tests/consumers/test_waiverdb.py @@ -23,7 +23,7 @@ def create_waiverdb_handler(greenwave_server): @pytest.mark.parametrize('subject_type', ('koji_build', 'brew-build')) -@mock.patch('greenwave.consumers.waiverdb.fedmsg.publish') +@mock.patch('greenwave.consumers.consumer.fedmsg.publish') def test_consume_new_waiver( mock_fedmsg, requests_session, greenwave_server, testdatabuilder, subject_type): diff --git a/greenwave/consumers/consumer.py b/greenwave/consumers/consumer.py new file mode 100644 index 0000000..75f0247 --- /dev/null +++ b/greenwave/consumers/consumer.py @@ -0,0 +1,222 @@ +# SPDX-License-Identifier: GPL-2.0+ +import fedmsg +import logging +import requests + +import fedmsg.consumers + +import greenwave.app_factory +from greenwave.api_v1 import subject_type_identifier_to_list +from greenwave.monitor import ( + publish_decision_exceptions_result_counter, + messaging_tx_to_send_counter, messaging_tx_stopped_counter, + messaging_tx_sent_ok_counter, messaging_tx_failed_counter) +from greenwave.policies import applicable_decision_context_product_version_pairs +from greenwave.utils import right_before_this_time + +import greenwave.resources + +try: + import fedora_messaging.api + import fedora_messaging.exceptions +except ImportError: + pass + +log = logging.getLogger(__name__) + + +def _equals_except_keys(lhs, rhs, except_keys): + keys = lhs.keys() - except_keys + return lhs.keys() == rhs.keys() \ + and all(lhs[key] == rhs[key] for key in keys) + + +def _is_decision_unchanged(old_decision, decision): + """ + Returns true only if new decision is same as old one + (ignores result_id values). + """ + if old_decision is None or decision is None: + return old_decision == decision + + requirements_keys = ('satisfied_requirements', 'unsatisfied_requirements') + if not _equals_except_keys(old_decision, decision, requirements_keys): + return False + + ignore_keys = ('result_id',) + for key in requirements_keys: + old_requirements = old_decision[key] + requirements = decision[key] + if len(old_requirements) != len(requirements): + return False + + for old_requirement, requirement in zip(old_requirements, requirements): + if not _equals_except_keys(old_requirement, requirement, ignore_keys): + return False + + return True + + +class Consumer(fedmsg.consumers.FedmsgConsumer): + """ + Base class for consumers. + """ + config_key = 'greenwave_handler' + hub_config_prefix = 'greenwave_consumer_' + default_topic = 'item.new' + monitor_labels = {'handler': 'greenwave_consumer'} + + def __init__(self, hub, *args, **kwargs): + """ + Initialize the consumer, subscribing it to the appropriate topics. + + Args: + hub (moksha.hub.hub.CentralMokshaHub): The hub from which this handler is consuming + messages. It is used to look up the hub config. + """ + + prefix = hub.config.get('topic_prefix') + env = hub.config.get('environment') + suffix = hub.config.get(f'{self.hub_config_prefix}topic_suffix', self.default_topic) + self.topic = ['.'.join([prefix, env, suffix])] + self.fedmsg_config = fedmsg.config.load_config() + + config = kwargs.pop('config', None) + + super().__init__(hub, *args, **kwargs) + + self.flask_app = greenwave.app_factory.create_app(config) + self.greenwave_api_url = self.flask_app.config['GREENWAVE_API_URL'] + log.info('Greenwave handler listening on: %s', self.topic) + + def consume(self, message): + """ + Process the given message and take action. + + Args: + message (munch.Munch): A fedmsg about a new item. + """ + message = message.get('body', message) + log.debug('Processing message "%s"', message) + + with self.flask_app.app_context(): + self._consume_message(message) + + def _inc(self, messaging_counter): + """Helper method to increase monitoring counter.""" + messaging_counter.labels(**self.monitor_labels).inc() + + def _publish_decision_update_fedmsg(self, decision): + try: + fedmsg.publish(topic='decision.update', msg=decision) + self._inc(messaging_tx_sent_ok_counter) + except Exception: + log.exception('Error sending fedmsg message') + self._inc(messaging_tx_failed_counter) + raise + + def _publish_decision_update_fedora_messaging(self, decision): + try: + msg = fedora_messaging.api.Message( + topic='greenwave.decision.update', + body=decision + ) + fedora_messaging.api.publish(msg) + self._inc(messaging_tx_sent_ok_counter) + except fedora_messaging.exceptions.PublishReturned as e: + log.warning( + 'Fedora Messaging broker rejected message %s: %s', + msg.id, e) + self._inc(messaging_tx_stopped_counter) + except fedora_messaging.exceptions.ConnectionException as e: + log.warning('Error sending message %s: %s', msg.id, e) + self._inc(messaging_tx_failed_counter) + except Exception: # pylint: disable=broad-except + log.exception('Error sending fedora-messaging message') + self._inc(messaging_tx_failed_counter) + + def _old_and_new_decisions(self, submit_time, **request_data): + """Returns decision before and after submit time.""" + greenwave_url = self.greenwave_api_url + '/decision' + log.debug('querying greenwave at: %s', greenwave_url) + + try: + decision = greenwave.resources.retrieve_decision(greenwave_url, request_data) + + request_data['when'] = right_before_this_time(submit_time) + old_decision = greenwave.resources.retrieve_decision(greenwave_url, request_data) + log.debug('old decision: %s', old_decision) + except requests.exceptions.HTTPError as e: + log.exception('Failed to retrieve decision for data=%s, error: %s', request_data, e) + self._inc(messaging_tx_stopped_counter) + return None, None + + return old_decision, decision + + @publish_decision_exceptions_result_counter.count_exceptions() + def _publish_decision_change( + self, + submit_time, + subject_type, + subject_identifier, + testcase, + product_version, + publish_testcase): + + policy_attributes = dict( + subject_type=subject_type, + subject_identifier=subject_identifier, + testcase=testcase, + ) + + if product_version: + policy_attributes['product_version'] = product_version + + policies = self.flask_app.config['policies'] + contexts_product_versions = applicable_decision_context_product_version_pairs( + policies, **policy_attributes) + + log.info('Getting greenwave info') + + for decision_context, product_version in sorted(contexts_product_versions): + self._inc(messaging_tx_to_send_counter) + + old_decision, decision = self._old_and_new_decisions( + submit_time, + decision_context=decision_context, + product_version=product_version, + subject_type=subject_type, + subject_identifier=subject_identifier, + ) + if decision is None: + continue + + if _is_decision_unchanged(old_decision, decision): + log.debug('Skipped emitting fedmsg, decision did not change: %s', decision) + self._inc(messaging_tx_stopped_counter) + continue + + decision.update({ + 'subject_type': subject_type, + 'subject_identifier': subject_identifier, + # subject is for backwards compatibility only: + 'subject': subject_type_identifier_to_list(subject_type, + subject_identifier), + 'decision_context': decision_context, + 'product_version': product_version, + 'previous': old_decision, + }) + if publish_testcase: + decision['testcase'] = testcase + + log.info( + 'Emitting a message on the bus, %r, with the topic ' + '"greenwave.decision.update"', decision) + if self.flask_app.config['MESSAGING'] == 'fedmsg': + log.debug(' - to fedmsg') + self._publish_decision_update_fedmsg(decision) + elif self.flask_app.config['MESSAGING'] == 'fedora-messaging': + log.debug(' - to fedora-messaging') + self._publish_decision_update_fedora_messaging(decision) + + self._inc(messaging_tx_stopped_counter) diff --git a/greenwave/consumers/resultsdb.py b/greenwave/consumers/resultsdb.py index 81ccd8c..1b0375d 100644 --- a/greenwave/consumers/resultsdb.py +++ b/greenwave/consumers/resultsdb.py @@ -12,28 +12,10 @@ to the message bus about the newly satisfied/unsatisfied policy. import logging import re -import fedmsg.consumers -import requests - -import greenwave.app_factory -import greenwave.resources -from greenwave.api_v1 import subject_type_identifier_to_list -from greenwave.monitor import ( - publish_decision_exceptions_result_counter, - messaging_tx_to_send_counter, messaging_tx_stopped_counter, - messaging_tx_sent_ok_counter, messaging_tx_failed_counter) -from greenwave.policies import applicable_decision_context_product_version_pairs -from greenwave.utils import right_before_this_time +from greenwave.consumers.consumer import Consumer import xmlrpc.client -try: - import fedora_messaging.api - import fedora_messaging.exceptions -except ImportError: - pass - - log = logging.getLogger(__name__) @@ -93,39 +75,7 @@ def _subject_product_version(subject_identifier, subject_type, koji_proxy=None): pass -def _equals_except_keys(lhs, rhs, except_keys): - keys = lhs.keys() - except_keys - return lhs.keys() == rhs.keys() \ - and all(lhs[key] == rhs[key] for key in keys) - - -def _is_decision_unchanged(old_decision, decision): - """ - Returns true only if new decision is same as old one - (ignores result_id values). - """ - if old_decision is None or decision is None: - return old_decision == decision - - requirements_keys = ('satisfied_requirements', 'unsatisfied_requirements') - if not _equals_except_keys(old_decision, decision, requirements_keys): - return False - - ignore_keys = ('result_id',) - for key in requirements_keys: - old_requirements = old_decision[key] - requirements = decision[key] - if len(old_requirements) != len(requirements): - return False - - for old_requirement, requirement in zip(old_requirements, requirements): - if not _equals_except_keys(old_requirement, requirement, ignore_keys): - return False - - return True - - -class ResultsDBHandler(fedmsg.consumers.FedmsgConsumer): +class ResultsDBHandler(Consumer): """ Handle a new result. @@ -134,29 +84,12 @@ class ResultsDBHandler(fedmsg.consumers.FedmsgConsumer): """ config_key = 'resultsdb_handler' + hub_config_prefix = 'resultsdb_' + default_topic = 'taskotron.result.new' + monitor_labels = {'handler': 'resultsdb'} - def __init__(self, hub, *args, **kwargs): - """ - Initialize the ResultsDBHandler, subscribing it to the appropriate topics. - - Args: - hub (moksha.hub.hub.CentralMokshaHub): The hub from which this handler is consuming - messages. It is used to look up the hub config. - """ - - prefix = hub.config.get('topic_prefix') - env = hub.config.get('environment') - suffix = hub.config.get('resultsdb_topic_suffix', 'taskotron.result.new') - self.topic = ['.'.join([prefix, env, suffix])] - self.fedmsg_config = fedmsg.config.load_config() - - config = kwargs.pop('config', None) - - super(ResultsDBHandler, self).__init__(hub, *args, **kwargs) - - self.flask_app = greenwave.app_factory.create_app(config) - self.greenwave_api_url = self.flask_app.config['GREENWAVE_API_URL'] - self.cache = self.flask_app.cache + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) koji_base_url = self.flask_app.config['KOJI_BASE_URL'] if koji_base_url: @@ -164,8 +97,6 @@ class ResultsDBHandler(fedmsg.consumers.FedmsgConsumer): else: self.koji_proxy = None - log.info('Greenwave resultsdb handler listening on: %s', self.topic) - @staticmethod def announcement_subjects(message): """ @@ -213,17 +144,9 @@ class ResultsDBHandler(fedmsg.consumers.FedmsgConsumer): elif 'item' in data and _type: yield (_type, _decode(data['item'])) - def consume(self, message): - """ - Process the given message and take action. - - Args: - message (munch.Munch): A fedmsg about a new result. - """ - message = message.get('body', message) - log.debug('Processing message "%s"', message) - + def _consume_message(self, message): msg = message['msg'] + try: testcase = msg['testcase']['name'] except KeyError: @@ -234,111 +157,17 @@ class ResultsDBHandler(fedmsg.consumers.FedmsgConsumer): except KeyError: submit_time = msg['result']['submit_time'] - with self.flask_app.app_context(): - for subject_type, subject_identifier in self.announcement_subjects(message): - log.debug('Considering subject %s: %r', subject_type, subject_identifier) - self._publish_decision_changes(subject_type, subject_identifier, - submit_time, testcase) - - @publish_decision_exceptions_result_counter.count_exceptions() - def _publish_decision_changes(self, subject_type, subject_identifier, submit_time, testcase): - """ - Process the given subject and publish a message if the decision is changed. + for subject_type, subject_identifier in self.announcement_subjects(message): + log.debug('Considering subject %s: %r', subject_type, subject_identifier) - Args: - subject_type (munch.Munch): subject type argument, used to query greenwave. - subject_identifier (munch.Munch): subject identifier argument, used to query greenwave. - submit_time (string): date. After this date, results will be ignored for comparison. - testcase (munch.Munch): the name of a testcase to consider. - """ - policy_attributes = dict( - subject_type=subject_type, - subject_identifier=subject_identifier, - testcase=testcase, - ) - - product_version = _subject_product_version( - subject_identifier, subject_type, self.koji_proxy) - if product_version: - policy_attributes['product_version'] = product_version - - policies = self.flask_app.config['policies'] - contexts_product_versions = applicable_decision_context_product_version_pairs( - policies, **policy_attributes) - - log.info('Getting greenwave info') - - for decision_context, product_version in sorted(contexts_product_versions): - messaging_tx_to_send_counter.labels(handler='resultsdb').inc() - greenwave_url = self.greenwave_api_url + '/decision' - - data = { - 'decision_context': decision_context, - 'product_version': product_version, - 'subject_type': subject_type, - 'subject_identifier': subject_identifier, - } + product_version = _subject_product_version( + subject_identifier, subject_type, self.koji_proxy) - try: - log.debug('querying greenwave at: %s', greenwave_url) - decision = greenwave.resources.retrieve_decision(greenwave_url, data) - - # get old decision - data.update({ - 'when': right_before_this_time(submit_time), - }) - old_decision = greenwave.resources.retrieve_decision(greenwave_url, data) - log.debug('old decision: %s', old_decision) - except requests.exceptions.HTTPError as e: - log.exception('Failed to retrieve decision for data=%s, error: %s', data, e) - messaging_tx_stopped_counter.labels(handler='resultsdb').inc() - continue - - if _is_decision_unchanged(old_decision, decision): - log.debug('Skipped emitting fedmsg, decision did not change: %s', decision) - messaging_tx_stopped_counter.labels(handler='resultsdb').inc() - continue - - decision.update({ - 'subject_type': subject_type, - 'subject_identifier': subject_identifier, - # subject is for backwards compatibility only: - 'subject': subject_type_identifier_to_list(subject_type, - subject_identifier), - 'decision_context': decision_context, - 'product_version': product_version, - 'previous': old_decision, - }) - log.info( - 'Emitting a message on the bus, %r, with the topic ' - '"greenwave.decision.update"', decision) - if self.flask_app.config['MESSAGING'] == 'fedmsg': - log.debug(' - to fedmsg') - try: - fedmsg.publish(topic='decision.update', msg=decision) - messaging_tx_sent_ok_counter.labels(handler='resultsdb').inc() - except Exception: - messaging_tx_failed_counter.labels(handler='resultsdb').inc() - raise - elif self.flask_app.config['MESSAGING'] == 'fedora-messaging': - log.debug(' - to fedora-messaging') - try: - msg = fedora_messaging.api.Message( - topic='greenwave.decision.update', - body=decision - ) - fedora_messaging.api.publish(msg) - messaging_tx_sent_ok_counter.labels(handler='resultsdb').inc() - except fedora_messaging.exceptions.PublishReturned as e: - log.warning( - 'Fedora Messaging broker rejected message %s: %s', - msg.id, e) - messaging_tx_stopped_counter.labels(handler='resultsdb').inc() - except fedora_messaging.exceptions.ConnectionException as e: - log.warning('Error sending message %s: %s', msg.id, e) - messaging_tx_failed_counter.labels(handler='resultsdb').inc() - except Exception: # pylint: disable=broad-except - log.exception('Error sending fedora-messaging message') - messaging_tx_failed_counter.labels(handler='resultsdb').inc() - - messaging_tx_stopped_counter.labels(handler='resultsdb').inc() + self._publish_decision_change( + submit_time=submit_time, + subject_type=subject_type, + subject_identifier=subject_identifier, + testcase=testcase, + product_version=product_version, + publish_testcase=False, + ) diff --git a/greenwave/consumers/waiverdb.py b/greenwave/consumers/waiverdb.py index f282fc6..d022787 100644 --- a/greenwave/consumers/waiverdb.py +++ b/greenwave/consumers/waiverdb.py @@ -9,35 +9,10 @@ and if the new waiver causes the decision to change it will publish a message to the message bus about the newly satisfied/unsatisfied policy. """ -import logging -import json +from greenwave.consumers.consumer import Consumer -import fedmsg.consumers -import greenwave.app_factory -from greenwave.api_v1 import subject_type_identifier_to_list -from greenwave.monitor import ( - publish_decision_exceptions_waiver_counter, - messaging_tx_to_send_counter, messaging_tx_stopped_counter, - messaging_tx_sent_ok_counter, messaging_tx_failed_counter) -from greenwave.policies import applicable_decision_context_product_version_pairs -from greenwave.utils import right_before_this_time -from greenwave.request_session import get_requests_session - -try: - import fedora_messaging.api - import fedora_messaging.exceptions -except ImportError: - pass - - -requests_session = get_requests_session() - - -log = logging.getLogger(__name__) - - -class WaiverDBHandler(fedmsg.consumers.FedmsgConsumer): +class WaiverDBHandler(Consumer): """ Handle a new waiver. @@ -46,39 +21,11 @@ class WaiverDBHandler(fedmsg.consumers.FedmsgConsumer): """ config_key = 'waiverdb_handler' + hub_config_prefix = 'waiverdb_' + default_topic = 'waiver.new' + monitor_labels = {'handler': 'waiverdb'} - def __init__(self, hub, *args, **kwargs): - """ - Initialize the WaiverDBHandler, subscribing it to the appropriate topics. - - Args: - hub (moksha.hub.hub.CentralMokshaHub): The hub from which this handler is consuming - messages. It is used to look up the hub config. - """ - - prefix = hub.config.get('topic_prefix') - env = hub.config.get('environment') - suffix = hub.config.get('waiverdb_topic_suffix', 'waiver.new') - self.topic = ['.'.join([prefix, env, suffix])] - self.fedmsg_config = fedmsg.config.load_config() - - config = kwargs.pop('config', None) - - super(WaiverDBHandler, self).__init__(hub, *args, **kwargs) - - self.flask_app = greenwave.app_factory.create_app(config) - self.greenwave_api_url = self.flask_app.config['GREENWAVE_API_URL'] - log.info('Greenwave waiverdb handler listening on: %s', self.topic) - - def consume(self, message): - """ - Process the given message and publish a message if the decision is changed. - - Args: - message (munch.Munch): A fedmsg about a new waiver. - """ - message = message.get('body', message) - log.debug('Processing message "%s"', message) + def _consume_message(self, message): msg = message['msg'] product_version = msg['product_version'] @@ -87,104 +34,11 @@ class WaiverDBHandler(fedmsg.consumers.FedmsgConsumer): subject_identifier = msg['subject_identifier'] submit_time = msg['timestamp'] - with self.flask_app.app_context(): - self._publish_decision_changes(subject_type, subject_identifier, submit_time, - product_version, testcase) - - @publish_decision_exceptions_waiver_counter.count_exceptions() - def _publish_decision_changes(self, subject_type, subject_identifier, submit_time, - product_version, testcase): - policies = self.flask_app.config['policies'] - contexts_product_versions = applicable_decision_context_product_version_pairs( - policies, + self._publish_decision_change( + submit_time=submit_time, subject_type=subject_type, subject_identifier=subject_identifier, testcase=testcase, - product_version=product_version) - - for decision_context, product_version in sorted(contexts_product_versions): - messaging_tx_to_send_counter.labels(handler='waiverdb').inc() - data = { - 'decision_context': decision_context, - 'product_version': product_version, - 'subject_type': subject_type, - 'subject_identifier': subject_identifier, - } - response = requests_session.post( - self.greenwave_api_url + '/decision', - headers={'Content-Type': 'application/json'}, - data=json.dumps(data)) - - if not response.ok: - log.error(response.text) - messaging_tx_stopped_counter.labels(handler='waiverdb').inc() - continue - - decision = response.json() - - # get old decision - data.update({ - 'when': right_before_this_time(submit_time), - }) - response = requests_session.post( - self.greenwave_api_url + '/decision', - headers={'Content-Type': 'application/json'}, - data=json.dumps(data)) - - if not response.ok: - log.error(response.text) - messaging_tx_stopped_counter.labels(handler='waiverdb').inc() - continue - - old_decision = response.json() - - if decision == old_decision: - log.debug('Skipped emitting fedmsg, decision did not change: %s', decision) - messaging_tx_stopped_counter.labels(handler='waiverdb').inc() - continue - - msg = decision - decision.update({ - 'subject_type': subject_type, - 'subject_identifier': subject_identifier, - # subject is for backwards compatibility only: - 'subject': subject_type_identifier_to_list(subject_type, - subject_identifier), - 'testcase': testcase, - 'decision_context': decision_context, - 'product_version': product_version, - 'previous': old_decision, - }) - log.info( - 'Emitting a message on the bus, %r, with the topic ' - '"greenwave.decision.update"', decision) - if self.flask_app.config['MESSAGING'] == 'fedmsg': - log.debug(' - to fedmsg') - try: - fedmsg.publish(topic='decision.update', msg=msg) - messaging_tx_sent_ok_counter.labels(handler='waiverdb').inc() - except Exception: - messaging_tx_failed_counter.labels(handler='waiverdb').inc() - raise - elif self.flask_app.config['MESSAGING'] == 'fedora-messaging': - log.debug(' - to fedora-messaging') - try: - msg = fedora_messaging.api.Message( - topic='greenwave.decision.update', - body=msg - ) - fedora_messaging.api.publish(msg) - messaging_tx_sent_ok_counter.labels(handler='waiverdb').inc() - except fedora_messaging.exceptions.PublishReturned as e: - log.warning( - 'Fedora Messaging broker rejected message %s: %s', - msg.id, e) - messaging_tx_stopped_counter.labels(handler='waiverdb').inc() - except fedora_messaging.exceptions.ConnectionException as e: - log.warning('Error sending message %s: %s', msg.id, e) - messaging_tx_failed_counter.labels(handler='waiverdb').inc() - except Exception: # pylint: disable=broad-except - log.exception('Error sending fedora-messaging message') - messaging_tx_failed_counter.labels(handler='waiverdb').inc() - - messaging_tx_stopped_counter.labels(handler='waiverdb').inc() + product_version=product_version, + publish_testcase=True, + ) diff --git a/greenwave/tests/test_resultsdb_consumer.py b/greenwave/tests/test_resultsdb_consumer.py index ec3e083..fe7fdf7 100644 --- a/greenwave/tests/test_resultsdb_consumer.py +++ b/greenwave/tests/test_resultsdb_consumer.py @@ -111,8 +111,8 @@ def test_no_announcement_subjects_for_old_compose_message(): parameters = [ - ('fedmsg', 'greenwave.consumers.resultsdb.fedmsg.publish'), - ('fedora-messaging', 'greenwave.consumers.resultsdb.fedora_messaging.api.publish'), + ('fedmsg', 'greenwave.consumers.consumer.fedmsg.publish'), + ('fedora-messaging', 'greenwave.consumers.consumer.fedora_messaging.api.publish'), ] @@ -546,7 +546,7 @@ def test_real_fedora_messaging_msg( """) config = 'fedora-messaging' - publish = 'greenwave.consumers.resultsdb.fedora_messaging.api.publish' + publish = 'greenwave.consumers.consumer.fedora_messaging.api.publish' with mock.patch('greenwave.config.Config.MESSAGING', config): with mock.patch(publish) as mock_fedmsg: