From 53e64758f42893232796abcfaca69bf7b35bcf9f Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
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 <rbean@redhat.com>
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 <rbean@redhat.com>
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 <rbean@redhat.com>
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 <rbean@redhat.com>
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 <rbean@redhat.com>
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 <rbean@redhat.com>
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 <rbean@redhat.com>
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)

