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

Add module cv_server_provision for integration with Arista CloudVision Portal. (#25450)

Browse source 
* Add module cv_server_provision for integration with Arista CloudVision Portal.

* Doc update.

* Remove shebang from test file. Update short description with company and product name.

* Update exception syntax to Python3 style.

* Remove blank line between imports.

* Remove newlines from RETURN documentation.

* Add cvprac to unittest requirements.

* Update unittest format. Add a few additional tests.

* Mock exceptions from cvprac so the library is not needed for unittests.

* Mock cvprac imports.

* Update unit tests to support python 3.5.

* Mock full cvprac library for unittests.

* Update Jinja2 import to pass updated CI checks.

* Update cvprac imports format for new CI tests.

* Add __metaclass__ and __future__.
This commit is contained in:
mharista 2017-08-02 10:24:52 -04:00 committed by John R Barker
parent 7fbd924cbb
commit c85f363aaa
4 changed files with 1532 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

0
lib/ansible/modules/network/cloudvision/__init__.py Normal file
View file

646
lib/ansible/modules/network/cloudvision/cv_server_provision.py Normal file
View file

@ -0,0 +1,646 @@
#!/usr/bin/python
# 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/>.
#
from __future__ import absolute_import, division, print_function
__metaclass__ = type
ANSIBLE_METADATA = {'metadata_version': '1.0',
'status': ['preview'],
'supported_by': 'community'}
DOCUMENTATION = '''
---
module: cv_server_provision
version_added: "2.4"
author: "EOS+ CS (ansible-dev@arista.com) (@mharista)"
short_description:
Provision server port by applying or removing template configuration to an
Arista CloudVision Portal configlet that is applied to a switch.
description:
- This module allows a server team to provision server network ports for
new servers without having to access Arista CVP or asking the network team
to do it for them. Provide the information for connecting to CVP, switch
rack, port the new server is connected to, optional vlan, and an action
and the module will apply the configuration to the switch port via CVP.
Actions are add (applies template config to port),
remove (defaults the interface config) and
show (returns the current port config).
options:
host:
description:
- The hostname or IP address of the CVP node being connected to.
required: true
port:
description:
- The port number to use when making API calls to the CVP node. This
will default to the default port for the specified protocol. Port 80
for http and port 443 for https.
default: None
protocol:
description:
- The protocol to use when making API calls to CVP. CVP defaults to https
and newer versions of CVP no longer support http.
default: https
choices: [https, http]
username:
description:
- The user that will be used to connect to CVP for making API calls.
required: true
password:
description:
- The password of the user that will be used to connect to CVP for API
calls.
required: true
server_name:
description:
- The hostname or identifier for the server that is having it's switch
port provisioned.
required: true
switch_name:
description:
- The hostname of the switch is being configured for the server being
provisioned.
required: true
switch_port:
description:
- The physical port number on the switch that the new server is
connected to.
required: true
port_vlan:
description:
- The vlan that should be applied to the port for this server.
This parameter is dependent on a proper template that supports single
vlan provisioning with it. If a port vlan is specified by the template
specified does not support this the module will exit out with no
changes. If a template is specified that requires a port vlan but no
port vlan is specified the module will exit out with no changes.
default: None
template:
description:
- A path to a Jinja formatted template file that contains the
configuration block that will be applied to the specified switch port.
This template will have variable fields replaced by the module before
being applied to the switch configuration.
required: true
action:
description:
- The action for the module to take. The actions are add, which applies
the specified template config to port, remove, which defaults the
specified interface configuration, and show, which will return the
current port configuration with no changes.
default: show
choices: [show, add, remove]
auto_run:
description:
- Flag that determines whether or not the module will execute the CVP
task spawned as a result of changes to a switch configlet. When an
add or remove action is taken which results in a change to a switch
configlet, CVP will spawn a task that needs to be executed for the
configuration to be applied to the switch. If this option is True then
the module will determined the task number created by the configuration
change, execute it and wait for the task to complete. If the option
is False then the task will remain in the Pending state in CVP for
a network administrator to review and execute.
default: False
type: bool
notes:
requirements: [Jinja2, cvprac >= 0.7.0]
'''
EXAMPLES = '''
- name: Get current configuration for interface Ethernet2
cv_server_provision:
host: cvp_node
username: cvp_user
password: cvp_pass
protocol: https
server_name: new_server
switch_name: eos_switch_1
switch_port: 2
template: template_file.j2
action: show
- name: Remove existing configuration from interface Ethernet2. Run task.
cv_server_provision:
host: cvp_node
username: cvp_user
password: cvp_pass
protocol: https
server_name: new_server
switch_name: eos_switch_1
switch_port: 2
template: template_file.j2
action: remove
auto_run: True
- name: Add template configuration to interface Ethernet2. No VLAN. Run task.
cv_server_provision:
host: cvp_node
username: cvp_user
password: cvp_pass
protocol: https
server_name: new_server
switch_name: eos_switch_1
switch_port: 2
template: single_attached_trunk.j2
action: add
auto_run: True
- name: Add template with VLAN configuration to interface Ethernet2. Run task.
cv_server_provision:
host: cvp_node
username: cvp_user
password: cvp_pass
protocol: https
server_name: new_server
switch_name: eos_switch_1
switch_port: 2
port_vlan: 22
template: single_attached_vlan.j2
action: add
auto_run: True
'''
RETURN = '''
changed:
description: Signifies if a change was made to the configlet
returned: success
type: bool
sample: true
currentConfigBlock:
description: The current config block for the user specified interface
returned: when action = show
type: string
sample: |
interface Ethernet4
!
newConfigBlock:
description: The new config block for the user specified interface
returned: when action = add or remove
type: string
sample: |
interface Ethernet3
description example
no switchport
!
oldConfigBlock:
description: The current config block for the user specified interface
before any changes are made
returned: when action = add or remove
type: string
sample: |
interface Ethernet3
!
fullConfig:
description: The full config of the configlet after being updated
returned: when action = add or remove
type: string
sample: |
!
interface Ethernet3
!
interface Ethernet4
!
updateConfigletResponse:
description: Response returned from CVP when configlet update is triggered
returned: when action = add or remove and configuration changes
type: string
sample: "Configlet veos1-server successfully updated and task initiated."
portConfigurable:
description: Signifies if the user specified port has an entry in the
configlet that Ansible has access to
returned: success
type: bool
sample: true
switchConfigurable:
description: Signifies if the user specified switch has a configlet
applied to it that CVP is allowed to edit
returned: success
type: bool
sample: true
switchInfo:
description: Information from CVP describing the switch being configured
returned: success
type: dictionary
sample: {"architecture": "i386",
"bootupTimeStamp": 1491264298.21,
"complianceCode": "0000",
"complianceIndication": "NONE",
"deviceInfo": "Registered",
"deviceStatus": "Registered",
"fqdn": "veos1",
"hardwareRevision": "",
"internalBuildId": "12-12",
"internalVersion": "4.17.1F-11111.4171F",
"ipAddress": "192.168.1.20",
"isDANZEnabled": "no",
"isMLAGEnabled": "no",
"key": "00:50:56:5d:e5:e0",
"lastSyncUp": 1496432895799,
"memFree": 472976,
"memTotal": 1893460,
"modelName": "vEOS",
"parentContainerId": "container_13_5776759195930",
"serialNumber": "",
"systemMacAddress": "00:50:56:5d:e5:e0",
"taskIdList": [],
"tempAction": null,
"type": "netelement",
"unAuthorized": false,
"version": "4.17.1F",
"ztpMode": "false"}
taskCompleted:
description: Signifies if the task created and executed has completed successfully
returned: when action = add or remove, and auto_run = true,
and configuration changes
type: bool
sample: true
taskCreated:
description: Signifies if a task was created due to configlet changes
returned: when action = add or remove, and auto_run = true or false,
and configuration changes
type: bool
sample: true
taskExecuted:
description: Signifies if the automation executed the spawned task
returned: when action = add or remove, and auto_run = true,
and configuration changes
type: bool
sample: true
taskId:
description: The task ID created by CVP because of changes to configlet
returned: when action = add or remove, and auto_run = true or false,
and configuration changes
type: string
sample: "500"
'''
import re
import time
from ansible.module_utils.basic import AnsibleModule
try:
import jinja2
from jinja2 import meta
HAS_JINJA2 = True
except ImportError:
HAS_JINJA2 = False
try:
from cvprac.cvp_client import CvpClient
from cvprac.cvp_client_errors import CvpLoginError, CvpApiError
HAS_CVPRAC = True
except ImportError:
HAS_CVPRAC = False
def connect(module):
''' Connects to CVP device using user provided credentials from playbook.
:param module: Ansible module with parameters and client connection.
:return: CvpClient object with connection instantiated.
'''
client = CvpClient()
try:
client.connect([module.params['host']],
module.params['username'],
module.params['password'],
protocol=module.params['protocol'],
port=module.params['port'])
except CvpLoginError as e:
module.fail_json(msg=str(e))
return client
def switch_info(module):
''' Get dictionary of switch info from CVP.
:param module: Ansible module with parameters and client connection.
:return: Dict of switch info from CVP or exit with failure if no
info for device is found.
'''
switch_name = module.params['switch_name']
switch_info = module.client.api.get_device_by_name(switch_name)
if not switch_info:
module.fail_json(msg=str("Device with name '%s' does not exist."
% switch_name))
return switch_info
def switch_in_compliance(module, sw_info):
''' Check if switch is currently in compliance.
:param module: Ansible module with parameters and client connection.
:param sw_info: Dict of switch info.
:return: Nothing or exit with failure if device is not in compliance.
'''
compliance = module.client.api.check_compliance(sw_info['key'],
sw_info['type'])
if compliance['complianceCode'] != '0000':
module.fail_json(msg=str('Switch %s is not in compliance. Returned'
' compliance code %s.'
% (sw_info['fqdn'],
compliance['complianceCode'])))
def server_configurable_configlet(module, sw_info):
''' Check CVP that the user specified switch has a configlet assigned to
it that Ansible is allowed to edit.
:param module: Ansible module with parameters and client connection.
:param sw_info: Dict of switch info.
:return: Dict of configlet information or None.
'''
configurable_configlet = None
configlet_name = module.params['switch_name'] + '-server'
switch_configlets = module.client.api.get_configlets_by_device_id(
sw_info['key'])
for configlet in switch_configlets:
if configlet['name'] == configlet_name:
configurable_configlet = configlet
return configurable_configlet
def port_configurable(module, configlet):
''' Check configlet if the user specified port has a configuration entry
in the configlet to determine if Ansible is allowed to configure the
port on this switch.
:param module: Ansible module with parameters and client connection.
:param configlet: Dict of configlet info.
:return: True or False.
'''
configurable = False
regex = r'^interface Ethernet%s' % module.params['switch_port']
for config_line in configlet['config'].split('\n'):
if re.match(regex, config_line):
configurable = True
return configurable
def configlet_action(module, configlet):
''' Take appropriate action based on current state of device and user
requested action.
Return current config block for specified port if action is show.
If action is add or remove make the appropriate changes to the
configlet and return the associated information.
:param module: Ansible module with parameters and client connection.
:param configlet: Dict of configlet info.
:return: Dict of information to updated results with.
'''
result = dict()
existing_config = current_config(module, configlet['config'])
if module.params['action'] == 'show':
result['currentConfigBlock'] = existing_config
return result
elif module.params['action'] == 'add':
result['newConfigBlock'] = config_from_template(module)
elif module.params['action'] == 'remove':
result['newConfigBlock'] = ('interface Ethernet%s\n!'
% module.params['switch_port'])
result['oldConfigBlock'] = existing_config
result['fullConfig'] = updated_configlet_content(module,
configlet['config'],
result['newConfigBlock'])
resp = module.client.api.update_configlet(result['fullConfig'],
configlet['key'],
configlet['name'])
if 'data' in resp:
result['updateConfigletResponse'] = resp['data']
if 'task' in resp['data']:
result['changed'] = True
result['taskCreated'] = True
return result
def current_config(module, config):
''' Parse the full port configuration for the user specified port out of
the full configlet configuration and return as a string.
:param module: Ansible module with parameters and client connection.
:param config: Full config to parse specific port config from.
:return: String of current config block for user specified port.
'''
regex = r'^interface Ethernet%s' % module.params['switch_port']
match = re.search(regex, config, re.M)
if not match:
module.fail_json(msg=str('interface section not found - %s'
% config))
block_start, line_end = match.regs[0]
match = re.search(r'!', config[line_end:], re.M)
if not match:
return config[block_start:]
_, block_end = match.regs[0]
block_end = line_end + block_end
return config[block_start:block_end]
def valid_template(port, template):
''' Test if the user provided Jinja template is valid.
:param port: User specified port.
:param template: Contents of Jinja template.
:return: True or False
'''
valid = True
regex = r'^interface Ethernet%s' % port
match = re.match(regex, template, re.M)
if not match:
valid = False
return valid
def config_from_template(module):
''' Load the Jinja template and apply user provided parameters in necessary
places. Fail if template is not found. Fail if rendered template does
not reference the correct port. Fail if the template requires a VLAN
but the user did not provide one with the port_vlan parameter.
:param module: Ansible module with parameters and client connection.
:return: String of Jinja template rendered with parameters or exit with
failure.
'''
template_loader = jinja2.FileSystemLoader('./templates')
env = jinja2.Environment(loader=template_loader,
undefined=jinja2.DebugUndefined)
template = env.get_template(module.params['template'])
if not template:
module.fail_json(msg=str('Could not find template - %s'
% module.params['template']))
data = {'switch_port': module.params['switch_port'],
'server_name': module.params['server_name']}
temp_source = env.loader.get_source(env, module.params['template'])[0]
parsed_content = env.parse(temp_source)
temp_vars = list(meta.find_undeclared_variables(parsed_content))
if 'port_vlan' in temp_vars:
if module.params['port_vlan']:
data['port_vlan'] = module.params['port_vlan']
else:
module.fail_json(msg=str('Template %s requires a vlan. Please'
' re-run with vlan number provided.'
% module.params['template']))
template = template.render(data)
if not valid_template(module.params['switch_port'], template):
module.fail_json(msg=str('Template content does not configure proper'
' interface - %s' % template))
return template
def updated_configlet_content(module, existing_config, new_config):
''' Update the configlet configuration with the new section for the port
specified by the user.
:param module: Ansible module with parameters and client connection.
:param existing_config: String of current configlet configuration.
:param new_config: String of configuration for user specified port to
replace in the existing config.
:return: String of the full updated configuration.
'''
regex = r'^interface Ethernet%s' % module.params['switch_port']
match = re.search(regex, existing_config, re.M)
if not match:
module.fail_json(msg=str('interface section not found - %s'
% existing_config))
block_start, line_end = match.regs[0]
updated_config = existing_config[:block_start] + new_config
match = re.search(r'!\n', existing_config[line_end:], re.M)
if match:
_, block_end = match.regs[0]
block_end = line_end + block_end
updated_config += '\n%s' % existing_config[block_end:]
return updated_config
def configlet_update_task(module):
''' Poll device info of switch from CVP up to three times to see if the
configlet updates have spawned a task. It sometimes takes a second for
the task to be spawned after configlet updates. If a task is found
return the task ID. Otherwise return None.
:param module: Ansible module with parameters and client connection.
:return: Task ID or None.
'''
for num in range(3):
device_info = switch_info(module)
if (('taskIdList' in device_info) and
(len(device_info['taskIdList']) > 0)):
for task in device_info['taskIdList']:
if ('Configlet Assign' in task['description'] and
task['data']['WORKFLOW_ACTION'] == 'Configlet Push'):
return task['workOrderId']
time.sleep(1)
return None
def wait_for_task_completion(module, task):
''' Poll CVP for the executed task to complete. There is currently no
timeout. Exits with failure if task status is Failed or Cancelled.
:param module: Ansible module with parameters and client connection.
:param task: Task ID to poll for completion.
:return: True or exit with failure if task is cancelled or fails.
'''
task_complete = False
while not task_complete:
task_info = module.client.api.get_task_by_id(task)
task_status = task_info['workOrderUserDefinedStatus']
if task_status == 'Completed':
return True
elif task_status in ['Failed', 'Cancelled']:
module.fail_json(msg=str('Task %s has reported status %s. Please'
' consult the CVP admins for more'
' information.' % (task, task_status)))
time.sleep(2)
def main():
""" main entry point for module execution
"""
argument_spec = dict(
host=dict(required=True),
port=dict(required=False, default=None),
protocol=dict(default='https', choices=['http', 'https']),
username=dict(required=True),
password=dict(required=True, no_log=True),
server_name=dict(required=True),
switch_name=dict(required=True),
switch_port=dict(required=True),
port_vlan=dict(required=False, default=None),
template=dict(require=True),
action=dict(default='show', choices=['show', 'add', 'remove']),
auto_run=dict(type='bool', default=False))
module = AnsibleModule(argument_spec=argument_spec,
supports_check_mode=False)
if not HAS_JINJA2:
module.fail_json(msg='The Jinja2 python module is required.')
if not HAS_CVPRAC:
module.fail_json(msg='The cvprac python module is required.')
result = dict(changed=False)
module.client = connect(module)
try:
result['switchInfo'] = switch_info(module)
if module.params['action'] in ['add', 'remove']:
switch_in_compliance(module, result['switchInfo'])
switch_configlet = server_configurable_configlet(module,
result['switchInfo'])
if not switch_configlet:
module.fail_json(msg=str('Switch %s has no configurable server'
' ports.' % module.params['switch_name']))
result['switchConfigurable'] = True
if not port_configurable(module, switch_configlet):
module.fail_json(msg=str('Port %s is not configurable as a server'
' port on switch %s.'
% (module.params['switch_port'],
module.params['switch_name'])))
result['portConfigurable'] = True
result['taskCreated'] = False
result['taskExecuted'] = False
result['taskCompleted'] = False
result.update(configlet_action(module, switch_configlet))
if module.params['auto_run'] and module.params['action'] != 'show':
task_id = configlet_update_task(module)
if task_id:
result['taskId'] = task_id
note = ('Update config on %s with %s action from Ansible.'
% (module.params['switch_name'],
module.params['action']))
module.client.api.add_note_to_task(task_id, note)
module.client.api.execute_task(task_id)
result['taskExecuted'] = True
task_completed = wait_for_task_completion(module, task_id)
if task_completed:
result['taskCompleted'] = True
else:
result['taskCreated'] = False
except CvpApiError as e:
module.fail_json(msg=str(e))
module.exit_json(**result)
if __name__ == '__main__':
main()

0
test/units/modules/network/cloudvision/__init__.py Normal file
View file

886
test/units/modules/network/cloudvision/test_cv_server_provision.py Normal file
View file

@ -0,0 +1,886 @@
# 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/>.
#
from ansible.compat.tests import unittest
from ansible.compat.tests.mock import patch, Mock
import sys
sys.modules['cvprac'] = Mock()
sys.modules['cvprac.cvp_client'] = Mock()
sys.modules['cvprac.cvp_client_errors'] = Mock()
import ansible.modules.network.cloudvision.cv_server_provision as cv_server_provision
class MockException(BaseException):
pass
class TestCvServerProvision(unittest.TestCase):
@patch('ansible.modules.network.cloudvision.cv_server_provision.CvpApiError',
new_callable=lambda: MockException)
@patch('ansible.modules.network.cloudvision.cv_server_provision.server_configurable_configlet')
@patch('ansible.modules.network.cloudvision.cv_server_provision.switch_in_compliance')
@patch('ansible.modules.network.cloudvision.cv_server_provision.switch_info')
@patch('ansible.modules.network.cloudvision.cv_server_provision.connect')
@patch('ansible.modules.network.cloudvision.cv_server_provision.AnsibleModule')
def test_main_module_args(self, mock_module, mock_connect, mock_info,
mock_comp, mock_server_conf, mock_exception):
''' Test main module args.
'''
mock_module_object = Mock()
mock_module_object.params = dict(action='show', switch_name='eos')
mock_module_object.fail_json.side_effect = SystemExit('Exiting')
mock_module.return_value = mock_module_object
mock_connect.return_value = 'Client'
mock_info.side_effect = mock_exception('Error Getting Info')
argument_spec = dict(
host=dict(required=True),
port=dict(required=False, default=None),
protocol=dict(default='https', choices=['http', 'https']),
username=dict(required=True),
password=dict(required=True, no_log=True),
server_name=dict(required=True),
switch_name=dict(required=True),
switch_port=dict(required=True),
port_vlan=dict(required=False, default=None),
template=dict(require=True),
action=dict(default='show', choices=['show', 'add', 'remove']),
auto_run=dict(type='bool', default=False),
)
self.assertRaises(SystemExit, cv_server_provision.main)
mock_module.assert_called_with(argument_spec=argument_spec,
supports_check_mode=False)
self.assertEqual(mock_connect.call_count, 1)
self.assertEqual(mock_info.call_count, 1)
mock_comp.assert_not_called()
mock_server_conf.assert_not_called()
mock_module_object.fail_json.assert_called_with(msg='Error Getting Info')
@patch('ansible.modules.network.cloudvision.cv_server_provision.CvpApiError',
new_callable=lambda: MockException)
@patch('ansible.modules.network.cloudvision.cv_server_provision.server_configurable_configlet')
@patch('ansible.modules.network.cloudvision.cv_server_provision.switch_in_compliance')
@patch('ansible.modules.network.cloudvision.cv_server_provision.switch_info')
@patch('ansible.modules.network.cloudvision.cv_server_provision.connect')
@patch('ansible.modules.network.cloudvision.cv_server_provision.AnsibleModule')
def test_main_no_switch_configlet(self, mock_module, mock_connect,
mock_info, mock_comp, mock_server_conf,
mock_exception):
''' Test main fails if switch has no configlet for Ansible to edit.
'''
mock_module_object = Mock()
mock_module_object.params = dict(action='add', switch_name='eos')
mock_module_object.fail_json.side_effect = SystemExit('Exiting')
mock_module.return_value = mock_module_object
mock_connect.return_value = 'Client'
mock_info.return_value = 'Info'
mock_server_conf.return_value = None
self.assertRaises(SystemExit, cv_server_provision.main)
self.assertEqual(mock_connect.call_count, 1)
self.assertEqual(mock_info.call_count, 1)
self.assertEqual(mock_comp.call_count, 1)
self.assertEqual(mock_server_conf.call_count, 1)
mock_module_object.fail_json.assert_called_with(
msg='Switch eos has no configurable server ports.')
@patch('ansible.modules.network.cloudvision.cv_server_provision.CvpApiError',
new_callable=lambda: MockException)
@patch('ansible.modules.network.cloudvision.cv_server_provision.port_configurable')
@patch('ansible.modules.network.cloudvision.cv_server_provision.server_configurable_configlet')
@patch('ansible.modules.network.cloudvision.cv_server_provision.switch_in_compliance')
@patch('ansible.modules.network.cloudvision.cv_server_provision.switch_info')
@patch('ansible.modules.network.cloudvision.cv_server_provision.connect')
@patch('ansible.modules.network.cloudvision.cv_server_provision.AnsibleModule')
def test_main_port_not_in_config(self, mock_module, mock_connect, mock_info,
mock_comp, mock_server_conf,
mock_port_conf, mock_exception):
''' Test main fails if user specified port not in configlet.
'''
mock_module_object = Mock()
mock_module_object.params = dict(action='add', switch_name='eos',
switch_port='3')
mock_module_object.fail_json.side_effect = SystemExit('Exiting')
mock_module.return_value = mock_module_object
mock_connect.return_value = 'Client'
mock_info.return_value = 'Info'
mock_server_conf.return_value = 'Configlet'
mock_port_conf.return_value = None
self.assertRaises(SystemExit, cv_server_provision.main)
self.assertEqual(mock_connect.call_count, 1)
self.assertEqual(mock_info.call_count, 1)
self.assertEqual(mock_comp.call_count, 1)
self.assertEqual(mock_server_conf.call_count, 1)
self.assertEqual(mock_port_conf.call_count, 1)
mock_module_object.fail_json.assert_called_with(
msg='Port 3 is not configurable as a server port on switch eos.')
@patch('ansible.modules.network.cloudvision.cv_server_provision.configlet_action')
@patch('ansible.modules.network.cloudvision.cv_server_provision.port_configurable')
@patch('ansible.modules.network.cloudvision.cv_server_provision.server_configurable_configlet')
@patch('ansible.modules.network.cloudvision.cv_server_provision.switch_in_compliance')
@patch('ansible.modules.network.cloudvision.cv_server_provision.switch_info')
@patch('ansible.modules.network.cloudvision.cv_server_provision.connect')
@patch('ansible.modules.network.cloudvision.cv_server_provision.AnsibleModule')
def test_main_show(self, mock_module, mock_connect, mock_info, mock_comp,
mock_server_conf, mock_port_conf, mock_conf_action):
''' Test main good with show action.
'''
mock_module_object = Mock()
mock_module_object.params = dict(action='show', switch_name='eos',
switch_port='3', auto_run=False)
mock_module.return_value = mock_module_object
mock_connect.return_value = 'Client'
mock_info.return_value = 'Info'
mock_server_conf.return_value = 'Configlet'
mock_port_conf.return_value = 'Port'
mock_conf_action.return_value = dict()
cv_server_provision.main()
self.assertEqual(mock_connect.call_count, 1)
self.assertEqual(mock_info.call_count, 1)
mock_comp.assert_not_called()
self.assertEqual(mock_server_conf.call_count, 1)
self.assertEqual(mock_port_conf.call_count, 1)
self.assertEqual(mock_conf_action.call_count, 1)
mock_module_object.fail_json.assert_not_called()
return_dict = dict(changed=False, switchInfo='Info',
switchConfigurable=True, portConfigurable=True,
taskCreated=False, taskExecuted=False,
taskCompleted=False)
mock_module_object.exit_json.assert_called_with(**return_dict)
@patch('ansible.modules.network.cloudvision.cv_server_provision.configlet_action')
@patch('ansible.modules.network.cloudvision.cv_server_provision.port_configurable')
@patch('ansible.modules.network.cloudvision.cv_server_provision.server_configurable_configlet')
@patch('ansible.modules.network.cloudvision.cv_server_provision.switch_in_compliance')
@patch('ansible.modules.network.cloudvision.cv_server_provision.switch_info')
@patch('ansible.modules.network.cloudvision.cv_server_provision.connect')
@patch('ansible.modules.network.cloudvision.cv_server_provision.AnsibleModule')
def test_main_add_no_auto_run(self, mock_module, mock_connect, mock_info,
mock_comp, mock_server_conf, mock_port_conf,
mock_conf_action):
''' Test main good with add action and no auto_run.
'''
mock_module_object = Mock()
mock_module_object.params = dict(action='add', switch_name='eos',
switch_port='3', auto_run=False)
mock_module.return_value = mock_module_object
mock_connect.return_value = 'Client'
mock_info.return_value = 'Info'
mock_server_conf.return_value = 'Configlet'
mock_port_conf.return_value = 'Port'
mock_conf_action.return_value = dict(taskCreated=True)
cv_server_provision.main()
self.assertEqual(mock_connect.call_count, 1)
self.assertEqual(mock_info.call_count, 1)
self.assertEqual(mock_comp.call_count, 1)
self.assertEqual(mock_server_conf.call_count, 1)
self.assertEqual(mock_port_conf.call_count, 1)
self.assertEqual(mock_conf_action.call_count, 1)
mock_module_object.fail_json.assert_not_called()
return_dict = dict(changed=False, switchInfo='Info',
switchConfigurable=True, portConfigurable=True,
taskCreated=True, taskExecuted=False,
taskCompleted=False)
mock_module_object.exit_json.assert_called_with(**return_dict)
@patch('ansible.modules.network.cloudvision.cv_server_provision.wait_for_task_completion')
@patch('ansible.modules.network.cloudvision.cv_server_provision.configlet_update_task')
@patch('ansible.modules.network.cloudvision.cv_server_provision.configlet_action')
@patch('ansible.modules.network.cloudvision.cv_server_provision.port_configurable')
@patch('ansible.modules.network.cloudvision.cv_server_provision.server_configurable_configlet')
@patch('ansible.modules.network.cloudvision.cv_server_provision.switch_in_compliance')
@patch('ansible.modules.network.cloudvision.cv_server_provision.switch_info')
@patch('ansible.modules.network.cloudvision.cv_server_provision.connect')
@patch('ansible.modules.network.cloudvision.cv_server_provision.AnsibleModule')
def test_main_add_auto_run(self, mock_module, mock_connect, mock_info,
mock_comp, mock_server_conf, mock_port_conf,
mock_conf_action, mock_conf_task, mock_wait):
''' Test main good with add and auto_run. Config updated, task created.
'''
mock_module_object = Mock()
mock_module_object.params = dict(action='add', switch_name='eos',
switch_port='3', auto_run=True)
mock_module.return_value = mock_module_object
mock_client_object = Mock()
mock_connect.return_value = mock_client_object
mock_info.return_value = 'Info'
mock_server_conf.return_value = 'Configlet'
mock_port_conf.return_value = 'Port'
mock_conf_action.return_value = dict(taskCreated=True, changed=True)
mock_conf_task.return_value = '7'
mock_wait.return_value = True
cv_server_provision.main()
self.assertEqual(mock_connect.call_count, 1)
self.assertEqual(mock_info.call_count, 1)
self.assertEqual(mock_comp.call_count, 1)
self.assertEqual(mock_server_conf.call_count, 1)
self.assertEqual(mock_port_conf.call_count, 1)
self.assertEqual(mock_conf_action.call_count, 1)
self.assertEqual(mock_conf_task.call_count, 1)
self.assertEqual(mock_wait.call_count, 1)
mock_module_object.fail_json.assert_not_called()
return_dict = dict(changed=True, switchInfo='Info', taskId='7',
switchConfigurable=True, portConfigurable=True,
taskCreated=True, taskExecuted=True,
taskCompleted=True)
mock_module_object.exit_json.assert_called_with(**return_dict)
@patch('ansible.modules.network.cloudvision.cv_server_provision.wait_for_task_completion')
@patch('ansible.modules.network.cloudvision.cv_server_provision.configlet_update_task')
@patch('ansible.modules.network.cloudvision.cv_server_provision.configlet_action')
@patch('ansible.modules.network.cloudvision.cv_server_provision.port_configurable')
@patch('ansible.modules.network.cloudvision.cv_server_provision.server_configurable_configlet')
@patch('ansible.modules.network.cloudvision.cv_server_provision.switch_in_compliance')
@patch('ansible.modules.network.cloudvision.cv_server_provision.switch_info')
@patch('ansible.modules.network.cloudvision.cv_server_provision.connect')
@patch('ansible.modules.network.cloudvision.cv_server_provision.AnsibleModule')
def test_main_add_auto_run_no_task(self, mock_module, mock_connect,
mock_info, mock_comp, mock_server_conf,
mock_port_conf, mock_conf_action, mock_conf_task,
mock_wait):
''' Test main good with add and auto_run. Config not updated, no task.
'''
mock_module_object = Mock()
mock_module_object.params = dict(action='add', switch_name='eos',
switch_port='3', auto_run=True)
mock_module.return_value = mock_module_object
mock_client_object = Mock()
mock_connect.return_value = mock_client_object
mock_info.return_value = 'Info'
mock_server_conf.return_value = 'Configlet'
mock_port_conf.return_value = 'Port'
mock_conf_action.return_value = dict(taskCreated=True, changed=False)
mock_conf_task.return_value = None
cv_server_provision.main()
self.assertEqual(mock_connect.call_count, 1)
self.assertEqual(mock_info.call_count, 1)
self.assertEqual(mock_comp.call_count, 1)
self.assertEqual(mock_server_conf.call_count, 1)
self.assertEqual(mock_port_conf.call_count, 1)
self.assertEqual(mock_conf_action.call_count, 1)
self.assertEqual(mock_conf_task.call_count, 1)
mock_wait.assert_not_called()
mock_module_object.fail_json.assert_not_called()
return_dict = dict(changed=False, switchInfo='Info',
switchConfigurable=True, portConfigurable=True,
taskCreated=False, taskExecuted=False,
taskCompleted=False)
mock_module_object.exit_json.assert_called_with(**return_dict)
@patch('ansible.modules.network.cloudvision.cv_server_provision.CvpClient')
def test_connect_good(self, mock_client):
''' Test connect success.
'''
module = Mock()
module.params = dict(host='host', username='username',
password='password', protocol='https', port='10')
connect_mock = Mock()
mock_client.return_value = connect_mock
client = cv_server_provision.connect(module)
self.assertIsInstance(client, Mock)
self.assertEqual(mock_client.call_count, 1)
connect_mock.connect.assert_called_once_with(['host'], 'username',
'password', port='10',
protocol='https')
module.fail_json.assert_not_called()
@patch('ansible.modules.network.cloudvision.cv_server_provision.CvpLoginError',
new_callable=lambda: MockException)
@patch('ansible.modules.network.cloudvision.cv_server_provision.CvpClient')
def test_connect_fail(self, mock_client, mock_exception):
''' Test connect failure with login error.
'''
module = Mock()
module.params = dict(host='host', username='username',
password='password', protocol='https', port='10')
module.fail_json.side_effect = SystemExit
connect_mock = Mock()
connect_mock.connect.side_effect = mock_exception('Login Error')
mock_client.return_value = connect_mock
self.assertRaises(SystemExit, cv_server_provision.connect, module)
self.assertEqual(connect_mock.connect.call_count, 1)
module.fail_json.assert_called_once_with(msg='Login Error')
def test_switch_info_good(self):
''' Test switch_info success.
'''
module = Mock()
module.params = dict(switch_name='eos')
module.client.api.get_device_by_name.return_value = dict(fqdn='eos')
info = cv_server_provision.switch_info(module)
self.assertEqual(module.client.api.get_device_by_name.call_count, 1)
self.assertEqual(info['fqdn'], 'eos')
module.fail_json.assert_not_called()
def test_switch_info_no_switch(self):
''' Test switch_info fails.
'''
module = Mock()
module.params = dict(switch_name='eos')
module.client.api.get_device_by_name.return_value = None
info = cv_server_provision.switch_info(module)
self.assertEqual(module.client.api.get_device_by_name.call_count, 1)
self.assertEqual(info, None)
module.fail_json.assert_called_once_with(
msg="Device with name 'eos' does not exist.")
def test_switch_in_compliance_good(self):
''' Test switch_in_compliance good.
'''
module = Mock()
module.client.api.check_compliance.return_value = dict(
complianceCode='0000')
sw_info = dict(key='key', type='type', fqdn='eos')
cv_server_provision.switch_in_compliance(module, sw_info)
self.assertEqual(module.client.api.check_compliance.call_count, 1)
module.fail_json.assert_not_called()
def test_switch_in_compliance_fail(self):
''' Test switch_in_compliance fail.
'''
module = Mock()
module.client.api.check_compliance.return_value = dict(
complianceCode='0001')
sw_info = dict(key='key', type='type', fqdn='eos')
cv_server_provision.switch_in_compliance(module, sw_info)
self.assertEqual(module.client.api.check_compliance.call_count, 1)
module.fail_json.assert_called_with(
msg='Switch eos is not in compliance.'
' Returned compliance code 0001.')
def test_server_configurable_configlet_good(self):
''' Test server_configurable_configlet good.
'''
module = Mock()
module.params = dict(switch_name='eos')
configlets = [dict(name='configlet1', info='line'),
dict(name='eos-server', info='info')]
module.client.api.get_configlets_by_device_id.return_value = configlets
sw_info = dict(key='key', type='type', fqdn='eos')
result = cv_server_provision.server_configurable_configlet(module,
sw_info)
self.assertEqual(module.client.api.get_configlets_by_device_id.call_count, 1)
self.assertIsNotNone(result)
self.assertEqual(result['name'], 'eos-server')
self.assertEqual(result['info'], 'info')
def test_server_configurable_configlet_not_configurable(self):
''' Test server_configurable_configlet fail. No server configlet.
'''
module = Mock()
module.params = dict(switch_name='eos')
configlets = [dict(name='configlet1', info='line'),
dict(name='configlet2', info='info')]
module.client.api.get_configlets_by_device_id.return_value = configlets
sw_info = dict(key='key', type='type', fqdn='eos')
result = cv_server_provision.server_configurable_configlet(module, sw_info)
self.assertEqual(module.client.api.get_configlets_by_device_id.call_count, 1)
self.assertIsNone(result)
def test_server_configurable_configlet_no_configlets(self):
''' Test server_configurable_configlet fail. No switch configlets.
'''
module = Mock()
module.params = dict(switch_name='eos')
module.client.api.get_configlets_by_device_id.return_value = []
sw_info = dict(key='key', type='type', fqdn='eos')
result = cv_server_provision.server_configurable_configlet(module,
sw_info)
self.assertEqual(module.client.api.get_configlets_by_device_id.call_count, 1)
self.assertIsNone(result)
def test_port_configurable_good(self):
''' Test port_configurable user provided switch port in configlet.
'''
module = Mock()
module.params = dict(switch_name='eos', switch_port='3')
config = '!\ninterface Ethernet3\n!\ninterface Ethernet4\n!'
configlet = dict(name='eos-server', config=config)
result = cv_server_provision.port_configurable(module, configlet)
self.assertTrue(result)
def test_port_configurable_fail(self):
''' Test port_configurable user provided switch port not in configlet.
'''
module = Mock()
module.params = dict(switch_name='eos', switch_port='2')
config = '!\ninterface Ethernet3\n!\ninterface Ethernet4\n!'
configlet = dict(name='eos-server', config=config)
result = cv_server_provision.port_configurable(module, configlet)
self.assertFalse(result)
def test_port_configurable_fail_no_config(self):
''' Test port_configurable configlet empty.
'''
module = Mock()
module.params = dict(switch_name='eos', switch_port='2')
config = ''
configlet = dict(name='eos-server', config=config)
result = cv_server_provision.port_configurable(module, configlet)
self.assertFalse(result)
def test_configlet_action_show_blank_config(self):
''' Test configlet_action show returns current port configuration.
'''
module = Mock()
module.params = dict(action='show', switch_name='eos', switch_port='3')
config = '!\ninterface Ethernet3\n!\ninterface Ethernet4\n!'
configlet = dict(name='eos-server', key='key', config=config)
result = cv_server_provision.configlet_action(module, configlet)
self.assertIsNotNone(result)
self.assertEqual(result['currentConfigBlock'], 'interface Ethernet3\n!')
module.client.api.update_configlet.assert_not_called()
@patch('ansible.modules.network.cloudvision.cv_server_provision.config_from_template')
def test_configlet_action_add_with_task(self, mock_template):
''' Test configlet_action add with change updates configlet and adds
proper info to return data. Including task spawned info.
'''
module = Mock()
module.params = dict(action='add', switch_name='eos', switch_port='3')
config = '!\ninterface Ethernet3\n!\ninterface Ethernet4\n!'
configlet = dict(name='eos-server', key='key', config=config)
template_config = ('interface Ethernet3\n description Host eos'
' managed by Ansible and Jinja template\n'
' load-interval 30\n'
' switchport\n'
' switchport mode trunk\n'
' no shutdown\n!')
mock_template.return_value = template_config
update_return = dict(data='Configlet eos-server successfully updated'
' and task initiated.')
module.client.api.update_configlet.return_value = update_return
result = cv_server_provision.configlet_action(module, configlet)
self.assertIsNotNone(result)
self.assertEqual(result['oldConfigBlock'], 'interface Ethernet3\n!')
full_config = '!\n' + template_config + '\ninterface Ethernet4\n!'
self.assertEqual(result['fullConfig'], full_config)
self.assertEqual(result['updateConfigletResponse'],
update_return['data'])
self.assertTrue(result['changed'])
self.assertTrue(result['taskCreated'])
self.assertEqual(module.client.api.update_configlet.call_count, 1)
@patch('ansible.modules.network.cloudvision.cv_server_provision.config_from_template')
def test_configlet_action_add_no_task(self, mock_template):
''' Test configlet_action add that doesn't change configlet adds proper
info to return data. Does not including any task info.
'''
module = Mock()
module.params = dict(action='add', switch_name='eos', switch_port='3')
config = ('!\ninterface Ethernet3\n description test\n'
'!\ninterface Ethernet4\n!')
configlet = dict(name='eos-server', key='key', config=config)
template_config = 'interface Ethernet3\n description test\n!'
mock_template.return_value = template_config
update_return = dict(data='Configlet eos-server successfully updated.')
module.client.api.update_configlet.return_value = update_return
result = cv_server_provision.configlet_action(module, configlet)
self.assertIsNotNone(result)
self.assertEqual(result['oldConfigBlock'],
'interface Ethernet3\n description test\n!')
self.assertEqual(result['fullConfig'], config)
self.assertEqual(result['updateConfigletResponse'],
update_return['data'])
self.assertNotIn('changed', result)
self.assertNotIn('taskCreated', result)
self.assertEqual(module.client.api.update_configlet.call_count, 1)
def test_configlet_action_remove_with_task(self):
''' Test configlet_action remove with change updates configlet and adds
proper info to return data. Including task spawned info.
'''
module = Mock()
module.params = dict(action='remove', switch_name='eos',
switch_port='3')
config = ('!\ninterface Ethernet3\n description test\n'
'!\ninterface Ethernet4\n!')
configlet = dict(name='eos-server', key='key', config=config)
update_return = dict(data='Configlet eos-server successfully updated'
' and task initiated.')
module.client.api.update_configlet.return_value = update_return
result = cv_server_provision.configlet_action(module, configlet)
self.assertIsNotNone(result)
self.assertEqual(result['oldConfigBlock'],
'interface Ethernet3\n description test\n!')
full_config = '!\ninterface Ethernet3\n!\ninterface Ethernet4\n!'
self.assertEqual(result['fullConfig'], full_config)
self.assertEqual(result['updateConfigletResponse'],
update_return['data'])
self.assertTrue(result['changed'])
self.assertTrue(result['taskCreated'])
self.assertEqual(module.client.api.update_configlet.call_count, 1)
def test_configlet_action_remove_no_task(self):
''' Test configlet_action with remove that doesn't change configlet and
adds proper info to return data. Does not including any task info.
'''
module = Mock()
module.params = dict(action='remove', switch_name='eos',
switch_port='3')
config = '!\ninterface Ethernet3\n!\ninterface Ethernet4\n!'
configlet = dict(name='eos-server', key='key', config=config)
update_return = dict(data='Configlet eos-server successfully updated.')
module.client.api.update_configlet.return_value = update_return
result = cv_server_provision.configlet_action(module, configlet)
self.assertIsNotNone(result)
self.assertEqual(result['oldConfigBlock'], 'interface Ethernet3\n!')
self.assertEqual(result['fullConfig'], config)
self.assertEqual(result['updateConfigletResponse'],
update_return['data'])
self.assertNotIn('changed', result)
self.assertNotIn('taskCreated', result)
self.assertEqual(module.client.api.update_configlet.call_count, 1)
def test_current_config_empty_config(self):
''' Test current_config with empty config for port
'''
module = Mock()
module.params = dict(switch_name='eos', switch_port='4')
config = '!\ninterface Ethernet3\n!\ninterface Ethernet4'
result = cv_server_provision.current_config(module, config)
self.assertIsNotNone(result)
self.assertEqual(result, 'interface Ethernet4')
def test_current_config_with_config(self):
''' Test current_config with config for port
'''
module = Mock()
module.params = dict(switch_name='eos', switch_port='3')
config = ('!\ninterface Ethernet3\n description test\n'
'!\ninterface Ethernet4\n!')
result = cv_server_provision.current_config(module, config)
self.assertIsNotNone(result)
self.assertEqual(result, 'interface Ethernet3\n description test\n!')
def test_current_config_no_match(self):
''' Test current_config with no entry for port
'''
module = Mock()
module.fail_json.side_effect = SystemExit
module.params = dict(switch_name='eos', switch_port='2')
config = '!\ninterface Ethernet3\n description test\n!'
self.assertRaises(SystemExit, cv_server_provision.current_config,
module, config)
def test_valid_template_true(self):
''' Test valid_template true
'''
template = 'interface Ethernet3\n description test\n!'
result = cv_server_provision.valid_template('3', template)
self.assertTrue(result)
def test_valid_template_false(self):
''' Test valid_template false
'''
template = 'interface Ethernet3\n description test\n!'
result = cv_server_provision.valid_template('4', template)
self.assertFalse(result)
@patch('jinja2.DebugUndefined')
@patch('jinja2.Environment')
@patch('jinja2.FileSystemLoader')
def test_config_from_template_no_template(self, mock_file_sys, mock_env,
mock_debug):
''' Test config_from_template good. No template.
'''
module = Mock()
module.fail_json.side_effect = SystemExit
module.params = dict(switch_name='eos', switch_port='3',
server_name='new', template='jinja.j2')
mock_file_sys.return_value = 'file'
mock_debug.return_value = 'debug'
env_mock = Mock()
env_mock.get_template.return_value = None
mock_env.return_value = env_mock
self.assertRaises(SystemExit, cv_server_provision.config_from_template,
module)
self.assertEqual(mock_file_sys.call_count, 1)
self.assertEqual(mock_env.call_count, 1)
self.assertEqual(module.fail_json.call_count, 1)
@patch('jinja2.meta.find_undeclared_variables')
@patch('jinja2.DebugUndefined')
@patch('jinja2.Environment')
@patch('jinja2.FileSystemLoader')
def test_config_from_template_good_no_vlan(self, mock_file_sys, mock_env, mock_debug,
mock_find):
''' Test config_from_template good. No port_vlan.
'''
module = Mock()
module.params = dict(switch_name='eos', switch_port='3',
server_name='new', template='jinja.j2')
mock_file_sys.return_value = 'file'
mock_debug.return_value = 'debug'
template_mock = Mock()
template_mock.render.return_value = ('interface Ethernet3\n'
' description test\n'
' switchport\n'
' switchport mode trunk\n'
' no shutdown\n!')
env_mock = Mock()
env_mock.loader.get_source.return_value = ['one', 'two']
env_mock.parse.return_value = 'parsed'
env_mock.get_template.return_value = template_mock
mock_env.return_value = env_mock
mock_find.return_value = dict(server_name=None, switch_port=None)
result = cv_server_provision.config_from_template(module)
self.assertIsNotNone(result)
expected = ('interface Ethernet3\n'
' description test\n'
' switchport\n'
' switchport mode trunk\n'
' no shutdown\n!')
self.assertEqual(result, expected)
self.assertEqual(mock_file_sys.call_count, 1)
self.assertEqual(mock_env.call_count, 1)
module.fail_json.assert_not_called()
@patch('jinja2.meta.find_undeclared_variables')
@patch('jinja2.DebugUndefined')
@patch('jinja2.Environment')
@patch('jinja2.FileSystemLoader')
def test_config_from_template_good_vlan(self, mock_file_sys, mock_env, mock_debug,
mock_find):
''' Test config_from_template good. With port_vlan.
'''
module = Mock()
module.params = dict(switch_name='eos', switch_port='3',
server_name='new', template='jinja.j2', port_vlan='7')
mock_file_sys.return_value = 'file'
mock_debug.return_value = 'debug'
template_mock = Mock()
template_mock.render.return_value = ('interface Ethernet3\n'
' description test\n'
' switchport\n'
' switchport access vlan 7\n'
' no shutdown\n!')
env_mock = Mock()
env_mock.loader.get_source.return_value = ['one', 'two']
env_mock.parse.return_value = 'parsed'
env_mock.get_template.return_value = template_mock
mock_env.return_value = env_mock
mock_find.return_value = dict(server_name=None, switch_port=None,
port_vlan=None)
result = cv_server_provision.config_from_template(module)
self.assertIsNotNone(result)
expected = ('interface Ethernet3\n'
' description test\n'
' switchport\n'
' switchport access vlan 7\n'
' no shutdown\n!')
self.assertEqual(result, expected)
self.assertEqual(mock_file_sys.call_count, 1)
self.assertEqual(mock_env.call_count, 1)
module.fail_json.assert_not_called()
@patch('jinja2.meta.find_undeclared_variables')
@patch('jinja2.DebugUndefined')
@patch('jinja2.Environment')
@patch('jinja2.FileSystemLoader')
def test_config_from_template_fail_wrong_port(self, mock_file_sys, mock_env,
mock_debug, mock_find):
''' Test config_from_template fail. Wrong port number in template.
'''
module = Mock()
module.params = dict(switch_name='eos', switch_port='4',
server_name='new', template='jinja.j2')
mock_file_sys.return_value = 'file'
mock_debug.return_value = 'debug'
template_mock = Mock()
template_mock.render.return_value = ('interface Ethernet3\n'
' description test\n!')
env_mock = Mock()
env_mock.loader.get_source.return_value = ['one', 'two']
env_mock.parse.return_value = 'parsed'
env_mock.get_template.return_value = template_mock
mock_env.return_value = env_mock
mock_find.return_value = dict(server_name=None, switch_port=None)
result = cv_server_provision.config_from_template(module)
self.assertIsNotNone(result)
expected = 'interface Ethernet3\n description test\n!'
self.assertEqual(result, expected)
self.assertEqual(mock_file_sys.call_count, 1)
self.assertEqual(mock_env.call_count, 1)
module.fail_json.assert_called_with(msg='Template content does not'
' configure proper interface'
' - %s' % expected)
@patch('jinja2.meta.find_undeclared_variables')
@patch('jinja2.DebugUndefined')
@patch('jinja2.Environment')
@patch('jinja2.FileSystemLoader')
def test_config_from_template_fail_no_vlan(self, mock_file_sys, mock_env,
mock_debug, mock_find):
''' Test config_from_template fail. Template needs vlan but none provided.
'''
module = Mock()
module.params = dict(switch_name='eos', switch_port='3',
server_name='new', template='jinja.j2',
port_vlan=None)
mock_file_sys.return_value = 'file'
mock_debug.return_value = 'debug'
template_mock = Mock()
template_mock.render.return_value = ('interface Ethernet3\n'
' description test\n!')
env_mock = Mock()
env_mock.loader.get_source.return_value = ['one', 'two']
env_mock.parse.return_value = 'parsed'
env_mock.get_template.return_value = template_mock
mock_env.return_value = env_mock
mock_find.return_value = dict(server_name=None, switch_port=None,
port_vlan=None)
result = cv_server_provision.config_from_template(module)
self.assertIsNotNone(result)
expected = 'interface Ethernet3\n description test\n!'
self.assertEqual(result, expected)
self.assertEqual(mock_file_sys.call_count, 1)
self.assertEqual(mock_env.call_count, 1)
module.fail_json.assert_called_with(msg='Template jinja.j2 requires a'
' vlan. Please re-run with vlan'
' number provided.')
def test_updated_configlet_content_add(self):
''' Test updated_configlet_content. Add config.
'''
module = Mock()
module.params = dict(switch_name='eos', switch_port='3')
existing_config = '!\ninterface Ethernet3\n!\ninterface Ethernet4\n!'
new_config_block = 'interface Ethernet3\n description test\n!'
result = cv_server_provision.updated_configlet_content(module,
existing_config,
new_config_block)
expected = ('!\ninterface Ethernet3\n description test\n'
'!\ninterface Ethernet4\n!')
self.assertEqual(result, expected)
module.fail_json.assert_not_called()
def test_updated_configlet_content_remove(self):
''' Test updated_configlet_content. Remove config.
'''
module = Mock()
module.params = dict(switch_name='eos', switch_port='3')
existing_config = ('!\ninterface Ethernet3\n description test\n'
'!\ninterface Ethernet4')
new_config_block = 'interface Ethernet3\n!'
result = cv_server_provision.updated_configlet_content(module,
existing_config,
new_config_block)
expected = '!\ninterface Ethernet3\n!\ninterface Ethernet4'
self.assertEqual(result, expected)
module.fail_json.assert_not_called()
def test_updated_configlet_content_no_match(self):
''' Test updated_configlet_content. Interface not in config.
'''
module = Mock()
module.fail_json.side_effect = SystemExit
module.params = dict(switch_name='eos', switch_port='2')
existing_config = '!\ninterface Ethernet3\n description test\n!'
new_config_block = 'interface Ethernet3\n!'
self.assertRaises(SystemExit,
cv_server_provision.updated_configlet_content,
module, existing_config, new_config_block)
@patch('time.sleep')
@patch('ansible.modules.network.cloudvision.cv_server_provision.switch_info')
def test_configlet_update_task_good_one_try(self, mock_info, mock_sleep):
''' Test configlet_update_task gets task after one try.
'''
module = Mock()
task = dict(data=dict(WORKFLOW_ACTION='Configlet Push'),
description='Configlet Assign',
workOrderId='7')
device_info = dict(taskIdList=[task])
mock_info.return_value = device_info
result = cv_server_provision.configlet_update_task(module)
self.assertEqual(result, '7')
mock_sleep.assert_not_called()
self.assertEqual(mock_info.call_count, 1)
@patch('time.sleep')
@patch('ansible.modules.network.cloudvision.cv_server_provision.switch_info')
def test_configlet_update_task_good_three_tries(self, mock_info, mock_sleep):
''' Test configlet_update_task gets task on third try.
'''
module = Mock()
task1 = dict(data=dict(WORKFLOW_ACTION='Configlet Push'),
description='Configlet Assign',
workOrderId='7')
task2 = dict(data=dict(WORKFLOW_ACTION='Nonsense'),
description='Configlet Assign',
workOrderId='700')
device_info = dict(taskIdList=[task1, task2])
mock_info.side_effect = [dict(), dict(), device_info]
result = cv_server_provision.configlet_update_task(module)
self.assertEqual(result, '7')
self.assertEqual(mock_sleep.call_count, 2)
self.assertEqual(mock_info.call_count, 3)
@patch('time.sleep')
@patch('ansible.modules.network.cloudvision.cv_server_provision.switch_info')
def test_configlet_update_task_no_task(self, mock_info, mock_sleep):
''' Test configlet_update_task does not get task after three tries.
'''
module = Mock()
mock_info.side_effect = [dict(), dict(), dict()]
result = cv_server_provision.configlet_update_task(module)
self.assertIsNone(result)
self.assertEqual(mock_sleep.call_count, 3)
self.assertEqual(mock_info.call_count, 3)
@patch('time.sleep')
def test_wait_for_task_completion_good_one_try(self, mock_time):
''' Test wait_for_task_completion completed. One Try.
'''
module = Mock()
module.client.api.get_task_by_id.return_value = dict(
workOrderUserDefinedStatus='Completed')
result = cv_server_provision.wait_for_task_completion(module, '7')
self.assertTrue(result)
self.assertEqual(module.client.api.get_task_by_id.call_count, 1)
module.fail_json.assert_not_called()
mock_time.assert_not_called()
@patch('time.sleep')
def test_wait_for_task_completion_good_three_tries(self, mock_time):
''' Test wait_for_task_completion completed. Three tries.
'''
module = Mock()
try_one_two = dict(workOrderUserDefinedStatus='Pending')
try_three = dict(workOrderUserDefinedStatus='Completed')
module.client.api.get_task_by_id.side_effect = [try_one_two,
try_one_two, try_three]
result = cv_server_provision.wait_for_task_completion(module, '7')
self.assertTrue(result)
self.assertEqual(module.client.api.get_task_by_id.call_count, 3)
module.fail_json.assert_not_called()
self.assertEqual(mock_time.call_count, 2)
@patch('time.sleep')
def test_wait_for_task_completion_fail(self, mock_time):
''' Test wait_for_task_completion failed.
'''
module = Mock()
try_one = dict(workOrderUserDefinedStatus='Failed')
try_two = dict(workOrderUserDefinedStatus='Completed')
module.client.api.get_task_by_id.side_effect = [try_one, try_two]
result = cv_server_provision.wait_for_task_completion(module, '7')
self.assertTrue(result)
self.assertEqual(module.client.api.get_task_by_id.call_count, 2)
text = ('Task 7 has reported status Failed. Please consult the CVP'
' admins for more information.')
module.fail_json.assert_called_with(msg=text)
self.assertEqual(mock_time.call_count, 1)