Add backup filename and dir path option for config network modules (#50801)

* Add configurable backup path option for network config modules

Fixes #50283
Fixes #32724

*  Add back_options in network config module argspec
*  Handle backup path options in network action plugin

* Fix review comments

* Add integration tests

* Update changelog

lib/ansible/plugins/action/network.py

@@ -25,7 +25,7 @@ import glob
 import re
 
 from ansible.errors import AnsibleError
-from ansible.module_utils._text import to_text
+from ansible.module_utils._text import to_text, to_bytes
 from ansible.module_utils.six.moves.urllib.parse import urlsplit
 from ansible.plugins.action.normal import ActionModule as _ActionModule
 from ansible.utils.display import Display
@@ -44,8 +44,8 @@ class ActionModule(_ActionModule):
 
         result = super(ActionModule, self).run(task_vars=task_vars)
 
-        if config_module and self._task.args.get('backup'):
-            self._handle_backup_option(result, task_vars['inventory_hostname'], result.get('__backup__', False))
+        if config_module and self._task.args.get('backup') and not result.get('failed'):
+            self._handle_backup_option(result, task_vars)
 
         return result
 
@@ -55,12 +55,60 @@ class ActionModule(_ActionModule):
         except ValueError as exc:
             return dict(failed=True, msg=to_text(exc))
 
-    def _handle_backup_option(self, result, hostname, backup):
-        if backup:
-            # User requested backup and no error occurred in module.
-            # NOTE: If there is a parameter error, _backup key may not be in results.
-            filepath = self._write_backup(hostname, backup)
-            result['backup_path'] = filepath
+    def _handle_backup_option(self, result, task_vars):
+
+        filename = None
+        backup_path = None
+        try:
+            content = result['__backup__']
+        except KeyError:
+            raise AnsibleError('Failed while reading configuration backup')
+
+        backup_options = self._task.args.get('backup_options')
+        if backup_options:
+            filename = backup_options.get('filename')
+            backup_path = backup_options.get('dir_path')
+
+        if not backup_path:
+            cwd = self._get_working_path()
+            backup_path = os.path.join(cwd, 'backup')
+        if not filename:
+            tstamp = time.strftime("%Y-%m-%d@%H:%M:%S", time.localtime(time.time()))
+            filename = '%s_config.%s' % (task_vars['inventory_hostname'], tstamp)
+
+        dest = os.path.join(backup_path, filename)
+        backup_path = os.path.expanduser(os.path.expandvars(to_bytes(backup_path, errors='surrogate_or_strict')))
+
+        if not os.path.exists(backup_path):
+            os.makedirs(backup_path)
+
+        new_task = self._task.copy()
+        for item in self._task.args:
+            if not item.startswith('_'):
+                new_task.args.pop(item, None)
+
+        new_task.args.update(
+            dict(
+                content=content,
+                dest=dest,
+            ),
+        )
+        copy_action = self._shared_loader_obj.action_loader.get('copy',
+                                                                task=new_task,
+                                                                connection=self._connection,
+                                                                play_context=self._play_context,
+                                                                loader=self._loader,
+                                                                templar=self._templar,
+                                                                shared_loader_obj=self._shared_loader_obj)
+        copy_result = copy_action.run(task_vars=task_vars)
+        if copy_result.get('failed'):
+            result['failed'] = copy_result['failed']
+            result['msg'] = copy_result.get('msg')
+            return
+
+        result['backup_path'] = copy_result['dest']
+        if copy_result.get('changed', False):
+            result['changed'] = copy_result['changed']
 
         # strip out any keys that have two leading and two trailing
         # underscore characters
@@ -74,20 +122,6 @@ class ActionModule(_ActionModule):
             cwd = self._task._role._role_path
         return cwd
 
-    def _write_backup(self, host, contents, encoding='utf-8'):
-        cwd = self._get_working_path()
-
-        backup_path = os.path.join(cwd, 'backup')
-        if not os.path.exists(backup_path):
-            os.mkdir(backup_path)
-        for existing_backup in glob.glob('%s/%s_config.*' % (backup_path, host)):
-            os.remove(existing_backup)
-        tstamp = time.strftime("%Y-%m-%d@%H:%M:%S", time.localtime(time.time()))
-        filename = '%s/%s_config.%s' % (backup_path, host, tstamp)
-        open(filename, 'w').write(to_text(contents, encoding=encoding))
-
-        return filename
-
     def _handle_template(self, convert_data=True):
         src = self._task.args.get('src')
         working_path = self._get_working_path()