Replace code for old API with hard-coded 'storagebox ID unknown'.

Author: felixfontein

changelogs/fragments/173-storagebox.yml

@@ -0,0 +1,2 @@
+minor_changes:
+  - "storagebox* modules - the code for the old API (that has been removed by Hetzner) has been replaced by hard-coding the result of the API, namely that no storagebox of this ID exists (https://github.com/ansible-collections/community.hrobot/pull/173)."

plugins/modules/storagebox.py

@@ -124,10 +124,8 @@
 from ansible.module_utils.common.text.converters import to_native
 
 from ansible_collections.community.hrobot.plugins.module_utils.robot import (
-    BASE_URL,
     ROBOT_DEFAULT_ARGUMENT_SPEC,
     _ROBOT_DEFAULT_ARGUMENT_SPEC_COMPAT_DEPRECATED,
-    fetch_url_json,
 )
 
 from ansible_collections.community.hrobot.plugins.module_utils.api import (
@@ -139,21 +137,6 @@
     api_fetch_url_json,
 )
 
-try:
-    from urllib.parse import urlencode
-except ImportError:
-    # Python 2.x fallback:
-    from urllib import urlencode
-
-
-PARAMETERS_LEGACY = {
-    'name': ('name', 'storagebox_name'),
-    'webdav': ('webdav', 'webdav'),
-    'samba': ('samba', 'samba'),
-    'ssh': ('ssh', 'ssh'),
-    'external_reachability': ('external_reachability', 'external_reachability'),
-    'zfs': ('zfs', 'zfs'),
-}
 
 UPDATE_PARAMETERS = {
     'name': ('name', ['name'], 'name'),
@@ -171,11 +154,6 @@
 PARAMETERS.update(ACTION_PARAMETERS)
 
 
-def extract_legacy(result):
-    sb = result['storagebox']
-    return {key: sb.get(key) for key, dummy in PARAMETERS_LEGACY.values()}
-
-
 def extract(result):
     sb = result['storage_box']
 
@@ -221,101 +199,68 @@ def main():
             collection_name="community.hrobot",
             version="3.0.0",
         )
-        # DEPRECATED: old API
-        url = "{0}/storagebox/{1}".format(BASE_URL, storagebox_id)
-        result, error = fetch_url_json(module, url, accept_errors=['STORAGEBOX_NOT_FOUND'])
-        if error:
-            module.fail_json(msg='Storagebox with ID {0} does not exist'.format(storagebox_id))
-
-        before = extract_legacy(result)
-        after = dict(before)
-
-        for option_name, (data_name, change_name) in PARAMETERS_LEGACY.items():
-            value = module.params[option_name]
-            if value is not None:
-                if before[data_name] != value:
-                    after[data_name] = value
-                    if isinstance(value, bool):
-                        changes[change_name] = str(value).lower()
-                    else:
-                        changes[change_name] = value
-
-        if changes and not module.check_mode:
-            headers = {"Content-type": "application/x-www-form-urlencoded"}
-            result, error = fetch_url_json(
-                module,
-                url,
-                data=urlencode(changes),
-                headers=headers,
-                method='POST',
-                accept_errors=['INVALID_INPUT'],
-            )
-            if error:
-                invalid = result['error'].get('invalid') or []
-                module.fail_json(msg='The values to update were invalid ({0})'.format(', '.join(invalid)))
-            after = extract_legacy(result)
-
-    else:
-        # NEW API!
-        url = "{0}/v1/storage_boxes/{1}".format(API_BASE_URL, storagebox_id)
-        result, dummy, error = api_fetch_url_json(module, url, accept_errors=['not_found'])
+        module.warn("The old storagebox API has been disabled by Hetzner. The supporting code has been removed.")
+        module.fail_json(msg='Storagebox with ID {0} does not exist'.format(storagebox_id))
+
+    url = "{0}/v1/storage_boxes/{1}".format(API_BASE_URL, storagebox_id)
+    result, dummy, error = api_fetch_url_json(module, url, accept_errors=['not_found'])
+    if error:
+        module.fail_json(msg='Storagebox with ID {0} does not exist'.format(storagebox_id))
+
+    before = extract(result)
+    after = dict(before)
+
+    update = {}
+    for option_name, (data_name, dummy, change_name) in UPDATE_PARAMETERS.items():
+        value = module.params[option_name]
+        if value is not None:
+            if before[data_name] != value:
+                after[data_name] = value
+                changes[change_name] = value
+                update[change_name] = value
+
+    action = {}
+    update_after_update = {}
+    for option_name, (data_name, dummy, change_name) in ACTION_PARAMETERS.items():
+        value = module.params[option_name]
+        if value is not None:
+            if before[data_name] != value:
+                after[data_name] = value
+                update_after_update[data_name] = value
+                changes[change_name] = value
+                action[change_name] = value
+
+    if update and not module.check_mode:
+        headers = {"Content-type": "application/json"}
+        result, dummy, error = api_fetch_url_json(
+            module,
+            url,
+            data=module.jsonify(update),
+            headers=headers,
+            method='PUT',
+            accept_errors=['invalid_input'],
+        )
         if error:
-            module.fail_json(msg='Storagebox with ID {0} does not exist'.format(storagebox_id))
-
-        before = extract(result)
-        after = dict(before)
-
-        update = {}
-        for option_name, (data_name, dummy, change_name) in UPDATE_PARAMETERS.items():
-            value = module.params[option_name]
-            if value is not None:
-                if before[data_name] != value:
-                    after[data_name] = value
-                    changes[change_name] = value
-                    update[change_name] = value
-
-        action = {}
-        update_after_update = {}
-        for option_name, (data_name, dummy, change_name) in ACTION_PARAMETERS.items():
-            value = module.params[option_name]
-            if value is not None:
-                if before[data_name] != value:
-                    after[data_name] = value
-                    update_after_update[data_name] = value
-                    changes[change_name] = value
-                    action[change_name] = value
-
-        if update and not module.check_mode:
-            headers = {"Content-type": "application/json"}
-            result, dummy, error = api_fetch_url_json(
+            details = result['error'].get('details') or {}
+            fields = details.get("fields") or []
+            details_str = ", ".join(['{0}: {1}'.format(to_native(field["name"]), to_native(field["message"])) for field in fields])
+            module.fail_json(msg='The values to update were invalid ({0})'.format(details_str or "no details"))
+        after = extract(result)
+
+    if action and not module.check_mode:
+        after.update(update_after_update)
+        action_url = "{0}/actions/update_access_settings".format(url)
+        try:
+            api_apply_action(
                 module,
-                url,
-                data=module.jsonify(update),
-                headers=headers,
-                method='PUT',
-                accept_errors=['invalid_input'],
+                action_url,
+                action,
+                lambda action_id: "{0}/v1/storage_boxes/actions/{1}".format(API_BASE_URL, action_id),
+                check_done_delay=1,
+                check_done_timeout=60,
             )
-            if error:
-                details = result['error'].get('details') or {}
-                fields = details.get("fields") or []
-                details_str = ", ".join(['{0}: {1}'.format(to_native(field["name"]), to_native(field["message"])) for field in fields])
-                module.fail_json(msg='The values to update were invalid ({0})'.format(details_str or "no details"))
-            after = extract(result)
-
-        if action and not module.check_mode:
-            after.update(update_after_update)
-            action_url = "{0}/actions/update_access_settings".format(url)
-            try:
-                api_apply_action(
-                    module,
-                    action_url,
-                    action,
-                    lambda action_id: "{0}/v1/storage_boxes/actions/{1}".format(API_BASE_URL, action_id),
-                    check_done_delay=1,
-                    check_done_timeout=60,
-                )
-            except ApplyActionError as exc:
-                module.fail_json(msg='Error while updating access settings: {0}'.format(exc))
+        except ApplyActionError as exc:
+            module.fail_json(msg='Error while updating access settings: {0}'.format(exc))
 
     result = dict(after)
     result['changed'] = bool(changes)

plugins/modules/storagebox_info.py

@@ -539,10 +539,8 @@
 from ansible.module_utils.basic import AnsibleModule
 
 from ansible_collections.community.hrobot.plugins.module_utils.robot import (
-    BASE_URL,
     ROBOT_DEFAULT_ARGUMENT_SPEC,
     _ROBOT_DEFAULT_ARGUMENT_SPEC_COMPAT_DEPRECATED,
-    fetch_url_json,
 )
 
 from ansible_collections.community.hrobot.plugins.module_utils.api import (
@@ -557,12 +555,6 @@
     deprecate_value,
 )
 
-try:
-    from urllib.parse import urlencode
-except ImportError:
-    # Python 2.x fallback:
-    from urllib import urlencode
-
 
 _CONVERT = {
     "login": ["username"],
@@ -621,49 +613,18 @@ def main():
             collection_name="community.hrobot",
             version="3.0.0",
         )
-        # DEPRECATED: old API
-        if storagebox_id is not None:
-            storagebox_ids = [storagebox_id]
-        else:
-            url = "{0}/storagebox".format(BASE_URL)
-            data = None
-            headers = None
-            if linked_server_number is not None:
-                data = urlencode({
-                    "linked_server": linked_server_number,
-                })
-                headers = {
-                    "Content-type": "application/x-www-form-urlencoded",
-                }
-            result, error = fetch_url_json(module, url, accept_errors=['STORAGEBOX_NOT_FOUND'], data=data)
-            storagebox_ids = []
-            if not error:
-                # When filtering by linked_server, the result should be a dictionary
-                if isinstance(result, dict):
-                    result = [result]
-                for entry in result:
-                    if full_info:
-                        storagebox_ids.append(entry['storagebox']['id'])
-                    else:
-                        storageboxes.append(entry['storagebox'])
-
-        for storagebox_id in storagebox_ids:
-            url = "{0}/storagebox/{1}".format(BASE_URL, storagebox_id)
-            result, error = fetch_url_json(module, url, accept_errors=['STORAGEBOX_NOT_FOUND'])
-            if not error:
-                storageboxes.append(result['storagebox'])
+        module.warn("The old storagebox API has been disabled by Hetzner. The supporting code has been removed.")
+        module.exit_json(changed=False, storageboxes=[])
 
+    if storagebox_id is not None:
+        url = "{0}/v1/storage_boxes/{1}".format(API_BASE_URL, storagebox_id)
+        result, dummy, error = api_fetch_url_json(module, url, accept_errors=["not_found"])
+        if error is None:
+            storageboxes = [result["storage_box"]]
     else:
-        # NEW API!
-        if storagebox_id is not None:
-            url = "{0}/v1/storage_boxes/{1}".format(API_BASE_URL, storagebox_id)
-            result, dummy, error = api_fetch_url_json(module, url, accept_errors=["not_found"])
-            if error is None:
-                storageboxes = [result["storage_box"]]
-        else:
-            url = "{0}/v1/storage_boxes".format(API_BASE_URL)
-            storageboxes, dummy = api_fetch_url_json_list(module, url, data_key="storage_boxes")
-        storageboxes = [add_hrobot_compat_shim(storagebox) for storagebox in storageboxes]
+        url = "{0}/v1/storage_boxes".format(API_BASE_URL)
+        storageboxes, dummy = api_fetch_url_json_list(module, url, data_key="storage_boxes")
+    storageboxes = [add_hrobot_compat_shim(storagebox) for storagebox in storageboxes]
 
     module.exit_json(
         changed=False,

plugins/modules/storagebox_set_password.py

@@ -83,10 +83,8 @@
 from ansible.module_utils.basic import AnsibleModule
 
 from ansible_collections.community.hrobot.plugins.module_utils.robot import (
-    BASE_URL,
     ROBOT_DEFAULT_ARGUMENT_SPEC,
     _ROBOT_DEFAULT_ARGUMENT_SPEC_COMPAT_DEPRECATED,
-    fetch_url_json,
 )
 
 from ansible_collections.community.hrobot.plugins.module_utils.api import (
@@ -97,12 +95,6 @@
     api_apply_action,
 )
 
-try:
-    from urllib.parse import urlencode
-except ImportError:
-    # Python 2.x fallback:
-    from urllib import urlencode
-
 
 def main():
     argument_spec = dict(
@@ -128,51 +120,30 @@ def main():
             collection_name="community.hrobot",
             version="3.0.0",
         )
-        # DEPRECATED: old API
-        url = "{0}/storagebox/{1}/password".format(BASE_URL, id)
-        accepted_errors = ["STORAGEBOX_NOT_FOUND", "STORAGEBOX_INVALID_PASSWORD"]
-
-        if password:
-            headers = {"Content-type": "application/x-www-form-urlencoded"}
-            result, error = fetch_url_json(
-                module, url, method="POST", accept_errors=accepted_errors, data=urlencode({"password": password}), headers=headers)
-        else:
-            result, error = fetch_url_json(
-                module, url, method="POST", accept_errors=accepted_errors)
-
-        if error == 'STORAGEBOX_NOT_FOUND':
-            module.fail_json(
-                msg='Storage Box with ID {0} not found'.format(id))
-
-        if error == 'STORAGEBOX_INVALID_PASSWORD':
-            module.fail_json(
-                msg="The chosen password has been considered insecure or does not comply with Hetzner's password guideline")
-
-        module.exit_json(changed=True, password=result["password"])
-
-    else:
-        # NEW API!
-        action_url = "{0}/v1/storage_boxes/{1}/actions/reset_password".format(API_BASE_URL, id)
-        action = {
-            "password": password,
-        }
-        try:
-            dummy, error = api_apply_action(
-                module,
-                action_url,
-                action,
-                lambda action_id: "{0}/v1/storage_boxes/actions/{1}".format(API_BASE_URL, action_id),
-                check_done_delay=1,
-                check_done_timeout=60,
-                accept_errors=["not_found"],
-            )
-        except ApplyActionError as exc:
-            module.fail_json(msg='Error while resetting password: {0}'.format(exc))
-
-        if error == "not_found":
-            module.fail_json(msg='Storage Box with ID {0} not found'.format(id))
-
-        module.exit_json(changed=True, password=password)
+        module.warn("The old storagebox API has been disabled by Hetzner. The supporting code has been removed.")
+        module.fail_json(msg='Storage Box with ID {0} not found'.format(id))
+
+    action_url = "{0}/v1/storage_boxes/{1}/actions/reset_password".format(API_BASE_URL, id)
+    action = {
+        "password": password,
+    }
+    try:
+        dummy, error = api_apply_action(
+            module,
+            action_url,
+            action,
+            lambda action_id: "{0}/v1/storage_boxes/actions/{1}".format(API_BASE_URL, action_id),
+            check_done_delay=1,
+            check_done_timeout=60,
+            accept_errors=["not_found"],
+        )
+    except ApplyActionError as exc:
+        module.fail_json(msg='Error while resetting password: {0}'.format(exc))
+
+    if error == "not_found":
+        module.fail_json(msg='Storage Box with ID {0} not found'.format(id))
+
+    module.exit_json(changed=True, password=password)
 
 
 if __name__ == '__main__':  # pragma: no cover