diff options
Diffstat (limited to '')
-rw-r--r-- | test/integration/targets/inventory_cache/plugins/inventory/cache_host.py | 56 | ||||
-rw-r--r-- | test/integration/targets/inventory_cache/plugins/inventory/exercise_cache.py | 344 |
2 files changed, 400 insertions, 0 deletions
diff --git a/test/integration/targets/inventory_cache/plugins/inventory/cache_host.py b/test/integration/targets/inventory_cache/plugins/inventory/cache_host.py new file mode 100644 index 0000000..628aba1 --- /dev/null +++ b/test/integration/targets/inventory_cache/plugins/inventory/cache_host.py @@ -0,0 +1,56 @@ +# Copyright (c) 2021 Ansible Project +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +DOCUMENTATION = ''' + inventory: cache_host + short_description: add a host to inventory and cache it + description: add a host to inventory and cache it + extends_documentation_fragment: + - inventory_cache + options: + plugin: + required: true + description: name of the plugin (cache_host) +''' + +from ansible.plugins.inventory import BaseInventoryPlugin, Cacheable +import random + + +class InventoryModule(BaseInventoryPlugin, Cacheable): + + NAME = 'cache_host' + + def verify_file(self, path): + if not path.endswith(('cache_host.yml', 'cache_host.yaml',)): + return False + return super(InventoryModule, self).verify_file(path) + + def parse(self, inventory, loader, path, cache=None): + super(InventoryModule, self).parse(inventory, loader, path) + self._read_config_data(path) + + cache_key = self.get_cache_key(path) + # user has enabled cache and the cache is not being flushed + read_cache = self.get_option('cache') and cache + # user has enabled cache and the cache is being flushed + update_cache = self.get_option('cache') and not cache + + host = None + if read_cache: + try: + host = self._cache[cache_key] + except KeyError: + # cache expired + update_cache = True + + if host is None: + host = 'testhost{0}'.format(random.randint(0, 50)) + + self.inventory.add_host(host, 'all') + + if update_cache: + self._cache[cache_key] = host diff --git a/test/integration/targets/inventory_cache/plugins/inventory/exercise_cache.py b/test/integration/targets/inventory_cache/plugins/inventory/exercise_cache.py new file mode 100644 index 0000000..cca2aa0 --- /dev/null +++ b/test/integration/targets/inventory_cache/plugins/inventory/exercise_cache.py @@ -0,0 +1,344 @@ +# Copyright (c) 2022 Ansible Project +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +DOCUMENTATION = ''' + inventory: exercise_cache + short_description: run tests against the specified cache plugin + description: + - This plugin doesn't modify inventory. + - Load a cache plugin and test the inventory cache interface is dict-like. + - Most inventory cache write methods only apply to the in-memory cache. + - The 'flush' and 'set_cache' methods should be used to apply changes to the backing cache plugin. + - The inventory cache read methods prefer the in-memory cache, and fall back to reading from the cache plugin. + extends_documentation_fragment: + - inventory_cache + options: + plugin: + required: true + description: name of the plugin (exercise_cache) + cache_timeout: + ini: [] + env: [] + cli: [] + default: 0 # never expire +''' + +from ansible.errors import AnsibleError +from ansible.plugins.inventory import BaseInventoryPlugin, Cacheable +from ansible.utils.display import Display + +from time import sleep + + +display = Display() + + +class InventoryModule(BaseInventoryPlugin, Cacheable): + + NAME = 'exercise_cache' + + test_cache_methods = [ + 'test_plugin_name', + 'test_update_cache_if_changed', + 'test_set_cache', + 'test_load_whole_cache', + 'test_iter', + 'test_len', + 'test_get_missing_key', + 'test_get_expired_key', + 'test_initial_get', + 'test_get', + 'test_items', + 'test_keys', + 'test_values', + 'test_pop', + 'test_del', + 'test_set', + 'test_update', + 'test_flush', + ] + + def verify_file(self, path): + if not path.endswith(('exercise_cache.yml', 'exercise_cache.yaml',)): + return False + return super(InventoryModule, self).verify_file(path) + + def parse(self, inventory, loader, path, cache=None): + super(InventoryModule, self).parse(inventory, loader, path) + self._read_config_data(path) + + try: + self.exercise_test_cache() + except AnsibleError: + raise + except Exception as e: + raise AnsibleError("Failed to run cache tests: {0}".format(e)) from e + + def exercise_test_cache(self): + failed = [] + for test_name in self.test_cache_methods: + try: + getattr(self, test_name)() + except AssertionError: + failed.append(test_name) + finally: + self.cache.flush() + self.cache.update_cache_if_changed() + + if failed: + raise AnsibleError(f"Cache tests failed: {', '.join(failed)}") + + def test_equal(self, a, b): + try: + assert a == b + except AssertionError: + display.warning(f"Assertion {a} == {b} failed") + raise + + def test_plugin_name(self): + self.test_equal(self.cache._plugin_name, self.get_option('cache_plugin')) + + def test_update_cache_if_changed(self): + self.cache._retrieved = {} + self.cache._cache = {'foo': 'bar'} + + self.cache.update_cache_if_changed() + + self.test_equal(self.cache._retrieved, {'foo': 'bar'}) + self.test_equal(self.cache._cache, {'foo': 'bar'}) + + def test_set_cache(self): + cache_key1 = 'key1' + cache1 = {'hosts': {'h1': {'foo': 'bar'}}} + cache_key2 = 'key2' + cache2 = {'hosts': {'h2': {}}} + + self.cache._cache = {cache_key1: cache1, cache_key2: cache2} + self.cache.set_cache() + + self.test_equal(self.cache._plugin.contains(cache_key1), True) + self.test_equal(self.cache._plugin.get(cache_key1), cache1) + self.test_equal(self.cache._plugin.contains(cache_key2), True) + self.test_equal(self.cache._plugin.get(cache_key2), cache2) + + def test_load_whole_cache(self): + cache_data = { + 'key1': {'hosts': {'h1': {'foo': 'bar'}}}, + 'key2': {'hosts': {'h2': {}}}, + } + self.cache._cache = cache_data + self.cache.set_cache() + self.cache._cache = {} + + self.cache.load_whole_cache() + self.test_equal(self.cache._cache, cache_data) + + def test_iter(self): + cache_data = { + 'key1': {'hosts': {'h1': {'foo': 'bar'}}}, + 'key2': {'hosts': {'h2': {}}}, + } + self.cache._cache = cache_data + self.test_equal(sorted(list(self.cache)), ['key1', 'key2']) + + def test_len(self): + cache_data = { + 'key1': {'hosts': {'h1': {'foo': 'bar'}}}, + 'key2': {'hosts': {'h2': {}}}, + } + self.cache._cache = cache_data + self.test_equal(len(self.cache), 2) + + def test_get_missing_key(self): + # cache should behave like a dictionary + # a missing key with __getitem__ should raise a KeyError + try: + self.cache['keyerror'] + except KeyError: + pass + else: + assert False + + # get should return the default instead + self.test_equal(self.cache.get('missing'), None) + self.test_equal(self.cache.get('missing', 'default'), 'default') + + def _setup_expired(self): + self.cache._cache = {'expired': True} + self.cache.set_cache() + + # empty the in-memory info to test loading the key + # keys that expire mid-use do not cause errors + self.cache._cache = {} + self.cache._retrieved = {} + self.cache._plugin._cache = {} + + self.cache._plugin.set_option('timeout', 1) + self.cache._plugin._timeout = 1 + sleep(2) + + def _cleanup_expired(self): + # Set cache timeout back to never + self.cache._plugin.set_option('timeout', 0) + self.cache._plugin._timeout = 0 + + def test_get_expired_key(self): + if not hasattr(self.cache._plugin, '_timeout'): + # DB-backed caches do not have a standard timeout interface + return + + self._setup_expired() + try: + self.cache['expired'] + except KeyError: + pass + else: + assert False + finally: + self._cleanup_expired() + + self._setup_expired() + try: + self.test_equal(self.cache.get('expired'), None) + self.test_equal(self.cache.get('expired', 'default'), 'default') + finally: + self._cleanup_expired() + + def test_initial_get(self): + # test cache behaves like a dictionary + + # set the cache to test getting a key that exists + k1 = {'hosts': {'h1': {'foo': 'bar'}}} + k2 = {'hosts': {'h2': {}}} + self.cache._cache = {'key1': k1, 'key2': k2} + self.cache.set_cache() + + # empty the in-memory info to test loading the key from the plugin + self.cache._cache = {} + self.cache._retrieved = {} + self.cache._plugin._cache = {} + + self.test_equal(self.cache['key1'], k1) + + # empty the in-memory info to test loading the key from the plugin + self.cache._cache = {} + self.cache._retrieved = {} + self.cache._plugin._cache = {} + + self.test_equal(self.cache.get('key1'), k1) + + def test_get(self): + # test cache behaves like a dictionary + + # set the cache to test getting a key that exists + k1 = {'hosts': {'h1': {'foo': 'bar'}}} + k2 = {'hosts': {'h2': {}}} + self.cache._cache = {'key1': k1, 'key2': k2} + self.cache.set_cache() + + self.test_equal(self.cache['key1'], k1) + self.test_equal(self.cache.get('key1'), k1) + + def test_items(self): + self.test_equal(self.cache.items(), {}.items()) + + test_items = {'hosts': {'host1': {'foo': 'bar'}}} + self.cache._cache = test_items + self.test_equal(self.cache.items(), test_items.items()) + + def test_keys(self): + self.test_equal(self.cache.keys(), {}.keys()) + + test_items = {'hosts': {'host1': {'foo': 'bar'}}} + self.cache._cache = test_items + self.test_equal(self.cache.keys(), test_items.keys()) + + def test_values(self): + self.test_equal(list(self.cache.values()), list({}.values())) + + test_items = {'hosts': {'host1': {'foo': 'bar'}}} + self.cache._cache = test_items + self.test_equal(list(self.cache.values()), list(test_items.values())) + + def test_pop(self): + try: + self.cache.pop('missing') + except KeyError: + pass + else: + assert False + + self.test_equal(self.cache.pop('missing', 'default'), 'default') + + self.cache._cache = {'cache_key': 'cache'} + self.test_equal(self.cache.pop('cache_key'), 'cache') + + # test backing plugin cache isn't modified + cache_key1 = 'key1' + cache1 = {'hosts': {'h1': {'foo': 'bar'}}} + cache_key2 = 'key2' + cache2 = {'hosts': {'h2': {}}} + + self.cache._cache = {cache_key1: cache1, cache_key2: cache2} + self.cache.set_cache() + + self.test_equal(self.cache.pop('key1'), cache1) + self.test_equal(self.cache._cache, {cache_key2: cache2}) + self.test_equal(self.cache._plugin._cache, {cache_key1: cache1, cache_key2: cache2}) + + def test_del(self): + try: + del self.cache['missing'] + except KeyError: + pass + else: + assert False + + cache_key1 = 'key1' + cache1 = {'hosts': {'h1': {'foo': 'bar'}}} + cache_key2 = 'key2' + cache2 = {'hosts': {'h2': {}}} + + self.cache._cache = {cache_key1: cache1, cache_key2: cache2} + self.cache.set_cache() + + del self.cache['key1'] + + self.test_equal(self.cache._cache, {cache_key2: cache2}) + self.test_equal(self.cache._plugin._cache, {cache_key1: cache1, cache_key2: cache2}) + + def test_set(self): + cache_key = 'key1' + hosts = {'hosts': {'h1': {'foo': 'bar'}}} + self.cache[cache_key] = hosts + + self.test_equal(self.cache._cache, {cache_key: hosts}) + self.test_equal(self.cache._plugin._cache, {}) + + def test_update(self): + cache_key1 = 'key1' + cache1 = {'hosts': {'h1': {'foo': 'bar'}}} + cache_key2 = 'key2' + cache2 = {'hosts': {'h2': {}}} + + self.cache._cache = {cache_key1: cache1} + self.cache.update({cache_key2: cache2}) + self.test_equal(self.cache._cache, {cache_key1: cache1, cache_key2: cache2}) + + def test_flush(self): + cache_key1 = 'key1' + cache1 = {'hosts': {'h1': {'foo': 'bar'}}} + cache_key2 = 'key2' + cache2 = {'hosts': {'h2': {}}} + + self.cache._cache = {cache_key1: cache1, cache_key2: cache2} + self.cache.set_cache() + + # Unlike the dict write methods, cache.flush() flushes the backing plugin + self.cache.flush() + + self.test_equal(self.cache._cache, {}) + self.test_equal(self.cache._plugin._cache, {}) |