From b6202dde3293f8395e05d43270cc0f398807cb1a Mon Sep 17 00:00:00 2001 From: Chris Archibald Date: Fri, 9 Aug 2019 13:04:46 -0700 Subject: [PATCH] Allow firmware update for (SP, ACP, Shelf, Disk) (#58597) * update * Revert "update" This reverts commit 84848f829108dde057cf63c9a402dcecab527202. * correct module * fixes --- .../netapp/na_ontap_firmware_upgrade.py | 461 ++++++++++++++++++ .../netapp/test_na_ontap_firmware_upgrade.py | 291 +++++++++++ 2 files changed, 752 insertions(+) create mode 100644 lib/ansible/modules/storage/netapp/na_ontap_firmware_upgrade.py create mode 100644 test/units/modules/storage/netapp/test_na_ontap_firmware_upgrade.py diff --git a/lib/ansible/modules/storage/netapp/na_ontap_firmware_upgrade.py b/lib/ansible/modules/storage/netapp/na_ontap_firmware_upgrade.py new file mode 100644 index 00000000000..dd50060c547 --- /dev/null +++ b/lib/ansible/modules/storage/netapp/na_ontap_firmware_upgrade.py @@ -0,0 +1,461 @@ +#!/usr/bin/python + +# (c) 2019, NetApp, Inc +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) +from __future__ import absolute_import, division, print_function +__metaclass__ = type + +ANSIBLE_METADATA = {'metadata_version': '1.1', + 'status': ['preview'], + 'supported_by': 'community'} + + +DOCUMENTATION = ''' +author: NetApp Ansible Team (@carchi8py) +description: + - Update ONTAP service-prosessor firmware +extends_documentation_fragment: + - netapp.na_ontap +module: na_ontap_firmware_upgrade +options: + state: + description: + - Whether the specified ONTAP firmware should be upgraded or not. + default: present + type: str + node: + description: + - Node on which the device is located. + type: str + required: true + clear_logs: + description: + - Clear logs on the device after update. Default value is true + type: bool + default: true + package: + description: + - Name of the package file containing the firmware to be installed. Not required when -baseline is true. + type: str + shelf_module_fw: + description: + - Shelf module firmware to be updated to. + type: str + disk_fw: + description: + - disk firmware to be updated to. + type: str + update_type: + description: + - Type of firmware update to be performed. Options include serial_full, serial_differential, network_full. + type: str + install_baseline_image: + description: + - Install the version packaged with ONTAP if this parameter is set to true. Otherwise, package must be used to specify the package to install. + type: bool + default: false + firmware_type: + description: + - Type of firmware to be upgraded. Options include shelf, ACP, service-processor, and disk. + - For shelf firmware upgrade the operation is asynchronous, and therefore returns no errors that might occur during the download process. + - Shelf firmware upgrade is idempotent if shelf_module_fw is provided . + - disk firmware upgrade is idempotent if disk_fw is provided . + - With check mode, SP, ACP, disk, and shelf firmware upgrade is not idempotent. + - This operation will only update firmware on shelves/disk that do not have the latest firmware-revision. + choices: ['service-processor', 'shelf', 'acp', 'disk'] + type: str +short_description: NetApp ONTAP firmware upgrade for SP, shelf, ACP, and disk. +version_added: "2.9" +''' + +EXAMPLES = """ + + - name: SP firmware upgrade + na_ontap_firmware_upgrade: + state: present + node: vsim1 + package: "{{ file name }}" + clear_logs: True + install_baseline_image: False + update_type: serial_full + firmware_type: service-processor + hostname: "{{ netapp_hostname }}" + username: "{{ netapp_username }}" + password: "{{ netapp_password }}" + - name: ACP firmware upgrade + na_ontap_firmware_upgrade: + state: present + node: vsim1 + firmware_type: acp + hostname: "{{ netapp_hostname }}" + username: "{{ netapp_username }}" + password: "{{ netapp_password }}" + - name: shelf firmware upgrade + na_ontap_firmware_upgrade: + state: present + firmware_type: shelf + shelf_module_fw: 1221 + hostname: "{{ netapp_hostname }}" + username: "{{ netapp_username }}" + password: "{{ netapp_password }}" + - name: disk firmware upgrade + na_ontap_firmware_upgrade: + state: present + firmware_type: disk + disk_fw: NA02 + hostname: "{{ netapp_hostname }}" + username: "{{ netapp_username }}" + password: "{{ netapp_password }}" +""" + +RETURN = """ +""" + +import traceback +from ansible.module_utils.basic import AnsibleModule +from ansible.module_utils._text import to_native +import ansible.module_utils.netapp as netapp_utils +from ansible.module_utils.netapp_module import NetAppModule +import time + +HAS_NETAPP_LIB = netapp_utils.has_netapp_lib() + + +class NetAppONTAPFirmwareUpgrade(object): + """ + Class with ONTAP firmware upgrade methods + """ + + def __init__(self): + self.argument_spec = netapp_utils.na_ontap_host_argument_spec() + self.argument_spec.update(dict( + state=dict(required=False, type='str', default='present'), + node=dict(required=False, type='str'), + firmware_type=dict(required=True, type='str', choices=['service-processor', 'shelf', 'acp', 'disk']), + clear_logs=dict(required=False, type='bool', default=True), + package=dict(required=False, type='str'), + install_baseline_image=dict(required=False, type='bool', default=False), + update_type=dict(required=False, type='str'), + shelf_module_fw=dict(required=False, type='str'), + disk_fw=dict(required=False, type='str') + + )) + + self.module = AnsibleModule( + argument_spec=self.argument_spec, + required_if=[ + ('firmware_type', 'acp', ['node']), + ('firmware_type', 'disk', ['node']), + ('firmware_type', 'service-processor', ['node', 'update_type']), + ], + supports_check_mode=True + ) + + self.na_helper = NetAppModule() + self.parameters = self.na_helper.set_parameters(self.module.params) + if self.parameters.get('firmware_type') == 'service-processor': + if self.parameters.get('install_baseline_image') and self.parameters.get('package') is not None: + self.module.fail_json(msg='Do not specify both package and install_baseline_image: true') + if not self.parameters.get('package') and self.parameters.get('install_baseline_image') == 'False': + self.module.fail_json(msg='Specify at least one of package or install_baseline_image') + if HAS_NETAPP_LIB is False: + self.module.fail_json(msg="the python NetApp-Lib module is required") + else: + self.server = netapp_utils.setup_na_ontap_zapi(module=self.module) + + def firmware_image_get_iter(self): + """ + Compose NaElement object to query current firmware version + :return: NaElement object for firmware_image_get_iter with query + """ + firmware_image_get = netapp_utils.zapi.NaElement('service-processor-get-iter') + query = netapp_utils.zapi.NaElement('query') + firmware_image_info = netapp_utils.zapi.NaElement('service-processor-info') + firmware_image_info.add_new_child('node', self.parameters['node']) + query.add_child_elem(firmware_image_info) + firmware_image_get.add_child_elem(query) + return firmware_image_get + + def firmware_image_get(self, node_name): + """ + Get current firmware image info + :return: True if query successful, else return None + """ + firmware_image_get_iter = self.firmware_image_get_iter() + try: + result = self.server.invoke_successfully(firmware_image_get_iter, enable_tunneling=True) + except netapp_utils.zapi.NaApiError as error: + self.module.fail_json(msg='Error fetching firmware image details: %s: %s' + % (self.parameters['node'], to_native(error)), + exception=traceback.format_exc()) + # return firmware image details + if result.get_child_by_name('num-records') and int(result.get_child_content('num-records')) > 0: + sp_info = result.get_child_by_name('attributes-list').get_child_by_name('service-processor-info') + firmware_version = sp_info.get_child_content('firmware-version') + return firmware_version + return None + + def acp_firmware_required_get(self): + """ + where acp firmware upgrade is required + :return: True is firmware upgrade is required else return None + """ + acp_firmware_get_iter = netapp_utils.zapi.NaElement('storage-shelf-acp-module-get-iter') + query = netapp_utils.zapi.NaElement('query') + acp_info = netapp_utils.zapi.NaElement('storage-shelf-acp-module') + query.add_child_elem(acp_info) + acp_firmware_get_iter.add_child_elem(query) + try: + result = self.server.invoke_successfully(acp_firmware_get_iter, enable_tunneling=True) + except netapp_utils.zapi.NaApiError as error: + self.module.fail_json(msg='Error fetching acp firmware details details: %s' + % (to_native(error)), exception=traceback.format_exc()) + if result.get_child_by_name('attributes-list').get_child_by_name('storage-shelf-acp-module'): + acp_module_info = result.get_child_by_name('attributes-list').get_child_by_name( + 'storage-shelf-acp-module') + state = acp_module_info.get_child_content('state') + if state == 'firmware_update_required': + # acp firmware version upgrade required + return True + return False + + def sp_firmware_image_update_progress_get(self, node_name): + """ + Get current firmware image update progress info + :return: Dictionary of firmware image update progress if query successful, else return None + """ + firmware_update_progress_get = netapp_utils.zapi.NaElement('service-processor-image-update-progress-get') + firmware_update_progress_get.add_new_child('node', self.parameters['node']) + + firmware_update_progress_info = dict() + try: + result = self.server.invoke_successfully(firmware_update_progress_get, enable_tunneling=True) + except netapp_utils.zapi.NaApiError as error: + self.module.fail_json(msg='Error fetching firmware image upgrade progress details: %s' + % (to_native(error)), exception=traceback.format_exc()) + # return firmware image update progress details + if result.get_child_by_name('attributes').get_child_by_name('service-processor-image-update-progress-info'): + update_progress_info = result.get_child_by_name('attributes').get_child_by_name('service-processor-image-update-progress-info') + firmware_update_progress_info['is-in-progress'] = update_progress_info.get_child_content('is-in-progress') + firmware_update_progress_info['node'] = update_progress_info.get_child_content('node') + return firmware_update_progress_info + + def shelf_firmware_info_get(self): + """ + Get the current firmware of shelf module + :return:dict with module id and firmware info + """ + shelf_id_fw_info = dict() + shelf_firmware_info_get = netapp_utils.zapi.NaElement('storage-shelf-info-get-iter') + desired_attributes = netapp_utils.zapi.NaElement('desired-attributes') + storage_shelf_info = netapp_utils.zapi.NaElement('storage-shelf-info') + shelf_module = netapp_utils.zapi.NaElement('shelf-modules') + shelf_module_info = netapp_utils.zapi.NaElement('storage-shelf-module-info') + shelf_module.add_child_elem(shelf_module_info) + storage_shelf_info.add_child_elem(shelf_module) + desired_attributes.add_child_elem(storage_shelf_info) + shelf_firmware_info_get.add_child_elem(desired_attributes) + + try: + result = self.server.invoke_successfully(shelf_firmware_info_get, enable_tunneling=True) + except netapp_utils.zapi.NaApiError as error: + self.module.fail_json(msg='Error fetching shelf module firmware details: %s' + % (to_native(error)), exception=traceback.format_exc()) + if result.get_child_by_name('num-records') and int(result.get_child_content('num-records')) > 0: + shelf_info = result.get_child_by_name('attributes-list').get_child_by_name('storage-shelf-info') + if (shelf_info.get_child_by_name('shelf-modules') and + shelf_info.get_child_by_name('shelf-modules').get_child_by_name('storage-shelf-module-info')): + shelves = shelf_info['shelf-modules'].get_children() + for shelf in shelves: + shelf_id_fw_info[shelf.get_child_content('module-id')] = shelf.get_child_content('module-fw-revision') + return shelf_id_fw_info + + def disk_firmware_info_get(self): + """ + Get the current firmware of disks module + :return: + """ + disk_id_fw_info = dict() + disk_firmware_info_get = netapp_utils.zapi.NaElement('storage-disk-get-iter') + desired_attributes = netapp_utils.zapi.NaElement('desired-attributes') + storage_disk_info = netapp_utils.zapi.NaElement('storage-disk-info') + disk_inv = netapp_utils.zapi.NaElement('disk-inventory-info') + storage_disk_info.add_child_elem(disk_inv) + desired_attributes.add_child_elem(storage_disk_info) + disk_firmware_info_get.add_child_elem(desired_attributes) + try: + result = self.server.invoke_successfully(disk_firmware_info_get, enable_tunneling=True) + except netapp_utils.zapi.NaApiError as error: + self.module.fail_json(msg='Error fetching disk module firmware details: %s' + % (to_native(error)), exception=traceback.format_exc()) + if result.get_child_by_name('num-records') and int(result.get_child_content('num-records')) > 0: + disk_info = result.get_child_by_name('attributes-list') + disks = disk_info.get_children() + for disk in disks: + disk_id_fw_info[disk.get_child_content('disk-uid')] = disk.get_child_by_name('disk-inventory-info').get_child_content('firmware-revision') + return disk_id_fw_info + + def disk_firmware_required_get(self): + """ + Check weather disk firmware upgrade is required or not + :return: True if the firmware upgrade is required + """ + disk_firmware_info = self.disk_firmware_info_get() + for disk in disk_firmware_info: + if (disk_firmware_info[disk]) != self.parameters['disk_fw']: + return True + return False + + def shelf_firmware_required_get(self): + """ + Check weather shelf firmware upgrade is required or not + :return: True if the firmware upgrade is required + """ + shelf_firmware_info = self.shelf_firmware_info_get() + for module in shelf_firmware_info: + if (shelf_firmware_info[module]) != self.parameters['shelf_module_fw']: + return True + return False + + def sp_firmware_image_update(self): + """ + Update current firmware image + """ + firmware_update_info = netapp_utils.zapi.NaElement('service-processor-image-update') + if self.parameters.get('package') is not None: + firmware_update_info.add_new_child('package', self.parameters['package']) + if self.parameters.get('clear_logs') is not None: + firmware_update_info.add_new_child('clear-logs', str(self.parameters['clear_logs'])) + if self.parameters.get('install_baseline_image') is not None: + firmware_update_info.add_new_child('install-baseline-image', str(self.parameters['install_baseline_image'])) + firmware_update_info.add_new_child('node', self.parameters['node']) + firmware_update_info.add_new_child('update-type', self.parameters['update_type']) + + try: + self.server.invoke_successfully(firmware_update_info, enable_tunneling=True) + except netapp_utils.zapi.NaApiError as error: + # Current firmware version matches the version to be installed + if to_native(error.code) == '13001' and (error.message.startswith('Service Processor update skipped')): + return False + self.module.fail_json(msg='Error updating firmware image for %s: %s' + % (self.parameters['node'], to_native(error)), + exception=traceback.format_exc()) + return True + + def shelf_firmware_upgrade(self): + """ + Upgrade shelf firmware image + """ + shelf_firmware_update_info = netapp_utils.zapi.NaElement('storage-shelf-firmware-update') + try: + self.server.invoke_successfully(shelf_firmware_update_info, enable_tunneling=True) + return True + except netapp_utils.zapi.NaApiError as error: + self.module.fail_json(msg='Error updating shelf firmware image : %s' + % (to_native(error)), exception=traceback.format_exc()) + + def acp_firmware_upgrade(self): + + """ + Upgrade shelf firmware image + """ + acp_firmware_update_info = netapp_utils.zapi.NaElement('storage-shelf-acp-firmware-update') + acp_firmware_update_info.add_new_child('node-name', self.parameters['node']) + try: + self.server.invoke_successfully(acp_firmware_update_info, enable_tunneling=True) + except netapp_utils.zapi.NaApiError as error: + self.module.fail_json(msg='Error updating acp firmware image : %s' + % (to_native(error)), exception=traceback.format_exc()) + + def disk_firmware_upgrade(self): + + """ + Upgrade disk firmware + """ + disk_firmware_update_info = netapp_utils.zapi.NaElement('disk-update-disk-fw') + disk_firmware_update_info.add_new_child('node-name', self.parameters['node']) + try: + self.server.invoke_successfully(disk_firmware_update_info, enable_tunneling=True) + except netapp_utils.zapi.NaApiError as error: + self.module.fail_json(msg='Error updating disk firmware image : %s' + % (to_native(error)), exception=traceback.format_exc()) + return True + + def autosupport_log(self): + """ + Autosupport log for software_update + :return: + """ + results = netapp_utils.get_cserver(self.server) + cserver = netapp_utils.setup_na_ontap_zapi(module=self.module, vserver=results) + netapp_utils.ems_log_event("na_ontap_firmware_upgrade", cserver) + + def apply(self): + """ + Apply action to upgrade firmware + """ + changed = False + self.autosupport_log() + firmware_update_progress = dict() + if self.parameters.get('firmware_type') == 'service-processor': + # service-processor firmware upgrade + current = self.firmware_image_get(self.parameters['node']) + + if self.parameters.get('state') == 'present' and current: + if not self.module.check_mode: + if self.sp_firmware_image_update(): + changed = True + firmware_update_progress = self.sp_firmware_image_update_progress_get(self.parameters['node']) + while firmware_update_progress.get('is-in-progress') == 'true': + time.sleep(25) + firmware_update_progress = self.sp_firmware_image_update_progress_get(self.parameters['node']) + else: + # we don't know until we try the upgrade + changed = True + + elif self.parameters.get('firmware_type') == 'shelf': + # shelf firmware upgrade + if self.parameters.get('shelf_module_fw'): + if self.shelf_firmware_required_get(): + if not self.module.check_mode: + changed = self.shelf_firmware_upgrade() + else: + changed = True + else: + if not self.module.check_mode: + changed = self.shelf_firmware_upgrade() + else: + # we don't know until we try the upgrade -- assuming the worst + changed = True + elif self.parameters.get('firmware_type') == 'acp': + # acp firmware upgrade + if self.acp_firmware_required_get(): + if not self.module.check_mode: + self.acp_firmware_upgrade() + changed = True + elif self.parameters.get('firmware_type') == 'disk': + # Disk firmware upgrade + if self.parameters.get('disk_fw'): + if self.disk_firmware_required_get(): + if not self.module.check_mode: + changed = self.disk_firmware_upgrade() + else: + changed = True + else: + if not self.module.check_mode: + changed = self.disk_firmware_upgrade() + else: + # we don't know until we try the upgrade -- assuming the worst + changed = True + + self.module.exit_json(changed=changed) + + +def main(): + """Execute action""" + community_obj = NetAppONTAPFirmwareUpgrade() + community_obj.apply() + + +if __name__ == '__main__': + main() diff --git a/test/units/modules/storage/netapp/test_na_ontap_firmware_upgrade.py b/test/units/modules/storage/netapp/test_na_ontap_firmware_upgrade.py new file mode 100644 index 00000000000..5c2399f0ddf --- /dev/null +++ b/test/units/modules/storage/netapp/test_na_ontap_firmware_upgrade.py @@ -0,0 +1,291 @@ +# (c) 2018, NetApp, Inc +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +''' unit tests ONTAP Ansible module: na_ontap_firmware_upgrade ''' + +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type +import json +import pytest + +from units.compat import unittest +from units.compat.mock import patch, Mock +from ansible.module_utils import basic +from ansible.module_utils._text import to_bytes +import ansible.module_utils.netapp as netapp_utils + +from ansible.modules.storage.netapp.na_ontap_firmware_upgrade\ + import NetAppONTAPFirmwareUpgrade as my_module # module under test + +if not netapp_utils.has_netapp_lib(): + pytestmark = pytest.mark.skip('skipping as missing required netapp_lib') + + +def set_module_args(args): + """prepare arguments so that they will be picked up during module creation""" + args = json.dumps({'ANSIBLE_MODULE_ARGS': args}) + basic._ANSIBLE_ARGS = to_bytes(args) # pylint: disable=protected-access + + +class AnsibleExitJson(Exception): + """Exception class to be raised by module.exit_json and caught by the test case""" + pass + + +class AnsibleFailJson(Exception): + """Exception class to be raised by module.fail_json and caught by the test case""" + pass + + +def exit_json(*args, **kwargs): # pylint: disable=unused-argument + """function to patch over exit_json; package return data into an exception""" + if 'changed' not in kwargs: + kwargs['changed'] = False + raise AnsibleExitJson(kwargs) + + +def fail_json(*args, **kwargs): # pylint: disable=unused-argument + """function to patch over fail_json; package return data into an exception""" + kwargs['failed'] = True + raise AnsibleFailJson(kwargs) + + +class MockONTAPConnection(object): + ''' mock server connection to ONTAP host ''' + + def __init__(self, kind=None, parm1=None, parm2=None, parm3=None): + ''' save arguments ''' + self.type = kind + self.parm1 = parm1 + self.parm2 = parm2 + # self.parm3 = parm3 + self.xml_in = None + self.xml_out = None + self.firmware_type = 'None' + + def invoke_successfully(self, xml, enable_tunneling): # pylint: disable=unused-argument + ''' mock invoke_successfully returning xml data ''' + self.xml_in = xml + if self.type == 'firmware_upgrade': + xml = self.build_firmware_upgrade_info(self.parm1, self.parm2) + if self.type == 'acp': + xml = self.build_acp_firmware_info(self.firmware_type) + self.xml_out = xml + return xml + + def autosupport_log(self): + ''' mock autosupport log''' + return None + + @staticmethod + def build_firmware_upgrade_info(version, node): + ''' build xml data for service-processor firmware info ''' + xml = netapp_utils.zapi.NaElement('xml') + data = { + 'num-records': 1, + 'attributes-list': {'service-processor-info': {'firmware-version': '3.4'}} + } + xml.translate_struct(data) + print(xml.to_string()) + return xml + + @staticmethod + def build_acp_firmware_info(firmware_type): + ''' build xml data for acp firmware info ''' + xml = netapp_utils.zapi.NaElement('xml') + data = { + # 'num-records': 1, + 'attributes-list': {'storage-shelf-acp-module': {'state': 'firmware_update_required'}} + } + xml.translate_struct(data) + print(xml.to_string()) + return xml + + +class TestMyModule(unittest.TestCase): + ''' a group of related Unit Tests ''' + + def setUp(self): + self.mock_module_helper = patch.multiple(basic.AnsibleModule, + exit_json=exit_json, + fail_json=fail_json) + self.mock_module_helper.start() + self.addCleanup(self.mock_module_helper.stop) + self.server = MockONTAPConnection() + self.use_vsim = False + + def set_default_args(self): + if self.use_vsim: + hostname = '10.10.10.10' + username = 'admin' + password = 'admin' + node = 'vsim1' + clear_logs = True + package = 'test1.zip' + install_baseline_image = False + update_type = 'serial_full' + else: + hostname = 'hostname' + username = 'username' + password = 'password' + node = 'abc' + package = 'test1.zip' + clear_logs = True + install_baseline_image = False + update_type = 'serial_full' + + return dict({ + 'hostname': hostname, + 'username': username, + 'password': password, + 'node': node, + 'package': package, + 'clear_logs': clear_logs, + 'install_baseline_image': install_baseline_image, + 'update_type': update_type, + 'https': 'true' + }) + + def test_module_fail_when_required_args_missing(self): + ''' required arguments are reported as errors ''' + with pytest.raises(AnsibleFailJson) as exc: + set_module_args({}) + my_module() + print('Info: %s' % exc.value.args[0]['msg']) + + def test_missing_parameters(self): + ''' fail if firmware_type is missing ''' + module_args = {} + module_args.update(self.set_default_args()) + set_module_args(module_args) + with pytest.raises(AnsibleFailJson) as exc: + set_module_args(module_args) + my_module() + msg = 'missing required arguments: firmware_type' + print('Info: %s' % exc.value.args[0]['msg']) + assert exc.value.args[0]['msg'] == msg + + def test_invalid_firmware_type_parameters(self): + ''' fail if firmware_type is missing ''' + module_args = {} + module_args.update(self.set_default_args()) + module_args.update({'firmware_type': 'service_test'}) + set_module_args(module_args) + with pytest.raises(AnsibleFailJson) as exc: + set_module_args(module_args) + my_module() + msg = 'value of firmware_type must be one of: service-processor, shelf, acp, disk, got: %s' % module_args['firmware_type'] + print('Info: %s' % exc.value.args[0]['msg']) + assert exc.value.args[0]['msg'] == msg + + def test_ensure_sp_firmware_get_called(self): + ''' a more interesting test ''' + module_args = {} + module_args.update(self.set_default_args()) + module_args.update({'firmware_type': 'service-processor'}) + set_module_args(module_args) + my_obj = my_module() + my_obj.server = self.server + firmware_image_get = my_obj.firmware_image_get('node') + print('Info: test_firmware_upgrade_get: %s' % repr(firmware_image_get)) + assert firmware_image_get is None + + def test_ensure_firmware_get_with_package_baseline_called(self): + ''' a more interesting test ''' + module_args = {} + module_args.update(self.set_default_args()) + module_args.update({'firmware_type': 'service-processor'}) + module_args.update({'package': 'test1.zip'}) + module_args.update({'install_baseline_image': True}) + with pytest.raises(AnsibleFailJson) as exc: + set_module_args(module_args) + my_module() + msg = 'Do not specify both package and install_baseline_image: true' + print('info: ' + exc.value.args[0]['msg']) + assert exc.value.args[0]['msg'] == msg + + def test_ensure_acp_firmware_required_get_called(self): + ''' a test tp verify acp firmware upgrade is required or not ''' + module_args = {} + module_args.update(self.set_default_args()) + module_args.update({'firmware_type': 'acp'}) + set_module_args(module_args) + my_obj = my_module() + # my_obj.server = self.server + my_obj.server = MockONTAPConnection(kind='acp') + acp_firmware_required_get = my_obj.acp_firmware_required_get() + print('Info: test_acp_firmware_upgrade_required_get: %s' % repr(acp_firmware_required_get)) + assert acp_firmware_required_get is True + + @patch('ansible.modules.storage.netapp.na_ontap_firmware_upgrade.NetAppONTAPFirmwareUpgrade.sp_firmware_image_update') + @patch('ansible.modules.storage.netapp.na_ontap_firmware_upgrade.NetAppONTAPFirmwareUpgrade.sp_firmware_image_update_progress_get') + def test_ensure_apply_for_firmware_upgrade_called(self, get_mock, update_mock): + ''' updgrading firmware and checking idempotency ''' + module_args = {} + module_args.update(self.set_default_args()) + module_args.update({'package': 'test1.zip'}) + module_args.update({'firmware_type': 'service-processor'}) + set_module_args(module_args) + my_obj = my_module() + my_obj.autosupport_log = Mock(return_value=None) + if not self.use_vsim: + my_obj.server = self.server + with pytest.raises(AnsibleExitJson) as exc: + my_obj.apply() + print('Info: test_firmware_upgrade_apply: %s' % repr(exc.value)) + assert not exc.value.args[0]['changed'] + if not self.use_vsim: + my_obj.server = MockONTAPConnection('firmware_upgrade', '3.5', 'true') + with pytest.raises(AnsibleExitJson) as exc: + my_obj.apply() + print('Info: test_firmware_upgrade_apply: %s' % repr(exc.value)) + assert exc.value.args[0]['changed'] + update_mock.assert_called_with() + + def test_shelf_firmware_upgrade(self): + ''' Test shelf firmware upgrade ''' + module_args = {} + module_args.update(self.set_default_args()) + module_args.update({'firmware_type': 'shelf'}) + set_module_args(module_args) + my_obj = my_module() + my_obj.autosupport_log = Mock(return_value=None) + if not self.use_vsim: + my_obj.server = self.server + with pytest.raises(AnsibleExitJson) as exc: + my_obj.apply() + print('Info: test_firmware_upgrade_apply: %s' % repr(exc.value)) + assert exc.value.args[0]['changed'] + + @patch('ansible.modules.storage.netapp.na_ontap_firmware_upgrade.NetAppONTAPFirmwareUpgrade.acp_firmware_upgrade') + @patch('ansible.modules.storage.netapp.na_ontap_firmware_upgrade.NetAppONTAPFirmwareUpgrade.acp_firmware_required_get') + def test_acp_firmware_upgrade(self, get_mock, update_mock): + ''' Test ACP firmware upgrade ''' + module_args = {} + module_args.update(self.set_default_args()) + module_args.update({'firmware_type': 'acp'}) + set_module_args(module_args) + my_obj = my_module() + my_obj.autosupport_log = Mock(return_value=None) + if not self.use_vsim: + my_obj.server = self.server + with pytest.raises(AnsibleExitJson) as exc: + my_obj.apply() + print('Info: test_firmware_upgrade_apply: %s' % repr(exc.value)) + assert exc.value.args[0]['changed'] + + @patch('ansible.modules.storage.netapp.na_ontap_firmware_upgrade.NetAppONTAPFirmwareUpgrade.disk_firmware_upgrade') + def test_disk_firmware_upgrade(self, get_mock): + ''' Test disk firmware upgrade ''' + module_args = {} + module_args.update(self.set_default_args()) + module_args.update({'firmware_type': 'disk'}) + set_module_args(module_args) + my_obj = my_module() + my_obj.autosupport_log = Mock(return_value=None) + if not self.use_vsim: + my_obj.server = self.server + with pytest.raises(AnsibleExitJson) as exc: + my_obj.apply() + print('Info: test_firmware_upgrade_apply: %s' % repr(exc.value)) + assert exc.value.args[0]['changed']