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

Adds module for managing bigip device connectivity (#32950)

Browse source 
This module is a critical part of the HA process for BIG-IPs.
This commit is contained in:
Tim Rupp 2017-11-15 16:51:51 -08:00 committed by GitHub
parent 1bc4940ee1
commit 0c1f493b6c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 1089 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

591
lib/ansible/modules/network/f5/bigip_device_connectivity.py Normal file
View file

@ -0,0 +1,591 @@
#!/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_device_connectivity
short_description: Manages device IP configuration settings for HA on a BIG-IP
description:
- Manages device IP configuration settings for HA on a BIG-IP. Each BIG-IP device
has synchronization and failover connectivity information (IP addresses) that
you define as part of HA pairing or clustering. This module allows you to configure
that information.
version_added: "2.5"
options:
config_sync_ip:
description:
- Local IP address that the system uses for ConfigSync operations.
mirror_primary_address:
description:
- Specifies the primary IP address for the system to use to mirror
connections.
mirror_secondary_address:
description:
- Specifies the secondary IP address for the system to use to mirror
connections.
unicast_failover:
description:
- Desired addresses to use for failover operations. Options C(address)
and C(port) are supported with dictionary structure where C(address) is the
local IP address that the system uses for failover operations. Port
specifies the port that the system uses for failover operations. If C(port)
is not specified, the default value C(1026) will be used. If you are
specifying the (recommended) management IP address, use 'management-ip' in
the address field.
failover_multicast:
description:
- When C(yes), ensures that the Failover Multicast configuration is enabled
and if no further multicast configuration is provided, ensures that
C(multicast_interface), C(multicast_address) and C(multicast_port) are
the defaults specified in each option's description. When C(no), ensures
that Failover Multicast configuration is disabled.
choices:
- yes
- no
multicast_interface:
description:
- Interface over which the system sends multicast messages associated
with failover. When C(failover_multicast) is C(yes) and this option is
not provided, a default of C(eth0) will be used.
multicast_address:
description:
- IP address for the system to send multicast messages associated with
failover. When C(failover_multicast) is C(yes) and this option is not
provided, a default of C(224.0.0.245) will be used.
multicast_port:
description:
- Port for the system to send multicast messages associated with
failover. When C(failover_multicast) is C(yes) and this option is not
provided, a default of C(62960) will be used. This value must be between
0 and 65535.
notes:
- Requires the f5-sdk Python package on the host. This is as easy as pip
install f5-sdk.
- This module is primarily used as a component of configuring HA pairs of
BIG-IP devices.
- Requires BIG-IP >= 12.0.0
requirements:
- f5-sdk >= 2.2.3
extends_documentation_fragment: f5
author:
- Tim Rupp (@caphrim007)
'''
EXAMPLES = r'''
- name: Configure device connectivity for standard HA pair
bigip_device_connectivity:
config_sync_ip: 10.1.30.1
mirror_primary_address: 10.1.30.1
unicast_failover:
- address: management-ip
- address: 10.1.30.1
server: lb.mydomain.com
user: admin
password: secret
delegate_to: localhost
'''
RETURN = r'''
changed:
description: Denotes if the F5 configuration was updated.
returned: always
type: bool
config_sync_ip:
description: The new value of the C(config_sync_ip) setting.
returned: changed
type: string
sample: 10.1.1.1
mirror_primary_address:
description: The new value of the C(mirror_primary_address) setting.
returned: changed
type: string
sample: 10.1.1.2
mirror_secondary_address:
description: The new value of the C(mirror_secondary_address) setting.
returned: changed
type: string
sample: 10.1.1.3
unicast_failover:
description: The new value of the C(unicast_failover) setting.
returned: changed
type: list
sample: [{'address': '10.1.1.2', 'port': 1026}]
failover_multicast:
description: Whether a failover multicast attribute has been changed or not.
returned: changed
type: bool
multicast_interface:
description: The new value of the C(multicast_interface) setting.
returned: changed
type: string
sample: eth0
multicast_address:
description: The new value of the C(multicast_address) setting.
returned: changed
type: string
sample: 224.0.0.245
multicast_port:
description: The new value of the C(multicast_port) setting.
returned: changed
type: string
sample: 1026
'''
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
from ansible.module_utils.f5_utils import F5ModuleError
from ansible.module_utils.six import iteritems
try:
from netaddr import IPAddress, AddrFormatError
HAS_NETADDR = True
except ImportError:
HAS_NETADDR = False
try:
from ansible.module_utils.f5_utils import iControlUnexpectedHTTPError
except ImportError:
HAS_F5SDK = False
class Parameters(AnsibleF5Parameters):
api_map = {
'configsyncIp': 'config_sync_ip',
'multicastInterface': 'multicast_interface',
'multicastIp': 'multicast_address',
'multicastPort': 'multicast_port',
'mirrorIp': 'mirror_primary_address',
'mirrorSecondaryIp': 'mirror_secondary_address',
'managementIp': 'management_ip'
}
api_attributes = [
'configsyncIp', 'multicastInterface', 'multicastIp', 'multicastPort',
'mirrorIp', 'mirrorSecondaryIp', 'unicastAddress'
]
returnables = [
'config_sync_ip', 'multicast_interface', 'multicast_address',
'multicast_port', 'mirror_primary_address', 'mirror_secondary_address',
'failover_multicast', 'unicast_failover'
]
updatables = [
'config_sync_ip', 'multicast_interface', 'multicast_address',
'multicast_port', 'mirror_primary_address', 'mirror_secondary_address',
'failover_multicast', 'unicast_failover'
]
@property
def multicast_port(self):
if self._values['multicast_port'] is None:
return None
result = int(self._values['multicast_port'])
if result < 0 or result > 65535:
raise F5ModuleError(
"The specified 'multicast_port' must be between 0 and 65535."
)
return result
@property
def multicast_address(self):
if self._values['multicast_address'] is None:
return None
elif self._values['multicast_address'] in ["none", "any6", '']:
return "any6"
elif self._values['multicast_address'] == 'any':
return 'any'
result = self._get_validated_ip_address('multicast_address')
return result
@property
def mirror_primary_address(self):
if self._values['mirror_primary_address'] is None:
return None
elif self._values['mirror_primary_address'] in ["none", "any6", '']:
return "any6"
result = self._get_validated_ip_address('mirror_primary_address')
return result
@property
def mirror_secondary_address(self):
if self._values['mirror_secondary_address'] is None:
return None
elif self._values['mirror_secondary_address'] in ["none", "any6", '']:
return "any6"
result = self._get_validated_ip_address('mirror_secondary_address')
return result
@property
def config_sync_ip(self):
if self._values['config_sync_ip'] is None:
return None
elif self._values['config_sync_ip'] in ["none", '']:
return "none"
result = self._get_validated_ip_address('config_sync_ip')
return result
@property
def unicastAddress(self):
return self.unicast_failover
@unicastAddress.setter
def unicastAddress(self, value):
result = []
for item in value:
item['address'] = item.pop('ip')
result.append(item)
if result:
self._values['unicast_failover'] = result
@property
def unicast_failover(self):
if self._values['unicast_failover'] is None:
return None
if self._values['unicast_failover'] == ['none']:
return []
result = []
for item in self._values['unicast_failover']:
address = item.get('address', None)
port = item.get('port', None)
address = self._validate_unicast_failover_address(address)
port = self._validate_unicast_failover_port(port)
result.append(
dict(
effectiveIp=address,
effectivePort=port,
ip=address,
port=port
)
)
if result:
return result
else:
return None
def _validate_unicast_failover_port(self, port):
try:
result = int(port)
except ValueError:
raise F5ModuleError(
"The provided 'port' for unicast failover is not a valid number"
)
except TypeError:
result = 1026
return result
def _validate_unicast_failover_address(self, address):
try:
if address != 'management-ip':
result = IPAddress(address)
return str(result)
else:
return address
except KeyError:
raise F5ModuleError(
"An 'address' must be supplied when configuring unicast failover"
)
except AddrFormatError:
raise F5ModuleError(
"'address' field in unicast failover is not a valid IP address"
)
def _get_validated_ip_address(self, address):
try:
IPAddress(self._values[address])
return self._values[address]
except AddrFormatError:
raise F5ModuleError(
"The specified '{0}' is not a valid IP address".format(
address
)
)
def api_params(self):
result = {}
for api_attribute in self.api_attributes:
if self.api_map is not None and api_attribute in self.api_map:
result[api_attribute] = getattr(self, self.api_map[api_attribute])
else:
result[api_attribute] = getattr(self, api_attribute)
result = self._filter_params(result)
return result
class Changes(Parameters):
def to_return(self):
result = {}
try:
for returnable in self.returnables:
result[returnable] = getattr(self, returnable)
except Exception:
pass
result = self._filter_params(result)
return result
class ReportableChanges(Changes):
returnables = [
'config_sync_ip', 'multicast_interface', 'multicast_address',
'multicast_port', 'mirror_primary_address', 'mirror_secondary_address',
'failover_multicast', 'unicast_failover'
]
@property
def mirror_secondary_address(self):
if self._values['mirror_secondary_address'] in ['none', 'any6']:
return 'none'
return self._values['mirror_secondary_address']
@property
def mirror_primary_address(self):
if self._values['mirror_primary_address'] in ['none', 'any6']:
return 'none'
return self._values['mirror_primary_address']
@property
def multicast_address(self):
if self._values['multicast_address'] in ['none', 'any6']:
return 'none'
return self._values['multicast_address']
class UsableChanges(Changes):
@property
def mirror_primary_address(self):
if self._values['mirror_primary_address'] == ['any6', 'none', 'any']:
return "any6"
else:
return self._values['mirror_primary_address']
@property
def mirror_secondary_address(self):
if self._values['mirror_secondary_address'] == ['any6', 'none', 'any']:
return "any6"
else:
return self._values['mirror_secondary_address']
@property
def multicast_address(self):
if self._values['multicast_address'] == ['any6', 'none', 'any']:
return "any"
else:
return self._values['multicast_address']
@property
def unicast_failover(self):
if self._values['unicast_failover'] is None:
return None
elif self._values['unicast_failover']:
return self._values['unicast_failover']
return "none"
class Difference(object):
def __init__(self, want, have=None):
self.want = want
self.have = have
def compare(self, param):
try:
result = getattr(self, param)
return result
except AttributeError:
return self.__default(param)
def __default(self, param):
attr1 = getattr(self.want, param)
try:
attr2 = getattr(self.have, param)
if attr1 != attr2:
return attr1
except AttributeError:
return attr1
def to_tuple(self, failovers):
result = []
for x in failovers:
for k, v in iteritems(x):
# Have to do this in cases where the BIG-IP stores the word
# "management-ip" when you specify the management IP address.
#
# Otherwise, a difference would be registered.
if v == self.have.management_ip:
v = 'management-ip'
result += [(str(k), str(v))]
return result
@property
def unicast_failover(self):
if self.want.unicast_failover == [] and self.have.unicast_failover is None:
return None
if self.want.unicast_failover is None:
return None
if self.have.unicast_failover is None:
return self.want.unicast_failover
want = self.to_tuple(self.want.unicast_failover)
have = self.to_tuple(self.have.unicast_failover)
if set(want) == set(have):
return None
else:
return self.want.unicast_failover
@property
def failover_multicast(self):
values = ['multicast_address', 'multicast_interface', 'multicast_port']
if self.want.failover_multicast is False:
if self.have.multicast_interface == 'eth0' and self.have.multicast_address == 'any' and self.have.multicast_port == 0:
return None
else:
result = dict(
failover_multicast=True,
multicast_port=0,
multicast_interface='eth0',
multicast_address='any'
)
return result
else:
if all(self.have._values[x] in [None, 'any6', 'any'] for x in values):
return True
class ModuleManager(object):
def __init__(self, client):
self.client = client
self.want = Parameters(self.client.module.params)
self.changes = UsableChanges()
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:
if isinstance(change, dict):
changed.update(change)
else:
changed[k] = change
if changed:
self.changes = UsableChanges(changed)
return True
return False
def should_update(self):
result = self._update_changed_options()
if result:
return True
return False
def exec_module(self):
changed = False
result = dict()
state = self.want.state
try:
if state == "present":
changed = self.update()
except iControlUnexpectedHTTPError as e:
raise F5ModuleError(str(e))
reportable = ReportableChanges(self.changes.to_return())
changes = reportable.to_return()
result.update(**changes)
result.update(dict(changed=changed))
return result
def update(self):
self.have = self.read_current_from_device()
if not self.should_update():
return False
if self.client.check_mode:
return True
self.update_on_device()
return True
def update_on_device(self):
params = self.changes.api_params()
collection = self.client.api.tm.cm.devices.get_collection()
for resource in collection:
if resource.selfDevice == 'true':
resource.modify(**params)
return
raise F5ModuleError(
"The host device was not found."
)
def read_current_from_device(self):
collection = self.client.api.tm.cm.devices.get_collection()
for resource in collection:
if resource.selfDevice == 'true':
result = resource.attrs
return Parameters(result)
raise F5ModuleError(
"The host device was not found."
)
class ArgumentSpec(object):
def __init__(self):
self.supports_check_mode = True
self.argument_spec = dict(
multicast_port=dict(
type='int'
),
multicast_address=dict(),
multicast_interface=dict(),
failover_multicast=dict(
type='bool'
),
unicast_failover=dict(
type='list'
),
mirror_primary_address=dict(),
mirror_secondary_address=dict(),
config_sync_ip=dict(),
state=dict(
default='present',
choices=['present']
)
)
self.f5_product_name = 'bigip'
self.required_together = [
['multicast_address', 'multicast_interface', 'multicast_port']
]
def main():
if not HAS_F5SDK:
raise F5ModuleError("The python f5-sdk module is required")
if not HAS_NETADDR:
raise F5ModuleError("The python netaddr module is required")
spec = ArgumentSpec()
client = AnsibleF5Client(
argument_spec=spec.argument_spec,
supports_check_mode=spec.supports_check_mode,
f5_product_name=spec.f5_product_name
)
try:
mm = ModuleManager(client)
results = mm.exec_module()
client.module.exit_json(**results)
except F5ModuleError as e:
client.module.fail_json(msg=str(e))
if __name__ == '__main__':
main()

76
test/units/modules/network/f5/fixtures/load_tm_cm_device.json Normal file
View file

@ -0,0 +1,76 @@
{
"kind": "tm:cm:device:devicestate",
"name": "bigip1",
"partition": "Common",
"fullPath": "/Common/bigip1",
"generation": 65,
"selfLink": "https://localhost/mgmt/tm/cm/device/~Common~bigip1?ver=12.1.2",
"activeModules": [
"APM, Max, VE (2500 CCU, 10000 Access Sessions)|P961057-1761515|Anti-Virus Checks|Base Endpoint Security Checks|Firewall Checks|Machine Certificate Checks|Network Access|Protected Workspace|Secure Virtual Keyboard|APM, Web Application|App Tunnel|Remote Desktop",
"LTM, 10 Gbps, VE|T487107-2453693|IPV6 Gateway|Rate Shaping|Ram Cache|External Interface and Network HSM, VE|SSL, Forward Proxy, VE|DENY-VER-GBB|Application Acceleration Manager, Core|PEM, Quota Management, VE|BIG-IP, iAppsLX (Node.js)|Max Compression, VE|BIG-IP VE, Multicast Routing|Recycle, BIG-IP, VE|APM, Limited|LTM to Best Bundle Upgrade, 10Gbps|BIG-IP, iRulesLX (Node.js)|SSL, VE|Anti-Virus Checks|Base Endpoint Security Checks|Firewall Checks|Machine Certificate Checks|Network Access|Protected Workspace|Secure Virtual Keyboard|APM, Web Application|App Tunnel|Remote Desktop|SDN Services, VE|Acceleration Manager, VE|AFM, VE|APM, Base, VE GBB (500 CCU)|ASM, VE|DNS-GTM, Base, 10Gbps|DNS Licensed Objects, Unlimited|GTM Licensed Objects, Unlimited|GTM Rate, 250K|DNS Rate Fallback, 250K|DNS Rate Limit, 250K QPS|GTM Rate Fallback, 250K|CGN, BIG-IP VE, AFM ONLY|PSM, VE|Routing Bundle, VE|DNSSEC",
"PEM, VE|X895364-1851682"
],
"baseMac": "08:00:27:27:74:82",
"build": "0.0.249",
"cert": "/Common/dtdi.crt",
"certReference": {
"link": "https://localhost/mgmt/tm/cm/cert/~Common~dtdi.crt?ver=12.1.2"
},
"chassisId": "2d37dfa6-c0e8-4e4a-ae983c67356d",
"chassisType": "individual",
"configsyncIp": "10.2.2.2",
"edition": "Final",
"failoverState": "active",
"haCapacity": 0,
"hostname": "bigip1",
"key": "/Common/dtdi.key",
"keyReference": {
"link": "https://localhost/mgmt/tm/cm/key/~Common~dtdi.key?ver=12.1.2"
},
"managementIp": "10.0.2.15",
"marketingName": "BIG-IP Virtual Edition",
"mirrorIp": "10.2.2.2",
"mirrorSecondaryIp": "10.2.3.2",
"multicastInterface": "eth0",
"multicastIp": "224.0.0.245",
"multicastPort": 62960,
"optionalModules": [
"APM, Base, VE (50 CCU / 200 Access Sessions)",
"App Mode (TMSH Only, No Root/Bash)",
"Concurrent Users",
"Concurrent Users and Access Sessions, VE",
"IPI Subscription, 1Yr, VE",
"IPI Subscription, 1Yr, VE-10G",
"IPI Subscription, 3Yr, VE-10G",
"LTM to Better Bundle Upgrade, 10Gbps",
"PEM URL Filtering, 1Yr, HIGH PERF",
"PEM URL Filtering, 3Yr, HIGH PERF",
"Routing Bundle",
"SWG Subscription, 1Yr, VE",
"URL Filtering Subscription, 1Yr, VE"
],
"platformId": "Z100",
"product": "BIG-IP",
"selfDevice": "true",
"timeLimitedModules": [
"IPI Subscription, 3Yr, VE|E430735-0717882|20170429|20170511|SUBSCRIPTION",
"SWG Subscription, 3Yr, VE|W797718-6984294|20170429|20170511|SUBSCRIPTION",
"URL Filtering Subscription, 3Yr, VE|G132953-9613041|20170429|20170511|SUBSCRIPTION"
],
"timeZone": "America/Los_Angeles",
"version": "12.1.2",
"unicastAddress": [
{
"effectiveIp": "management-ip",
"effectivePort": 1026,
"ip": "management-ip",
"port": 1026
},
{
"effectiveIp": "10.2.2.2",
"effectivePort": 1026,
"ip": "10.2.2.2",
"port": 1026
}
]
}

61
test/units/modules/network/f5/fixtures/load_tm_cm_device_default.json Normal file
View file

@ -0,0 +1,61 @@
{
"kind": "tm:cm:device:devicestate",
"name": "bigip1",
"partition": "Common",
"fullPath": "/Common/bigip1",
"generation": 1,
"selfLink": "https://localhost/mgmt/tm/cm/device/~Common~bigip1?ver=12.1.2",
"activeModules": [
"APM, Max, VE (2500 CCU, 10000 Access Sessions)|P961057-1761515|Anti-Virus Checks|Base Endpoint Security Checks|Firewall Checks|Machine Certificate Checks|Network Access|Protected Workspace|Secure Virtual Keyboard|APM, Web Application|App Tunnel|Remote Desktop",
"LTM, 10 Gbps, VE|T487107-2453693|IPV6 Gateway|Rate Shaping|Ram Cache|External Interface and Network HSM, VE|SSL, Forward Proxy, VE|DENY-VER-GBB|Application Acceleration Manager, Core|PEM, Quota Management, VE|BIG-IP, iAppsLX (Node.js)|Max Compression, VE|BIG-IP VE, Multicast Routing|Recycle, BIG-IP, VE|APM, Limited|LTM to Best Bundle Upgrade, 10Gbps|BIG-IP, iRulesLX (Node.js)|SSL, VE|Anti-Virus Checks|Base Endpoint Security Checks|Firewall Checks|Machine Certificate Checks|Network Access|Protected Workspace|Secure Virtual Keyboard|APM, Web Application|App Tunnel|Remote Desktop|SDN Services, VE|Acceleration Manager, VE|AFM, VE|APM, Base, VE GBB (500 CCU)|ASM, VE|DNS-GTM, Base, 10Gbps|DNS Licensed Objects, Unlimited|GTM Licensed Objects, Unlimited|GTM Rate, 250K|DNS Rate Fallback, 250K|DNS Rate Limit, 250K QPS|GTM Rate Fallback, 250K|CGN, BIG-IP VE, AFM ONLY|PSM, VE|Routing Bundle, VE|DNSSEC",
"PEM, VE|X895364-1851682"
],
"baseMac": "08:00:27:27:74:82",
"build": "0.0.249",
"cert": "/Common/dtdi.crt",
"certReference": {
"link": "https://localhost/mgmt/tm/cm/cert/~Common~dtdi.crt?ver=12.1.2"
},
"chassisId": "42d93eba-35bb-4f01-4663fb03951a",
"chassisType": "individual",
"configsyncIp": "none",
"edition": "Final",
"failoverState": "active",
"haCapacity": 0,
"hostname": "bigip1",
"key": "/Common/dtdi.key",
"keyReference": {
"link": "https://localhost/mgmt/tm/cm/key/~Common~dtdi.key?ver=12.1.2"
},
"managementIp": "10.0.2.15",
"marketingName": "BIG-IP Virtual Edition",
"mirrorIp": "any6",
"mirrorSecondaryIp": "any6",
"multicastIp": "any6",
"multicastPort": 0,
"optionalModules": [
"APM, Base, VE (50 CCU / 200 Access Sessions)",
"App Mode (TMSH Only, No Root/Bash)",
"Concurrent Users",
"Concurrent Users and Access Sessions, VE",
"IPI Subscription, 1Yr, VE",
"IPI Subscription, 1Yr, VE-10G",
"IPI Subscription, 3Yr, VE-10G",
"LTM to Better Bundle Upgrade, 10Gbps",
"PEM URL Filtering, 1Yr, HIGH PERF",
"PEM URL Filtering, 3Yr, HIGH PERF",
"Routing Bundle",
"SWG Subscription, 1Yr, VE",
"URL Filtering Subscription, 1Yr, VE"
],
"platformId": "Z100",
"product": "BIG-IP",
"selfDevice": "true",
"timeLimitedModules": [
"IPI Subscription, 3Yr, VE|E430735-0717882|20170502|20170511|SUBSCRIPTION",
"SWG Subscription, 3Yr, VE|W797718-6984294|20170502|20170511|SUBSCRIPTION",
"URL Filtering Subscription, 3Yr, VE|G132953-9613041|20170502|20170511|SUBSCRIPTION"
],
"timeZone": "America/Los_Angeles",
"version": "12.1.2"
}

361
test/units/modules/network/f5/test_bigip_device_connectivity.py Normal file
View file

@ -0,0 +1,361 @@
# -*- 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 pytest
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 patch, Mock
from ansible.module_utils import basic
from ansible.module_utils._text import to_bytes
from ansible.module_utils.f5_utils import AnsibleF5Client
from ansible.module_utils.f5_utils import F5ModuleError
try:
from library.bigip_device_connectivity import Parameters
from library.bigip_device_connectivity import ModuleManager
from library.bigip_device_connectivity import ArgumentSpec
from ansible.module_utils.f5_utils import iControlUnexpectedHTTPError
except ImportError:
try:
from ansible.modules.network.f5.bigip_device_connectivity import Parameters
from ansible.modules.network.f5.bigip_device_connectivity import ModuleManager
from ansible.modules.network.f5.bigip_device_connectivity import ArgumentSpec
from ansible.module_utils.f5_utils import iControlUnexpectedHTTPError
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 set_module_args(args):
args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
basic._ANSIBLE_ARGS = to_bytes(args)
def load_fixture(name):
path = os.path.join(fixture_path, name)
with open(path) as f:
data = f.read()
try:
data = json.loads(data)
except Exception:
pass
return data
class TestParameters(unittest.TestCase):
def test_module_parameters(self):
args = dict(
multicast_port='1010',
multicast_address='10.10.10.10',
multicast_interface='eth0',
failover_multicast=True,
unicast_failover=[
dict(
address='20.20.20.20',
port='1234'
)
],
mirror_primary_address='1.2.3.4',
mirror_secondary_address='5.6.7.8',
config_sync_ip='4.3.2.1',
state='present',
server='localhost',
user='admin',
password='password'
)
p = Parameters(args)
assert p.multicast_port == 1010
assert p.multicast_address == '10.10.10.10'
assert p.multicast_interface == 'eth0'
assert p.failover_multicast is True
assert p.mirror_primary_address == '1.2.3.4'
assert p.mirror_secondary_address == '5.6.7.8'
assert p.config_sync_ip == '4.3.2.1'
assert len(p.unicast_failover) == 1
assert 'effectiveIp' in p.unicast_failover[0]
assert 'effectivePort' in p.unicast_failover[0]
assert 'port' in p.unicast_failover[0]
assert 'ip' in p.unicast_failover[0]
assert p.unicast_failover[0]['effectiveIp'] == '20.20.20.20'
assert p.unicast_failover[0]['ip'] == '20.20.20.20'
assert p.unicast_failover[0]['port'] == 1234
assert p.unicast_failover[0]['effectivePort'] == 1234
def test_api_parameters(self):
params = load_fixture('load_tm_cm_device.json')
p = Parameters(params)
assert p.multicast_port == 62960
assert p.multicast_address == '224.0.0.245'
assert p.multicast_interface == 'eth0'
assert p.mirror_primary_address == '10.2.2.2'
assert p.mirror_secondary_address == '10.2.3.2'
assert p.config_sync_ip == '10.2.2.2'
assert len(p.unicast_failover) == 2
assert 'effectiveIp' in p.unicast_failover[0]
assert 'effectivePort' in p.unicast_failover[0]
assert 'port' in p.unicast_failover[0]
assert 'ip' in p.unicast_failover[0]
assert p.unicast_failover[0]['effectiveIp'] == 'management-ip'
assert p.unicast_failover[0]['ip'] == 'management-ip'
assert p.unicast_failover[0]['port'] == 1026
assert p.unicast_failover[0]['effectivePort'] == 1026
@patch('ansible.module_utils.f5_utils.AnsibleF5Client._get_mgmt_root',
return_value=True)
class TestManager(unittest.TestCase):
def setUp(self):
self.spec = ArgumentSpec()
def test_update_settings(self, *args):
set_module_args(dict(
config_sync_ip="10.1.30.1",
mirror_primary_address="10.1.30.1",
unicast_failover=[
dict(
address="10.1.30.1"
)
],
server='localhost',
user='admin',
password='password'
))
# Configure the parameters that would be returned by querying the
# remote device
current = Parameters(load_fixture('load_tm_cm_device_default.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
)
mm = ModuleManager(client)
# Override methods to force specific logic in the module to happen
mm.update_on_device = Mock(return_value=True)
mm.read_current_from_device = Mock(return_value=current)
results = mm.exec_module()
assert results['changed'] is True
assert results['config_sync_ip'] == '10.1.30.1'
assert results['mirror_primary_address'] == '10.1.30.1'
assert len(results.keys()) == 3
def test_set_primary_mirror_address_none(self, *args):
set_module_args(dict(
mirror_primary_address="none",
server='localhost',
user='admin',
password='password'
))
# Configure the parameters that would be returned by querying the
# remote device
current = Parameters(load_fixture('load_tm_cm_device.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
)
mm = ModuleManager(client)
# Override methods to force specific logic in the module to happen
mm.update_on_device = Mock(return_value=True)
mm.read_current_from_device = Mock(return_value=current)
results = mm.exec_module()
assert results['changed'] is True
assert results['mirror_primary_address'] == 'none'
assert len(results.keys()) == 2
def test_set_secondary_mirror_address_none(self, *args):
set_module_args(dict(
mirror_secondary_address="none",
server='localhost',
user='admin',
password='password'
))
# Configure the parameters that would be returned by querying the
# remote device
current = Parameters(load_fixture('load_tm_cm_device.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
)
mm = ModuleManager(client)
# Override methods to force specific logic in the module to happen
mm.update_on_device = Mock(return_value=True)
mm.read_current_from_device = Mock(return_value=current)
results = mm.exec_module()
assert results['changed'] is True
assert results['mirror_secondary_address'] == 'none'
assert len(results.keys()) == 2
def test_set_multicast_address_none(self, *args):
set_module_args(dict(
multicast_address="none",
server='localhost',
user='admin',
password='password'
))
# Configure the parameters that would be returned by querying the
# remote device
current = Parameters(load_fixture('load_tm_cm_device.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
)
mm = ModuleManager(client)
# Override methods to force specific logic in the module to happen
mm.update_on_device = Mock(return_value=True)
mm.read_current_from_device = Mock(return_value=current)
results = mm.exec_module()
assert results['changed'] is True
assert results['multicast_address'] == 'none'
assert len(results.keys()) == 2
def test_set_multicast_port_negative(self, *args):
set_module_args(dict(
multicast_port=-1,
server='localhost',
user='admin',
password='password'
))
# Configure the parameters that would be returned by querying the
# remote device
current = Parameters(load_fixture('load_tm_cm_device.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
)
mm = ModuleManager(client)
# Override methods to force specific logic in the module to happen
mm.update_on_device = Mock(return_value=True)
mm.read_current_from_device = Mock(return_value=current)
with pytest.raises(F5ModuleError) as ex:
mm.exec_module()
assert 'must be between' in str(ex)
def test_set_multicast_address(self, *args):
set_module_args(dict(
multicast_address="10.1.1.1",
server='localhost',
user='admin',
password='password'
))
# Configure the parameters that would be returned by querying the
# remote device
current = Parameters(load_fixture('load_tm_cm_device.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
)
mm = ModuleManager(client)
# Override methods to force specific logic in the module to happen
mm.update_on_device = Mock(return_value=True)
mm.read_current_from_device = Mock(return_value=current)
results = mm.exec_module()
assert results['changed'] is True
assert results['multicast_address'] == '10.1.1.1'
assert len(results.keys()) == 2
def test_unset_unicast_failover(self, *args):
set_module_args(dict(
unicast_failover="none",
server='localhost',
user='admin',
password='password'
))
# Configure the parameters that would be returned by querying the
# remote device
current = Parameters(load_fixture('load_tm_cm_device.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
)
mm = ModuleManager(client)
# Override methods to force specific logic in the module to happen
mm.update_on_device = Mock(return_value=True)
mm.read_current_from_device = Mock(return_value=current)
results = mm.exec_module()
assert results['changed'] is True
assert results['unicast_failover'] == 'none'
assert len(results.keys()) == 2
def test_unset_config_sync_ip(self, *args):
set_module_args(dict(
config_sync_ip="none",
server='localhost',
user='admin',
password='password'
))
# Configure the parameters that would be returned by querying the
# remote device
current = Parameters(load_fixture('load_tm_cm_device.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
)
mm = ModuleManager(client)
# Override methods to force specific logic in the module to happen
mm.update_on_device = Mock(return_value=True)
mm.read_current_from_device = Mock(return_value=current)
results = mm.exec_module()
assert results['changed'] is True
assert results['config_sync_ip'] == 'none'
assert len(results.keys()) == 2