add a new module to configure mdn on interface. (#63755)

* add a new module to confiure mdn on interface.

* doc-default-does-not-match-spec.

* add unittest

* Update test_ce_mdn_interface.py

* add integration.

* update for review.

* add notes.
pull/43435/head
Xu Yuandong 5 years ago committed by John R Barker
parent f543e72d0a
commit 9f58e915d9

@ -0,0 +1,403 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# Copyright 2019 Red Hat
# 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 = '''
---
module: ce_mdn_interface
version_added: "2.10"
short_description: Manages MDN configuration on HUAWEI CloudEngine switches.
description:
- Manages MDN configuration on HUAWEI CloudEngine switches.
author: xuxiaowei0512 (@CloudEngine-Ansible)
options:
lldpenable:
description:
- Set global LLDP enable state.
type: str
choices: ['enabled', 'disabled']
mdnstatus:
description:
- Set interface MDN enable state.
type: str
choices: ['rxOnly', 'disabled']
ifname:
description:
- Interface name.
type: str
state:
description:
- Manage the state of the resource.
default: present
type: str
choices: ['present','absent']
notes:
- This module requires the netconf system service be enabled on
the remote device being managed.
- This module works with connection C(netconf).
'''
EXAMPLES = '''
- name: "Configure global LLDP enable state"
ce_mdn_interface:
lldpenable: enabled
- name: "Configure interface MDN enable state"
ce_mdn_interface:
ifname: 10GE1/0/1
mdnstatus: rxOnly
'''
RETURN = '''
proposed:
description: k/v pairs of parameters passed into module
returned: always
type: dict
sample: {
"lldpenable": "enabled",
"ifname": "10GE1/0/1",
"mdnstatus": "rxOnly",
"state":"present"
}
existing:
description: k/v pairs of existing global LLDP configration
returned: always
type: dict
sample: {
"lldpenable": "enabled",
"ifname": "10GE1/0/1",
"mdnstatus": "disabled"
}
end_state:
description: k/v pairs of global LLDP configration after module execution
returned: always
type: dict
sample: {
"lldpenable": "enabled",
"ifname": "10GE1/0/1",
"mdnstatus": "rxOnly"
}
updates:
description: command sent to the device
returned: always
type: list
sample: [
"interface 10ge 1/0/1",
"lldp mdn enable",
]
changed:
description: check to see if a change was made on the device
returned: always
type: bool
sample: true
'''
import copy
import re
from xml.etree import ElementTree
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.network.cloudengine.ce import set_nc_config, get_nc_config, execute_nc_action
CE_NC_GET_GLOBAL_LLDPENABLE_CONFIG = """
<filter type="subtree">
<lldp xmlns="http://www.huawei.com/netconf/vrp" content-version="1.0" format-version="1.0">
<lldpSys>
<lldpEnable></lldpEnable>
</lldpSys>
</lldp>
</filter>
"""
CE_NC_MERGE_GLOBA_LLDPENABLE_CONFIG = """
<config>
<lldp xmlns="http://www.huawei.com/netconf/vrp" content-version="1.0" format-version="1.0">
<lldpSys operation="merge">
<lldpEnable>%s</lldpEnable>
</lldpSys>
</lldp>
</config>
"""
CE_NC_GET_INTERFACE_MDNENABLE_CONFIG = """
<filter type="subtree">
<lldp xmlns="http://www.huawei.com/netconf/vrp" content-version="1.0" format-version="1.0">
<mdnInterfaces>
<mdnInterface>
<ifName></ifName>
<mdnStatus></mdnStatus>
</mdnInterface>
</mdnInterfaces>
</lldp>
</filter>
"""
CE_NC_MERGE_INTERFACE_MDNENABLE_CONFIG = """
<config>
<lldp xmlns="http://www.huawei.com/netconf/vrp" content-version="1.0" format-version="1.0">
<mdnInterfaces>
<mdnInterface operation="merge">
<ifName>%s</ifName>
<mdnStatus>%s</mdnStatus>
</mdnInterface>
</mdnInterfaces>
</lldp>
</config>
"""
def get_interface_type(interface):
"""Gets the type of interface, such as 10GE, ..."""
if interface is None:
return None
iftype = None
if interface.upper().startswith('GE'):
iftype = 'ge'
elif interface.upper().startswith('10GE'):
iftype = '10ge'
elif interface.upper().startswith('25GE'):
iftype = '25ge'
elif interface.upper().startswith('40GE'):
iftype = '40ge'
elif interface.upper().startswith('100GE'):
iftype = '100ge'
elif interface.upper().startswith('PORT-GROUP'):
iftype = 'stack-Port'
elif interface.upper().startswith('NULL'):
iftype = 'null'
else:
return None
return iftype.lower()
class Interface_mdn(object):
"""Manage global lldp enable configration"""
def __init__(self, argument_spec):
self.spec = argument_spec
self.module = None
self.init_module()
# LLDP global configration info
self.lldpenable = self.module.params['lldpenable'] or None
self.ifname = self.module.params['ifname']
self.mdnstatus = self.module.params['mdnstatus'] or None
self.state = self.module.params['state']
self.lldp_conf = dict()
self.conf_exsit = False
self.enable_flag = 0
self.check_params()
# state
self.changed = False
self.proposed_changed = dict()
self.updates_cmd = list()
self.results = dict()
self.proposed = dict()
self.existing = dict()
self.end_state = dict()
def check_params(self):
"""Check all input params"""
if self.ifname:
intf_type = get_interface_type(self.ifname)
if not intf_type:
self.module.fail_json(
msg='Error: ifname name of %s '
'is error.' % self.ifname)
if (len(self.ifname) < 1) or (len(self.ifname) > 63):
self.module.fail_json(
msg='Error: Ifname length is beetween 1 and 63.')
def init_module(self):
"""Init module object"""
self.module = AnsibleModule(
argument_spec=self.spec, supports_check_mode=True)
def check_response(self, xml_str, xml_name):
"""Check if response message is already succeed"""
if "<ok/>" not in xml_str:
self.module.fail_json(msg='Error: %s failed.' % xml_name)
def config_interface_mdn(self):
"""Configure lldp enabled and interface mdn enabled parameters"""
if self.state == 'present':
if self.enable_flag == 0 and self.lldpenable == 'enabled':
xml_str = CE_NC_MERGE_GLOBA_LLDPENABLE_CONFIG % self.lldpenable
ret_xml = set_nc_config(self.module, xml_str)
self.check_response(ret_xml, "LLDP_ENABLE_CONFIG")
self.changed = True
elif self.enable_flag == 1 and self.lldpenable == 'disabled':
xml_str = CE_NC_MERGE_GLOBA_LLDPENABLE_CONFIG % self.lldpenable
ret_xml = set_nc_config(self.module, xml_str)
self.check_response(ret_xml, "LLDP_ENABLE_CONFIG")
self.changed = True
elif self.enable_flag == 1 and self.conf_exsit:
xml_str = CE_NC_MERGE_INTERFACE_MDNENABLE_CONFIG % (self.ifname, self.mdnstatus)
ret_xml = set_nc_config(self.module, xml_str)
self.check_response(ret_xml, "INTERFACE_MDN_ENABLE_CONFIG")
self.changed = True
def show_result(self):
"""Show result"""
self.results['changed'] = self.changed
self.results['proposed'] = self.proposed
self.results['existing'] = self.existing
self.results['end_state'] = self.end_state
if self.changed:
self.results['updates'] = self.updates_cmd
else:
self.results['updates'] = list()
self.module.exit_json(**self.results)
def get_interface_mdn_exist_config(self):
"""Get lldp existed configure"""
lldp_config = list()
lldp_dict = dict()
conf_enable_str = CE_NC_GET_GLOBAL_LLDPENABLE_CONFIG
conf_enable_obj = get_nc_config(self.module, conf_enable_str)
xml_enable_str = conf_enable_obj.replace('\r', '').replace('\n', '').\
replace('xmlns="urn:ietf:params:xml:ns:netconf:base:1.0"', "").\
replace('xmlns="http://www.huawei.com/netconf/vrp"', "")
# get lldp enable config info
root_enable = ElementTree.fromstring(xml_enable_str)
ntpsite_enable = root_enable.findall("lldp/lldpSys")
for nexthop_enable in ntpsite_enable:
for ele_enable in nexthop_enable:
if ele_enable.tag in ["lldpEnable"]:
lldp_dict[ele_enable.tag] = ele_enable.text
if self.state == "present":
if lldp_dict['lldpEnable'] == 'enabled':
self.enable_flag = 1
lldp_config.append(dict(lldpenable=lldp_dict['lldpEnable']))
if self.enable_flag == 1:
conf_str = CE_NC_GET_INTERFACE_MDNENABLE_CONFIG
conf_obj = get_nc_config(self.module, conf_str)
if "<data/>" in conf_obj:
return lldp_config
xml_str = conf_obj.replace('\r', '').replace('\n', '').\
replace('xmlns="urn:ietf:params:xml:ns:netconf:base:1.0"', "").\
replace('xmlns="http://www.huawei.com/netconf/vrp"', "")
# get all ntp config info
root = ElementTree.fromstring(xml_str)
ntpsite = root.findall("lldp/mdnInterfaces/mdnInterface")
for nexthop in ntpsite:
for ele in nexthop:
if ele.tag in ["ifName", "mdnStatus"]:
lldp_dict[ele.tag] = ele.text
if self.state == "present":
cur_interface_mdn_cfg = dict(ifname=lldp_dict['ifName'], mdnstatus=lldp_dict['mdnStatus'])
exp_interface_mdn_cfg = dict(ifname=self.ifname, mdnstatus=self.mdnstatus)
if self.ifname == lldp_dict['ifName']:
if cur_interface_mdn_cfg != exp_interface_mdn_cfg:
self.conf_exsit = True
lldp_config.append(dict(ifname=lldp_dict['ifName'], mdnstatus=lldp_dict['mdnStatus']))
return lldp_config
lldp_config.append(dict(ifname=lldp_dict['ifName'], mdnstatus=lldp_dict['mdnStatus']))
return lldp_config
def get_existing(self):
"""Get existing info"""
self.existing = self.get_interface_mdn_exist_config()
def get_proposed(self):
"""Get proposed info"""
if self.lldpenable:
self.proposed = dict(lldpenable=self.lldpenable)
if self.enable_flag == 1:
if self.ifname:
self.proposed = dict(ifname=self.ifname, mdnstatus=self.mdnstatus)
def get_end_state(self):
"""Get end state info"""
self.end_state = self.get_interface_mdn_exist_config()
def get_update_cmd(self):
"""Get updated commands"""
update_list = list()
if self.state == "present":
if self.lldpenable == "enabled":
cli_str = "lldp enable"
update_list.append(cli_str)
if self.ifname:
cli_str = "%s %s" % ("interface", self.ifname)
update_list.append(cli_str)
if self.mdnstatus:
if self.mdnstatus == "rxOnly":
cli_str = "lldp mdn enable"
update_list.append(cli_str)
else:
cli_str = "undo lldp mdn enable"
update_list.append(cli_str)
elif self.lldpenable == "disabled":
cli_str = "undo lldp enable"
update_list.append(cli_str)
else:
if self.enable_flag == 1:
if self.ifname:
cli_str = "%s %s" % ("interface", self.ifname)
update_list.append(cli_str)
if self.mdnstatus:
if self.mdnstatus == "rxOnly":
cli_str = "lldp mdn enable"
update_list.append(cli_str)
else:
cli_str = "undo lldp mdn enable"
update_list.append(cli_str)
self.updates_cmd.append(update_list)
def work(self):
"""Excute task"""
self.check_params()
self.get_existing()
self.get_proposed()
self.config_interface_mdn()
self.get_update_cmd()
self.get_end_state()
self.show_result()
def main():
"""Main function entry"""
argument_spec = dict(
lldpenable=dict(type='str', choices=['enabled', 'disabled']),
mdnstatus=dict(type='str', choices=['rxOnly', 'disabled']),
ifname=dict(type='str'),
state=dict(choices=['absent', 'present'], default='present'),
)
lldp_obj = Interface_mdn(argument_spec)
lldp_obj.work()
if __name__ == '__main__':
main()

@ -0,0 +1,3 @@
---
testcase: "[^_].*"
test_items: []

@ -0,0 +1,2 @@
---
- { include: netconf.yaml, tags: ['netconf'] }

@ -0,0 +1,17 @@
---
- name: collect all netconf test cases
find:
paths: "{{ role_path }}/tests/netconf"
patterns: "{{ testcase }}.yaml"
use_regex: true
connection: local
register: test_cases
- name: set test_items
set_fact: test_items="{{ test_cases.files | map(attribute='path') | list }}"
- name: run test case (connection=netconf)
include: "{{ test_case_to_run }} ansible_connection=netconf"
with_items: "{{ test_items }}"
loop_control:
loop_var: test_case_to_run

@ -0,0 +1,97 @@
---
- debug:
msg: "START ce_mdn_interface presented integration tests on connection={{ ansible_connection }}"
# set up default before test
- name: clean up default configuration with the exisiting running configuration
ce_mdn_interface:
lldpenable: disabled
mdnstatus: disabled
ifname: 10GE1/0/1
- name: present the provided configuration with the exisiting running configuration
ce_mdn_interface: &present
lldpenable: enabled
mdnstatus: rxOnly
ifname: 10GE1/0/1
register: result
- name: Assert the configuration is reflected on host
assert:
that:
- "result['changed'] == true"
- name: Get mdnInterface config by ce_netconf.
ce_netconf: &get_config
rpc: get
cfg_xml: "<filter type=\"subtree\">
<lldp xmlns=\"http://www.huawei.com/netconf/vrp\" content-version=\"1.0\" format-version=\"1.0\">
<mdnInterfaces>
<mdnInterface>
<ifName>10GE1/0/1</ifName>
<mdnStatus></mdnStatus>
</mdnInterface>
</mdnInterfaces>
</lldp>
</filter>"
register: result_xml
- name: Get lldp enabled config by ce_netconf.
ce_netconf: &get_config_lldp
rpc: get
cfg_xml: "<filter type=\"subtree\">
<lldp xmlns=\"http://www.huawei.com/netconf/vrp\" content-version=\"1.0\" format-version=\"1.0\">
<lldpSys>
<lldpEnable></lldpEnable>
</lldpSys>
</lldp>
/filter>"
register: result_xml_lldp
- name: present the provided configuration with the existing running configuration (IDEMPOTENT)
ce_mdn_interface: *present
register: repeat
- name: Assert that the previous task was idempotent
assert:
that:
- "repeat.changed == false"
- "'<mdnStatus>rxOnly</mdnStatus>' in result_xml.end_state.result"
- "'<lldpEnable>enabled</lldpEnable>' in result_xml_lldp.end_state.result"
- name: absent the provided configuration with the exisiting running configuration
ce_mdn_interface: &absent
lldpenable: disabled
mdnstatus: disabled
ifname: 10GE1/0/1
state: absent
register: result
- name: Assert the configuration is reflected on host
assert:
that:
- "result['changed'] == true"
- name: absent the provided configuration with the existing running configuration (REPEAT)
ce_mdn_interface: *absent
register: repeat
- name: Get mdnInterface config by ce_netconf.
ce_netconf: *get_config
register: result_xml
- name: Get lldp enabled config by ce_netconf.
ce_netconf: *get_config
register: result_xml_lldp
- name: Assert that the previous task was idempotent
assert:
that:
- "result['changed'] == false"
- "'<mdnStatus>disabled</mdnStatus>' not in result_xml.end_state.result"
- "'<lldpEnable>disabled</lldpEnable>' in result_xml_lldp.end_state.result"
# after present, isis 100 should be deleted
- debug:
msg: "END ce_mdn_interface resentd integration tests on connection={{ ansible_connection }}"

@ -0,0 +1,14 @@
<data>
<lldp>
<lldpSys>
<lldpEnable>enabled</lldpEnable>
<mdnStatus>enabled</mdnStatus>
</lldpSys>
<mdnInterfaces>
<mdnInterface>
<ifName>10GE1/0/1</ifName>
<mdnStatus>rxOnly</mdnStatus>
</mdnInterface>
</mdnInterfaces>
</lldp>
</data>

@ -0,0 +1,14 @@
<data>
<lldp>
<lldpSys>
<lldpEnable>disabled</lldpEnable>
<mdnStatus>disabled</mdnStatus>
</lldpSys>
<mdnInterfaces>
<mdnInterface>
<ifName>10GE1/0/1</ifName>
<mdnStatus>disabled</mdnStatus>
</mdnInterface>
</mdnInterfaces>
</lldp>
</data>

@ -0,0 +1,67 @@
# (c) 2019 Red Hat Inc.
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
# Make coding more python3-ish
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from units.compat.mock import patch
from ansible.modules.network.cloudengine import ce_mdn_interface
from units.modules.network.cloudengine.ce_module import TestCloudEngineModule, load_fixture
from units.modules.utils import set_module_args
class TestCloudEngineLacpModule(TestCloudEngineModule):
module = ce_mdn_interface
def setUp(self):
super(TestCloudEngineLacpModule, self).setUp()
self.mock_get_config = patch('ansible.modules.network.cloudengine.ce_mdn_interface.get_nc_config')
self.get_nc_config = self.mock_get_config.start()
self.mock_set_config = patch('ansible.modules.network.cloudengine.ce_mdn_interface.set_nc_config')
self.set_nc_config = self.mock_set_config.start()
self.set_nc_config.return_value = "<ok/>"
self.before = load_fixture('ce_mdn_interface', 'before.txt')
self.after = load_fixture('ce_mdn_interface', 'after.txt')
def tearDown(self):
super(TestCloudEngineLacpModule, self).tearDown()
self.mock_set_config.stop()
self.mock_get_config.stop()
def test_mdn_enable(self):
update = [['lldp enable', 'interface 10GE1/0/1', 'lldp mdn enable']]
self.get_nc_config.side_effect = (self.before, self.before, self.after, self.after)
set_module_args(dict(
lldpenable='enabled',
mdnstatus='rxOnly',
ifname='10GE1/0/1')
)
result = self.execute_module(changed=True)
self.assertEquals(sorted(result['updates']), sorted(update))
def test_repeat_enable(self):
self.get_nc_config.side_effect = (self.after, self.after, self.after, self.after, )
set_module_args(dict(
lldpenable='enabled',
mdnstatus='rxOnly',
ifname='10GE1/0/1')
)
self.execute_module(changed=False)
Loading…
Cancel
Save