archive
/
ansible
mirror of https://github.com/ansible/ansible.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Removes deprecated code from bigip_monitor_tcp (#34520)

Browse Source 
* Removes deprecated code from bigip_monitor_tcp

The deprecated code was moved into its own modules. We let it lay
here for a release, and now I'm removing it.

* Fixing upstreaming tests
pull/34524/head
Tim Rupp 7 years ago committed by GitHub
parent f189106ef6
commit 803e4124b6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 109 additions and 1149 deletions
Show Stats Download Patch File Download Diff File

503
lib/ansible/modules/network/f5/bigip_monitor_tcp.py
Unescape Escape View File

@ -190,6 +190,26 @@ except ImportError:
class Parameters(AnsibleF5Parameters):
api_map = {
'timeUntilUp': 'time_until_up',
'defaultsFrom': 'parent',
'recv': 'receive'
}
api_attributes = [
'timeUntilUp', 'defaultsFrom', 'interval', 'timeout', 'recv', 'send',
'destination'
]
returnables = [
'parent', 'send', 'receive', 'ip', 'port', 'interval', 'timeout',
'time_until_up'
]
updatables = [
'destination', 'send', 'receive', 'interval', 'timeout', 'time_until_up'
]
def __init__(self, params=None):
self._values = defaultdict(lambda: None)
self._values['__warnings'] = []
@ -220,6 +240,11 @@ class Parameters(AnsibleF5Parameters):
# If the mapped value is not a @property
self._values[map_key] = v
def _fqdn_name(self, value):
if value is not None and not value.startswith('/'):
return '/{0}/{1}'.format(self.partition, value)
return value
def to_return(self):
result = {}
try:
@ -278,57 +303,10 @@ class Parameters(AnsibleF5Parameters):
@property
def parent(self):
if self._values['parent_partition']:
if self._values['parent_partition'] is None:
return None
if self._values['parent_partition'].startswith('/'):
partition = os.path.basename(self._values['parent_partition'])
result = '/{0}/{1}'.format(partition, self.type)
else:
result = '/{0}/{1}'.format(self.parent_partition, self.type)
else:
if self._values['parent'] is None:
return None
if self._values['parent'].startswith('/'):
parent = os.path.basename(self._values['parent'])
result = '/{0}/{1}'.format(self.partition, parent)
else:
result = '/{0}/{1}'.format(self.partition, self._values['parent'])
return result
@property
def parent_partition(self):
if self._values['parent_partition'] is None:
if self._values['parent'] is None:
return None
self._values['__warnings'].append(
dict(
msg="The parent_partition param is deprecated",
version='2.4'
)
)
return self._values['parent_partition']
class ParametersTcp(Parameters):
api_map = {
'timeUntilUp': 'time_until_up',
'defaultsFrom': 'parent',
'recv': 'receive'
}
api_attributes = [
'timeUntilUp', 'defaultsFrom', 'interval', 'timeout', 'recv', 'send',
'destination'
]
returnables = [
'parent', 'send', 'receive', 'ip', 'port', 'interval', 'timeout',
'time_until_up'
]
updatables = [
'destination', 'send', 'receive', 'interval', 'timeout', 'time_until_up'
]
result = self._fqdn_name(self._values['parent'])
return result
@property
def port(self):
@ -355,148 +333,6 @@ class ParametersTcp(Parameters):
def type(self):
return 'tcp'
@type.setter
def type(self, value):
if value:
self._values['__warnings'].append(
dict(
msg="The type param is deprecated",
version='2.4'
)
)
class ParametersEcho(Parameters):
api_map = {
'timeUntilUp': 'time_until_up',
'defaultsFrom': 'parent',
'destination': 'ip'
}
api_attributes = [
'timeUntilUp', 'defaultsFrom', 'interval', 'timeout', 'destination'
]
returnables = [
'parent', 'ip', 'interval', 'timeout', 'time_until_up'
]
updatables = [
'destination', 'interval', 'timeout', 'time_until_up'
]
@property
def type(self):
return 'tcp_echo'
@type.setter
def type(self, value):
if value:
self._values['__warnings'].append(
dict(
msg="The type param is deprecated",
version='2.4'
)
)
@property
def destination(self):
return self.ip
@destination.setter
def destination(self, value):
self._values['ip'] = value
@property
def send(self):
if self._values['send'] is None:
return None
raise F5ModuleError(
"The 'send' parameter is not available for TCP echo"
)
@property
def receive(self):
if self._values['receive'] is None:
return None
raise F5ModuleError(
"The 'receive' parameter is not available for TCP echo"
)
@property
def port(self):
return None
class ParametersHalfOpen(Parameters):
api_map = {
'timeUntilUp': 'time_until_up',
'defaultsFrom': 'parent'
}
api_attributes = [
'timeUntilUp', 'defaultsFrom', 'interval', 'timeout', 'destination'
]
returnables = [
'parent', 'ip', 'port', 'interval', 'timeout', 'time_until_up'
]
updatables = [
'destination', 'interval', 'timeout', 'time_until_up'
]
@property
def destination(self):
if self.ip is None and self.port is None:
return None
result = '{0}:{1}'.format(self.ip, self.port)
return result
@destination.setter
def destination(self, value):
ip, port = value.split(':')
self._values['ip'] = ip
self._values['port'] = port
@property
def port(self):
if self._values['port'] is None:
return None
elif self._values['port'] == '*':
return '*'
return int(self._values['port'])
@property
def type(self):
return 'tcp_half_open'
@type.setter
def type(self, value):
if value:
self._values['__warnings'].append(
dict(
msg="The type param is deprecated",
version='2.4'
)
)
@property
def send(self):
if self._values['send'] is None:
return None
raise F5ModuleError(
"The 'send' parameter is not available for TCP half open"
)
@property
def receive(self):
if self._values['receive'] is None:
return None
raise F5ModuleError(
"The 'receive' parameter is not available for TCP half open"
)
class Difference(object):
def __init__(self, want, have=None):
@ -513,28 +349,24 @@ class Difference(object):
@property
def parent(self):
if self.want.parent != self.want.parent:
if self.want.parent != self.have.parent:
raise F5ModuleError(
"The parent monitor cannot be changed"
)
@property
def destination(self):
if self.want.type == 'tcp_echo':
if self.want.ip is None:
return None
else:
if self.want.ip is None and self.want.port is None:
return None
if self.want.port is None:
self.want.update({'port': self.have.port})
if self.want.ip is None:
self.want.update({'ip': self.have.ip})
if self.want.port in [None, '*'] and self.want.ip != '*':
raise F5ModuleError(
"Specifying an IP address requires that a port number be specified"
)
if self.want.ip is None and self.want.port is None:
return None
if self.want.port is None:
self.want.update({'port': self.have.port})
if self.want.ip is None:
self.want.update({'ip': self.have.ip})
if self.want.port in [None, '*'] and self.want.ip != '*':
raise F5ModuleError(
"Specifying an IP address requires that a port number be specified"
)
if self.want.destination != self.have.destination:
return self.want.destination
@ -569,26 +401,36 @@ class Difference(object):
return attr1
# TODO: Remove all of this in 2.5
class ModuleManager(object):
def __init__(self, client):
self.client = client
self.have = None
self.want = Parameters(self.client.module.params)
self.changes = Parameters()
def exec_module(self):
type = self.client.module.params.get('type', 'tcp')
manager = self.get_manager(type)
return manager.exec_module()
def get_manager(self, type):
if type in [None, 'tcp', 'TTYPE_TCP']:
return TcpManager(self.client)
elif type in ['tcp_echo', 'TTYPE_TCP_ECHO']:
return TcpEchoManager(self.client)
elif type in ['tcp_half_open', 'TTYPE_TCP_HALF_OPEN']:
return TcpHalfOpenManager(self.client)
def _set_changed_options(self):
changed = {}
for key in Parameters.returnables:
if getattr(self.want, key) is not None:
changed[key] = getattr(self.want, key)
if changed:
self.changes = Parameters(changed)
def _update_changed_options(self):
diff = Difference(self.want, self.have)
updatables = Parameters.updatables
changed = dict()
for k in updatables:
change = diff.compare(k)
if change is None:
continue
else:
changed[k] = change
if changed:
self.changes = Parameters(changed)
return True
return False
class BaseManager(object):
def _announce_deprecations(self):
warnings = []
if self.want:
@ -662,37 +504,6 @@ class BaseManager(object):
raise F5ModuleError("Failed to delete the monitor.")
return True
class TcpManager(BaseManager):
def __init__(self, client):
self.client = client
self.have = None
self.want = ParametersTcp(self.client.module.params)
self.changes = ParametersTcp()
def _set_changed_options(self):
changed = {}
for key in ParametersTcp.returnables:
if getattr(self.want, key) is not None:
changed[key] = getattr(self.want, key)
if changed:
self.changes = ParametersTcp(changed)
def _update_changed_options(self):
diff = Difference(self.want, self.have)
updatables = ParametersTcp.updatables
changed = dict()
for k in updatables:
change = diff.compare(k)
if change is None:
continue
else:
changed[k] = change
if changed:
self.changes = ParametersTcp(changed)
return True
return False
def _set_default_creation_values(self):
if self.want.timeout is None:
self.want.update({'timeout': 16})
@ -711,7 +522,7 @@ class TcpManager(BaseManager):
partition=self.want.partition
)
result = resource.attrs
return ParametersTcp(result)
return Parameters(result)
def exists(self):
result = self.client.api.tm.ltm.monitor.tcps.tcp.exists(
@ -745,199 +556,19 @@ class TcpManager(BaseManager):
result.delete()
# TODO: Remove this in 2.5 and put it its own module
class TcpEchoManager(BaseManager):
def __init__(self, client):
self.client = client
self.have = None
self.want = ParametersEcho(self.client.module.params)
self.changes = ParametersEcho()
def _set_default_creation_values(self):
if self.want.timeout is None:
self.want.update({'timeout': 16})
if self.want.interval is None:
self.want.update({'interval': 5})
if self.want.time_until_up is None:
self.want.update({'time_until_up': 0})
if self.want.ip is None:
self.want.update({'ip': '*'})
def _set_changed_options(self):
changed = {}
for key in ParametersEcho.returnables:
if getattr(self.want, key) is not None:
changed[key] = getattr(self.want, key)
if changed:
self.changes = ParametersEcho(changed)
def _update_changed_options(self):
diff = Difference(self.want, self.have)
updatables = ParametersEcho.updatables
changed = dict()
for k in updatables:
change = diff.compare(k)
if change is None:
continue
else:
changed[k] = change
if changed:
self.changes = ParametersEcho(changed)
return True
return False
def read_current_from_device(self):
resource = self.client.api.tm.ltm.monitor.tcp_echos.tcp_echo.load(
name=self.want.name,
partition=self.want.partition
)
result = resource.attrs
return ParametersEcho(result)
def exists(self):
result = self.client.api.tm.ltm.monitor.tcp_echos.tcp_echo.exists(
name=self.want.name,
partition=self.want.partition
)
return result
def update_on_device(self):
params = self.want.api_params()
result = self.client.api.tm.ltm.monitor.tcp_echos.tcp_echo.load(
name=self.want.name,
partition=self.want.partition
)
result.modify(**params)
def create_on_device(self):
params = self.want.api_params()
self.client.api.tm.ltm.monitor.tcp_echos.tcp_echo.create(
name=self.want.name,
partition=self.want.partition,
**params
)
def remove_from_device(self):
result = self.client.api.tm.ltm.monitor.tcp_echos.tcp_echo.load(
name=self.want.name,
partition=self.want.partition
)
if result:
result.delete()
# TODO: Remove this in 2.5 and put it its own module
class TcpHalfOpenManager(BaseManager):
def __init__(self, client):
self.client = client
self.have = None
self.want = ParametersHalfOpen(self.client.module.params)
self.changes = ParametersHalfOpen()
def _set_changed_options(self):
changed = {}
for key in ParametersHalfOpen.returnables:
if getattr(self.want, key) is not None:
changed[key] = getattr(self.want, key)
if changed:
self.changes = ParametersHalfOpen(changed)
def _update_changed_options(self):
diff = Difference(self.want, self.have)
updatables = ParametersHalfOpen.updatables
changed = dict()
for k in updatables:
change = diff.compare(k)
if change is None:
continue
else:
changed[k] = change
if changed:
self.changes = ParametersHalfOpen(changed)
return True
return False
def _set_default_creation_values(self):
if self.want.timeout is None:
self.want.update({'timeout': 16})
if self.want.interval is None:
self.want.update({'interval': 5})
if self.want.time_until_up is None:
self.want.update({'time_until_up': 0})
if self.want.ip is None:
self.want.update({'ip': '*'})
if self.want.port is None:
self.want.update({'port': '*'})
def read_current_from_device(self):
resource = self.client.api.tm.ltm.monitor.tcp_half_opens.tcp_half_open.load(
name=self.want.name,
partition=self.want.partition
)
result = resource.attrs
return ParametersHalfOpen(result)
def exists(self):
result = self.client.api.tm.ltm.monitor.tcp_half_opens.tcp_half_open.exists(
name=self.want.name,
partition=self.want.partition
)
return result
def update_on_device(self):
params = self.want.api_params()
result = self.client.api.tm.ltm.monitor.tcp_half_opens.tcp_half_open.load(
name=self.want.name,
partition=self.want.partition
)
result.modify(**params)
def create_on_device(self):
params = self.want.api_params()
self.client.api.tm.ltm.monitor.tcp_half_opens.tcp_half_open.create(
name=self.want.name,
partition=self.want.partition,
**params
)
def remove_from_device(self):
result = self.client.api.tm.ltm.monitor.tcp_half_opens.tcp_half_open.load(
name=self.want.name,
partition=self.want.partition
)
if result:
result.delete()
class ArgumentSpec(object):
def __init__(self):
self.supports_check_mode = True
self.argument_spec = dict(
name=dict(required=True),
# Make this assume "tcp" in the partition specified. The user
# is required to specify the full path if they want to use a different
# partition.
parent=dict(default='tcp'),
parent=dict(default='/Common/tcp'),
send=dict(),
receive=dict(),
ip=dict(),
port=dict(type='int'),
interval=dict(type='int'),
timeout=dict(type='int'),
time_until_up=dict(type='int'),
# Deprecated params
type=dict(
removed_in_version='2.4',
choices=[
'tcp', 'TTYPE_TCP', 'TTYPE_TCP_ECHO', 'TTYPE_TCP_HALF_OPEN'
]
),
parent_partition=dict(
removed_in_version='2.4'
)
time_until_up=dict(type='int')
)
self.f5_product_name = 'bigip'
self.mutually_exclusive = [

755
test/units/modules/network/f5/test_bigip_monitor_tcp.py
Unescape Escape View File

@ -22,26 +22,16 @@ from ansible.module_utils.f5_utils import AnsibleF5Client
from ansible.module_utils.f5_utils import F5ModuleError
try:
from library.bigip_monitor_tcp import ParametersTcp
from library.bigip_monitor_tcp import ParametersHalfOpen
from library.bigip_monitor_tcp import ParametersEcho
from library.bigip_monitor_tcp import Parameters
from library.bigip_monitor_tcp import ModuleManager
from library.bigip_monitor_tcp import ArgumentSpec
from library.bigip_monitor_tcp import TcpManager
from library.bigip_monitor_tcp import TcpEchoManager
from library.bigip_monitor_tcp import TcpHalfOpenManager
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_monitor_tcp import ParametersTcp
from ansible.modules.network.f5.bigip_monitor_tcp import ParametersHalfOpen
from ansible.modules.network.f5.bigip_monitor_tcp import ParametersEcho
from ansible.modules.network.f5.bigip_monitor_tcp import Parameters
from ansible.modules.network.f5.bigip_monitor_tcp import ModuleManager
from ansible.modules.network.f5.bigip_monitor_tcp import ArgumentSpec
from ansible.modules.network.f5.bigip_monitor_tcp import TcpManager
from ansible.modules.network.f5.bigip_monitor_tcp import TcpEchoManager
from ansible.modules.network.f5.bigip_monitor_tcp import TcpHalfOpenManager
from ansible.module_utils.f5_utils import iControlUnexpectedHTTPError
from units.modules.utils import set_module_args
except ImportError:
@ -85,7 +75,7 @@ class TestParameters(unittest.TestCase):
partition='Common'
)
p = ParametersTcp(args)
p = Parameters(args)
assert p.name == 'foo'
assert p.parent == '/Common/parent'
assert p.send == 'this is a send string'
@ -113,7 +103,7 @@ class TestParameters(unittest.TestCase):
partition='Common'
)
p = ParametersTcp(args)
p = Parameters(args)
assert p.name == 'foo'
assert p.parent == '/Common/parent'
assert p.send == 'this is a send string'
@ -138,7 +128,7 @@ class TestParameters(unittest.TestCase):
timeUntilUp=60
)
p = ParametersTcp(args)
p = Parameters(args)
assert p.name == 'foo'
assert p.parent == '/Common/parent'
assert p.send == 'this is a send string'
@ -182,14 +172,10 @@ class TestManager(unittest.TestCase):
f5_product_name=self.spec.f5_product_name
)
# Override methods in the specific type of manager
tm = TcpManager(client)
tm.exists = Mock(side_effect=[False, True])
tm.create_on_device = Mock(return_value=True)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
mm.exists = Mock(side_effect=[False, True])
mm.create_on_device = Mock(return_value=True)
results = mm.exec_module()
@ -213,7 +199,7 @@ class TestManager(unittest.TestCase):
user='admin'
))
current = ParametersTcp(load_fixture('load_ltm_monitor_tcp.json'))
current = Parameters(load_fixture('load_ltm_monitor_tcp.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
@ -221,13 +207,9 @@ class TestManager(unittest.TestCase):
)
# Override methods in the specific type of manager
tm = TcpManager(client)
tm.exists = Mock(return_value=True)
tm.read_current_from_device = Mock(return_value=current)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
mm.exists = Mock(return_value=True)
mm.read_current_from_device = Mock(return_value=current)
results = mm.exec_module()
@ -243,7 +225,7 @@ class TestManager(unittest.TestCase):
user='admin'
))
current = ParametersTcp(load_fixture('load_ltm_monitor_tcp.json'))
current = Parameters(load_fixture('load_ltm_monitor_tcp.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
@ -251,14 +233,10 @@ class TestManager(unittest.TestCase):
)
# Override methods in the specific type of manager
tm = TcpManager(client)
tm.exists = Mock(return_value=True)
tm.read_current_from_device = Mock(return_value=current)
tm.update_on_device = Mock(return_value=True)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
mm.exists = Mock(return_value=True)
mm.read_current_from_device = Mock(return_value=current)
mm.update_on_device = Mock(return_value=True)
results = mm.exec_module()
@ -275,7 +253,7 @@ class TestManager(unittest.TestCase):
user='admin'
))
current = ParametersTcp(load_fixture('load_ltm_monitor_tcp.json'))
current = Parameters(load_fixture('load_ltm_monitor_tcp.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
@ -283,14 +261,10 @@ class TestManager(unittest.TestCase):
)
# Override methods in the specific type of manager
tm = TcpManager(client)
tm.exists = Mock(return_value=True)
tm.read_current_from_device = Mock(return_value=current)
tm.update_on_device = Mock(return_value=True)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
mm.exists = Mock(return_value=True)
mm.read_current_from_device = Mock(return_value=current)
mm.update_on_device = Mock(return_value=True)
results = mm.exec_module()
@ -307,7 +281,7 @@ class TestManager(unittest.TestCase):
user='admin'
))
current = ParametersTcp(load_fixture('load_ltm_monitor_tcp.json'))
current = Parameters(load_fixture('load_ltm_monitor_tcp.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
@ -315,14 +289,10 @@ class TestManager(unittest.TestCase):
)
# Override methods in the specific type of manager
tm = TcpManager(client)
tm.exists = Mock(return_value=True)
tm.read_current_from_device = Mock(return_value=current)
tm.update_on_device = Mock(return_value=True)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
mm.exists = Mock(return_value=True)
mm.read_current_from_device = Mock(return_value=current)
mm.update_on_device = Mock(return_value=True)
with pytest.raises(F5ModuleError) as ex:
mm.exec_module()
@ -340,7 +310,7 @@ class TestManager(unittest.TestCase):
user='admin'
))
current = ParametersTcp(load_fixture('load_ltm_monitor_tcp.json'))
current = Parameters(load_fixture('load_ltm_monitor_tcp.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
@ -348,14 +318,10 @@ class TestManager(unittest.TestCase):
)
# Override methods in the specific type of manager
tm = TcpManager(client)
tm.exists = Mock(return_value=True)
tm.read_current_from_device = Mock(return_value=current)
tm.update_on_device = Mock(return_value=True)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
mm.exists = Mock(return_value=True)
mm.read_current_from_device = Mock(return_value=current)
mm.update_on_device = Mock(return_value=True)
with pytest.raises(F5ModuleError) as ex:
mm.exec_module()
@ -372,7 +338,7 @@ class TestManager(unittest.TestCase):
user='admin'
))
current = ParametersTcp(load_fixture('load_ltm_monitor_tcp.json'))
current = Parameters(load_fixture('load_ltm_monitor_tcp.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
@ -380,14 +346,10 @@ class TestManager(unittest.TestCase):
)
# Override methods in the specific type of manager
tm = TcpManager(client)
tm.exists = Mock(return_value=True)
tm.read_current_from_device = Mock(return_value=current)
tm.update_on_device = Mock(return_value=True)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
mm.exists = Mock(return_value=True)
mm.read_current_from_device = Mock(return_value=current)
mm.update_on_device = Mock(return_value=True)
results = mm.exec_module()
@ -404,7 +366,7 @@ class TestManager(unittest.TestCase):
user='admin'
))
current = ParametersTcp(load_fixture('load_ltm_monitor_tcp.json'))
current = Parameters(load_fixture('load_ltm_monitor_tcp.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
@ -412,14 +374,10 @@ class TestManager(unittest.TestCase):
)
# Override methods in the specific type of manager
tm = TcpManager(client)
tm.exists = Mock(return_value=True)
tm.read_current_from_device = Mock(return_value=current)
tm.update_on_device = Mock(return_value=True)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
mm.exists = Mock(return_value=True)
mm.read_current_from_device = Mock(return_value=current)
mm.update_on_device = Mock(return_value=True)
results = mm.exec_module()
@ -436,7 +394,7 @@ class TestManager(unittest.TestCase):
user='admin'
))
current = ParametersTcp(load_fixture('load_ltm_monitor_tcp.json'))
current = Parameters(load_fixture('load_ltm_monitor_tcp.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
@ -444,14 +402,10 @@ class TestManager(unittest.TestCase):
)
# Override methods in the specific type of manager
tm = TcpManager(client)
tm.exists = Mock(return_value=True)
tm.read_current_from_device = Mock(return_value=current)
tm.update_on_device = Mock(return_value=True)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
mm.exists = Mock(return_value=True)
mm.read_current_from_device = Mock(return_value=current)
mm.update_on_device = Mock(return_value=True)
results = mm.exec_module()
@ -468,116 +422,7 @@ class TestManager(unittest.TestCase):
user='admin'
))
current = ParametersTcp(load_fixture('load_ltm_monitor_tcp.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
# Override methods in the specific type of manager
tm = TcpManager(client)
tm.exists = Mock(return_value=True)
tm.read_current_from_device = Mock(return_value=current)
tm.update_on_device = Mock(return_value=True)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
results = mm.exec_module()
assert results['changed'] is True
assert results['time_until_up'] == 300
class TestParametersEcho(unittest.TestCase):
def test_module_parameters(self):
args = dict(
name='foo',
parent='parent',
ip='10.10.10.10',
type='TTYPE_TCP_ECHO',
interval=20,
timeout=30,
time_until_up=60,
partition='Common'
)
p = ParametersEcho(args)
assert p.name == 'foo'
assert p.parent == '/Common/parent'
assert p.ip == '10.10.10.10'
assert p.type == 'tcp_echo'
assert p.destination == '10.10.10.10'
assert p.interval == 20
assert p.timeout == 30
assert p.time_until_up == 60
def test_module_parameters_ints_as_strings(self):
args = dict(
name='foo',
parent='parent',
ip='10.10.10.10',
type='TTYPE_TCP_ECHO',
interval='20',
timeout='30',
time_until_up='60',
partition='Common'
)
p = ParametersEcho(args)
assert p.name == 'foo'
assert p.parent == '/Common/parent'
assert p.ip == '10.10.10.10'
assert p.type == 'tcp_echo'
assert p.destination == '10.10.10.10'
assert p.interval == 20
assert p.timeout == 30
assert p.time_until_up == 60
def test_api_parameters(self):
args = dict(
name='foo',
defaultsFrom='/Common/parent',
destination='10.10.10.10',
interval=20,
timeout=30,
timeUntilUp=60
)
p = ParametersEcho(args)
assert p.name == 'foo'
assert p.parent == '/Common/parent'
assert p.ip == '10.10.10.10'
assert p.type == 'tcp_echo'
assert p.destination == '10.10.10.10'
assert p.interval == 20
assert p.timeout == 30
assert p.time_until_up == 60
@patch('ansible.module_utils.f5_utils.AnsibleF5Client._get_mgmt_root',
return_value=True)
class TestManagerEcho(unittest.TestCase):
def setUp(self):
self.spec = ArgumentSpec()
def test_create_monitor(self, *args):
set_module_args(dict(
name='foo',
ip='10.10.10.10',
interval=20,
timeout=30,
time_until_up=60,
type='TTYPE_TCP_ECHO',
parent_partition='Common',
server='localhost',
password='password',
user='admin'
))
current = Parameters(load_fixture('load_ltm_monitor_tcp.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
@ -585,526 +430,10 @@ class TestManagerEcho(unittest.TestCase):
)
# Override methods in the specific type of manager
tm = TcpEchoManager(client)
tm.exists = Mock(side_effect=[False, True])
tm.create_on_device = Mock(return_value=True)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
results = mm.exec_module()
assert results['changed'] is True
assert results['parent'] == '/Common/tcp_echo'
def test_create_monitor_idempotent(self, *args):
set_module_args(dict(
name='foo',
ip='10.10.10.10',
interval=20,
timeout=30,
time_until_up=60,
type='TTYPE_TCP_ECHO',
parent_partition='Common',
server='localhost',
password='password',
user='admin'
))
current = ParametersEcho(load_fixture('load_ltm_monitor_tcp_echo.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
# Override methods in the specific type of manager
tm = TcpEchoManager(client)
tm.exists = Mock(return_value=True)
tm.read_current_from_device = Mock(return_value=current)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
results = mm.exec_module()
assert results['changed'] is False
def test_update_interval(self, *args):
set_module_args(dict(
name='foo',
interval=10,
parent_partition='Common',
type='TTYPE_TCP_ECHO',
server='localhost',
password='password',
user='admin'
))
current = ParametersEcho(load_fixture('load_ltm_monitor_tcp_echo.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
# Override methods in the specific type of manager
tm = TcpEchoManager(client)
tm.exists = Mock(return_value=True)
tm.read_current_from_device = Mock(return_value=current)
tm.update_on_device = Mock(return_value=True)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
results = mm.exec_module()
assert results['changed'] is True
assert results['interval'] == 10
def test_update_interval_larger_than_existing_timeout(self, *args):
set_module_args(dict(
name='foo',
interval=30,
parent_partition='Common',
type='TTYPE_TCP_ECHO',
server='localhost',
password='password',
user='admin'
))
current = ParametersEcho(load_fixture('load_ltm_monitor_tcp_echo.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
# Override methods in the specific type of manager
tm = TcpEchoManager(client)
tm.exists = Mock(return_value=True)
tm.read_current_from_device = Mock(return_value=current)
tm.update_on_device = Mock(return_value=True)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
with pytest.raises(F5ModuleError) as ex:
mm.exec_module()
assert "must be less than" in str(ex)
def test_update_interval_larger_than_new_timeout(self, *args):
set_module_args(dict(
name='foo',
interval=10,
timeout=5,
parent_partition='Common',
type='TTYPE_TCP_ECHO',
server='localhost',
password='password',
user='admin'
))
current = ParametersEcho(load_fixture('load_ltm_monitor_tcp_echo.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
# Override methods in the specific type of manager
tm = TcpEchoManager(client)
tm.exists = Mock(return_value=True)
tm.read_current_from_device = Mock(return_value=current)
tm.update_on_device = Mock(return_value=True)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
with pytest.raises(F5ModuleError) as ex:
mm.exec_module()
assert "must be less than" in str(ex)
def test_update_timeout(self, *args):
set_module_args(dict(
name='foo',
timeout=300,
parent_partition='Common',
type='TTYPE_TCP_ECHO',
server='localhost',
password='password',
user='admin'
))
current = ParametersEcho(load_fixture('load_ltm_monitor_tcp_echo.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
# Override methods in the specific type of manager
tm = TcpEchoManager(client)
tm.exists = Mock(return_value=True)
tm.read_current_from_device = Mock(return_value=current)
tm.update_on_device = Mock(return_value=True)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
results = mm.exec_module()
assert results['changed'] is True
assert results['timeout'] == 300
def test_update_time_until_up(self, *args):
set_module_args(dict(
name='foo',
time_until_up=300,
parent_partition='Common',
type='TTYPE_TCP_ECHO',
server='localhost',
password='password',
user='admin'
))
current = ParametersEcho(load_fixture('load_ltm_monitor_tcp_echo.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
# Override methods in the specific type of manager
tm = TcpEchoManager(client)
tm.exists = Mock(return_value=True)
tm.read_current_from_device = Mock(return_value=current)
tm.update_on_device = Mock(return_value=True)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
results = mm.exec_module()
assert results['changed'] is True
assert results['time_until_up'] == 300
class TestParametersHalfOpen(unittest.TestCase):
def test_module_parameters(self):
args = dict(
name='foo',
parent='parent',
ip='10.10.10.10',
port=80,
type='TTYPE_TCP_HALF_OPEN',
interval=20,
timeout=30,
time_until_up=60,
partition='Common'
)
p = ParametersHalfOpen(args)
assert p.name == 'foo'
assert p.parent == '/Common/parent'
assert p.ip == '10.10.10.10'
assert p.port == 80
assert p.type == 'tcp_half_open'
assert p.destination == '10.10.10.10:80'
assert p.interval == 20
assert p.timeout == 30
assert p.time_until_up == 60
def test_module_parameters_ints_as_strings(self):
args = dict(
name='foo',
parent='parent',
ip='10.10.10.10',
port=80,
type='TTYPE_TCP_HALF_OPEN',
interval='20',
timeout='30',
time_until_up='60',
partition='Common'
)
p = ParametersHalfOpen(args)
assert p.name == 'foo'
assert p.parent == '/Common/parent'
assert p.ip == '10.10.10.10'
assert p.port == 80
assert p.type == 'tcp_half_open'
assert p.destination == '10.10.10.10:80'
assert p.interval == 20
assert p.timeout == 30
assert p.time_until_up == 60
def test_api_parameters(self):
args = dict(
name='foo',
defaultsFrom='/Common/parent',
destination='10.10.10.10:80',
interval=20,
timeout=30,
timeUntilUp=60
)
p = ParametersHalfOpen(args)
assert p.name == 'foo'
assert p.parent == '/Common/parent'
assert p.ip == '10.10.10.10'
assert p.port == 80
assert p.type == 'tcp_half_open'
assert p.destination == '10.10.10.10:80'
assert p.interval == 20
assert p.timeout == 30
assert p.time_until_up == 60
@patch('ansible.module_utils.f5_utils.AnsibleF5Client._get_mgmt_root',
return_value=True)
class TestManagerHalfOpen(unittest.TestCase):
def setUp(self):
self.spec = ArgumentSpec()
def test_create_monitor(self, *args):
set_module_args(dict(
name='foo',
ip='10.10.10.10',
port=80,
interval=20,
timeout=30,
time_until_up=60,
type='TTYPE_TCP_HALF_OPEN',
parent_partition='Common',
server='localhost',
password='password',
user='admin'
))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
# Override methods in the specific type of manager
tm = TcpHalfOpenManager(client)
tm.exists = Mock(side_effect=[False, True])
tm.create_on_device = Mock(return_value=True)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
results = mm.exec_module()
assert results['changed'] is True
assert results['parent'] == '/Common/tcp_half_open'
def test_create_monitor_idempotent(self, *args):
set_module_args(dict(
name='foo',
ip='10.10.10.10',
port=80,
interval=20,
timeout=30,
time_until_up=60,
type='TTYPE_TCP_HALF_OPEN',
parent_partition='Common',
server='localhost',
password='password',
user='admin'
))
current = ParametersHalfOpen(load_fixture('load_ltm_monitor_tcp_half_open.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
# Override methods in the specific type of manager
tm = TcpHalfOpenManager(client)
tm.exists = Mock(return_value=True)
tm.read_current_from_device = Mock(return_value=current)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
results = mm.exec_module()
assert results['changed'] is False
def test_update_interval(self, *args):
set_module_args(dict(
name='foo',
interval=10,
parent_partition='Common',
type='TTYPE_TCP_HALF_OPEN',
server='localhost',
password='password',
user='admin'
))
current = ParametersHalfOpen(load_fixture('load_ltm_monitor_tcp_half_open.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
# Override methods in the specific type of manager
tm = TcpHalfOpenManager(client)
tm.exists = Mock(return_value=True)
tm.read_current_from_device = Mock(return_value=current)
tm.update_on_device = Mock(return_value=True)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
results = mm.exec_module()
assert results['changed'] is True
assert results['interval'] == 10
def test_update_interval_larger_than_existing_timeout(self, *args):
set_module_args(dict(
name='foo',
interval=30,
parent_partition='Common',
type='TTYPE_TCP_HALF_OPEN',
server='localhost',
password='password',
user='admin'
))
current = ParametersHalfOpen(load_fixture('load_ltm_monitor_tcp_half_open.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
# Override methods in the specific type of manager
tm = TcpHalfOpenManager(client)
tm.exists = Mock(return_value=True)
tm.read_current_from_device = Mock(return_value=current)
tm.update_on_device = Mock(return_value=True)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
with pytest.raises(F5ModuleError) as ex:
mm.exec_module()
assert "must be less than" in str(ex)
def test_update_interval_larger_than_new_timeout(self, *args):
set_module_args(dict(
name='foo',
interval=10,
timeout=5,
parent_partition='Common',
type='TTYPE_TCP_HALF_OPEN',
server='localhost',
password='password',
user='admin'
))
current = ParametersHalfOpen(load_fixture('load_ltm_monitor_tcp_half_open.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
# Override methods in the specific type of manager
tm = TcpHalfOpenManager(client)
tm.exists = Mock(return_value=True)
tm.read_current_from_device = Mock(return_value=current)
tm.update_on_device = Mock(return_value=True)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
with pytest.raises(F5ModuleError) as ex:
mm.exec_module()
assert "must be less than" in str(ex)
def test_update_timeout(self, *args):
set_module_args(dict(
name='foo',
timeout=300,
parent_partition='Common',
type='TTYPE_TCP_HALF_OPEN',
server='localhost',
password='password',
user='admin'
))
current = ParametersHalfOpen(load_fixture('load_ltm_monitor_tcp_half_open.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
# Override methods in the specific type of manager
tm = TcpHalfOpenManager(client)
tm.exists = Mock(return_value=True)
tm.read_current_from_device = Mock(return_value=current)
tm.update_on_device = Mock(return_value=True)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
results = mm.exec_module()
assert results['changed'] is True
assert results['timeout'] == 300
def test_update_time_until_up(self, *args):
set_module_args(dict(
name='foo',
time_until_up=300,
parent_partition='Common',
type='TTYPE_TCP_HALF_OPEN',
server='localhost',
password='password',
user='admin'
))
current = ParametersHalfOpen(load_fixture('load_ltm_monitor_tcp_half_open.json'))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
# Override methods in the specific type of manager
tm = TcpHalfOpenManager(client)
tm.exists = Mock(return_value=True)
tm.read_current_from_device = Mock(return_value=current)
tm.update_on_device = Mock(return_value=True)
# Override methods to force specific logic in the module to happen
mm = ModuleManager(client)
mm.get_manager = Mock(return_value=tm)
mm.exists = Mock(return_value=True)
mm.read_current_from_device = Mock(return_value=current)
mm.update_on_device = Mock(return_value=True)
results = mm.exec_module()

Write Preview
Loading…
Cancel
Save