From 53e64758f42893232796abcfaca69bf7b35bcf9f Mon Sep 17 00:00:00 2001 From: Ralph Bean Date: Sep 25 2017 16:54:37 +0000 Subject: [PATCH 1/8] Add a new test config for testing with caching. --- diff --git a/functional-tests/conftest.py b/functional-tests/conftest.py index 5aaf9df..28e232c 100644 --- a/functional-tests/conftest.py +++ b/functional-tests/conftest.py @@ -82,7 +82,16 @@ def waiverdb_server(request): @pytest.fixture(scope='session') def greenwave_server(request): app = greenwave.app_factory.create_app('greenwave.config.TestingConfig') - server = WSGIServerThread(app, init_func=lambda: None, port=5005) + server = WSGIServerThread(app, init_func=lambda: None, port=app.config['PORT']) + server.start() + request.addfinalizer(server.stop) + return server + + +@pytest.fixture(scope='session') +def cached_greenwave_server(request): + app = greenwave.app_factory.create_app('greenwave.config.CachedTestingConfig') + server = WSGIServerThread(app, init_func=lambda: None, port=app.config['PORT']) server.start() request.addfinalizer(server.stop) return server diff --git a/greenwave/config.py b/greenwave/config.py index 0d297dc..07ac4fa 100644 --- a/greenwave/config.py +++ b/greenwave/config.py @@ -42,3 +42,9 @@ class TestingConfig(Config): os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'conf', 'policies' ) + + +class CachedTestingConfig(TestingConfig): + PORT = 6005 + # Cache in memory + CACHE = {'backend': 'dogpile.cache.memory'} From 2df9b2205dc8aed5e185258b751d7c57e6e0be10 Mon Sep 17 00:00:00 2001 From: Ralph Bean Date: Sep 25 2017 16:54:37 +0000 Subject: [PATCH 2/8] Add a test to ensure that our caching really works. --- diff --git a/functional-tests/test_api_v1.py b/functional-tests/test_api_v1.py index 567e71f..2aae000 100644 --- a/functional-tests/test_api_v1.py +++ b/functional-tests/test_api_v1.py @@ -457,3 +457,43 @@ def test_ignore_waiver(requests_session, greenwave_server, testdatabuilder): ] assert res_data['policies_satisified'] is False assert res_data['unsatisfied_requirements'] == expected_unsatisfied_requirements + + +def test_cached_false_positive(requests_session, cached_greenwave_server, testdatabuilder): + """ Test that caching without invalidation produces false positives. + + This just tests that our caching works in the first place. + - Check a decision, it passes. + - Insert a failing result. + - Check the decision again, it passes + + (but it shouldn't) which means caching works. + """ + nvr = testdatabuilder.unique_nvr() + for testcase_name in all_rpmdiff_testcase_names: + testdatabuilder.create_result(item=nvr, + testcase_name=testcase_name, + outcome='PASSED') + data = { + 'decision_context': 'errata_newfile_to_qe', + 'product_version': 'rhel-7', + 'subject': [{'item': nvr, 'type': 'koji_build'}] + } + r = requests_session.post(cached_greenwave_server.url + 'api/v1.0/decision', + headers={'Content-Type': 'application/json'}, + data=json.dumps(data)) + assert r.status_code == 200 + res_data = r.json() + assert res_data['policies_satisified'] is True + + # Now, insert a *failing* result. The cache should return the old results + # that exclude the failing one (erroneously). + testdatabuilder.create_result(item=nvr, + testcase_name=testcase_name, + outcome='FAILED') + r = requests_session.post(cached_greenwave_server.url + 'api/v1.0/decision', + headers={'Content-Type': 'application/json'}, + data=json.dumps(data)) + assert r.status_code == 200 + res_data = r.json() + assert res_data['policies_satisified'] is True From 4c5f14d3d95387d7e90de8f6ec097194b0a042a4 Mon Sep 17 00:00:00 2001 From: Ralph Bean Date: Sep 25 2017 16:54:37 +0000 Subject: [PATCH 3/8] Fix caching to be testable. The old way didn't work because our ``cache`` object was global. It would get configured by the first test and then never be reconfigurable for other tests. This gets fixed here by hanging the cache region on `flask.current_app` so that it gets reinstantiated and reconfigured for each test run (and each WSGI process). --- diff --git a/greenwave/app_factory.py b/greenwave/app_factory.py index cb86e47..44b53a4 100644 --- a/greenwave/app_factory.py +++ b/greenwave/app_factory.py @@ -4,9 +4,9 @@ from flask import Flask from greenwave.logger import init_logging from greenwave.api_v1 import api -from greenwave.cache import cache from greenwave.utils import json_error, load_config +from dogpile.cache import make_region from requests import ConnectionError, Timeout from werkzeug.exceptions import default_exceptions @@ -32,8 +32,8 @@ def create_app(config_obj=None): app.add_url_rule('/healthcheck', view_func=healthcheck) # Initialize the cache. - if not cache.is_configured: - cache.configure(**app.config['CACHE']) + app.cache = make_region() + app.cache.configure(**app.config['CACHE']) return app diff --git a/greenwave/cache.py b/greenwave/cache.py index 27fb3b6..00bcf50 100644 --- a/greenwave/cache.py +++ b/greenwave/cache.py @@ -1,9 +1,19 @@ # SPDX-License-Identifier: GPL-2.0+ import dogpile.cache - -# Our globally available cache region. Gets initialized in app_factory. -cache = dogpile.cache.make_region() +import flask # Provide a convenient alias for the key generator we want to use key_generator = dogpile.cache.util.function_key_generator + + +def cached(fn): + """ Cache arguments with a region hung on the flask app. """ + def wrapper(*args): + decoration = flask.current_app.cache.cache_on_arguments + decorator = decoration(function_key_generator=key_generator) + return decorator(fn)(*args) + wrapper.__name__ = fn.__name__ + wrapper.__module__ = fn.__module__ + wrapper.__doc__ = fn.__doc__ + return wrapper diff --git a/greenwave/resources.py b/greenwave/resources.py index 2a95f2f..11335f0 100644 --- a/greenwave/resources.py +++ b/greenwave/resources.py @@ -9,12 +9,12 @@ waiverdb, etc..). import requests from flask import current_app -from greenwave.cache import cache, key_generator +from greenwave.cache import cached requests_session = requests.Session() -@cache.cache_on_arguments(function_key_generator=key_generator) +@cached def retrieve_results(item): """ Retrieve cached results from resultsdb for a given item. """ # XXX make this more efficient than just fetching everything From f5c904a7133f04b71b04ab1ce0e41047defdf152 Mon Sep 17 00:00:00 2001 From: Ralph Bean Date: Sep 25 2017 16:54:37 +0000 Subject: [PATCH 4/8] Use a file-based cache for the tests... and they work! --- diff --git a/.gitignore b/.gitignore index 823cb0b..81e0689 100644 --- a/.gitignore +++ b/.gitignore @@ -24,3 +24,4 @@ conf/settings.py .cache coverage.xml htmlcov +greenwave-test-cache.dbm* diff --git a/fedmsg.d/config.py b/fedmsg.d/config.py index 3c4d1e4..b0aa370 100644 --- a/fedmsg.d/config.py +++ b/fedmsg.d/config.py @@ -128,5 +128,10 @@ config = dict( routing_nitpicky=False, # Greenwave API url - greenwave_api_url='https://greenwave.domain.local/api/v1.0' + 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 28e232c..bc400b6 100644 --- a/functional-tests/conftest.py +++ b/functional-tests/conftest.py @@ -94,7 +94,11 @@ def cached_greenwave_server(request): server = WSGIServerThread(app, init_func=lambda: None, port=app.config['PORT']) server.start() request.addfinalizer(server.stop) - return server + try: + yield server + finally: + # Remove the cache file so the next test can start afresh. + os.remove(app.config['CACHE']['arguments']['filename']) @pytest.fixture(scope='session') diff --git a/greenwave/config.py b/greenwave/config.py index 07ac4fa..58e51ae 100644 --- a/greenwave/config.py +++ b/greenwave/config.py @@ -47,4 +47,8 @@ class TestingConfig(Config): class CachedTestingConfig(TestingConfig): PORT = 6005 # Cache in memory - CACHE = {'backend': 'dogpile.cache.memory'} + CACHE = dict( + backend="dogpile.cache.dbm", + expiration_time=300, + arguments={"filename": "greenwave-test-cache.dbm"} + ) From 1576c068d4cd81a4cc1dd8975e4090b4dadbedaa Mon Sep 17 00:00:00 2001 From: Ralph Bean Date: Sep 25 2017 16:54:37 +0000 Subject: [PATCH 5/8] A cache invalidator, and tests. --- diff --git a/functional-tests/consumers/test_cache.py b/functional-tests/consumers/test_cache.py new file mode 100644 index 0000000..45c0dee --- /dev/null +++ b/functional-tests/consumers/test_cache.py @@ -0,0 +1,123 @@ +# SPDX-License-Identifier: GPL-2.0+ + +import json +import mock +import pprint + +from greenwave.config import CachedTestingConfig +from greenwave.consumers import cache + + +def test_consume_new_result_with_mocked_cache( + requests_session, cached_greenwave_server, testdatabuilder, monkeypatch): + """ Consume a result, and ensure that `delete` is called. """ + monkeypatch.setenv('TEST', 'true') + nvr = testdatabuilder.unique_nvr() + result = testdatabuilder.create_result( + item=nvr, testcase_name='dist.rpmdeplint', outcome='PASSED') + message = { + 'topic': 'taskotron.result.new', + 'msg': { + 'result': { + 'id': result['id'], + 'outcome': 'PASSED' + }, + 'task': { + 'item': nvr, + 'type': 'koji_build', + 'name': 'dist.rpmdeplint' + } + } + } + hub = mock.MagicMock() + hub.config = { + 'environment': 'environment', + 'topic_prefix': 'topic_prefix', + 'greenwave_cache': {'backend': 'dogpile.cache.memory'}, + } + handler = cache.CacheInvalidatorExtraordinaire(hub) + handler.cache = mock.MagicMock() + assert handler.topic == [ + 'topic_prefix.environment.taskotron.result.new', + # Not ready to handle waiverdb yet. + #'topic_prefix.environment.waiver.new', + ] + handler.consume(message) + expected = ("greenwave.resources:retrieve_results|" + "{'item': '%s', 'type': 'koji_build'}" % nvr) + handler.cache.delete.assert_called_once_with(expected) + + +def test_consume_new_result_with_real_cache( + requests_session, cached_greenwave_server, testdatabuilder, monkeypatch): + monkeypatch.setenv('TEST', 'true') + nvr = testdatabuilder.unique_nvr() + for testcase_name in ['dist.rpmdeplint', 'dist.upgradepath', 'dist.abicheck']: + testdatabuilder.create_result( + item=nvr, testcase_name=testcase_name, outcome='PASSED') + + # get first passing decision + query = { + 'decision_context': 'bodhi_update_push_stable', + 'product_version': 'fedora-26', + 'subject': [{'item': nvr, 'type': 'koji_build'}], + } + r = requests_session.post(cached_greenwave_server.url + 'api/v1.0/decision', + headers={'Content-Type': 'application/json'}, + data=json.dumps(query)) + assert r.status_code == 200 + response = r.json() + # Ensure it is passing... + assert response['policies_satisified'], pprint.pformat(response) + + # Now, insert a new result and ensure that caching has made it such that + # even though the new result fails, our decision still passes (bad) + testdatabuilder.create_result( + item=nvr, testcase_name='dist.abicheck', outcome='FAILED') + r = requests_session.post(cached_greenwave_server.url + 'api/v1.0/decision', + headers={'Content-Type': 'application/json'}, + data=json.dumps(query)) + assert r.status_code == 200 + response = r.json() + # Ensure it is passing... BUT IT SHOULDN'T BE! + assert response['policies_satisified'], pprint.pformat(response) + + # Now, handle a message about the new failing result + message = { + u'topic': u'taskotron.result.new', + u'msg': { + u'result': { + u'id': u'whatever', + u'outcome': u'doesn\'t matter', + }, + u'task': { + u'item': nvr.decode('utf-8'), + u'type': u'koji_build', + u'name': u'dist.rpmdeplint' + } + } + } + hub = mock.MagicMock() + hub.config = { + 'environment': 'environment', + 'topic_prefix': 'topic_prefix', + 'greenwave_cache': CachedTestingConfig().CACHE, + } + handler = cache.CacheInvalidatorExtraordinaire(hub) + assert handler.topic == [ + 'topic_prefix.environment.taskotron.result.new', + # Not ready to handle waiverdb yet. + #'topic_prefix.environment.waiver.new', + ] + handler.consume(message) + + # At this point, the invalidator should have invalidated the cache. If we + # ask again, the decision should be correct now. It should be a stone cold + # "no". + r = requests_session.post(cached_greenwave_server.url + 'api/v1.0/decision', + headers={'Content-Type': 'application/json'}, + data=json.dumps(query)) + assert r.status_code == 200 + response = r.json() + # Ensure it is failing -- as it should be. + assert not response['policies_satisified'], pprint.pformat(response) diff --git a/greenwave/consumers/cache.py b/greenwave/consumers/cache.py new file mode 100644 index 0000000..bb78b07 --- /dev/null +++ b/greenwave/consumers/cache.py @@ -0,0 +1,80 @@ +# -*- coding: utf-8 -*- +# SPDX-License-Identifier: GPL-2.0+ +""" +The "cache handler". + +This module is responsible for listening for new results from ResultsDB (and +eventually waiverdb also). When a new result or waiver is received, this code +will lookup any possible cache values we have for that item and destroy them -- +invalidate them. + +https://pagure.io/greenwave/issue/77 +""" + +import logging +import requests +import dogpile.cache +import fedmsg.consumers + +import greenwave.cache +import greenwave.resources + +requests_session = requests.Session() + + +log = logging.getLogger(__name__) + + +class CacheInvalidatorExtraordinaire(fedmsg.consumers.FedmsgConsumer): + """ + Handle a new result or waiver. + + Attributes: + topic (list): A list of strings that indicate which fedmsg topics this consumer listens to. + """ + + config_key = 'cache_invalidator' + + def __init__(self, hub, *args, **kwargs): + """ + Initialize the CacheInvalidatorExtraordinaire, 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') + self.topic = [ + prefix + '.' + env + '.taskotron.result.new', + # Not ready to handle waivers yet... + #prefix + '.' + env + '.waiver.new', + ] + self.fedmsg_config = fedmsg.config.load_config() + super(CacheInvalidatorExtraordinaire, self).__init__(hub, *args, **kwargs) + log.info('Greenwave cache invalidator listening on:\n' + '%r' % self.topic) + + # Initialize the cache. + self.cache = dogpile.cache.make_region() + self.cache.configure(**hub.config['greenwave_cache']) + + def consume(self, message): + """ + Process the given message and delete cache keys as necessary. + + Args: + message (munch.Munch): A fedmsg about a new result or waiver. + """ + log.debug('Processing message "{0}"'.format(message)) + msg = message['msg'] + task = msg['task'] + del task['name'] + # here, task is {"item": "nodejs-ansi-black-0.1.1-1.fc28", "type": "koji_build" } + namespace = None + fn = greenwave.resources.retrieve_results + key = greenwave.cache.key_generator(namespace, fn)(task) + if not self.cache.get(key): + raise KeyError(key) + self.cache.delete(key) From 6078915f2eff9a2a10b8a87112f704117baf52bf Mon Sep 17 00:00:00 2001 From: Ralph Bean Date: Sep 25 2017 16:54:37 +0000 Subject: [PATCH 6/8] Use functools.wraps. ...like a pro. --- diff --git a/greenwave/cache.py b/greenwave/cache.py index 00bcf50..37d04ee 100644 --- a/greenwave/cache.py +++ b/greenwave/cache.py @@ -1,5 +1,6 @@ # SPDX-License-Identifier: GPL-2.0+ +import functools import dogpile.cache import flask @@ -9,11 +10,9 @@ key_generator = dogpile.cache.util.function_key_generator def cached(fn): """ Cache arguments with a region hung on the flask app. """ + @functools.wraps(fn) def wrapper(*args): decoration = flask.current_app.cache.cache_on_arguments decorator = decoration(function_key_generator=key_generator) return decorator(fn)(*args) - wrapper.__name__ = fn.__name__ - wrapper.__module__ = fn.__module__ - wrapper.__doc__ = fn.__doc__ return wrapper From 79c02383f7f9feffd0fa7847883c68e3674eaaf3 Mon Sep 17 00:00:00 2001 From: Ralph Bean Date: Sep 25 2017 16:54:37 +0000 Subject: [PATCH 7/8] Ensure that invalidating an unknown value is sane. --- diff --git a/functional-tests/consumers/test_cache.py b/functional-tests/consumers/test_cache.py index 45c0dee..f24d138 100644 --- a/functional-tests/consumers/test_cache.py +++ b/functional-tests/consumers/test_cache.py @@ -121,3 +121,41 @@ def test_consume_new_result_with_real_cache( response = r.json() # Ensure it is failing -- as it should be. assert not response['policies_satisified'], pprint.pformat(response) + + +def test_consume_new_result_with_no_preexisting_cache( + requests_session, cached_greenwave_server, testdatabuilder, monkeypatch): + """ Ensure that invalidating an unknown value is sane. """ + monkeypatch.setenv('TEST', 'true') + nvr = testdatabuilder.unique_nvr() + result = testdatabuilder.create_result( + item=nvr, testcase_name='dist.rpmdeplint', outcome='PASSED') + message = { + 'topic': 'taskotron.result.new', + 'msg': { + 'result': { + 'id': result['id'], + 'outcome': 'PASSED' + }, + 'task': { + 'item': nvr, + 'type': 'koji_build', + 'name': 'dist.rpmdeplint' + } + } + } + hub = mock.MagicMock() + hub.config = { + 'environment': 'environment', + 'topic_prefix': 'topic_prefix', + 'greenwave_cache': {'backend': 'dogpile.cache.memory'}, + } + handler = cache.CacheInvalidatorExtraordinaire(hub) + handler.cache.delete = mock.MagicMock() + assert handler.topic == [ + 'topic_prefix.environment.taskotron.result.new', + # Not ready to handle waiverdb yet. + #'topic_prefix.environment.waiver.new', + ] + handler.consume(message) + handler.cache.delete.assert_not_called() diff --git a/greenwave/consumers/cache.py b/greenwave/consumers/cache.py index bb78b07..e5c2ce1 100644 --- a/greenwave/consumers/cache.py +++ b/greenwave/consumers/cache.py @@ -76,5 +76,7 @@ class CacheInvalidatorExtraordinaire(fedmsg.consumers.FedmsgConsumer): fn = greenwave.resources.retrieve_results key = greenwave.cache.key_generator(namespace, fn)(task) if not self.cache.get(key): - raise KeyError(key) - self.cache.delete(key) + log.debug("No cache value found for %r" % key) + else: + log.debug("Invalidating cache for %r" % key) + self.cache.delete(key) From 580591ee761c31c9be2b3b4487445a20b87ce2fc Mon Sep 17 00:00:00 2001 From: Ralph Bean Date: Sep 25 2017 16:56:22 +0000 Subject: [PATCH 8/8] Move cache invalidation inside the resultsdb handler. Do this to ensure that invalidation happens *first* before any decisions are published. We want to make sure those evaluations are made on only the most recent data. --- diff --git a/functional-tests/consumers/test_cache.py b/functional-tests/consumers/test_cache.py deleted file mode 100644 index f24d138..0000000 --- a/functional-tests/consumers/test_cache.py +++ /dev/null @@ -1,161 +0,0 @@ -# SPDX-License-Identifier: GPL-2.0+ - -import json -import mock -import pprint - -from greenwave.config import CachedTestingConfig -from greenwave.consumers import cache - - -def test_consume_new_result_with_mocked_cache( - requests_session, cached_greenwave_server, testdatabuilder, monkeypatch): - """ Consume a result, and ensure that `delete` is called. """ - monkeypatch.setenv('TEST', 'true') - nvr = testdatabuilder.unique_nvr() - result = testdatabuilder.create_result( - item=nvr, testcase_name='dist.rpmdeplint', outcome='PASSED') - message = { - 'topic': 'taskotron.result.new', - 'msg': { - 'result': { - 'id': result['id'], - 'outcome': 'PASSED' - }, - 'task': { - 'item': nvr, - 'type': 'koji_build', - 'name': 'dist.rpmdeplint' - } - } - } - hub = mock.MagicMock() - hub.config = { - 'environment': 'environment', - 'topic_prefix': 'topic_prefix', - 'greenwave_cache': {'backend': 'dogpile.cache.memory'}, - } - handler = cache.CacheInvalidatorExtraordinaire(hub) - handler.cache = mock.MagicMock() - assert handler.topic == [ - 'topic_prefix.environment.taskotron.result.new', - # Not ready to handle waiverdb yet. - #'topic_prefix.environment.waiver.new', - ] - handler.consume(message) - expected = ("greenwave.resources:retrieve_results|" - "{'item': '%s', 'type': 'koji_build'}" % nvr) - handler.cache.delete.assert_called_once_with(expected) - - -def test_consume_new_result_with_real_cache( - requests_session, cached_greenwave_server, testdatabuilder, monkeypatch): - monkeypatch.setenv('TEST', 'true') - nvr = testdatabuilder.unique_nvr() - for testcase_name in ['dist.rpmdeplint', 'dist.upgradepath', 'dist.abicheck']: - testdatabuilder.create_result( - item=nvr, testcase_name=testcase_name, outcome='PASSED') - - # get first passing decision - query = { - 'decision_context': 'bodhi_update_push_stable', - 'product_version': 'fedora-26', - 'subject': [{'item': nvr, 'type': 'koji_build'}], - } - r = requests_session.post(cached_greenwave_server.url + 'api/v1.0/decision', - headers={'Content-Type': 'application/json'}, - data=json.dumps(query)) - assert r.status_code == 200 - response = r.json() - # Ensure it is passing... - assert response['policies_satisified'], pprint.pformat(response) - - # Now, insert a new result and ensure that caching has made it such that - # even though the new result fails, our decision still passes (bad) - testdatabuilder.create_result( - item=nvr, testcase_name='dist.abicheck', outcome='FAILED') - r = requests_session.post(cached_greenwave_server.url + 'api/v1.0/decision', - headers={'Content-Type': 'application/json'}, - data=json.dumps(query)) - assert r.status_code == 200 - response = r.json() - # Ensure it is passing... BUT IT SHOULDN'T BE! - assert response['policies_satisified'], pprint.pformat(response) - - # Now, handle a message about the new failing result - message = { - u'topic': u'taskotron.result.new', - u'msg': { - u'result': { - u'id': u'whatever', - u'outcome': u'doesn\'t matter', - }, - u'task': { - u'item': nvr.decode('utf-8'), - u'type': u'koji_build', - u'name': u'dist.rpmdeplint' - } - } - } - hub = mock.MagicMock() - hub.config = { - 'environment': 'environment', - 'topic_prefix': 'topic_prefix', - 'greenwave_cache': CachedTestingConfig().CACHE, - } - handler = cache.CacheInvalidatorExtraordinaire(hub) - assert handler.topic == [ - 'topic_prefix.environment.taskotron.result.new', - # Not ready to handle waiverdb yet. - #'topic_prefix.environment.waiver.new', - ] - handler.consume(message) - - # At this point, the invalidator should have invalidated the cache. If we - # ask again, the decision should be correct now. It should be a stone cold - # "no". - r = requests_session.post(cached_greenwave_server.url + 'api/v1.0/decision', - headers={'Content-Type': 'application/json'}, - data=json.dumps(query)) - assert r.status_code == 200 - response = r.json() - # Ensure it is failing -- as it should be. - assert not response['policies_satisified'], pprint.pformat(response) - - -def test_consume_new_result_with_no_preexisting_cache( - requests_session, cached_greenwave_server, testdatabuilder, monkeypatch): - """ Ensure that invalidating an unknown value is sane. """ - monkeypatch.setenv('TEST', 'true') - nvr = testdatabuilder.unique_nvr() - result = testdatabuilder.create_result( - item=nvr, testcase_name='dist.rpmdeplint', outcome='PASSED') - message = { - 'topic': 'taskotron.result.new', - 'msg': { - 'result': { - 'id': result['id'], - 'outcome': 'PASSED' - }, - 'task': { - 'item': nvr, - 'type': 'koji_build', - 'name': 'dist.rpmdeplint' - } - } - } - hub = mock.MagicMock() - hub.config = { - 'environment': 'environment', - 'topic_prefix': 'topic_prefix', - 'greenwave_cache': {'backend': 'dogpile.cache.memory'}, - } - handler = cache.CacheInvalidatorExtraordinaire(hub) - handler.cache.delete = mock.MagicMock() - assert handler.topic == [ - 'topic_prefix.environment.taskotron.result.new', - # Not ready to handle waiverdb yet. - #'topic_prefix.environment.waiver.new', - ] - handler.consume(message) - handler.cache.delete.assert_not_called() diff --git a/functional-tests/consumers/test_resultsdb.py b/functional-tests/consumers/test_resultsdb.py index edc9aee..720e9da 100644 --- a/functional-tests/consumers/test_resultsdb.py +++ b/functional-tests/consumers/test_resultsdb.py @@ -1,15 +1,18 @@ # SPDX-License-Identifier: GPL-2.0+ -import mock import json +import mock +import pprint +from greenwave.config import CachedTestingConfig from greenwave.consumers import resultsdb @mock.patch('greenwave.consumers.resultsdb.fedmsg.config.load_config') @mock.patch('greenwave.consumers.resultsdb.fedmsg.publish') def test_consume_new_result( - mock_fedmsg, load_config, requests_session, greenwave_server, testdatabuilder, monkeypatch): + mock_fedmsg, load_config, requests_session, greenwave_server, + testdatabuilder, monkeypatch): monkeypatch.setenv('TEST', 'true') load_config.return_value = {'greenwave_api_url': greenwave_server.url + 'api/v1.0'} nvr = testdatabuilder.unique_nvr() @@ -31,7 +34,11 @@ def test_consume_new_result( } } hub = mock.MagicMock() - hub.config = {'environment': 'environment', 'topic_prefix': 'topic_prefix'} + 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'] handler.consume(message) @@ -88,7 +95,8 @@ def test_consume_new_result( @mock.patch('greenwave.consumers.resultsdb.fedmsg.config.load_config') @mock.patch('greenwave.consumers.resultsdb.fedmsg.publish') def test_no_message_for_unchanged_decision( - mock_fedmsg, load_config, requests_session, greenwave_server, testdatabuilder, monkeypatch): + mock_fedmsg, load_config, requests_session, greenwave_server, + testdatabuilder, monkeypatch): monkeypatch.setenv('TEST', 'true') load_config.return_value = {'greenwave_api_url': greenwave_server.url + 'api/v1.0'} nvr = testdatabuilder.unique_nvr() @@ -115,10 +123,179 @@ def test_no_message_for_unchanged_decision( } } hub = mock.MagicMock() - hub.config = {'environment': 'environment', 'topic_prefix': 'topic_prefix'} + 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'] handler.consume(message) # No message should be published as the decision is unchanged since we # are still missing the required tests. mock_fedmsg.assert_not_called() + + +@mock.patch('greenwave.consumers.resultsdb.fedmsg.config.load_config') +@mock.patch('greenwave.consumers.resultsdb.fedmsg.publish') +def test_invalidate_new_result_with_mocked_cache( + mock_fedmsg, load_config, requests_session, cached_greenwave_server, + testdatabuilder, monkeypatch): + """ Consume a result, and ensure that `delete` is called. """ + monkeypatch.setenv('TEST', 'true') + load_config.return_value = {'greenwave_api_url': cached_greenwave_server.url + 'api/v1.0'} + nvr = testdatabuilder.unique_nvr() + result = testdatabuilder.create_result( + item=nvr, testcase_name='dist.rpmdeplint', outcome='PASSED') + message = { + 'topic': 'taskotron.result.new', + 'msg': { + 'result': { + 'id': result['id'], + 'outcome': 'PASSED' + }, + 'task': { + 'item': nvr, + 'type': 'koji_build', + 'name': 'dist.rpmdeplint' + } + } + } + hub = mock.MagicMock() + hub.config = { + 'environment': 'environment', + 'topic_prefix': 'topic_prefix', + 'greenwave_cache': {'backend': 'dogpile.cache.memory'}, + } + handler = resultsdb.ResultsDBHandler(hub) + handler.cache = mock.MagicMock() + assert handler.topic == [ + 'topic_prefix.environment.taskotron.result.new', + # Not ready to handle waiverdb yet. + #'topic_prefix.environment.waiver.new', + ] + handler.consume(message) + expected = ("greenwave.resources:retrieve_results|" + "{'item': '%s', 'type': 'koji_build'}" % nvr) + handler.cache.delete.assert_called_once_with(expected) + + +@mock.patch('greenwave.consumers.resultsdb.fedmsg.config.load_config') +@mock.patch('greenwave.consumers.resultsdb.fedmsg.publish') +def test_invalidate_new_result_with_real_cache( + mock_fedmsg, load_config, requests_session, cached_greenwave_server, + testdatabuilder, monkeypatch): + monkeypatch.setenv('TEST', 'true') + load_config.return_value = {'greenwave_api_url': cached_greenwave_server.url + 'api/v1.0'} + nvr = testdatabuilder.unique_nvr() + for testcase_name in ['dist.rpmdeplint', 'dist.upgradepath', 'dist.abicheck']: + testdatabuilder.create_result( + item=nvr, testcase_name=testcase_name, outcome='PASSED') + + # get first passing decision + query = { + 'decision_context': 'bodhi_update_push_stable', + 'product_version': 'fedora-26', + 'subject': [{'item': nvr, 'type': 'koji_build'}], + } + r = requests_session.post(cached_greenwave_server.url + 'api/v1.0/decision', + headers={'Content-Type': 'application/json'}, + data=json.dumps(query)) + assert r.status_code == 200 + response = r.json() + # Ensure it is passing... + assert response['policies_satisified'], pprint.pformat(response) + + # Now, insert a new result and ensure that caching has made it such that + # even though the new result fails, our decision still passes (bad) + testdatabuilder.create_result( + item=nvr, testcase_name='dist.abicheck', outcome='FAILED') + r = requests_session.post(cached_greenwave_server.url + 'api/v1.0/decision', + headers={'Content-Type': 'application/json'}, + data=json.dumps(query)) + assert r.status_code == 200 + response = r.json() + # Ensure it is passing... BUT IT SHOULDN'T BE! + assert response['policies_satisified'], pprint.pformat(response) + + # Now, handle a message about the new failing result + message = { + u'topic': u'taskotron.result.new', + u'msg': { + u'result': { + u'id': u'whatever', + u'outcome': u'doesn\'t matter', + }, + u'task': { + u'item': nvr.decode('utf-8'), + u'type': u'koji_build', + u'name': u'dist.rpmdeplint' + } + } + } + hub = mock.MagicMock() + hub.config = { + 'environment': 'environment', + 'topic_prefix': 'topic_prefix', + 'greenwave_cache': CachedTestingConfig().CACHE, + } + handler = resultsdb.ResultsDBHandler(hub) + assert handler.topic == [ + 'topic_prefix.environment.taskotron.result.new', + # Not ready to handle waiverdb yet. + #'topic_prefix.environment.waiver.new', + ] + handler.consume(message) + + # At this point, the invalidator should have invalidated the cache. If we + # ask again, the decision should be correct now. It should be a stone cold + # "no". + r = requests_session.post(cached_greenwave_server.url + 'api/v1.0/decision', + headers={'Content-Type': 'application/json'}, + data=json.dumps(query)) + assert r.status_code == 200 + response = r.json() + # Ensure it is failing -- as it should be. + assert not response['policies_satisified'], pprint.pformat(response) + + +@mock.patch('greenwave.consumers.resultsdb.fedmsg.config.load_config') +@mock.patch('greenwave.consumers.resultsdb.fedmsg.publish') +def test_invalidate_new_result_with_no_preexisting_cache( + mock_fedmsg, load_config, requests_session, cached_greenwave_server, + testdatabuilder, monkeypatch): + """ Ensure that invalidating an unknown value is sane. """ + monkeypatch.setenv('TEST', 'true') + load_config.return_value = {'greenwave_api_url': cached_greenwave_server.url + 'api/v1.0'} + nvr = testdatabuilder.unique_nvr() + result = testdatabuilder.create_result( + item=nvr, testcase_name='dist.rpmdeplint', outcome='PASSED') + message = { + 'topic': 'taskotron.result.new', + 'msg': { + 'result': { + 'id': result['id'], + 'outcome': 'PASSED' + }, + 'task': { + 'item': nvr, + 'type': 'koji_build', + 'name': 'dist.rpmdeplint' + } + } + } + hub = mock.MagicMock() + hub.config = { + 'environment': 'environment', + 'topic_prefix': 'topic_prefix', + 'greenwave_cache': {'backend': 'dogpile.cache.memory'}, + } + handler = resultsdb.ResultsDBHandler(hub) + handler.cache.delete = mock.MagicMock() + assert handler.topic == [ + 'topic_prefix.environment.taskotron.result.new', + # Not ready to handle waiverdb yet. + #'topic_prefix.environment.waiver.new', + ] + handler.consume(message) + handler.cache.delete.assert_not_called() diff --git a/greenwave/consumers/cache.py b/greenwave/consumers/cache.py deleted file mode 100644 index e5c2ce1..0000000 --- a/greenwave/consumers/cache.py +++ /dev/null @@ -1,82 +0,0 @@ -# -*- coding: utf-8 -*- -# SPDX-License-Identifier: GPL-2.0+ -""" -The "cache handler". - -This module is responsible for listening for new results from ResultsDB (and -eventually waiverdb also). When a new result or waiver is received, this code -will lookup any possible cache values we have for that item and destroy them -- -invalidate them. - -https://pagure.io/greenwave/issue/77 -""" - -import logging -import requests -import dogpile.cache -import fedmsg.consumers - -import greenwave.cache -import greenwave.resources - -requests_session = requests.Session() - - -log = logging.getLogger(__name__) - - -class CacheInvalidatorExtraordinaire(fedmsg.consumers.FedmsgConsumer): - """ - Handle a new result or waiver. - - Attributes: - topic (list): A list of strings that indicate which fedmsg topics this consumer listens to. - """ - - config_key = 'cache_invalidator' - - def __init__(self, hub, *args, **kwargs): - """ - Initialize the CacheInvalidatorExtraordinaire, 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') - self.topic = [ - prefix + '.' + env + '.taskotron.result.new', - # Not ready to handle waivers yet... - #prefix + '.' + env + '.waiver.new', - ] - self.fedmsg_config = fedmsg.config.load_config() - super(CacheInvalidatorExtraordinaire, self).__init__(hub, *args, **kwargs) - log.info('Greenwave cache invalidator listening on:\n' - '%r' % self.topic) - - # Initialize the cache. - self.cache = dogpile.cache.make_region() - self.cache.configure(**hub.config['greenwave_cache']) - - def consume(self, message): - """ - Process the given message and delete cache keys as necessary. - - Args: - message (munch.Munch): A fedmsg about a new result or waiver. - """ - log.debug('Processing message "{0}"'.format(message)) - msg = message['msg'] - task = msg['task'] - del task['name'] - # here, task is {"item": "nodejs-ansi-black-0.1.1-1.fc28", "type": "koji_build" } - namespace = None - fn = greenwave.resources.retrieve_results - key = greenwave.cache.key_generator(namespace, fn)(task) - if not self.cache.get(key): - log.debug("No cache value found for %r" % key) - else: - log.debug("Invalidating cache for %r" % key) - self.cache.delete(key) diff --git a/greenwave/consumers/resultsdb.py b/greenwave/consumers/resultsdb.py index bf4b9e5..41212ef 100644 --- a/greenwave/consumers/resultsdb.py +++ b/greenwave/consumers/resultsdb.py @@ -9,11 +9,16 @@ and if the new result causes the decision to change it will publish a message to the message bus about the newly satisfied/unsatisfied policy. """ -import logging -import requests +import copy import json +import logging + +import dogpile.cache import fedmsg.consumers +import requests +import greenwave.cache +import greenwave.resources from greenwave.utils import load_config requests_session = requests.Session() @@ -47,18 +52,34 @@ class ResultsDBHandler(fedmsg.consumers.FedmsgConsumer): prefix + '.' + env + '.taskotron.result.new', ] self.fedmsg_config = fedmsg.config.load_config() + super(ResultsDBHandler, self).__init__(hub, *args, **kwargs) + + # Initialize the cache. + self.cache = dogpile.cache.make_region() + self.cache.configure(**hub.config['greenwave_cache']) + log.info('Greenwave resultsdb handler listening on: %s', self.topic) def consume(self, message): """ - Process the given message and publish a message if the decision is changed. + Process the given message and take action. Args: message (munch.Munch): A fedmsg about a new result. """ log.debug('Processing message "%s"', message) - msg = message['msg'] + self._invalidate_cache(message) + self._publish_decision_changes(message) + + def _publish_decision_changes(self, message): + """ + Process the given message and publish a message if the decision is changed. + + Args: + message (munch.Munch): A fedmsg about a new result. + """ + msg = copy.deepcopy(message['msg']) task = msg['task'] testcase = task['name'] del task['name'] @@ -102,3 +123,23 @@ class ResultsDBHandler(fedmsg.consumers.FedmsgConsumer): log.debug('Emitted a fedmsg, %r, on the "%s" topic', msg, 'greenwave.decision.update') fedmsg.publish(topic='greenwave.decision.update', msg=msg) + + def _invalidate_cache(self, message): + """ + Process the given message and delete cache keys as necessary. + + Args: + message (munch.Munch): A fedmsg about a new result or waiver. + """ + msg = copy.deepcopy(message['msg']) + task = msg['task'] + del task['name'] + # here, task is {"item": "nodejs-ansi-black-0.1.1-1.fc28", "type": "koji_build" } + namespace = None + fn = greenwave.resources.retrieve_results + key = greenwave.cache.key_generator(namespace, fn)(task) + if not self.cache.get(key): + log.debug("No cache value found for %r" % key) + else: + log.debug("Invalidating cache for %r" % key) + self.cache.delete(key)