ansible/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2024-09-14 20:13:21 +02:00
Code Releases Activity

Adds bigip_ucs_fetch module (#35113)

Browse source 
This module can be used to download UCS files from a BIG-IP
This commit is contained in:
Tim Rupp 2018-01-19 20:25:47 -08:00 committed by GitHub
parent 2cd8a3a9a3
commit 585d8cf4c7
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 779 additions and 33 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
lib/ansible/module_utils/network/f5/common.py
View file

@ -34,6 +34,7 @@ f5_provider_spec = {
),
'password': dict(
no_log=True,
aliases=['pass', 'pwd'],
fallback=(env_fallback, ['F5_PASSWORD', 'ANSIBLE_NET_PASSWORD'])
),
'ssh_keyfile': dict(
@ -67,6 +68,7 @@ f5_top_spec = {
'password': dict(
removed_in_version=2.9,
no_log=True,
aliases=['pass', 'pwd'],
fallback=(env_fallback, ['F5_PASSWORD', 'ANSIBLE_NET_PASSWORD'])
),
'validate_certs': dict(

22
lib/ansible/modules/network/f5/bigip_device_group.py
View file

@ -76,10 +76,6 @@ options:
incremental synchronization operations can reduce the per-device sync/load
time for configuration changes. This setting is relevant only when
C(full_sync) is C(false).
partition:
description:
- Device partition to manage resources on.
default: Common
state:
description:
- When C(state) is C(present), ensures the device group exists.
@ -151,7 +147,6 @@ max_incremental_sync_size:
'''
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.basic import env_fallback
from ansible.module_utils.parsing.convert_bool import BOOLEANS_TRUE
HAS_DEVEL_IMPORTS = False
@ -339,8 +334,7 @@ class ModuleManager(object):
def exists(self):
result = self.client.api.tm.cm.device_groups.device_group.exists(
name=self.want.name,
partition=self.want.partition
name=self.want.name
)
return result
@ -372,15 +366,13 @@ class ModuleManager(object):
params = self.want.api_params()
self.client.api.tm.cm.device_groups.device_group.create(
name=self.want.name,
partition=self.want.partition,
**params
)
def update_on_device(self):
params = self.want.api_params()
resource = self.client.api.tm.cm.device_groups.device_group.load(
name=self.want.name,
partition=self.want.partition
name=self.want.name
)
resource.modify(**params)
@ -391,16 +383,14 @@ class ModuleManager(object):
def remove_from_device(self):
resource = self.client.api.tm.cm.device_groups.device_group.load(
name=self.want.name,
partition=self.want.partition
name=self.want.name
)
if resource:
resource.delete()
def read_current_from_device(self):
resource = self.client.api.tm.cm.device_groups.device_group.load(
name=self.want.name,
partition=self.want.partition
name=self.want.name
)
result = resource.attrs
return Parameters(params=result)
@ -431,10 +421,6 @@ class ArgumentSpec(object):
state=dict(
default='present',
choices=['absent', 'present']
),
partition=dict(
default='Common',
fallback=(env_fallback, ['F5_PARTITION'])
)
)
self.argument_spec = {}

14
lib/ansible/modules/network/f5/bigip_gtm_virtual_server.py
View file

@ -76,9 +76,19 @@ except ImportError:
pass # Handled by f5_utils.bigsuds_found
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.f5_utils import bigip_api, bigsuds_found, f5_argument_spec
from ansible.module_utils.f5_utils import bigip_api, bigsuds_found
from ansible.module_utils._text import to_native
HAS_DEVEL_IMPORTS = False
try:
# Sideband repository used for dev
from library.module_utils.network.f5.common import f5_argument_spec
HAS_DEVEL_IMPORTS = True
except ImportError:
# Upstream Ansible
from ansible.module_utils.network.f5.common import f5_argument_spec
def server_exists(api, server):
# hack to determine if virtual server exists
@ -136,7 +146,7 @@ def set_virtual_server_state(api, name, server, state):
def main():
argument_spec = f5_argument_spec()
argument_spec = f5_argument_spec
meta_args = dict(
state=dict(type='str', default='present', choices=['present', 'absent', 'enabled', 'disabled']),

8
lib/ansible/modules/network/f5/bigip_pool_member.py
View file

@ -179,6 +179,7 @@ except ImportError:
pass # Handled by f5_utils.bigsuds_found
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.basic import env_fallback
from ansible.module_utils.f5_utils import bigip_api, bigsuds_found
HAS_DEVEL_IMPORTS = False
@ -399,7 +400,12 @@ def main():
rate_limit=dict(type='int'),
ratio=dict(type='int'),
preserve_node=dict(type='bool', default=False),
priority_group=dict(type='int')
priority_group=dict(type='int'),
state=dict(default='present', choices=['absent', 'present']),
partition=dict(
default='Common',
fallback=(env_fallback, ['F5_PARTITION'])
)
)
argument_spec.update(meta_args)

38
lib/ansible/modules/network/f5/bigip_routedomain.py
View file

@ -179,15 +179,30 @@ except ImportError:
pass # Handled via f5_utils.HAS_F5SDK
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.basic import env_fallback
from ansible.module_utils.ec2 import camel_dict_to_snake_dict
from ansible.module_utils.f5_utils import F5ModuleError
from ansible.module_utils.f5_utils import HAS_F5SDK
from ansible.module_utils.f5_utils import f5_argument_spec
HAS_DEVEL_IMPORTS = False
try:
from ansible.module_utils.f5_utils import iControlUnexpectedHTTPError
# Sideband repository used for dev
from library.module_utils.network.f5.bigip import HAS_F5SDK
from library.module_utils.network.f5.common import F5ModuleError
from library.module_utils.network.f5.common import f5_argument_spec
try:
from library.module_utils.network.f5.common import iControlUnexpectedHTTPError
except ImportError:
HAS_F5SDK = False
HAS_DEVEL_IMPORTS = True
except ImportError:
HAS_F5SDK = False
# Upstream Ansible
from ansible.module_utils.network.f5.bigip import HAS_F5SDK
from ansible.module_utils.network.f5.common import F5ModuleError
from ansible.module_utils.network.f5.common import f5_argument_spec
try:
from ansible.module_utils.network.f5.common import iControlUnexpectedHTTPError
except ImportError:
HAS_F5SDK = False
PROTOCOLS = [
@ -521,7 +536,7 @@ class BigIpRouteDomain(object):
def main():
argument_spec = f5_argument_spec()
argument_spec = f5_argument_spec
meta_args = dict(
name=dict(),
@ -529,13 +544,20 @@ def main():
description=dict(),
strict=dict(choices=STRICTS),
parent=dict(type='int'),
partition=dict(default='Common'),
vlans=dict(type='list'),
routing_protocol=dict(type='list'),
bwc_policy=dict(),
connection_limit=dict(type='int',),
flow_eviction_policy=dict(),
service_policy=dict()
service_policy=dict(),
partition=dict(
default='Common',
fallback=(env_fallback, ['F5_PARTITION'])
),
state=dict(
default='present',
choices=['present', 'absent']
)
)
argument_spec.update(meta_args)

533
lib/ansible/modules/network/f5/bigip_ucs_fetch.py Normal file
View file

@ -0,0 +1,533 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# Copyright (c) 2017 F5 Networks Inc.
# 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
ANSIBLE_METADATA = {'metadata_version': '1.1',
'status': ['preview'],
'supported_by': 'community'}
DOCUMENTATION = r'''
---
module: bigip_ucs_fetch
short_description: Fetches a UCS file from remote nodes
description:
- This module is used for fetching UCS files from remote machines and
storing them locally in a file tree, organized by hostname. Note that
this module is written to transfer UCS files that might not be present,
so a missing remote UCS won't be an error unless fail_on_missing is
set to 'yes'.
version_added: 2.5
options:
backup:
description:
- Create a backup file including the timestamp information so you can
get the original file back if you somehow clobbered it incorrectly.
default: no
type: bool
create_on_missing:
description:
- Creates the UCS based on the value of C(src) if the file does not already
exist on the remote system.
default: yes
type: bool
dest:
description:
- A directory to save the UCS file into.
required: yes
encryption_password:
description:
- Password to use to encrypt the UCS file if desired
fail_on_missing:
description:
- Make the module fail if the UCS file on the remote system is missing.
default: no
type: bool
force:
description:
- If C(no), the file will only be transferred if the destination does not
exist.
default: yes
type: bool
src:
description:
- The name of the UCS file to create on the remote server for downloading
notes:
- BIG-IP provides no way to get a checksum of the UCS files on the system
via any interface except, perhaps, logging in directly to the box (which
would not support appliance mode). Therefore, the best this module can
do is check for the existence of the file on disk; no check-summing.
extends_documentation_fragment: f5
author:
- Tim Rupp (@caphrim007)
'''
EXAMPLES = r'''
- name: Download a new UCS
bigip_ucs_fetch:
server: lb.mydomain.com
user: admin
password: secret
src: cs_backup.ucs
dest: /tmp/cs_backup.ucs
delegate_to: localhost
'''
RETURN = r'''
checksum:
description: The SHA1 checksum of the downloaded file
returned: success or changed
type: string
sample: 7b46bbe4f8ebfee64761b5313855618f64c64109
dest:
description: Location on the ansible host that the UCS was saved to
returned: success
type: string
sample: /path/to/file.txt
src:
description:
- Name of the UCS file on the remote BIG-IP to download. If not
specified, then this will be a randomly generated filename
returned: changed
type: string
sample: cs_backup.ucs
backup_file:
description: Name of backup file created
returned: changed and if backup=yes
type: string
sample: /path/to/file.txt.2015-02-12@22:09~
gid:
description: Group id of the UCS file, after execution
returned: success
type: int
sample: 100
group:
description: Group of the UCS file, after execution
returned: success
type: string
sample: httpd
owner:
description: Owner of the UCS file, after execution
returned: success
type: string
sample: httpd
uid:
description: Owner id of the UCS file, after execution
returned: success
type: int
sample: 100
md5sum:
description: The MD5 checksum of the downloaded file
returned: changed or success
type: string
sample: 96cacab4c259c4598727d7cf2ceb3b45
mode:
description: Permissions of the target UCS, after execution
returned: success
type: string
sample: 0644
size:
description: Size of the target UCS, after execution
returned: success
type: int
sample: 1220
'''
import os
import re
import tempfile
from ansible.module_utils.basic import AnsibleModule
from distutils.version import LooseVersion
HAS_DEVEL_IMPORTS = False
try:
# Sideband repository used for dev
from library.module_utils.network.f5.bigip import HAS_F5SDK
from library.module_utils.network.f5.bigip import F5Client
from library.module_utils.network.f5.common import F5ModuleError
from library.module_utils.network.f5.common import AnsibleF5Parameters
from library.module_utils.network.f5.common import cleanup_tokens
from library.module_utils.network.f5.common import fqdn_name
from library.module_utils.network.f5.common import f5_argument_spec
try:
from library.module_utils.network.f5.common import iControlUnexpectedHTTPError
except ImportError:
HAS_F5SDK = False
HAS_DEVEL_IMPORTS = True
except ImportError:
# Upstream Ansible
from ansible.module_utils.network.f5.bigip import HAS_F5SDK
from ansible.module_utils.network.f5.bigip import F5Client
from ansible.module_utils.network.f5.common import F5ModuleError
from ansible.module_utils.network.f5.common import AnsibleF5Parameters
from ansible.module_utils.network.f5.common import cleanup_tokens
from ansible.module_utils.network.f5.common import fqdn_name
from ansible.module_utils.network.f5.common import f5_argument_spec
try:
from ansible.module_utils.network.f5.common import iControlUnexpectedHTTPError
except ImportError:
HAS_F5SDK = False
class Parameters(AnsibleF5Parameters):
updatables = []
returnables = ['dest', 'src', 'md5sum', 'checksum', 'backup_file']
api_attributes = []
api_map = {}
@property
def options(self):
result = []
if self.passphrase:
result.append(dict(
passphrase=self.want.passphrase
))
return result
@property
def src(self):
if self._values['src'] is not None:
return self._values['src']
result = next(tempfile._get_candidate_names()) + '.ucs'
self._values['src'] = result
return result
@property
def fulldest(self):
result = None
if os.path.isdir(self.dest):
result = os.path.join(self.dest, self.src)
else:
if os.path.exists(os.path.dirname(self.dest)):
result = self.dest
else:
try:
# os.path.exists() can return false in some
# circumstances where the directory does not have
# the execute bit for the current user set, in
# which case the stat() call will raise an OSError
os.stat(os.path.dirname(result))
except OSError as e:
if "permission denied" in str(e).lower():
raise F5ModuleError(
"Destination directory {0} is not accessible".format(os.path.dirname(result))
)
raise F5ModuleError(
"Destination directory {0} does not exist".format(os.path.dirname(result))
)
if not os.access(os.path.dirname(result), os.W_OK):
raise F5ModuleError(
"Destination {0} not writable".format(os.path.dirname(result))
)
return result
class Changes(Parameters):
def to_return(self):
result = {}
try:
for returnable in self.returnables:
result[returnable] = getattr(self, returnable)
result = self._filter_params(result)
except Exception:
pass
return result
class UsableChanges(Changes):
pass
class ReportableChanges(Changes):
pass
class ModuleManager(object):
def __init__(self, *args, **kwargs):
self.client = kwargs.get('client', None)
self.kwargs = kwargs
def exec_module(self):
if self.is_version_v1():
manager = self.get_manager('v1')
else:
manager = self.get_manager('v2')
return manager.exec_module()
def get_manager(self, type):
if type == 'v1':
return V1Manager(**self.kwargs)
elif type == 'v2':
return V2Manager(**self.kwargs)
def is_version_v1(self):
"""Checks to see if the TMOS version is less than 12.1.0
Versions prior to 12.1.0 have a bug which prevents the REST
API from properly listing any UCS files when you query the
/mgmt/tm/sys/ucs endpoint. Therefore you need to do everything
through tmsh over REST.
:return: bool
"""
version = self.client.api.tmos_version
if LooseVersion(version) < LooseVersion('12.1.0'):
return True
else:
return False
class BaseManager(object):
def __init__(self, *args, **kwargs):
self.module = kwargs.get('module', None)
self.client = kwargs.get('client', None)
self.want = Parameters(params=self.module.params)
self.changes = UsableChanges()
def exec_module(self):
result = dict()
try:
self.present()
except iControlUnexpectedHTTPError as e:
raise F5ModuleError(str(e))
reportable = ReportableChanges(params=self.changes.to_return())
changes = reportable.to_return()
result.update(**changes)
result.update(dict(changed=True))
return result
def present(self):
if self.exists():
self.update()
else:
self.create()
def update(self):
if os.path.exists(self.want.fulldest):
if not self.want.force:
raise F5ModuleError(
"File '{0}' already exists".format(self.want.fulldest)
)
self.execute()
def _get_backup_file(self):
return self.module.backup_local(self.want.fulldest)
def execute(self):
try:
if self.want.backup:
if os.path.exists(self.want.fulldest):
backup_file = self._get_backup_file()
self.changes.update({'backup_file': backup_file})
self.download()
except IOError:
raise F5ModuleError(
"Failed to copy: {0} to {1}".format(self.want.src, self.want.fulldest)
)
self._set_checksum()
self._set_md5sum()
file_args = self.module.load_file_common_arguments(self.module.params)
return self.module.set_fs_attributes_if_different(file_args, True)
def _set_checksum(self):
try:
result = self.module.sha1(self.want.fulldest)
self.want.update({'checksum': result})
except ValueError:
pass
def _set_md5sum(self):
try:
result = self.module.md5(self.want.fulldest)
self.want.update({'md5sum': result})
except ValueError:
pass
def create(self):
if self.want.fail_on_missing:
raise F5ModuleError(
"UCS '{0}' was not found".format(self.want.src)
)
if not self.want.create_on_missing:
raise F5ModuleError(
"UCS '{0}' was not found".format(self.want.src)
)
if self.module.check_mode:
return True
if self.want.create_on_missing:
self.create_on_device()
self.execute()
return True
def create_on_device(self):
if self.want.passphrase:
self.client.api.tm.sys.ucs.exec_cmd(
'save',
name=self.want.src,
options=[{'passphrase': self.want.encryption_password}]
)
else:
self.client.api.tm.sys.ucs.exec_cmd(
'save',
name=self.want.src
)
def download(self):
self.download_from_device()
if os.path.exists(self.want.dest):
return True
raise F5ModuleError(
"Failed to download the remote file"
)
class V1Manager(BaseManager):
def __init__(self, *args, **kwargs):
super(V1Manager, self).__init__(**kwargs)
self.remote_dir = '/var/config/rest/madm'
def read_current(self):
result = None
output = self.read_current_from_device()
if hasattr(output, 'commandResult'):
result = self._read_ucs_files_from_output(output.commandResult)
return result
def read_current_from_device(self):
output = self.client.api.tm.util.bash.exec_cmd(
'run',
utilCmdArgs='-c "tmsh list sys ucs"'
)
return output
def _read_ucs_files_from_output(self, output):
search = re.compile(r'filename\s+(.*)').search
lines = output.split("\n")
result = [m.group(1) for m in map(search, lines) if m]
return result
def exists(self):
collection = self.read_current()
base = os.path.basename(self.want.src)
if any(base == os.path.basename(x) for x in collection):
return True
return False
def download_from_device(self):
madm = self.client.api.shared.file_transfer.madm
madm.download_file(self.want.filename, self.want.dest)
if os.path.exists(self.want.dest):
return True
return False
def _move_to_download(self):
try:
move_path = '/var/local/ucs/{0} {1}/{0}'.format(
self.want.filename, self.remote_dir
)
self.client.api.tm.util.unix_mv.exec_cmd(
'run',
utilCmdArgs=move_path
)
return True
except Exception:
return False
class V2Manager(BaseManager):
def read_current(self):
collection = self.read_current_from_device()
if 'items' not in collection.attrs:
return []
resources = collection.attrs['items']
result = [x['apiRawValues']['filename'] for x in resources]
return result
def read_current_from_device(self):
collection = self.client.api.tm.sys.ucs.load()
return collection
def exists(self):
collection = self.read_current()
base = os.path.basename(self.want.src)
if any(base == os.path.basename(x) for x in collection):
return True
return False
def download_from_device(self):
ucs = self.client.api.shared.file_transfer.ucs_downloads
ucs.download_file(self.want.src, self.want.dest)
if os.path.exists(self.want.dest):
return True
return False
class ArgumentSpec(object):
def __init__(self):
self.supports_check_mode = True
argument_spec = dict(
backup=dict(
default='no',
type='bool'
),
create_on_missing=dict(
default='yes',
type='bool'
),
encryption_password=dict(no_log=True),
dest=dict(
required=True,
type='path'
),
force=dict(
default='yes',
type='bool'
),
fail_on_missing=dict(
default='no',
type='bool'
),
src=dict()
)
self.argument_spec = {}
self.argument_spec.update(f5_argument_spec)
self.argument_spec.update(argument_spec)
self.add_file_common_args = True
def main():
spec = ArgumentSpec()
module = AnsibleModule(
argument_spec=spec.argument_spec,
supports_check_mode=spec.supports_check_mode,
add_file_common_args=spec.add_file_common_args
)
if not HAS_F5SDK:
module.fail_json(msg="The python f5-sdk module is required")
try:
client = F5Client(**module.params)
mm = ModuleManager(module=module, client=client)
results = mm.exec_module()
cleanup_tokens(client)
module.exit_json(**results)
except F5ModuleError as ex:
cleanup_tokens(client)
module.fail_json(msg=str(ex))
if __name__ == '__main__':
main()

65
lib/ansible/utils/module_docs_fragments/f5.py
View file

@ -51,13 +51,70 @@ options:
You can omit this option if the environment variable
C(F5_VALIDATE_CERTS) is set.
default: yes
choices:
- yes
- no
type: bool
version_added: 2.0
provider:
description:
- A dict object containing connection details.
default: null
version_added: 2.5
suboptions:
password:
description:
- The password for the user account used to connect to the BIG-IP.
You can omit this option if the environment variable C(F5_PASSWORD)
is set.
required: true
aliases: ['pass', 'pwd']
server:
description:
- The BIG-IP host. You can omit this option if the environment
variable C(F5_SERVER) is set.
required: true
server_port:
description:
- The BIG-IP server port. You can omit this option if the environment
variable C(F5_SERVER_PORT) is set.
default: 443
user:
description:
- The username to connect to the BIG-IP with. This user must have
administrative privileges on the device. You can omit this option
if the environment variable C(F5_USER) is set.
required: true
validate_certs:
description:
- If C(no), SSL certificates will not be validated. Use this only
on personally controlled sites using self-signed certificates.
You can omit this option if the environment variable
C(F5_VALIDATE_CERTS) is set.
default: yes
type: bool
timeout:
description:
- Specifies the timeout in seconds for communicating with the network device
for either connecting or sending commands. If the timeout is
exceeded before the operation is completed, the module will error.
default: 10
ssh_keyfile:
description:
- Specifies the SSH keyfile to use to authenticate the connection to
the remote device. This argument is only used for I(cli) transports.
If the value is not specified in the task, the value of environment
variable C(ANSIBLE_NET_SSH_KEYFILE) will be used instead.
transport:
description:
- Configures the transport connection to use when connecting to the
remote device.
required: true
choices:
- rest
- cli
default: cli
notes:
- For more information on using Ansible to manage F5 Networks devices see U(https://www.ansible.com/integrations/networks/f5).
- Requires the f5-sdk Python package on the host. This is as easy as C(pip install f5-sdk).
requirements:
- f5-sdk >= 3.0.6
- f5-sdk >= 3.0.9
'''

130
test/units/modules/network/f5/test_bigip_ucs_fetch.py Normal file
View file

@ -0,0 +1,130 @@
# -*- coding: utf-8 -*-
#
# Copyright (c) 2017 F5 Networks Inc.
# 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
import os
import json
import sys
from nose.plugins.skip import SkipTest
if sys.version_info < (2, 7):
raise SkipTest("F5 Ansible modules require Python >= 2.7")
from ansible.compat.tests import unittest
from ansible.compat.tests.mock import Mock
from ansible.compat.tests.mock import patch
from ansible.module_utils.basic import AnsibleModule
try:
from library.bigip_ucs_fetch import Parameters
from library.bigip_ucs_fetch import ModuleManager
from library.bigip_ucs_fetch import V1Manager
from library.bigip_ucs_fetch import V2Manager
from library.bigip_ucs_fetch import ArgumentSpec
from library.module_utils.network.f5.common import F5ModuleError
from library.module_utils.network.f5.common import iControlUnexpectedHTTPError
from test.unit.modules.utils import set_module_args
except ImportError:
try:
from ansible.modules.network.f5.bigip_ucs_fetch import Parameters
from ansible.modules.network.f5.bigip_ucs_fetch import ModuleManager
from ansible.modules.network.f5.bigip_ucs_fetch import V1Manager
from ansible.modules.network.f5.bigip_ucs_fetch import V2Manager
from ansible.modules.network.f5.bigip_ucs_fetch import ArgumentSpec
from ansible.module_utils.network.f5.common import F5ModuleError
from ansible.module_utils.network.f5.common import iControlUnexpectedHTTPError
from units.modules.utils import set_module_args
except ImportError:
raise SkipTest("F5 Ansible modules require the f5-sdk Python library")
fixture_path = os.path.join(os.path.dirname(__file__), 'fixtures')
fixture_data = {}
def load_fixture(name):
path = os.path.join(fixture_path, name)
if path in fixture_data:
return fixture_data[path]
with open(path) as f:
data = f.read()
try:
data = json.loads(data)
except Exception:
pass
fixture_data[path] = data
return data
class TestParameters(unittest.TestCase):
def test_module_parameters(self):
args = dict(
backup='yes',
create_on_missing='yes',
encryption_password='my-password',
dest='/tmp/foo.ucs',
force='yes',
fail_on_missing='no',
src='remote.ucs',
password='password',
server='localhost',
user='admin'
)
p = Parameters(params=args)
assert p.backup == 'yes'
class TestV1Manager(unittest.TestCase):
def setUp(self):
self.spec = ArgumentSpec()
def test_create(self, *args):
set_module_args(dict(
backup='yes',
create_on_missing='yes',
dest='/tmp/foo.ucs',
force='yes',
fail_on_missing='no',
src='remote.ucs',
password='passsword',
server='localhost',
user='admin'
))
module = AnsibleModule(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode
)
# Override methods to force specific logic in the module to happen
m1 = V1Manager(module=module)
m1.exists = Mock(return_value=False)
m1.create_on_device = Mock(return_value=True)
m1._get_backup_file = Mock(return_value='/tmp/foo.backup')
m1.download_from_device = Mock(return_value=True)
m1._set_checksum = Mock(return_value=12345)
m1._set_md5sum = Mock(return_value=54321)
mm = ModuleManager(module=module)
mm.get_manager = Mock(return_value=m1)
mm.is_version_v1 = Mock(return_value=True)
p1 = patch('os.path.exists', return_value=True)
p1.start()
p2 = patch('os.path.isdir', return_value=False)
p2.start()
results = mm.exec_module()
p1.stop()
p2.stop()
assert results['changed'] is True