From 2916ff0a1a2482caad387767eb1bded85bb77714 Mon Sep 17 00:00:00 2001
From: Tim Rupp <caphrim007@gmail.com>
Date: Sat, 6 Jan 2018 23:23:32 -0800
Subject: [PATCH] Introduces numerous fixes for bigip command (#34550)
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
---
.../modules/network/f5/bigip_command.py | 90 +++++++++++++++----
.../modules/network/f5/test_bigip_command.py | 60 ++++++++++---
2 files changed, 122 insertions(+), 28 deletions(-)
diff --git a/lib/ansible/modules/network/f5/bigip_command.py b/lib/ansible/modules/network/f5/bigip_command.py
index 058bf06601a..4125d3662c1 100644
--- a/lib/ansible/modules/network/f5/bigip_command.py
+++ b/lib/ansible/modules/network/f5/bigip_command.py
@@ -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))
diff --git a/test/units/modules/network/f5/test_bigip_command.py b/test/units/modules/network/f5/test_bigip_command.py
index 39d4c2333d7..8f354916424 100644
--- a/test/units/modules/network/f5/test_bigip_command.py
+++ b/test/units/modules/network/f5/test_bigip_command.py
@@ -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