cmueller/ansible
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Introduces numerous fixes for bigip command (#34550)

Browse source 
A bug in the parsing of single commands with commas
Token cleanup
Password argument now defaults to false
Addition of coding conventions from v3 conventions
This commit is contained in:
Tim Rupp 2018-01-06 23:23:32 -08:00 committed by GitHub
parent 460deb08cb
commit 2916ff0a1a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 122 additions and 28 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

90
lib/ansible/modules/network/f5/bigip_command.py
View file

@ -79,11 +79,6 @@ options:
- cli
default: rest
version_added: "2.5"
notes:
- Requires the f5-sdk Python package on the host. This is as easy as pip
install f5-sdk.
requirements:
- f5-sdk >= 2.2.3
extends_documentation_fragment: f5
author:
- Tim Rupp (@caphrim007)
@ -167,6 +162,7 @@ failed_conditions:
import re
import time
from ansible.module_utils.basic import env_fallback
from ansible.module_utils.f5_utils import AnsibleF5Client
from ansible.module_utils.f5_utils import AnsibleF5Parameters
from ansible.module_utils.f5_utils import HAS_F5SDK
@ -183,6 +179,8 @@ from ansible.module_utils.network.common.parsing import FailedConditionsError
from ansible.module_utils.network.common.parsing import Conditional
from ansible.module_utils.network.common.utils import ComplexList
from ansible.module_utils.network.common.utils import to_list
from ansible.module_utils.six import iteritems
from collections import defaultdict
from collections import deque
try:
@ -194,6 +192,36 @@ except ImportError:
class Parameters(AnsibleF5Parameters):
returnables = ['stdout', 'stdout_lines', 'warnings']
def __init__(self, params=None):
self._values = defaultdict(lambda: None)
self._values['__warnings'] = []
if params:
self.update(params=params)
def update(self, params=None):
if params:
for k, v in iteritems(params):
if self.api_map is not None and k in self.api_map:
map_key = self.api_map[k]
else:
map_key = k
# Handle weird API parameters like `dns.proxy.__iter__` by
# using a map provided by the module developer
class_attr = getattr(type(self), map_key, None)
if isinstance(class_attr, property):
# There is a mapped value for the api_map key
if class_attr.fset is None:
# If the mapped value does not have
# an associated setter
self._values[map_key] = v
else:
# The mapped value has a setter
setattr(self, map_key, v)
else:
# If the mapped value is not a @property
self._values[map_key] = v
def to_return(self):
result = {}
for returnable in self.returnables:
@ -201,19 +229,28 @@ class Parameters(AnsibleF5Parameters):
result = self._filter_params(result)
return result
def _listify(self, item):
if isinstance(item, string_types):
result = [item]
else:
result = item
return result
@property
def commands(self):
commands = deque(self._values['commands'])
commands = self._listify(self._values['commands'])
commands = deque(commands)
if self._values['transport'] != 'cli':
commands.appendleft(
'tmsh modify cli preference pager disabled'
)
commands = map(self._ensure_tmsh_prefix, list(commands))
commands = map(self._ensure_tmsh_prefix, list(commands))
return list(commands)
@property
def user_commands(self):
return map(self._ensure_tmsh_prefix, list(self._values['commands']))
commands = self._listify(self._values['commands'])
return map(self._ensure_tmsh_prefix, commands)
def _ensure_tmsh_prefix(self, cmd):
cmd = cmd.strip()
@ -259,12 +296,13 @@ class ModuleManager(object):
result.update(dict(changed=changed))
return result
def _run_commands(self, module, commands):
return run_commands(module, commands)
def execute(self):
warnings = list()
changed = ('tmsh modify', 'tmsh create', 'tmsh delete')
commands = self.parse_commands(warnings)
wait_for = self.want.wait_for or list()
retries = self.want.retries
@ -275,7 +313,7 @@ class ModuleManager(object):
while retries > 0:
if self.client.module.params['transport'] == 'cli' and HAS_CLI_TRANSPORT:
responses = run_commands(self.client.module, self.want.commands)
responses = self._run_commands(self.client.module, commands)
else:
responses = self.execute_on_device(commands)
@ -327,10 +365,15 @@ class ModuleManager(object):
'module does not exist, then please file a bug. The command '
'in question is "%s..."' % item['command'][0:40]
)
if item['output'] == 'one-line' and 'one-line' not in item['command']:
# This needs to be removed so that the ComplexList used in to_commands
# will work correctly.
output = item.pop('output', None)
if output == 'one-line' and 'one-line' not in item['command']:
item['command'] += ' one-line'
elif item['output'] == 'text' and 'one-line' in item['command']:
elif output == 'text' and 'one-line' in item['command']:
item['command'] = item['command'].replace('one-line', '')
results.append(item)
return results
@ -353,7 +396,7 @@ class ArgumentSpec(object):
self.supports_check_mode = True
self.argument_spec = dict(
commands=dict(
type='list',
type='raw',
required=True
),
wait_for=dict(
@ -376,11 +419,26 @@ class ArgumentSpec(object):
type='str',
default='rest',
choices=['cli', 'rest']
),
password=dict(
required=False,
fallback=(env_fallback, ['F5_PASSWORD']),
no_log=True
)
)
self.f5_product_name = 'bigip'
def cleanup_tokens(client):
try:
resource = client.api.shared.authz.tokens_s.token.load(
name=client.api.icrs.token
)
resource.delete()
except Exception:
pass
def main():
spec = ArgumentSpec()
@ -396,8 +454,10 @@ def main():
try:
mm = ModuleManager(client)
results = mm.exec_module()
cleanup_tokens(client)
client.module.exit_json(**results)
except (FailedConditionsError, AttributeError) as e:
except F5ModuleError as e:
cleanup_tokens(client)
client.module.fail_json(msg=str(e))

60
test/units/modules/network/f5/test_bigip_command.py
View file

@ -15,21 +15,23 @@ 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 patch, Mock
from ansible.compat.tests.mock import patch
from ansible.compat.tests.mock import Mock
from ansible.module_utils.f5_utils import AnsibleF5Client
from units.modules.utils import set_module_args
try:
from library.bigip_command import Parameters
from library.bigip_command import ModuleManager
from library.bigip_command import ArgumentSpec
from ansible.module_utils.f5_utils import iControlUnexpectedHTTPError
from test.unit.modules.utils import set_module_args
except ImportError:
try:
from ansible.modules.network.f5.bigip_command import Parameters
from ansible.modules.network.f5.bigip_command import ModuleManager
from ansible.modules.network.f5.bigip_command import ArgumentSpec
from ansible.module_utils.f5_utils import iControlUnexpectedHTTPError
from units.modules.utils import set_module_args
except ImportError:
raise SkipTest("F5 Ansible modules require the f5-sdk Python library")
@ -68,10 +70,6 @@ class TestParameters(unittest.TestCase):
class TestManager(unittest.TestCase):
def setUp(self):
self.mock_run_commands = patch('ansible.modules.network.f5.bigip_command.run_commands')
self.run_commands = self.mock_run_commands.start()
self.mock_execute_on_device = patch('ansible.modules.network.f5.bigip_command.ModuleManager.execute_on_device')
self.execute_on_device = self.mock_execute_on_device.start()
self.spec = ArgumentSpec()
def test_run_single_command(self, *args):
@ -90,12 +88,14 @@ class TestManager(unittest.TestCase):
f5_product_name=self.spec.f5_product_name
)
mm = ModuleManager(client)
mm._run_commands = Mock(return_value=[])
mm.execute_on_device = Mock(return_value=[])
results = mm.exec_module()
assert results['changed'] is False
self.assertEqual(self.run_commands.call_count, 0)
self.assertEqual(self.execute_on_device.call_count, 1)
assert mm._run_commands.call_count == 0
assert mm.execute_on_device.call_count == 1
def test_run_single_modification_command(self, *args):
set_module_args(dict(
@ -113,12 +113,14 @@ class TestManager(unittest.TestCase):
f5_product_name=self.spec.f5_product_name
)
mm = ModuleManager(client)
mm._run_commands = Mock(return_value=[])
mm.execute_on_device = Mock(return_value=[])
results = mm.exec_module()
assert results['changed'] is True
self.assertEqual(self.run_commands.call_count, 0)
self.assertEqual(self.execute_on_device.call_count, 1)
assert mm._run_commands.call_count == 0
assert mm.execute_on_device.call_count == 1
def test_cli_command(self, *args):
set_module_args(dict(
@ -136,6 +138,38 @@ class TestManager(unittest.TestCase):
f5_product_name=self.spec.f5_product_name
)
mm = ModuleManager(client)
mm.exec_module()
self.assertEqual(self.run_commands.call_count, 1)
self.assertEqual(self.execute_on_device.call_count, 0)
mm._run_commands = Mock(return_value=[])
mm.execute_on_device = Mock(return_value=[])
results = mm.exec_module()
assert results['changed'] is False
assert mm._run_commands.call_count == 1
assert mm.execute_on_device.call_count == 0
def test_command_with_commas(self, *args):
set_module_args(dict(
commands="""
tmsh create /auth ldap system-auth {bind-dn uid=binduser,
cn=users,dc=domain,dc=com bind-pw $ENCRYPTEDPW check-roles-group
enabled search-base-dn cn=users,dc=domain,dc=com servers add {
ldap.server.com } }"
""",
server='localhost',
user='admin',
password='password'
))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
mm = ModuleManager(client)
mm._run_commands = Mock(return_value=[])
mm.execute_on_device = Mock(return_value=[])
results = mm.exec_module()
assert results['changed'] is True
assert mm._run_commands.call_count == 0
assert mm.execute_on_device.call_count == 1