Make cache adjudicator's flush call plugin flush (#70987)

Change:
- Previously CachePluginAdjudicator#flush only removed entries from the
  cache backend that it knew about by using them earlier. Now it calls
  the underlying plugin's flush() method.

Test Plan:
- New unit tests

Tickets:
- Fixes #68770

Signed-off-by: Rick Elrod <rick@elrod.me>

changelogs/fragments/68770_cache_adjudicator_flush.yml

@@ -0,0 +1,2 @@
+bugfixes:
+  - The ``flush()`` method of ``CachePluginAdjudicator`` now calls the plugin's ``flush()`` method instead of iterating over the keys that the adjudicator knows about and deleting those from the cache. (https://github.com/ansible/ansible/issues/68770)

docs/docsite/rst/dev_guide/developing_inventory.rst

@@ -274,7 +274,7 @@ After the ``parse`` method is complete, the contents of ``self._cache`` is used
 You have three other cache methods available:
   - ``set_cache_plugin`` forces the cache plugin to be set with the contents of ``self._cache`` before the ``parse`` method completes
   - ``update_cache_if_changed`` sets the cache plugin only if ``self._cache`` has been modified before the ``parse`` method completes
-  - ``clear_cache`` deletes the keys in ``self._cache`` from your cache plugin
+  - ``clear_cache`` flushes the cache, ultimately by calling the cache plugin's ``flush()`` method, whose implementation is dependent upon the particular cache plugin in use. Note that if the user is using the same cache backend for facts and inventory, both will get flushed. To avoid this, the user can specify a distinct cache backend in their inventory plugin configuration.
 
 .. _inventory_source_common_format:
 

docs/docsite/rst/porting_guides/porting_guide_base_2.11.rst

@@ -137,8 +137,7 @@ Noteworthy module changes
 Plugins
 =======
 
-No notable changes
-
+* inventory plugins - ``CachePluginAdjudicator.flush()`` now calls the underlying cache plugin's ``flush()`` instead of only deleting keys that it knows about. Inventory plugins should use ``delete()`` to remove any specific keys. As a user, this means that when an inventory plugin calls its ``clear_cache()`` method, facts could also be flushed from the cache. To work around this, users can configure inventory plugins to use a cache backend that is independent of the facts cache.
 
 Porting custom scripts
 ======================

lib/ansible/plugins/cache/__init__.py

@@ -368,8 +368,7 @@ class CachePluginAdjudicator(MutableMapping):
         self._cache[key] = value
 
     def flush(self):
-        for key in self._cache.keys():
-            self._plugin.delete(key)
+        self._plugin.flush()
         self._cache = {}
 
     def update(self, value):

test/units/plugins/cache/test_cache.py

@@ -29,11 +29,12 @@ from ansible.plugins.loader import cache_loader
 import pytest
 
 
-class TestCachePluginAdjudicator:
-    # memory plugin cache
-    cache = CachePluginAdjudicator()
-    cache['cache_key'] = {'key1': 'value1', 'key2': 'value2'}
-    cache['cache_key_2'] = {'key': 'value'}
+class TestCachePluginAdjudicator(unittest.TestCase):
+    def setUp(self):
+        # memory plugin cache
+        self.cache = CachePluginAdjudicator()
+        self.cache['cache_key'] = {'key1': 'value1', 'key2': 'value2'}
+        self.cache['cache_key_2'] = {'key': 'value'}
 
     def test___setitem__(self):
         self.cache['new_cache_key'] = {'new_key1': ['new_value1', 'new_value2']}
@@ -77,6 +78,23 @@ class TestCachePluginAdjudicator:
         self.cache.update({'cache_key': {'key2': 'updatedvalue'}})
         assert self.cache['cache_key']['key2'] == 'updatedvalue'
 
+    def test_flush(self):
+        # Fake that the cache already has some data in it but the adjudicator
+        # hasn't loaded it in.
+        self.cache._plugin.set('monkey', 'animal')
+        self.cache._plugin.set('wolf', 'animal')
+        self.cache._plugin.set('another wolf', 'another animal')
+
+        # The adjudicator does't know about the new entries
+        assert len(self.cache) == 2
+        # But the cache itself does
+        assert len(self.cache._plugin._cache) == 3
+
+        # If we call flush, both the adjudicator and the cache should flush
+        self.cache.flush()
+        assert len(self.cache) == 0
+        assert len(self.cache._plugin._cache) == 0
+
 
 class TestFactCache(unittest.TestCase):