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

Refactors bigip_vlan (#33489)

This patch adds tests and refactors the code to be inline with current
f5 coding standards
This commit is contained in:
Tim Rupp 2017-12-02 18:58:51 -08:00 committed by GitHub
parent f9b138b0ab
commit 277d416b5b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 1257 additions and 261 deletions

View file

@ -4,11 +4,15 @@
# Copyright (c) 2017 F5 Networks Inc. # Copyright (c) 2017 F5 Networks Inc.
# GNU General Public License v3.0 (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) # 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', ANSIBLE_METADATA = {'metadata_version': '1.1',
'status': ['preview'], 'status': ['preview'],
'supported_by': 'community'} 'supported_by': 'community'}
DOCUMENTATION = ''' DOCUMENTATION = r'''
--- ---
module: bigip_vlan module: bigip_vlan
short_description: Manage VLANs on a BIG-IP system short_description: Manage VLANs on a BIG-IP system
@ -24,27 +28,24 @@ options:
- Specifies a list of tagged interfaces and trunks that you want to - Specifies a list of tagged interfaces and trunks that you want to
configure for the VLAN. Use tagged interfaces or trunks when configure for the VLAN. Use tagged interfaces or trunks when
you want to assign a single interface or trunk to multiple VLANs. you want to assign a single interface or trunk to multiple VLANs.
required: false
aliases: aliases:
- tagged_interface - tagged_interface
untagged_interfaces: untagged_interfaces:
description: description:
- Specifies a list of untagged interfaces and trunks that you want to - Specifies a list of untagged interfaces and trunks that you want to
configure for the VLAN. configure for the VLAN.
required: false
aliases: aliases:
- untagged_interface - untagged_interface
name: name:
description: description:
- The VLAN to manage. If the special VLAN C(ALL) is specified with - The VLAN to manage. If the special VLAN C(ALL) is specified with
the C(state) value of C(absent) then all VLANs will be removed. the C(state) value of C(absent) then all VLANs will be removed.
required: true required: True
state: state:
description: description:
- The state of the VLAN on the system. When C(present), guarantees - The state of the VLAN on the system. When C(present), guarantees
that the VLAN exists with the provided attributes. When C(absent), that the VLAN exists with the provided attributes. When C(absent),
removes the VLAN from the system. removes the VLAN from the system.
required: false
default: present default: present
choices: choices:
- absent - absent
@ -63,9 +64,10 @@ requirements:
- f5-sdk - f5-sdk
author: author:
- Tim Rupp (@caphrim007) - Tim Rupp (@caphrim007)
- Wojciech Wypior (@wojtek0806)
''' '''
EXAMPLES = ''' EXAMPLES = r'''
- name: Create VLAN - name: Create VLAN
bigip_vlan: bigip_vlan:
name: "net1" name: "net1"
@ -110,7 +112,7 @@ EXAMPLES = '''
delegate_to: localhost delegate_to: localhost
''' '''
RETURN = ''' RETURN = r'''
description: description:
description: The description set on the VLAN description: The description set on the VLAN
returned: changed returned: changed
@ -138,255 +140,141 @@ tag:
sample: 2345 sample: 2345
''' '''
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
from collections import defaultdict
try: try:
from f5.bigip import ManagementRoot from ansible.module_utils.f5_utils import iControlUnexpectedHTTPError
from icontrol.session import iControlUnexpectedHTTPError
HAS_F5SDK = True
except ImportError: except ImportError:
HAS_F5SDK = False HAS_F5SDK = False
class BigIpVlan(object): class Parameters(AnsibleF5Parameters):
def __init__(self, *args, **kwargs): def __init__(self, params=None):
if not HAS_F5SDK: self._values = defaultdict(lambda: None)
raise F5ModuleError("The python f5-sdk module is required")
# The params that change in the module
self.cparams = dict()
# Stores the params that are sent to the module
self.params = kwargs
self.api = ManagementRoot(kwargs['server'],
kwargs['user'],
kwargs['password'],
port=kwargs['server_port'])
def present(self):
if self.exists():
return self.update()
else:
return self.create()
def absent(self):
changed = False
if self.exists():
changed = self.delete()
return changed
def read(self):
"""Read information and transform it
The values that are returned by BIG-IP in the f5-sdk can have encoding
attached to them as well as be completely missing in some cases.
Therefore, this method will transform the data from the BIG-IP into a
format that is more easily consumable by the rest of the class and the
parameters that are supported by the module.
"""
p = dict()
name = self.params['name']
partition = self.params['partition']
r = self.api.tm.net.vlans.vlan.load(
name=name,
partition=partition
)
ifcs = r.interfaces_s.get_collection()
if hasattr(r, 'tag'):
p['tag'] = int(r.tag)
if hasattr(r, 'description'):
p['description'] = str(r.description)
if len(ifcs) is not 0:
untagged = []
tagged = []
for x in ifcs:
if hasattr(x, 'tagged'):
tagged.append(str(x.name))
elif hasattr(x, 'untagged'):
untagged.append(str(x.name))
if untagged:
p['untagged_interfaces'] = list(set(untagged))
if tagged:
p['tagged_interfaces'] = list(set(tagged))
p['name'] = name
return p
def create(self):
params = dict()
check_mode = self.params['check_mode']
description = self.params['description']
name = self.params['name']
untagged_interfaces = self.params['untagged_interfaces']
tagged_interfaces = self.params['tagged_interfaces']
partition = self.params['partition']
tag = self.params['tag']
if tag is not None:
params['tag'] = tag
if untagged_interfaces is not None or tagged_interfaces is not None:
tmp = []
ifcs = self.api.tm.net.interfaces.get_collection()
ifcs = [str(x.name) for x in ifcs]
if len(ifcs) is 0:
raise F5ModuleError(
'No interfaces were found'
)
pinterfaces = []
if untagged_interfaces:
interfaces = untagged_interfaces
elif tagged_interfaces:
interfaces = tagged_interfaces
for ifc in interfaces:
ifc = str(ifc)
if ifc in ifcs:
pinterfaces.append(ifc)
if tagged_interfaces:
tmp = [dict(name=x, tagged=True) for x in pinterfaces]
elif untagged_interfaces:
tmp = [dict(name=x, untagged=True) for x in pinterfaces]
if tmp:
params['interfaces'] = tmp
if description is not None:
params['description'] = self.params['description']
params['name'] = name
params['partition'] = partition
self.cparams = camel_dict_to_snake_dict(params)
if check_mode:
return True
d = self.api.tm.net.vlans.vlan
d.create(**params)
if self.exists():
return True
else:
raise F5ModuleError("Failed to create the VLAN")
def update(self):
changed = False
params = dict()
current = self.read()
check_mode = self.params['check_mode']
description = self.params['description']
name = self.params['name']
tag = self.params['tag']
partition = self.params['partition']
tagged_interfaces = self.params['tagged_interfaces']
untagged_interfaces = self.params['untagged_interfaces']
if untagged_interfaces is not None or tagged_interfaces is not None:
ifcs = self.api.tm.net.interfaces.get_collection()
ifcs = [str(x.name) for x in ifcs]
if len(ifcs) is 0:
raise F5ModuleError(
'No interfaces were found'
)
pinterfaces = []
if untagged_interfaces:
interfaces = untagged_interfaces
elif tagged_interfaces:
interfaces = tagged_interfaces
for ifc in interfaces:
ifc = str(ifc)
if ifc in ifcs:
pinterfaces.append(ifc)
else:
raise F5ModuleError(
'The specified interface "%s" was not found' % (ifc)
)
if tagged_interfaces:
tmp = [dict(name=x, tagged=True) for x in pinterfaces]
if 'tagged_interfaces' in current:
if pinterfaces != current['tagged_interfaces']:
params['interfaces'] = tmp
else:
params['interfaces'] = tmp
elif untagged_interfaces:
tmp = [dict(name=x, untagged=True) for x in pinterfaces]
if 'untagged_interfaces' in current:
if pinterfaces != current['untagged_interfaces']:
params['interfaces'] = tmp
else:
params['interfaces'] = tmp
if description is not None:
if 'description' in current:
if description != current['description']:
params['description'] = description
else:
params['description'] = description
if tag is not None:
if 'tag' in current:
if tag != current['tag']:
params['tag'] = tag
else:
params['tag'] = tag
if params: if params:
changed = True self.update(params=params)
params['name'] = name
params['partition'] = partition
if check_mode:
return changed
self.cparams = camel_dict_to_snake_dict(params)
else:
return changed
r = self.api.tm.net.vlans.vlan.load( def update(self, params=None):
name=name, if params:
partition=partition for k, v in iteritems(params):
) if self.api_map is not None and k in self.api_map:
r.update(**params) map_key = self.api_map[k]
r.refresh() else:
map_key = k
return True # 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 delete(self): updatables = [
params = dict() 'tagged_interfaces', 'untagged_interfaces', 'tag',
check_mode = self.params['check_mode'] 'description'
]
params['name'] = self.params['name'] returnables = [
params['partition'] = self.params['partition'] 'description', 'partition', 'name', 'tag', 'interfaces',
'tagged_interfaces', 'untagged_interfaces'
]
self.cparams = camel_dict_to_snake_dict(params) api_attributes = [
if check_mode: 'description', 'interfaces', 'partition', 'name', 'tag'
return True ]
api_map = {}
dc = self.api.tm.net.vlans.vlan.load(**params) @property
dc.delete() def interfaces(self):
tagged = self._values['tagged_interfaces']
untagged = self._values['untagged_interfaces']
if tagged:
return [dict(name=x, tagged=True) for x in tagged]
if untagged:
return [dict(name=x, untagged=True) for x in untagged]
if self.exists(): @property
raise F5ModuleError("Failed to delete the VLAN") def tagged_interfaces(self):
return True value = self._values['tagged_interfaces']
if value is None:
return None
ifcs = self._parse_return_ifcs()
for ifc in value:
if ifc not in ifcs:
err = 'The specified interface "%s" was not found' % ifc
raise F5ModuleError(err)
return value
def exists(self): @property
name = self.params['name'] def untagged_interfaces(self):
partition = self.params['partition'] value = self._values['untagged_interfaces']
return self.api.tm.net.vlans.vlan.exists( if value is None:
name=name, return None
partition=partition ifcs = self._parse_return_ifcs()
) for ifc in value:
if ifc not in ifcs:
err = 'The specified interface "%s" was not found' % ifc
raise F5ModuleError(err)
return value
def flush(self): def _get_interfaces_from_device(self):
lst = self.client.api.tm.net.interfaces.get_collection()
return lst
def _parse_return_ifcs(self):
ifclst = self._get_interfaces_from_device()
ifcs = [str(x.name) for x in ifclst]
if not ifcs:
err = 'No interfaces were found'
raise F5ModuleError(err)
return ifcs
def to_return(self):
result = {}
for returnable in self.returnables:
result[returnable] = getattr(self, returnable)
result = self._filter_params(result)
return result
def api_params(self):
result = {}
for api_attribute in self.api_attributes:
if 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 ModuleManager(object):
def __init__(self, client):
self.client = client
self.have = None
self.want = Parameters()
self.want.client = self.client
self.want.update(self.client.module.params)
self.changes = Parameters()
def exec_module(self):
changed = False
result = dict() result = dict()
state = self.params['state'] state = self.want.state
try: try:
if state == "present": if state == "present":
@ -396,42 +284,161 @@ class BigIpVlan(object):
except iControlUnexpectedHTTPError as e: except iControlUnexpectedHTTPError as e:
raise F5ModuleError(str(e)) raise F5ModuleError(str(e))
result.update(**self.cparams) changes = self.changes.to_return()
result.update(**changes)
result.update(dict(changed=changed)) result.update(dict(changed=changed))
return result return result
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):
changed = {}
for key in Parameters.updatables:
if getattr(self.want, key) is not None:
attr1 = getattr(self.want, key)
attr2 = getattr(self.have, key)
if attr1 != attr2:
changed[key] = attr1
if changed:
self.changes = Parameters(changed)
return True
return False
def _have_interfaces(self, ifcs):
untagged = [str(x.name) for x in ifcs if hasattr(x, 'untagged')]
tagged = [str(x.name) for x in ifcs if hasattr(x, 'tagged')]
if untagged:
self.have.update({'untagged_interfaces': untagged})
if tagged:
self.have.update({'tagged_interfaces': tagged})
def present(self):
if self.exists():
return self.update()
else:
return self.create()
def absent(self):
if self.exists():
return self.remove()
return False
def should_update(self):
result = self._update_changed_options()
if result:
return True
return False
def update(self):
self.have, ifcs = self.read_current_from_device()
if ifcs:
self._have_interfaces(ifcs)
if not self.should_update():
return False
if self.client.check_mode:
return True
self.update_on_device()
return True
def remove(self):
if self.client.check_mode:
return True
self.remove_from_device()
if self.exists():
raise F5ModuleError("Failed to delete the VLAN")
return True
def create(self):
self._set_changed_options()
if self.client.check_mode:
return True
self.create_on_device()
return True
def create_on_device(self):
params = self.want.api_params()
self.client.api.tm.net.vlans.vlan.create(**params)
def update_on_device(self):
params = self.want.api_params()
result = self.client.api.tm.net.vlans.vlan.load(
name=self.want.name, partition=self.want.partition
)
result.modify(**params)
def exists(self):
return self.client.api.tm.net.vlans.vlan.exists(
name=self.want.name, partition=self.want.partition
)
def remove_from_device(self):
result = self.client.api.tm.net.vlans.vlan.load(
name=self.want.name, partition=self.want.partition
)
if result:
result.delete()
def read_current_from_device(self):
tmp_res = self.client.api.tm.net.vlans.vlan.load(
name=self.want.name, partition=self.want.partition
)
ifcs = tmp_res.interfaces_s.get_collection()
result = tmp_res.attrs
return Parameters(result), ifcs
class ArgumentSpec(object):
def __init__(self):
self.supports_check_mode = True
self.argument_spec = dict(
name=dict(
required=True,
),
tagged_interfaces=dict(
type='list',
aliases=['tagged_interface']
),
untagged_interfaces=dict(
type='list',
aliases=['untagged_interface']
),
description=dict(),
tag=dict(
type='int'
)
)
self.f5_product_name = 'bigip'
def main(): def main():
argument_spec = f5_argument_spec() if not HAS_F5SDK:
raise F5ModuleError("The python f5-sdk module is required")
meta_args = dict( spec = ArgumentSpec()
description=dict(required=False, default=None),
tagged_interfaces=dict(required=False, default=None, type='list', aliases=['tagged_interface']),
untagged_interfaces=dict(required=False, default=None, type='list', aliases=['untagged_interface']),
name=dict(required=True),
tag=dict(required=False, default=None, type='int')
)
argument_spec.update(meta_args)
module = AnsibleModule( client = AnsibleF5Client(
argument_spec=argument_spec, argument_spec=spec.argument_spec,
supports_check_mode=True, supports_check_mode=spec.supports_check_mode,
f5_product_name=spec.f5_product_name,
mutually_exclusive=[ mutually_exclusive=[
['tagged_interfaces', 'untagged_interfaces'] ['tagged_interfaces', 'untagged_interfaces']
] ]
) )
try: try:
obj = BigIpVlan(check_mode=module.check_mode, **module.params) mm = ModuleManager(client)
result = obj.flush() results = mm.exec_module()
client.module.exit_json(**results)
module.exit_json(**result)
except F5ModuleError as e: except F5ModuleError as e:
module.fail_json(msg=str(e)) client.module.fail_json(msg=str(e))
from ansible.module_utils.basic import *
from ansible.module_utils.ec2 import camel_dict_to_snake_dict
from ansible.module_utils.f5_utils import *
if __name__ == '__main__': if __name__ == '__main__':
main() main()

View file

@ -0,0 +1,424 @@
[
{
"kind": "tm:net:interface:interfacestate",
"name": "1.1",
"fullPath": "1.1",
"generation": 7767,
"selfLink": "https://localhost/mgmt/tm/net/interface/1.1?ver=12.1.0",
"bundle": "not-supported",
"bundleSpeed": "not-supported",
"enabled": true,
"flowControl": "tx-rx",
"forceGigabitFiber": "disabled",
"forwardErrorCorrection": "not-supported",
"ifIndex": 176,
"lldpAdmin": "txonly",
"lldpTlvmap": 130943,
"macAddress": "00:23:e9:f1:e4:02",
"mediaActive": "none",
"mediaFixed": "auto",
"mediaMax": "1000T-FD",
"mediaSfp": "auto",
"mtu": 9198,
"preferPort": "sfp",
"qinqEthertype": "0x8100",
"sflow": {
"pollInterval": 0,
"pollIntervalGlobal": "yes"
},
"stp": "enabled",
"stpAutoEdgePort": "enabled",
"stpEdgePort": "true",
"stpLinkType": "auto"
},
{
"kind": "tm:net:interface:interfacestate",
"name": "1.2",
"fullPath": "1.2",
"generation": 7770,
"selfLink": "https://localhost/mgmt/tm/net/interface/1.2?ver=12.1.0",
"bundle": "not-supported",
"bundleSpeed": "not-supported",
"enabled": true,
"flowControl": "tx-rx",
"forceGigabitFiber": "disabled",
"forwardErrorCorrection": "not-supported",
"ifIndex": 192,
"lldpAdmin": "txonly",
"lldpTlvmap": 130943,
"macAddress": "00:23:e9:f1:e4:03",
"mediaActive": "none",
"mediaFixed": "auto",
"mediaMax": "1000T-FD",
"mediaSfp": "auto",
"mtu": 9198,
"preferPort": "sfp",
"qinqEthertype": "0x8100",
"sflow": {
"pollInterval": 0,
"pollIntervalGlobal": "yes"
},
"stp": "enabled",
"stpAutoEdgePort": "enabled",
"stpEdgePort": "true",
"stpLinkType": "auto"
},
{
"kind": "tm:net:interface:interfacestate",
"name": "1.3",
"fullPath": "1.3",
"generation": 7773,
"selfLink": "https://localhost/mgmt/tm/net/interface/1.3?ver=12.1.0",
"bundle": "not-supported",
"bundleSpeed": "not-supported",
"enabled": true,
"flowControl": "tx-rx",
"forceGigabitFiber": "disabled",
"forwardErrorCorrection": "not-supported",
"ifIndex": 208,
"lldpAdmin": "txonly",
"lldpTlvmap": 130943,
"macAddress": "00:23:e9:f1:e4:04",
"mediaActive": "none",
"mediaFixed": "auto",
"mediaMax": "1000T-FD",
"mediaSfp": "auto",
"mtu": 9198,
"preferPort": "sfp",
"qinqEthertype": "0x8100",
"sflow": {
"pollInterval": 0,
"pollIntervalGlobal": "yes"
},
"stp": "enabled",
"stpAutoEdgePort": "enabled",
"stpEdgePort": "true",
"stpLinkType": "auto"
},
{
"kind": "tm:net:interface:interfacestate",
"name": "1.4",
"fullPath": "1.4",
"generation": 7776,
"selfLink": "https://localhost/mgmt/tm/net/interface/1.4?ver=12.1.0",
"bundle": "not-supported",
"bundleSpeed": "not-supported",
"enabled": true,
"flowControl": "tx-rx",
"forceGigabitFiber": "disabled",
"forwardErrorCorrection": "not-supported",
"ifIndex": 224,
"lldpAdmin": "txonly",
"lldpTlvmap": 130943,
"macAddress": "00:23:e9:f1:e4:05",
"mediaActive": "none",
"mediaFixed": "auto",
"mediaMax": "1000T-FD",
"mediaSfp": "auto",
"mtu": 9198,
"preferPort": "sfp",
"qinqEthertype": "0x8100",
"sflow": {
"pollInterval": 0,
"pollIntervalGlobal": "yes"
},
"stp": "enabled",
"stpAutoEdgePort": "enabled",
"stpEdgePort": "true",
"stpLinkType": "auto"
},
{
"kind": "tm:net:interface:interfacestate",
"name": "2.1",
"fullPath": "2.1",
"generation": 7859,
"selfLink": "https://localhost/mgmt/tm/net/interface/2.1?ver=12.1.0",
"bundle": "not-supported",
"bundleSpeed": "not-supported",
"enabled": true,
"flowControl": "tx-rx",
"forceGigabitFiber": "disabled",
"forwardErrorCorrection": "not-supported",
"ifIndex": 240,
"lldpAdmin": "txonly",
"lldpTlvmap": 130943,
"macAddress": "00:23:e9:f1:e4:06",
"mediaActive": "10000SR-FD",
"mediaFixed": "auto",
"mediaMax": "10000T-FD",
"mediaSfp": "auto",
"moduleDescription": "F5 compatible optics",
"mtu": 9198,
"preferPort": "sfp",
"qinqEthertype": "0x8100",
"serial": "ARP2LGU",
"sflow": {
"pollInterval": 0,
"pollIntervalGlobal": "yes"
},
"stp": "enabled",
"stpAutoEdgePort": "enabled",
"stpEdgePort": "true",
"stpLinkType": "auto",
"vendor": "F5 NETWORKS INC.",
"vendorOui": "009065",
"vendorPartnum": "OPT-0016",
"vendorRevision": "A0"
},
{
"kind": "tm:net:interface:interfacestate",
"name": "2.2",
"fullPath": "2.2",
"generation": 7746,
"selfLink": "https://localhost/mgmt/tm/net/interface/2.2?ver=12.1.0",
"bundle": "not-supported",
"bundleSpeed": "not-supported",
"enabled": true,
"flowControl": "tx-rx",
"forceGigabitFiber": "disabled",
"forwardErrorCorrection": "not-supported",
"ifIndex": 256,
"lldpAdmin": "txonly",
"lldpTlvmap": 130943,
"macAddress": "00:23:e9:f1:e4:07",
"mediaActive": "none",
"mediaFixed": "auto",
"mediaMax": "10000T-FD",
"mediaSfp": "auto",
"mtu": 9198,
"preferPort": "sfp",
"qinqEthertype": "0x8100",
"sflow": {
"pollInterval": 0,
"pollIntervalGlobal": "yes"
},
"stp": "enabled",
"stpAutoEdgePort": "enabled",
"stpEdgePort": "true",
"stpLinkType": "auto"
},
{
"kind": "tm:net:interface:interfacestate",
"name": "2.3",
"fullPath": "2.3",
"generation": 7749,
"selfLink": "https://localhost/mgmt/tm/net/interface/2.3?ver=12.1.0",
"bundle": "not-supported",
"bundleSpeed": "not-supported",
"enabled": true,
"flowControl": "tx-rx",
"forceGigabitFiber": "disabled",
"forwardErrorCorrection": "not-supported",
"ifIndex": 272,
"lldpAdmin": "txonly",
"lldpTlvmap": 130943,
"macAddress": "00:23:e9:f1:e4:08",
"mediaActive": "none",
"mediaFixed": "auto",
"mediaMax": "10000T-FD",
"mediaSfp": "auto",
"mtu": 9198,
"preferPort": "sfp",
"qinqEthertype": "0x8100",
"sflow": {
"pollInterval": 0,
"pollIntervalGlobal": "yes"
},
"stp": "enabled",
"stpAutoEdgePort": "enabled",
"stpEdgePort": "true",
"stpLinkType": "auto"
},
{
"kind": "tm:net:interface:interfacestate",
"name": "2.4",
"fullPath": "2.4",
"generation": 7752,
"selfLink": "https://localhost/mgmt/tm/net/interface/2.4?ver=12.1.0",
"bundle": "not-supported",
"bundleSpeed": "not-supported",
"enabled": true,
"flowControl": "tx-rx",
"forceGigabitFiber": "disabled",
"forwardErrorCorrection": "not-supported",
"ifIndex": 288,
"lldpAdmin": "txonly",
"lldpTlvmap": 130943,
"macAddress": "00:23:e9:f1:e4:09",
"mediaActive": "none",
"mediaFixed": "auto",
"mediaMax": "10000T-FD",
"mediaSfp": "auto",
"mtu": 9198,
"preferPort": "sfp",
"qinqEthertype": "0x8100",
"sflow": {
"pollInterval": 0,
"pollIntervalGlobal": "yes"
},
"stp": "enabled",
"stpAutoEdgePort": "enabled",
"stpEdgePort": "true",
"stpLinkType": "auto"
},
{
"kind": "tm:net:interface:interfacestate",
"name": "2.5",
"fullPath": "2.5",
"generation": 7755,
"selfLink": "https://localhost/mgmt/tm/net/interface/2.5?ver=12.1.0",
"bundle": "not-supported",
"bundleSpeed": "not-supported",
"enabled": true,
"flowControl": "tx-rx",
"forceGigabitFiber": "disabled",
"forwardErrorCorrection": "not-supported",
"ifIndex": 304,
"lldpAdmin": "txonly",
"lldpTlvmap": 130943,
"macAddress": "00:23:e9:f1:e4:0a",
"mediaActive": "none",
"mediaFixed": "auto",
"mediaMax": "10000T-FD",
"mediaSfp": "auto",
"mtu": 9198,
"preferPort": "sfp",
"qinqEthertype": "0x8100",
"sflow": {
"pollInterval": 0,
"pollIntervalGlobal": "yes"
},
"stp": "enabled",
"stpAutoEdgePort": "enabled",
"stpEdgePort": "true",
"stpLinkType": "auto"
},
{
"kind": "tm:net:interface:interfacestate",
"name": "2.6",
"fullPath": "2.6",
"generation": 7758,
"selfLink": "https://localhost/mgmt/tm/net/interface/2.6?ver=12.1.0",
"bundle": "not-supported",
"bundleSpeed": "not-supported",
"enabled": true,
"flowControl": "tx-rx",
"forceGigabitFiber": "disabled",
"forwardErrorCorrection": "not-supported",
"ifIndex": 320,
"lldpAdmin": "txonly",
"lldpTlvmap": 130943,
"macAddress": "00:23:e9:f1:e4:0b",
"mediaActive": "none",
"mediaFixed": "auto",
"mediaMax": "10000T-FD",
"mediaSfp": "auto",
"mtu": 9198,
"preferPort": "sfp",
"qinqEthertype": "0x8100",
"sflow": {
"pollInterval": 0,
"pollIntervalGlobal": "yes"
},
"stp": "enabled",
"stpAutoEdgePort": "enabled",
"stpEdgePort": "true",
"stpLinkType": "auto"
},
{
"kind": "tm:net:interface:interfacestate",
"name": "2.7",
"fullPath": "2.7",
"generation": 7761,
"selfLink": "https://localhost/mgmt/tm/net/interface/2.7?ver=12.1.0",
"bundle": "not-supported",
"bundleSpeed": "not-supported",
"enabled": true,
"flowControl": "tx-rx",
"forceGigabitFiber": "disabled",
"forwardErrorCorrection": "not-supported",
"ifIndex": 336,
"lldpAdmin": "txonly",
"lldpTlvmap": 130943,
"macAddress": "00:23:e9:f1:e4:0c",
"mediaActive": "none",
"mediaFixed": "auto",
"mediaMax": "10000T-FD",
"mediaSfp": "auto",
"mtu": 9198,
"preferPort": "sfp",
"qinqEthertype": "0x8100",
"sflow": {
"pollInterval": 0,
"pollIntervalGlobal": "yes"
},
"stp": "enabled",
"stpAutoEdgePort": "enabled",
"stpEdgePort": "true",
"stpLinkType": "auto"
},
{
"kind": "tm:net:interface:interfacestate",
"name": "2.8",
"fullPath": "2.8",
"generation": 7764,
"selfLink": "https://localhost/mgmt/tm/net/interface/2.8?ver=12.1.0",
"bundle": "not-supported",
"bundleSpeed": "not-supported",
"enabled": true,
"flowControl": "tx-rx",
"forceGigabitFiber": "disabled",
"forwardErrorCorrection": "not-supported",
"ifIndex": 352,
"lldpAdmin": "txonly",
"lldpTlvmap": 130943,
"macAddress": "00:23:e9:f1:e4:0d",
"mediaActive": "none",
"mediaFixed": "auto",
"mediaMax": "10000T-FD",
"mediaSfp": "auto",
"mtu": 9198,
"preferPort": "sfp",
"qinqEthertype": "0x8100",
"sflow": {
"pollInterval": 0,
"pollIntervalGlobal": "yes"
},
"stp": "enabled",
"stpAutoEdgePort": "enabled",
"stpEdgePort": "true",
"stpLinkType": "auto"
},
{
"kind": "tm:net:interface:interfacestate",
"name": "mgmt",
"fullPath": "mgmt",
"generation": 7651,
"selfLink": "https://localhost/mgmt/tm/net/interface/mgmt?ver=12.1.0",
"bundle": "not-supported",
"bundleSpeed": "not-supported",
"enabled": true,
"flowControl": "tx-rx",
"forceGigabitFiber": "disabled",
"forwardErrorCorrection": "not-supported",
"ifIndex": 96,
"lldpAdmin": "txonly",
"lldpTlvmap": 130943,
"macAddress": "00:23:e9:f1:e4:01",
"mediaActive": "1000T-FD",
"mediaFixed": "auto",
"mediaMax": "1000T-FD",
"mediaSfp": "auto",
"mtu": 1500,
"preferPort": "sfp",
"qinqEthertype": "0x8100",
"sflow": {
"pollInterval": 0,
"pollIntervalGlobal": "yes"
},
"stp": "enabled",
"stpAutoEdgePort": "enabled",
"stpEdgePort": "true",
"stpLinkType": "auto"
}
]

View file

@ -0,0 +1,30 @@
{
"kind": "tm:net:vlan:vlanstate",
"name": "somevlan",
"partition": "Common",
"fullPath": "/Common/somevlan",
"generation": 1,
"selfLink": "https://localhost/mgmt/tm/net/vlan/~Common~somevlan?ver=12.1.0",
"autoLasthop": "default",
"cmpHash": "default",
"dagRoundRobin": "disabled",
"dagTunnel": "outer",
"failsafe": "disabled",
"failsafeAction": "failover-restart-tm",
"failsafeTimeout": 90,
"ifIndex": 480,
"learning": "enable-forward",
"mtu": 1500,
"sflow": {
"pollInterval": 0,
"pollIntervalGlobal": "yes",
"samplingRate": 0,
"samplingRateGlobal": "yes"
},
"sourceChecking": "disabled",
"tag": 4094,
"interfacesReference": {
"link": "https://localhost/mgmt/tm/net/vlan/~Common~somevlan/interfaces?ver=12.1.0",
"isSubcollection": true
}
}

View file

@ -0,0 +1,12 @@
[
{
"kind": "tm:net:vlan:interfaces:interfacesstate",
"name": "1.2",
"fullPath": "1.2",
"generation": 2,
"selfLink": "https://localhost/mgmt/tm/net/vlan/~Common~internal/interfaces/1.2?ver=12.1.0",
"tagMode": "none",
"tagged": true
}
]

View file

@ -0,0 +1,11 @@
[
{
"kind": "tm:net:vlan:interfaces:interfacesstate",
"name": "1.1",
"fullPath": "1.1",
"generation": 1,
"selfLink": "https://localhost/mgmt/tm/net/vlan/~Common~internal/interfaces/1.1?ver=12.1.0",
"tagMode": "none",
"untagged": true
}
]

View file

@ -0,0 +1,31 @@
{
"kind": "tm:net:vlan:vlanstate",
"name": "somevlan",
"partition": "Common",
"fullPath": "/Common/somevlan",
"description": "changed_this",
"generation": 1,
"selfLink": "https://localhost/mgmt/tm/net/vlan/~Common~somevlan?ver=12.1.0",
"autoLasthop": "default",
"cmpHash": "default",
"dagRoundRobin": "disabled",
"dagTunnel": "outer",
"failsafe": "disabled",
"failsafeAction": "failover-restart-tm",
"failsafeTimeout": 90,
"ifIndex": 480,
"learning": "enable-forward",
"mtu": 1500,
"sflow": {
"pollInterval": 0,
"pollIntervalGlobal": "yes",
"samplingRate": 0,
"samplingRateGlobal": "yes"
},
"sourceChecking": "disabled",
"tag": 4094,
"interfacesReference": {
"link": "https://localhost/mgmt/tm/net/vlan/~Common~somevlan/interfaces?ver=12.1.0",
"isSubcollection": true
}
}

View file

@ -0,0 +1,481 @@
# -*- 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 Mock
from ansible.compat.tests.mock import patch
from ansible.module_utils.f5_utils import AnsibleF5Client
from ansible.module_utils.f5_utils import F5ModuleError
try:
from library.bigip_vlan import Parameters
from library.bigip_vlan import ModuleManager
from library.bigip_vlan 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_vlan import Parameters
from ansible.modules.network.f5.bigip_vlan import ModuleManager
from ansible.modules.network.f5.bigip_vlan 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")
fixture_path = os.path.join(os.path.dirname(__file__), 'fixtures')
fixture_data = {}
def load_fixture(name):
path = os.path.join(fixture_path, name)
if path in fixture_data:
return fixture_data[path]
with open(path) as f:
data = f.read()
try:
data = json.loads(data)
except Exception:
pass
fixture_data[path] = data
return data
class BigIpObj(object):
def __init__(self, **kwargs):
self.__dict__.update(kwargs)
class TestParameters(unittest.TestCase):
def setUp(self):
self.loaded_ifcs = []
ifcs_json = load_fixture('load_net_interfaces.json')
for item in ifcs_json:
self.loaded_ifcs.append(BigIpObj(**item))
def test_module_parameters(self):
args = dict(
name='somevlan',
tag=213,
description='fakevlan',
untagged_interfaces=['1.1'],
)
with patch.object(Parameters, '_get_interfaces_from_device') as obj:
obj.return_value = self.loaded_ifcs
p = Parameters(args)
assert p.name == 'somevlan'
assert p.tag == 213
assert p.description == 'fakevlan'
assert p.untagged_interfaces == ['1.1']
def test_api_parameters(self):
args = dict(
name='somevlan',
description='fakevlan',
tag=213,
tagged_interfaces=['1.2']
)
with patch.object(Parameters, '_get_interfaces_from_device') as obj:
obj.return_value = self.loaded_ifcs
p = Parameters(args)
assert p.name == 'somevlan'
assert p.tag == 213
assert p.interfaces == [{'tagged': True, 'name': '1.2'}]
assert p.description == 'fakevlan'
@patch('ansible.module_utils.f5_utils.AnsibleF5Client._get_mgmt_root',
return_value=True)
class TestManager(unittest.TestCase):
def setUp(self):
self.spec = ArgumentSpec()
self.loaded_ifcs = []
self.loaded_vlan_ifc_tag = []
self.loaded_vlan_ifc_untag = []
ifcs_tag = load_fixture('load_vlan_tagged_ifcs.json')
ifcs_untag = load_fixture('load_vlan_untag_ifcs.json')
ifcs_json = load_fixture('load_net_interfaces.json')
for item in ifcs_json:
self.loaded_ifcs.append(BigIpObj(**item))
for item in ifcs_tag:
self.loaded_vlan_ifc_tag.append(BigIpObj(**item))
for item in ifcs_untag:
self.loaded_vlan_ifc_untag.append(BigIpObj(**item))
def test_create_vlan(self, *args):
set_module_args(dict(
name='somevlan',
description='fakevlan',
server='localhost',
password='password',
user='admin',
partition='Common'
))
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 to force specific logic in the module to happen
with patch.object(Parameters, '_get_interfaces_from_device') as obj:
obj.return_value = self.loaded_ifcs
mm = ModuleManager(client)
mm.create_on_device = Mock(return_value=True)
mm.exists = Mock(return_value=False)
results = mm.exec_module()
assert results['changed'] is True
assert results['name'] == 'somevlan'
assert results['description'] == 'fakevlan'
def test_create_vlan_tagged_interface(self, *args):
set_module_args(dict(
name='somevlan',
tagged_interface=['2.1'],
tag=213,
server='localhost',
password='password',
user='admin',
partition='Common'
))
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 to force specific logic in the module to happen
with patch.object(Parameters, '_get_interfaces_from_device') as obj:
obj.return_value = self.loaded_ifcs
mm = ModuleManager(client)
mm.create_on_device = Mock(return_value=True)
mm.exists = Mock(return_value=False)
results = mm.exec_module()
assert results['changed'] is True
assert results['interfaces'] == [{'tagged': True, 'name': '2.1'}]
assert results['tag'] == 213
assert results['name'] == 'somevlan'
def test_create_vlan_untagged_interface(self, *args):
set_module_args(dict(
name='somevlan',
untagged_interface=['2.1'],
server='localhost',
password='password',
user='admin',
partition='Common'
))
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 to force specific logic in the module to happen
with patch.object(Parameters, '_get_interfaces_from_device') as obj:
obj.return_value = self.loaded_ifcs
mm = ModuleManager(client)
mm.create_on_device = Mock(return_value=True)
mm.exists = Mock(return_value=False)
results = mm.exec_module()
assert results['changed'] is True
assert results['interfaces'] == [{'untagged': True, 'name': '2.1'}]
assert results['name'] == 'somevlan'
def test_create_vlan_tagged_interfaces(self, *args):
set_module_args(dict(
name='somevlan',
tagged_interface=['2.1', '1.1'],
tag=213,
server='localhost',
password='password',
user='admin',
partition='Common'
))
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 to force specific logic in the module to happen
with patch.object(Parameters, '_get_interfaces_from_device') as obj:
obj.return_value = self.loaded_ifcs
mm = ModuleManager(client)
mm.create_on_device = Mock(return_value=True)
mm.exists = Mock(return_value=False)
results = mm.exec_module()
assert results['changed'] is True
assert results['interfaces'] == [{'tagged': True, 'name': '2.1'},
{'tagged': True, 'name': '1.1'}]
assert results['tag'] == 213
assert results['name'] == 'somevlan'
def test_create_vlan_untagged_interfaces(self, *args):
set_module_args(dict(
name='somevlan',
untagged_interface=['2.1', '1.1'],
server='localhost',
password='password',
user='admin',
partition='Common',
))
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 to force specific logic in the module to happen
with patch.object(Parameters, '_get_interfaces_from_device') as obj:
obj.return_value = self.loaded_ifcs
mm = ModuleManager(client)
mm.create_on_device = Mock(return_value=True)
mm.exists = Mock(return_value=False)
results = mm.exec_module()
assert results['changed'] is True
assert results['interfaces'] == [{'untagged': True, 'name': '2.1'},
{'untagged': True, 'name': '1.1'}]
assert results['name'] == 'somevlan'
def test_update_vlan_untag_interface(self, *args):
set_module_args(dict(
name='somevlan',
untagged_interface=['2.1'],
server='localhost',
password='password',
user='admin',
partition='Common',
))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
ifcs = self.loaded_vlan_ifc_untag
# Override methods to force specific logic in the module to happen
with patch.object(Parameters, '_get_interfaces_from_device') as obj:
obj.return_value = self.loaded_ifcs
mm = ModuleManager(client)
current = (
Parameters(
load_fixture('load_vlan.json')
),
ifcs
)
mm.update_on_device = Mock(return_value=True)
mm.exists = Mock(return_value=True)
mm.read_current_from_device = Mock(return_value=current)
results = mm.exec_module()
assert results['changed'] is True
assert results['interfaces'] == [{'untagged': True, 'name': '2.1'}]
def test_update_vlan_tag_interface(self, *args):
set_module_args(dict(
name='somevlan',
tagged_interface=['2.1'],
server='localhost',
password='password',
user='admin',
partition='Common',
))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
ifcs = self.loaded_vlan_ifc_tag
# Override methods to force specific logic in the module to happen
with patch.object(Parameters, '_get_interfaces_from_device') as obj:
obj.return_value = self.loaded_ifcs
mm = ModuleManager(client)
current = (
Parameters(
load_fixture('load_vlan.json')
),
ifcs
)
mm.update_on_device = Mock(return_value=True)
mm.exists = Mock(return_value=True)
mm.read_current_from_device = Mock(return_value=current)
results = mm.exec_module()
assert results['changed'] is True
assert results['interfaces'] == [{'tagged': True, 'name': '2.1'}]
def test_update_vlan_description(self, *args):
set_module_args(dict(
name='somevlan',
description='changed_that',
server='localhost',
password='password',
user='admin',
partition='Common',
))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
ifcs = self.loaded_vlan_ifc_tag
# Override methods to force specific logic in the module to happen
with patch.object(Parameters, '_get_interfaces_from_device') as obj:
obj.return_value = self.loaded_ifcs
mm = ModuleManager(client)
current = (
Parameters(
load_fixture('update_vlan_description.json')
),
ifcs
)
mm.update_on_device = Mock(return_value=True)
mm.exists = Mock(return_value=True)
mm.read_current_from_device = Mock(return_value=current)
results = mm.exec_module()
assert results['changed'] is True
assert results['description'] == 'changed_that'
def test_untagged_ifc_raises(self, *args):
set_module_args(dict(
name='somevlan',
untagged_interface=['10.2'],
server='localhost',
password='password',
user='admin',
partition='Common'
))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
msg = 'The specified interface "10.2" was not found'
# Override methods to force specific logic in the module to happen
with patch.object(Parameters, '_get_interfaces_from_device') as obj:
obj.return_value = self.loaded_ifcs
mm = ModuleManager(client)
mm.create_on_device = Mock(return_value=True)
mm.exists = Mock(return_value=False)
with pytest.raises(F5ModuleError) as err:
mm.exec_module()
assert str(err.value) == msg
def test_tagged_ifc_raises(self, *args):
set_module_args(dict(
name='somevlan',
tagged_interface=['10.2'],
tag=213,
server='localhost',
password='password',
user='admin',
partition='Common'
))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
msg = 'The specified interface "10.2" was not found'
# Override methods to force specific logic in the module to happen
with patch.object(Parameters, '_get_interfaces_from_device') as obj:
obj.return_value = self.loaded_ifcs
mm = ModuleManager(client)
mm.create_on_device = Mock(return_value=True)
mm.exists = Mock(return_value=False)
with pytest.raises(F5ModuleError) as err:
mm.exec_module()
assert str(err.value) == msg
def test_parse_return_ifcs_raises(self, *args):
set_module_args(dict(
name='somevlan',
untagged_interface=['1.2'],
server='localhost',
password='password',
user='admin',
partition='Common'
))
client = AnsibleF5Client(
argument_spec=self.spec.argument_spec,
supports_check_mode=self.spec.supports_check_mode,
f5_product_name=self.spec.f5_product_name
)
msg = 'No interfaces were found'
# Override methods to force specific logic in the module to happen
with patch.object(Parameters, '_get_interfaces_from_device') as obj:
obj.return_value = []
mm = ModuleManager(client)
mm.create_on_device = Mock(return_value=True)
mm.exists = Mock(return_value=False)
with pytest.raises(F5ModuleError) as err:
mm.exec_module()
assert str(err.value) == msg