Add vault_kv2_delete module

plugins/modules/vault_kv2_delete.py

@@ -0,0 +1,71 @@
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+DOCUMENTATION = r'''
+module: vault_kv2_get
+'''
+
+
+import traceback
+
+from ansible.module_utils._text import to_native
+from ansible.module_utils.basic import missing_required_lib
+
+from ansible_collections.community.hashi_vault.plugins.module_utils._hashi_vault_module import HashiVaultModule
+from ansible_collections.community.hashi_vault.plugins.module_utils._hashi_vault_common import HashiVaultValueError
+
+try:
+    import hvac
+except ImportError:
+    HAS_HVAC = False
+    HVAC_IMPORT_ERROR = traceback.format_exc()
+else:
+    HVAC_IMPORT_ERROR = None
+    HAS_HVAC = True
+
+def run_module():
+
+
+    argspec = HashiVaultModule.generate_argspec(
+        engine_mount_point=dict(type='str', default='kv'),
+        path=dict(type='str', required=True)
+    )
+
+    module = HashiVaultModule(
+        argument_spec=argspec,
+        supports_check_mode=True
+    )
+
+    if not HAS_HVAC:
+        module.fail_json(
+            msg=missing_required_lib('hvac'),
+            exception=HVAC_IMPORT_ERROR
+        )
+
+    engine_mount_point = module.params.get('engine_mount_point')
+    path = module.params.get('path')
+
+    module.connection_options.process_connection_options()
+    client_args = module.connection_options.get_hvac_connection_options()
+    client = module.helper.get_vault_client(**client_args)
+
+    try:
+        module.authenticator.validate()
+        module.authenticator.authenticate(client)
+    except (NotImplementedError, HashiVaultValueError) as e:
+        module.fail_json(msg=to_native(e), exception=traceback.format_exc())
+
+    try:
+        raw = client.secrets.kv.v2.delete_latest_version_of_secret(path=path, mount_point=engine_mount_point)
+    except hvac.exceptions.Forbidden as e:
+        module.fail_json(msg="Forbidden: Permission Denied to path ['%s']." % path, exception=traceback.format_exc())
+
+    module.exit_json(changed=True, response_code=raw.status_code)
+
+def main():
+    run_module()
+
+
+if __name__ == '__main__':
+    main()

tests/unit/plugins/modules/test_vault_kv2_delete.py

@@ -0,0 +1,154 @@
+# -*- coding: utf-8 -*-
+# Copyright (c) 2022 Brian Scholer (@briantist)
+# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+import pytest
+import re
+import json
+
+from ansible.module_utils.basic import missing_required_lib
+
+from ...compat import mock
+from .....plugins.modules import vault_kv2_delete
+from .....plugins.module_utils._hashi_vault_common import HashiVaultValueError
+
+
+hvac = pytest.importorskip('hvac')
+
+
+pytestmark = pytest.mark.usefixtures(
+    'patch_ansible_module',
+    'patch_authenticator',
+    'patch_get_vault_client',
+)
+
+
+def _connection_options():
+    return {
+        'auth_method': 'token',
+        'url': 'http://myvault',
+        'token': 'beep-boop',
+    }
+
+
+def _sample_options():
+    return {
+        'engine_mount_point': 'secret',
+        'path': 'endpoint',
+    }
+
+
+def _combined_options(**kwargs):
+    opt = _connection_options()
+    opt.update(_sample_options())
+    opt.update(kwargs)
+    return opt
+
+
+@pytest.fixture
+def kv2_delete_response():
+    class request_response:
+        status_code = 204
+
+    return request_response
+
+
+class TestModuleVaultKv2Delete():
+
+    @pytest.mark.parametrize('patch_ansible_module', [_combined_options()], indirect=True)
+    @pytest.mark.parametrize('exc', [HashiVaultValueError('throwaway msg'), NotImplementedError('throwaway msg')])
+    def test_vault_kv2_delete_authentication_error(self, authenticator, exc, capfd):
+        authenticator.authenticate.side_effect = exc
+
+        with pytest.raises(SystemExit) as e:
+            vault_kv2_delete.main()
+
+        out, err = capfd.readouterr()
+        result = json.loads(out)
+
+        assert e.value.code != 0, "result: %r" % (result,)
+        assert result['msg'] == 'throwaway msg', "result: %r" % result
+
+    @pytest.mark.parametrize('patch_ansible_module', [_combined_options()], indirect=True)
+    @pytest.mark.parametrize('exc', [HashiVaultValueError('throwaway msg'), NotImplementedError('throwaway msg')])
+    def test_vault_kv2_get_auth_validation_error(self, authenticator, exc, capfd):
+        authenticator.validate.side_effect = exc
+
+        with pytest.raises(SystemExit) as e:
+            vault_kv2_delete.main()
+
+        out, err = capfd.readouterr()
+        result = json.loads(out)
+
+        assert e.value.code != 0, "result: %r" % (result,)
+        assert result['msg'] == 'throwaway msg'
+
+    @pytest.mark.parametrize('opt_engine_mount_point', ['secret', 'other'])
+    @pytest.mark.parametrize('patch_ansible_module', [[_combined_options(), 'engine_mount_point']], indirect=True)
+    def test_vault_kv2_delete_return_data(self, patch_ansible_module, kv2_delete_response, vault_client, opt_engine_mount_point, capfd):
+        client = vault_client
+        client.secrets.kv.v2.delete_latest_version_of_secret.return_value = kv2_delete_response
+
+        expected_response = 204
+
+        with pytest.raises(SystemExit) as e:
+            vault_kv2_delete.main()
+
+        out, err = capfd.readouterr()
+        result = json.loads(out)
+
+        assert e.value.code == 0, "result: %r" % (result,)
+
+        client.secrets.kv.v2.delete_latest_version_of_secret.assert_called_once_with(
+            path=patch_ansible_module['path'],
+            mount_point=patch_ansible_module['engine_mount_point'],
+        )
+
+        assert result['response_code'] == expected_response, (
+            "module result did not match expected result:\nmodule: %r\nexpected: %r" % (
+                result['response_code'], expected_response)
+        )
+
+    @pytest.mark.parametrize('patch_ansible_module', [_combined_options()], indirect=True)
+    def test_vault_kv2_delete_no_hvac(self, capfd):
+        with mock.patch.multiple(vault_kv2_delete, HAS_HVAC=False, HVAC_IMPORT_ERROR=None, create=True):
+            with pytest.raises(SystemExit) as e:
+                vault_kv2_delete.main()
+
+        out, err = capfd.readouterr()
+        result = json.loads(out)
+
+        assert e.value.code != 0, "result: %r" % (result,)
+        assert result['msg'] == missing_required_lib('hvac')
+
+    @pytest.mark.parametrize(
+        'exc',
+        [
+            (hvac.exceptions.Forbidden, "",
+             r"^Forbidden: Permission Denied to path \['([^']+)'\]"),
+        ]
+    )
+    @pytest.mark.parametrize('patch_ansible_module', [[_combined_options(), 'path']], indirect=True)
+    @pytest.mark.parametrize('opt_path', ['path/1', 'second/path'])
+    def test_vault_kv2_get_vault_exception(self, vault_client, exc, opt_path, capfd):
+
+        client = vault_client
+        client.secrets.kv.v2.delete_latest_version_of_secret.side_effect = exc[0](
+            exc[1])
+
+        with pytest.raises(SystemExit) as e:
+            vault_kv2_delete.main()
+
+        out, err = capfd.readouterr()
+        result = json.loads(out)
+
+        assert e.value.code != 0, "result: %r" % (result,)
+        match = re.search(exc[2], result['msg'])
+        assert match is not None, "result: %r\ndid not match: %s" % (
+            result, exc[2])
+
+        assert opt_path == match.group(1)