archive
/
ansible
mirror of https://github.com/ansible/ansible.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Add netapp_e_firmware module (#59529)

Browse Source 
* Add na_santricity_firmware module.

Manages NetApp E-Series firmware upgrades.
Includes unit and integration tests.

* Add legacy support to na_santricity_firmware module.

* Rename na_santricity_firmware to netapp_e_firmware

* Improved netapp_e_firmware example documentation.
pull/61499/head
Nathan Swartz 5 years ago committed by Jake Jackson
parent f02a62db50
commit 1d9b473e56
5 changed files with 1407 additions and 0 deletions
Show Stats Download Patch File Download Diff File

488
lib/ansible/modules/storage/netapp/netapp_e_firmware.py
Unescape Escape View File

@ -0,0 +1,488 @@
#!/usr/bin/python
# (c) 2016, 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 = """
---
module: netapp_e_firmware
version_added: "2.9"
short_description: NetApp E-Series manage firmware.
description:
- Ensure specific firmware versions are activated on E-Series storage system.
author:
- Nathan Swartz (@ndswartz)
extends_documentation_fragment:
- netapp.eseries
options:
nvsram:
description:
- Path to the NVSRAM file.
type: str
required: true
firmware:
description:
- Path to the firmware file.
type: str
required: true
wait_for_completion:
description:
- This flag will cause module to wait for any upgrade actions to complete.
type: bool
default: false
ignore_health_check:
description:
- This flag will force firmware to be activated in spite of the health check.
- Use at your own risk. Certain non-optimal states could result in data loss.
type: bool
default: false
"""
EXAMPLES = """
- name: Ensure correct firmware versions
netapp_e_firmware:
ssid: "1"
api_url: "https://192.168.1.100:8443/devmgr/v2"
api_username: "admin"
api_password: "adminpass"
validate_certs: true
nvsram: "path/to/nvsram"
bundle: "path/to/bundle"
wait_for_completion: true
- name: Ensure correct firmware versions
netapp_e_firmware:
ssid: "1"
api_url: "https://192.168.1.100:8443/devmgr/v2"
api_username: "admin"
api_password: "adminpass"
validate_certs: true
nvsram: "path/to/nvsram"
firmware: "path/to/firmware"
"""
RETURN = """
msg:
description: Status and version of firmware and NVSRAM.
type: str
returned: always
sample:
"""
import os
from time import sleep
from ansible.module_utils import six
from ansible.module_utils.netapp import NetAppESeriesModule, create_multipart_formdata, request
from ansible.module_utils._text import to_native, to_text, to_bytes
class NetAppESeriesFirmware(NetAppESeriesModule):
HEALTH_CHECK_TIMEOUT_MS = 120000
REBOOT_TIMEOUT_SEC = 15 * 60
FIRMWARE_COMPATIBILITY_CHECK_TIMEOUT_SEC = 60
DEFAULT_TIMEOUT = 60 * 15 # This will override the NetAppESeriesModule request method timeout.
def __init__(self):
ansible_options = dict(
nvsram=dict(type="str", required=True),
firmware=dict(type="str", required=True),
wait_for_completion=dict(type="bool", default=False),
ignore_health_check=dict(type="bool", default=False))
super(NetAppESeriesFirmware, self).__init__(ansible_options=ansible_options,
web_services_version="02.00.0000.0000",
supports_check_mode=True)
args = self.module.params
self.nvsram = args["nvsram"]
self.firmware = args["firmware"]
self.wait_for_completion = args["wait_for_completion"]
self.ignore_health_check = args["ignore_health_check"]
self.nvsram_name = None
self.firmware_name = None
self.is_bundle_cache = None
self.firmware_version_cache = None
self.nvsram_version_cache = None
self.upgrade_required = False
self.upgrade_in_progress = False
self.module_info = dict()
self.nvsram_name = os.path.basename(self.nvsram)
self.firmware_name = os.path.basename(self.firmware)
def is_firmware_bundled(self):
"""Determine whether supplied firmware is bundle."""
if self.is_bundle_cache is None:
with open(self.firmware, "rb") as fh:
signature = fh.read(16).lower()
if b"firmware" in signature:
self.is_bundle_cache = False
elif b"combined_content" in signature:
self.is_bundle_cache = True
else:
self.module.fail_json(msg="Firmware file is invalid. File [%s]. Array [%s]" % (self.firmware, self.ssid))
return self.is_bundle_cache
def firmware_version(self):
"""Retrieve firmware version of the firmware file. Return: bytes string"""
if self.firmware_version_cache is None:
# Search firmware file for bundle or firmware version
with open(self.firmware, "rb") as fh:
line = fh.readline()
while line:
if self.is_firmware_bundled():
if b'displayableAttributeList=' in line:
for item in line[25:].split(b','):
key, value = item.split(b"|")
if key == b'VERSION':
self.firmware_version_cache = value.strip(b"\n")
break
elif b"Version:" in line:
self.firmware_version_cache = line.split()[-1].strip(b"\n")
break
line = fh.readline()
else:
self.module.fail_json(msg="Failed to determine firmware version. File [%s]. Array [%s]." % (self.firmware, self.ssid))
return self.firmware_version_cache
def nvsram_version(self):
"""Retrieve NVSRAM version of the NVSRAM file. Return: byte string"""
if self.nvsram_version_cache is None:
with open(self.nvsram, "rb") as fh:
line = fh.readline()
while line:
if b".NVSRAM Configuration Number" in line:
self.nvsram_version_cache = line.split(b'"')[-2]
break
line = fh.readline()
else:
self.module.fail_json(msg="Failed to determine NVSRAM file version. File [%s]. Array [%s]." % (self.nvsram, self.ssid))
return self.nvsram_version_cache
def check_system_health(self):
"""Ensure E-Series storage system is healthy. Works for both embedded and proxy web services."""
try:
rc, request_id = self.request("health-check", method="POST", data={"onlineOnly": True, "storageDeviceIds": [self.ssid]})
while True:
sleep(1)
try:
rc, response = self.request("health-check?requestId=%s" % request_id["requestId"])
if not response["healthCheckRunning"]:
return response["results"][0]["successful"]
elif int(response["results"][0]["processingTimeMS"]) > self.HEALTH_CHECK_TIMEOUT_MS:
self.module.fail_json(msg="Health check failed to complete. Array Id [%s]." % self.ssid)
except Exception as error:
self.module.fail_json(msg="Failed to retrieve health check status. Array Id [%s]. Error[%s]." % (self.ssid, to_native(error)))
except Exception as error:
self.module.fail_json(msg="Failed to initiate health check. Array Id [%s]. Error[%s]." % (self.ssid, to_native(error)))
self.module.fail_json(msg="Failed to retrieve health check status. Array Id [%s]. Error[%s]." % self.ssid)
def embedded_check_compatibility(self):
"""Verify files are compatible with E-Series storage system."""
self.embedded_check_nvsram_compatibility()
self.embedded_check_bundle_compatibility()
def embedded_check_nvsram_compatibility(self):
"""Verify the provided NVSRAM is compatible with E-Series storage system."""
# Check nvsram compatibility
try:
files = [("nvsramimage", self.nvsram_name, self.nvsram)]
headers, data = create_multipart_formdata(files=files)
rc, nvsram_compatible = self.request("firmware/embedded-firmware/%s/nvsram-compatibility-check" % self.ssid,
method="POST", data=data, headers=headers)
if not nvsram_compatible["signatureTestingPassed"]:
self.module.fail_json(msg="Invalid NVSRAM file. File [%s]." % self.nvsram)
if not nvsram_compatible["fileCompatible"]:
self.module.fail_json(msg="Incompatible NVSRAM file. File [%s]." % self.nvsram)
# Determine whether nvsram is required
for module in nvsram_compatible["versionContents"]:
if module["bundledVersion"] != module["onboardVersion"]:
self.upgrade_required = True
# Update bundle info
self.module_info.update({module["module"]: {"onboard_version": module["onboardVersion"], "bundled_version": module["bundledVersion"]}})
except Exception as error:
self.module.fail_json(msg="Failed to retrieve NVSRAM compatibility results. Array Id [%s]. Error[%s]." % (self.ssid, to_native(error)))
def embedded_check_bundle_compatibility(self):
"""Verify the provided firmware bundle is compatible with E-Series storage system."""
try:
files = [("files[]", "blob", self.firmware)]
headers, data = create_multipart_formdata(files=files, send_8kb=True)
rc, bundle_compatible = self.request("firmware/embedded-firmware/%s/bundle-compatibility-check" % self.ssid,
method="POST", data=data, headers=headers)
# Determine whether valid and compatible firmware
if not bundle_compatible["signatureTestingPassed"]:
self.module.fail_json(msg="Invalid firmware bundle file. File [%s]." % self.firmware)
if not bundle_compatible["fileCompatible"]:
self.module.fail_json(msg="Incompatible firmware bundle file. File [%s]." % self.firmware)
# Determine whether upgrade is required
for module in bundle_compatible["versionContents"]:
bundle_module_version = module["bundledVersion"].split(".")
onboard_module_version = module["onboardVersion"].split(".")
version_minimum_length = min(len(bundle_module_version), len(onboard_module_version))
if bundle_module_version[:version_minimum_length] != onboard_module_version[:version_minimum_length]:
self.upgrade_required = True
# Check whether downgrade is being attempted
bundle_version = module["bundledVersion"].split(".")[:2]
onboard_version = module["onboardVersion"].split(".")[:2]
if bundle_version[0] < onboard_version[0] or (bundle_version[0] == onboard_version[0] and bundle_version[1] < onboard_version[1]):
self.module.fail_json(msg="Downgrades are not permitted. onboard [%s] > bundled[%s]."
% (module["onboardVersion"], module["bundledVersion"]))
# Update bundle info
self.module_info.update({module["module"]: {"onboard_version": module["onboardVersion"], "bundled_version": module["bundledVersion"]}})
except Exception as error:
self.module.fail_json(msg="Failed to retrieve bundle compatibility results. Array Id [%s]. Error[%s]." % (self.ssid, to_native(error)))
def embedded_wait_for_upgrade(self):
"""Wait for SANtricity Web Services Embedded to be available after reboot."""
for count in range(0, self.REBOOT_TIMEOUT_SEC):
try:
rc, response = self.request("storage-systems/%s/graph/xpath-filter?query=/sa/saData" % self.ssid)
bundle_display = [m["versionString"] for m in response[0]["extendedSAData"]["codeVersions"] if m["codeModule"] == "bundleDisplay"][0]
if rc == 200 and six.b(bundle_display) == self.firmware_version() and six.b(response[0]["nvsramVersion"]) == self.nvsram_version():
self.upgrade_in_progress = False
break
except Exception as error:
pass
sleep(1)
else:
self.module.fail_json(msg="Timeout waiting for Santricity Web Services Embedded. Array [%s]" % self.ssid)
def embedded_upgrade(self):
"""Upload and activate both firmware and NVSRAM."""
files = [("nvsramfile", self.nvsram_name, self.nvsram),
("dlpfile", self.firmware_name, self.firmware)]
headers, data = create_multipart_formdata(files=files)
try:
rc, response = self.request("firmware/embedded-firmware?staged=false&nvsram=true", method="POST", data=data, headers=headers)
self.upgrade_in_progress = True
except Exception as error:
self.module.fail_json(msg="Failed to upload and activate firmware. Array Id [%s]. Error[%s]." % (self.ssid, to_native(error)))
if self.wait_for_completion:
self.embedded_wait_for_upgrade()
def proxy_check_nvsram_compatibility(self):
"""Verify nvsram is compatible with E-Series storage system."""
data = {"storageDeviceIds": [self.ssid]}
try:
rc, check = self.request("firmware/compatibility-check", method="POST", data=data)
for count in range(0, int((self.FIRMWARE_COMPATIBILITY_CHECK_TIMEOUT_SEC / 5))):
sleep(5)
try:
rc, response = self.request("firmware/compatibility-check?requestId=%s" % check["requestId"])
if not response["checkRunning"]:
for result in response["results"][0]["nvsramFiles"]:
if result["filename"] == self.nvsram_name:
return
self.module.fail_json(msg="NVSRAM is not compatible. NVSRAM [%s]. Array [%s]." % (self.nvsram_name, self.ssid))
except Exception as error:
self.module.fail_json(msg="Failed to retrieve NVSRAM status update from proxy. Array [%s]. Error [%s]." % (self.ssid, to_native(error)))
except Exception as error:
self.module.fail_json(msg="Failed to receive NVSRAM compatibility information. Array [%s]. Error [%s]." % (self.ssid, to_native(error)))
def proxy_check_firmware_compatibility(self):
"""Verify firmware is compatible with E-Series storage system."""
data = {"storageDeviceIds": [self.ssid]}
try:
rc, check = self.request("firmware/compatibility-check", method="POST", data=data)
for count in range(0, int((self.FIRMWARE_COMPATIBILITY_CHECK_TIMEOUT_SEC / 5))):
sleep(5)
try:
rc, response = self.request("firmware/compatibility-check?requestId=%s" % check["requestId"])
if not response["checkRunning"]:
for result in response["results"][0]["cfwFiles"]:
if result["filename"] == self.firmware_name:
return
self.module.fail_json(msg="Firmware bundle is not compatible. firmware [%s]. Array [%s]." % (self.firmware_name, self.ssid))
except Exception as error:
self.module.fail_json(msg="Failed to retrieve firmware status update from proxy. Array [%s]. Error [%s]." % (self.ssid, to_native(error)))
except Exception as error:
self.module.fail_json(msg="Failed to receive firmware compatibility information. Array [%s]. Error [%s]." % (self.ssid, to_native(error)))
def proxy_upload_and_check_compatibility(self):
"""Ensure firmware is uploaded and verify compatibility."""
try:
rc, cfw_files = self.request("firmware/cfw-files")
for file in cfw_files:
if file["filename"] == self.nvsram_name:
break
else:
fields = [("validate", "true")]
files = [("firmwareFile", self.nvsram_name, self.nvsram)]
headers, data = create_multipart_formdata(files=files, fields=fields)
try:
rc, response = self.request("firmware/upload", method="POST", data=data, headers=headers)
except Exception as error:
self.module.fail_json(msg="Failed to upload NVSRAM file. File [%s]. Array [%s]. Error [%s]."
% (self.nvsram_name, self.ssid, to_native(error)))
self.proxy_check_nvsram_compatibility()
for file in cfw_files:
if file["filename"] == self.firmware_name:
break
else:
fields = [("validate", "true")]
files = [("firmwareFile", self.firmware_name, self.firmware)]
headers, data = create_multipart_formdata(files=files, fields=fields)
try:
rc, response = self.request("firmware/upload", method="POST", data=data, headers=headers)
except Exception as error:
self.module.fail_json(msg="Failed to upload firmware bundle file. File [%s]. Array [%s]. Error [%s]."
% (self.firmware_name, self.ssid, to_native(error)))
self.proxy_check_firmware_compatibility()
except Exception as error:
self.module.fail_json(msg="Failed to retrieve existing existing firmware files. Error [%s]" % to_native(error))
def proxy_check_upgrade_required(self):
"""Staging is required to collect firmware information from the web services proxy."""
# Verify controller consistency and get firmware versions
try:
# Retrieve current bundle version
if self.is_firmware_bundled():
rc, response = self.request("storage-systems/%s/graph/xpath-filter?query=/controller/codeVersions[codeModule='bundleDisplay']" % self.ssid)
current_firmware_version = six.b(response[0]["versionString"])
else:
rc, response = self.request("storage-systems/%s/graph/xpath-filter?query=/sa/saData/fwVersion" % self.ssid)
current_firmware_version = six.b(response[0])
# Determine whether upgrade is required
if current_firmware_version != self.firmware_version():
current = current_firmware_version.split(b".")[:2]
upgrade = self.firmware_version().split(b".")[:2]
if current[0] < upgrade[0] or (current[0] == upgrade[0] and current[1] <= upgrade[1]):
self.upgrade_required = True
else:
self.module.fail_json(msg="Downgrades are not permitted. Firmware [%s]. Array [%s]." % (self.firmware, self.ssid))
except Exception as error:
self.module.fail_json(msg="Failed to retrieve controller firmware information. Array [%s]. Error [%s]" % (self.ssid, to_native(error)))
# Determine current NVSRAM version and whether change is required
try:
rc, response = self.request("storage-systems/%s/graph/xpath-filter?query=/sa/saData/nvsramVersion" % self.ssid)
if six.b(response[0]) != self.nvsram_version():
self.upgrade_required = True
except Exception as error:
self.module.fail_json(msg="Failed to retrieve storage system's NVSRAM version. Array [%s]. Error [%s]" % (self.ssid, to_native(error)))
def proxy_wait_for_upgrade(self, request_id):
"""Wait for SANtricity Web Services Proxy to report upgrade complete"""
if self.is_firmware_bundled():
while True:
try:
sleep(5)
rc, response = self.request("batch/cfw-upgrade/%s" % request_id)
if response["status"] == "complete":
self.upgrade_in_progress = False
break
elif response["status"] in ["failed", "cancelled"]:
self.module.fail_json(msg="Firmware upgrade failed to complete. Array [%s]." % self.ssid)
except Exception as error:
self.module.fail_json(msg="Failed to retrieve firmware upgrade status. Array [%s]. Error [%s]." % (self.ssid, to_native(error)))
else:
for count in range(0, int(self.REBOOT_TIMEOUT_SEC / 5)):
try:
sleep(5)
rc_firmware, firmware = self.request("storage-systems/%s/graph/xpath-filter?query=/sa/saData/fwVersion" % self.ssid)
rc_nvsram, nvsram = self.request("storage-systems/%s/graph/xpath-filter?query=/sa/saData/nvsramVersion" % self.ssid)
if six.b(firmware[0]) == self.firmware_version() and six.b(nvsram[0]) == self.nvsram_version():
self.upgrade_in_progress = False
break
except Exception as error:
pass
else:
self.module.fail_json(msg="Timed out waiting for firmware upgrade to complete. Array [%s]." % self.ssid)
def proxy_upgrade(self):
"""Activate previously uploaded firmware related files."""
request_id = None
if self.is_firmware_bundled():
data = {"activate": True,
"firmwareFile": self.firmware_name,
"nvsramFile": self.nvsram_name,
"systemInfos": [{"systemId": self.ssid,
"allowNonOptimalActivation": self.ignore_health_check}]}
try:
rc, response = self.request("batch/cfw-upgrade", method="POST", data=data)
request_id = response["requestId"]
except Exception as error:
self.module.fail_json(msg="Failed to initiate firmware upgrade. Array [%s]. Error [%s]." % (self.ssid, to_native(error)))
else:
data = {"stageFirmware": False,
"skipMelCheck": self.ignore_health_check,
"cfwFile": self.firmware_name,
"nvsramFile": self.nvsram_name}
try:
rc, response = self.request("storage-systems/%s/cfw-upgrade" % self.ssid, method="POST", data=data)
request_id = response["requestId"]
except Exception as error:
self.module.fail_json(msg="Failed to initiate firmware upgrade. Array [%s]. Error [%s]." % (self.ssid, to_native(error)))
self.upgrade_in_progress = True
if self.wait_for_completion:
self.proxy_wait_for_upgrade(request_id)
def apply(self):
"""Upgrade controller firmware."""
self.check_system_health()
# Verify firmware compatibility and whether changes are required
if self.is_embedded():
self.embedded_check_compatibility()
else:
self.proxy_check_upgrade_required()
# This will upload the firmware files to the web services proxy but not to the controller
if self.upgrade_required:
self.proxy_upload_and_check_compatibility()
# Perform upgrade
if self.upgrade_required and not self.module.check_mode:
if self.is_embedded():
self.embedded_upgrade()
else:
self.proxy_upgrade()
self.module.exit_json(changed=self.upgrade_required, upgrade_in_process=self.upgrade_in_progress, status=self.module_info)
def main():
firmware = NetAppESeriesFirmware()
firmware.apply()
if __name__ == '__main__':
main()

15
test/integration/targets/netapp_eseries_firmware/aliases
Unescape Escape View File

@ -0,0 +1,15 @@
# This test is not enabled by default, but can be utilized by defining required variables in integration_config.yml
# Example integration_config.yml:
# ---
netapp_e_embedded_api_host: 192.168.1.1
netapp_e_embedded_api_username: admin
netapp_e_embedded_api_password: adminPass
netapp_e_embedded_ssid: 1
netapp_e_proxy_api_host: 192.168.1.100
netapp_e_proxy_api_username: admin
netapp_e_proxy_api_password: adminPass
netapp_e_proxy_ssid: 10
unsupported
netapp/eseries

1
test/integration/targets/netapp_eseries_firmware/tasks/main.yml
Unescape Escape View File

@ -0,0 +1 @@
- include_tasks: run.yml

348
test/integration/targets/netapp_eseries_firmware/tasks/run.yml
Unescape Escape View File

@ -0,0 +1,348 @@
# Test code for the netapp_e_firmware module
# (c) 2018, NetApp, Inc
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
# TODO: MUST BE DOWNGRADE BEFORE EXECUTING INTEGRATION TO RCB_11.40.3R2_280x_5c7d81b3.dlp and N280X-842834-D02.dlp
# loadControllerFirmware_MT swartzn@10.113.1.250 /home/swartzn/Downloads/RCB_11.40.3R2_280x_5c7d81b3.dlp /home/swartzn/Downloads/N280X-842834-D02.dlp
# This integration test will validate upgrade functionality for firmware-only, firmware-and-nvsram, and check mode.
- name: NetApp Test ASUP module
fail:
msg: "Please define netapp_e_embedded_api_host, netapp_e_embedded_api_username, netapp_e_embedded_api_password, netapp_e_embedded_ssid,
netapp_e_proxy_api_host, netapp_e_proxy_api_username, netapp_e_proxy_api_password, and netapp_e_proxy_ssid."
when: "netapp_e_embedded_api_host is undefined or netapp_e_embedded_api_username is undefined or netapp_e_embedded_api_password is undefined or
netapp_e_embedded_ssid is undefined or netapp_e_proxy_api_host is undefined or netapp_e_proxy_api_username is undefined or
netapp_e_proxy_api_password is undefined or netapp_e_proxy_ssid is undefined"
- set_fact:
path: "/home/swartzn/Downloads/"
upgrades:
- firmware: "RCB_11.40.3R2_280x_5c7d81b3.dlp"
nvsram: "N280X-842834-D02.dlp"
expected_firmware_version: "08.42.30.05"
expected_nvsram_version: "N280X-842834-D02"
- firmware: "RCB_11.40.5_280x_5ceef00e.dlp"
nvsram: "N280X-842834-D02.dlp"
expected_firmware_version: "08.42.50.00"
expected_nvsram_version: "N280X-842834-D02"
- firmware: "RCB_11.50.2_280x_5ce8501f.dlp"
nvsram: "N280X-852834-D02.dlp"
expected_firmware_version: "08.52.00.00"
expected_nvsram_version: "N280X-852834-D02"
- name: Perform firmware upgrade using the Web Services REST API (checkmode-no change, firmware only)
netapp_e_firmware:
ssid: "{{ netapp_e_embedded_ssid }}"
api_url: "https://{{ netapp_e_embedded_api_host }}:8443/devmgr/v2"
api_username: "{{ netapp_e_embedded_api_username }}"
api_password: "{{ netapp_e_embedded_api_password }}"
validate_certs: no
nvsram: "{{ path }}{{ upgrades[0]['nvsram'] }}"
firmware: "{{ path }}{{ upgrades[0]['firmware'] }}"
wait_for_completion: true
ignore_health_check: true
check_mode: true
register: netapp_e_firmware
- name: Retrieve current firmware version
uri:
url: "https://{{ netapp_e_embedded_api_host }}:8443/devmgr/v2/storage-systems/{{ netapp_e_embedded_ssid }}/graph/xpath-filter?query=/sa/saData/fwVersion"
user: "{{ netapp_e_embedded_api_username }}"
password: "{{ netapp_e_embedded_api_password }}"
validate_certs: no
register: current_firmware
- name: Retrieve current nvsram version
uri:
url: "https://{{ netapp_e_embedded_api_host }}:8443/devmgr/v2/storage-systems/{{ netapp_e_embedded_ssid }}/graph/xpath-filter?query=/sa/saData/nvsramVersion"
user: "{{ netapp_e_embedded_api_username }}"
password: "{{ netapp_e_embedded_api_password }}"
validate_certs: no
register: current_nvsram
- name: Verify change status
assert:
that: "{{ netapp_e_firmware.changed == False }}"
msg: "Failed to return unchanged."
- name: Verify current firmware version
assert:
that: "{{ current_firmware['json'][0] == upgrades[0]['expected_firmware_version'] }}"
msg: "Unexpected firmware version."
- name: Verify current nvsram version
assert:
that: "{{ current_nvsram['json'][0] == upgrades[0]['expected_nvsram_version'] }}"
msg: "Unexpected nvsram version."
- name: Perform firmware upgrade using the Web Services REST API (no change, firmware only)
netapp_e_firmware:
ssid: "{{ netapp_e_embedded_ssid }}"
api_url: "https://{{ netapp_e_embedded_api_host }}:8443/devmgr/v2"
api_username: "{{ netapp_e_embedded_api_username }}"
api_password: "{{ netapp_e_embedded_api_password }}"
validate_certs: no
nvsram: "{{ path }}{{ upgrades[0]['nvsram'] }}"
firmware: "{{ path }}{{ upgrades[0]['firmware'] }}"
wait_for_completion: true
ignore_health_check: true
register: netapp_e_firmware
- name: Retrieve current firmware version
uri:
url: "https://{{ netapp_e_embedded_api_host }}:8443/devmgr/v2/storage-systems/{{ netapp_e_embedded_ssid }}/graph/xpath-filter?query=/sa/saData/fwVersion"
user: "{{ netapp_e_embedded_api_username }}"
password: "{{ netapp_e_embedded_api_password }}"
validate_certs: no
register: current_firmware
- name: Retrieve current nvsram version
uri:
url: "https://{{ netapp_e_embedded_api_host }}:8443/devmgr/v2/storage-systems/{{ netapp_e_embedded_ssid }}/graph/xpath-filter?query=/sa/saData/nvsramVersion"
user: "{{ netapp_e_embedded_api_username }}"
password: "{{ netapp_e_embedded_api_password }}"
validate_certs: no
register: current_nvsram
- name: Verify change status
assert:
that: "{{ netapp_e_firmware.changed == False }}"
msg: "Failed to return changed."
- name: Verify current firmware version
assert:
that: "{{ current_firmware['json'][0] == upgrades[0]['expected_firmware_version'] }}"
msg: "Unexpected firmware version."
- name: Verify current nvsram version
assert:
that: "{{ current_nvsram['json'][0] == upgrades[0]['expected_nvsram_version'] }}"
msg: "Unexpected nvsram version."
- name: Perform firmware upgrade using the Web Services REST API (checkmode-change, firmware)
netapp_e_firmware:
ssid: "{{ netapp_e_embedded_ssid }}"
api_url: "https://{{ netapp_e_embedded_api_host }}:8443/devmgr/v2"
api_username: "{{ netapp_e_embedded_api_username }}"
api_password: "{{ netapp_e_embedded_api_password }}"
validate_certs: no
nvsram: "{{ path }}{{ upgrades[1]['nvsram'] }}"
firmware: "{{ path }}{{ upgrades[1]['firmware'] }}"
wait_for_completion: true
ignore_health_check: true
register: netapp_e_firmware
check_mode: true
- name: Retrieve current firmware version
uri:
url: "https://{{ netapp_e_embedded_api_host }}:8443/devmgr/v2/storage-systems/{{ netapp_e_embedded_ssid }}/graph/xpath-filter?query=/sa/saData/fwVersion"
user: "{{ netapp_e_embedded_api_username }}"
password: "{{ netapp_e_embedded_api_password }}"
validate_certs: no
register: current_firmware
- name: Retrieve current nvsram version
uri:
url: "https://{{ netapp_e_embedded_api_host }}:8443/devmgr/v2/storage-systems/{{ netapp_e_embedded_ssid }}/graph/xpath-filter?query=/sa/saData/nvsramVersion"
user: "{{ netapp_e_embedded_api_username }}"
password: "{{ netapp_e_embedded_api_password }}"
validate_certs: no
register: current_nvsram
- name: Verify change status
assert:
that: "{{ netapp_e_firmware.changed == True }}"
msg: "Failed to return changed."
- name: Verify current firmware version
assert:
that: "{{ current_firmware['json'][0] == upgrades[0]['expected_firmware_version'] }}"
msg: "Unexpected firmware version."
- name: Verify current nvsram version
assert:
that: "{{ current_nvsram['json'][0] == upgrades[0]['expected_nvsram_version'] }}"
msg: "Unexpected nvsram version."
- name: Perform firmware upgrade using the Web Services REST API (change, firmware)
netapp_e_firmware:
ssid: "{{ netapp_e_embedded_ssid }}"
api_url: "https://{{ netapp_e_embedded_api_host }}:8443/devmgr/v2"
api_username: "{{ netapp_e_embedded_api_username }}"
api_password: "{{ netapp_e_embedded_api_password }}"
validate_certs: no
nvsram: "{{ path }}{{ upgrades[1]['nvsram'] }}"
firmware: "{{ path }}{{ upgrades[1]['firmware'] }}"
wait_for_completion: true
ignore_health_check: true
register: netapp_e_firmware
- name: Retrieve current firmware version
uri:
url: "https://{{ netapp_e_embedded_api_host }}:8443/devmgr/v2/storage-systems/{{ netapp_e_embedded_ssid }}/graph/xpath-filter?query=/sa/saData/fwVersion"
user: "{{ netapp_e_embedded_api_username }}"
password: "{{ netapp_e_embedded_api_password }}"
validate_certs: no
register: current_firmware
- name: Retrieve current nvsram version
uri:
url: "https://{{ netapp_e_embedded_api_host }}:8443/devmgr/v2/storage-systems/{{ netapp_e_embedded_ssid }}/graph/xpath-filter?query=/sa/saData/nvsramVersion"
user: "{{ netapp_e_embedded_api_username }}"
password: "{{ netapp_e_embedded_api_password }}"
validate_certs: no
register: current_nvsram
- name: Verify change status
assert:
that: "{{ netapp_e_firmware.changed == True }}"
msg: "Failed to return changed."
- name: Verify current firmware version
assert:
that: "{{ current_firmware['json'][0] == upgrades[1]['expected_firmware_version'] }}"
msg: "Unexpected firmware version. {{ current_firmware['json'][0] }} != {{ upgrades[1]['expected_firmware_version'] }}"
- name: Verify current nvsram version
assert:
that: "{{ current_nvsram['json'][0] == upgrades[1]['expected_nvsram_version'] }}"
msg: "Unexpected nvsram version. {{ current_nvsram['json'][0] }} != {{ upgrades[1]['expected_nvsram_version'] }}"
- name: Perform firmware upgrade using the Web Services REST API (changed, firmware)
netapp_e_firmware:
ssid: "{{ netapp_e_proxy_ssid }}"
api_url: "https://{{ netapp_e_proxy_api_host }}:8443/devmgr/v2"
api_username: "{{ netapp_e_proxy_api_username }}"
api_password: "{{ netapp_e_proxy_api_password }}"
validate_certs: no
nvsram: "{{ path }}{{ upgrades[0]['nvsram'] }}"
firmware: "{{ path }}{{ upgrades[0]['firmware'] }}"
wait_for_completion: true
ignore_health_check: true
register: netapp_e_firmware
- name: Retrieve current firmware version
uri:
url: "https://{{ netapp_e_proxy_api_host }}:8443/devmgr/v2/storage-systems/{{ netapp_e_proxy_ssid }}/graph/xpath-filter?query=/sa/saData/fwVersion"
user: "{{ netapp_e_proxy_api_username }}"
password: "{{ netapp_e_proxy_api_password }}"
validate_certs: no
register: current_firmware
- name: Retrieve current nvsram version
uri:
url: "https://{{ netapp_e_proxy_api_host }}:8443/devmgr/v2/storage-systems/{{ netapp_e_proxy_ssid }}/graph/xpath-filter?query=/sa/saData/nvsramVersion"
user: "{{ netapp_e_proxy_api_username }}"
password: "{{ netapp_e_proxy_api_password }}"
validate_certs: no
register: current_nvsram
- name: Verify change status
assert:
that: "{{ netapp_e_firmware.changed == True }}"
msg: "Failed to return changed."
- name: Verify current firmware version
assert:
that: "{{ current_firmware['json'][0] == upgrades[0]['expected_firmware_version'] }}"
msg: "Failed to change the firmware version."
- name: Verify current nvsram version
assert:
that: "{{ current_nvsram['json'][0] == upgrades[0]['expected_nvsram_version'] }}"
msg: "Failed to change the nvsram version."
- name: Perform firmware upgrade using the Web Services REST API (checkmode-unchanged, firmware)
netapp_e_firmware:
ssid: "{{ netapp_e_proxy_ssid }}"
api_url: "https://{{ netapp_e_proxy_api_host }}:8443/devmgr/v2"
api_username: "{{ netapp_e_proxy_api_username }}"
api_password: "{{ netapp_e_proxy_api_password }}"
validate_certs: no
nvsram: "{{ path }}{{ upgrades[0]['nvsram'] }}"
firmware: "{{ path }}{{ upgrades[0]['firmware'] }}"
wait_for_completion: true
ignore_health_check: true
check_mode: true
register: netapp_e_firmware
- name: Retrieve current firmware version
uri:
url: "https://{{ netapp_e_proxy_api_host }}:8443/devmgr/v2/storage-systems/{{ netapp_e_proxy_ssid }}/graph/xpath-filter?query=/sa/saData/fwVersion"
user: "{{ netapp_e_proxy_api_username }}"
password: "{{ netapp_e_proxy_api_password }}"
validate_certs: no
register: current_firmware
- name: Retrieve current nvsram version
uri:
url: "https://{{ netapp_e_proxy_api_host }}:8443/devmgr/v2/storage-systems/{{ netapp_e_proxy_ssid }}/graph/xpath-filter?query=/sa/saData/nvsramVersion"
user: "{{ netapp_e_proxy_api_username }}"
password: "{{ netapp_e_proxy_api_password }}"
validate_certs: no
register: current_nvsram
- name: Verify change status
assert:
that: "{{ netapp_e_firmware.changed == False }}"
msg: "Failed to return unchanged."
- name: Verify current firmware version
assert:
that: "{{ current_firmware['json'][0] == upgrades[0]['expected_firmware_version'] }}"
msg: "Failed to change the firmware version."
- name: Verify current nvsram version
assert:
that: "{{ current_nvsram['json'][0] == upgrades[0]['expected_nvsram_version'] }}"
msg: "Failed to change the nvsram version."
- name: Perform firmware upgrade using the Web Services REST API (checkmode-change, firmware and nvsram)
netapp_e_firmware:
ssid: "{{ netapp_e_proxy_ssid }}"
api_url: "https://{{ netapp_e_proxy_api_host }}:8443/devmgr/v2"
api_username: "{{ netapp_e_proxy_api_username }}"
api_password: "{{ netapp_e_proxy_api_password }}"
validate_certs: no
nvsram: "{{ path }}{{ upgrades[2]['nvsram'] }}"
firmware: "{{ path }}{{ upgrades[2]['firmware'] }}"
wait_for_completion: true
ignore_health_check: true
check_mode: true
register: netapp_e_firmware
- name: Retrieve current firmware version
uri:
url: "https://{{ netapp_e_proxy_api_host }}:8443/devmgr/v2/storage-systems/{{ netapp_e_proxy_ssid }}/graph/xpath-filter?query=/sa/saData/fwVersion"
user: "{{ netapp_e_proxy_api_username }}"
password: "{{ netapp_e_proxy_api_password }}"
validate_certs: no
register: current_firmware
- name: Retrieve current nvsram version
uri:
url: "https://{{ netapp_e_proxy_api_host }}:8443/devmgr/v2/storage-systems/{{ netapp_e_proxy_ssid }}/graph/xpath-filter?query=/sa/saData/nvsramVersion"
user: "{{ netapp_e_proxy_api_username }}"
password: "{{ netapp_e_proxy_api_password }}"
validate_certs: no
register: current_nvsram
- name: Verify change status
assert:
that: "{{ netapp_e_firmware.changed == True }}"
msg: "Failed to return changed."
- name: Verify current firmware version
assert:
that: "{{ current_firmware['json'][0] == upgrades[0]['expected_firmware_version'] }}"
msg: "Failed to change the firmware version."
- name: Verify current nvsram version
assert:
that: "{{ current_nvsram['json'][0] == upgrades[0]['expected_nvsram_version'] }}"
msg: "Failed to change the nvsram version."
- name: Perform firmware upgrade using the Web Services REST API (changed, firmware and nvsram)
netapp_e_firmware:
ssid: "{{ netapp_e_proxy_ssid }}"
api_url: "https://{{ netapp_e_proxy_api_host }}:8443/devmgr/v2"
api_username: "{{ netapp_e_proxy_api_username }}"
api_password: "{{ netapp_e_proxy_api_password }}"
validate_certs: no
nvsram: "{{ path }}{{ upgrades[2]['nvsram'] }}"
firmware: "{{ path }}{{ upgrades[2]['firmware'] }}"
wait_for_completion: true
ignore_health_check: true
register: netapp_e_firmware
- name: Retrieve current firmware version
uri:
url: "https://{{ netapp_e_proxy_api_host }}:8443/devmgr/v2/storage-systems/{{ netapp_e_proxy_ssid }}/graph/xpath-filter?query=/sa/saData/fwVersion"
user: "{{ netapp_e_proxy_api_username }}"
password: "{{ netapp_e_proxy_api_password }}"
validate_certs: no
register: current_firmware
- name: Retrieve current nvsram version
uri:
url: "https://{{ netapp_e_proxy_api_host }}:8443/devmgr/v2/storage-systems/{{ netapp_e_proxy_ssid }}/graph/xpath-filter?query=/sa/saData/nvsramVersion"
user: "{{ netapp_e_proxy_api_username }}"
password: "{{ netapp_e_proxy_api_password }}"
validate_certs: no
register: current_nvsram
- name: Verify change status
assert:
that: "{{ netapp_e_firmware.changed == True }}"
msg: "Failed to return changed."
- name: Verify current firmware version
assert:
that: "{{ current_firmware['json'][0] == upgrades[2]['expected_firmware_version'] }}"
msg: "Failed to change the firmware version."
- name: Verify current nvsram version
assert:
that: "{{ current_nvsram['json'][0] == upgrades[2]['expected_nvsram_version'] }}"
msg: "Failed to change the nvsram version."

555
test/units/modules/storage/netapp/test_netapp_e_firmware.py
Unescape Escape View File

@ -0,0 +1,555 @@
# (c) 2018, 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
try:
from unittest.mock import patch, mock_open
except ImportError:
from mock import patch, mock_open
from ansible.module_utils import six
from ansible.modules.storage.netapp.netapp_e_firmware import NetAppESeriesFirmware
from units.modules.utils import AnsibleExitJson, AnsibleFailJson, ModuleTestCase, set_module_args
if six.PY2:
builtin_path = "__builtin__.open"
else:
builtin_path = "builtins.open"
def mock_open_with_iter(*args, **kwargs):
mock = mock_open(*args, **kwargs)
if six.PY2:
mock.return_value.__iter__ = lambda x: iter(x.readline, "")
else:
mock.return_value.__iter__ = lambda x: x
mock.return_value.__next__ = lambda x: iter(x.readline, "")
return mock
class FirmwareTest(ModuleTestCase):
REQUIRED_PARAMS = {"api_username": "username",
"api_password": "password",
"api_url": "http://localhost/devmgr/v2",
"ssid": "1",
"validate_certs": "no"}
REQUEST_FUNC = "ansible.modules.storage.netapp.netapp_e_firmware.NetAppESeriesFirmware.request"
BASE_REQUEST_FUNC = "ansible.modules.storage.netapp.netapp_e_firmware.request"
CREATE_MULTIPART_FORMDATA_FUNC = "ansible.modules.storage.netapp.netapp_e_firmware.create_multipart_formdata"
SLEEP_FUNC = "ansible.modules.storage.netapp.netapp_e_firmware.sleep"
BUNDLE_HEADER = b'combined_content\x00\x00\x00\x04\x00\x00\x07\xf8#Engenio Downloadable Package\n#Tue Jun 04 11:46:48 CDT 2019\ncheckList=compatibleBoard' \
b'Map,compatibleSubmodelMap,compatibleFirmwareMap,fileManifest\ncompatibleSubmodelMap=261|true,262|true,263|true,264|true,276|true,277|t' \
b'rue,278|true,282|true,300|true,301|true,302|true,318|true,319|true,320|true,321|true,322|true,323|true,324|true,325|true,326|true,328|t' \
b'rue,329|true,330|true,331|true,332|true,333|true,338|true,339|true,340|true,341|true,342|true,343|true,344|true,345|true,346|true,347|t' \
b'rue,356|true,357|true,390|true\nnonDisplayableAttributeList=512\ndisplayableAttributeList=FILENAME|RCB_11.40.5_280x_5ceef00e.dlp,VERSI' \
b'ON|11.40.5\ndacStoreLimit=512\nfileManifest=metadata.tar|metadata|08.42.50.00.000|c04275f98fc2f07bd63126fc57cb0569|bundle|10240,084250' \
b'00_m3_e30_842_root.img|linux|08.42.50.00|367c5216e5c4b15b904a025bff69f039|linux|1342177280,RC_08425000_m3_e30_842_280x.img|linux_cfw|0' \
b'8.42.50.00|e6589b0a50b29ff34b34d3ced8ae3ccb|eos|1073741824,msw.img|sam|11.42.0000.0028|ef3ee5589ab4a019a3e6f83768364aa1|linux|41943040' \
b'0,iom.img|iom|11.42.0G00.0003|9bb740f8d3a4e62a0f2da2ec83c254c4|linux|8177664\nmanagementVersionList=devmgr.v1142api8.Manager\ncompatib' \
b'leFirmwareMap=08.30.*.*|true,08.30.*.30|false,08.30.*.31|false,08.30.*.32|false,08.30.*.33|false,08.30.*.34|false,08.30.*.35|false,08.' \
b'30.*.36|false,08.30.*.37|false,08.30.*.38|false,08.30.*.39|false,08.40.*.*|true,08.40.*.30|false,08.40.*.31|false,08.40.*.32|false,08.4' \
b'0.*.33|false,08.40.*.34|false,08.40.*.35|false,08.40.*.36|false,08.40.*.37|false,08.40.*.38|false,08.40.*.39|false,08.41.*.*|true,08.4' \
b'1.*.30|false,08.41.*.31|false,08.41.*.32|false,08.41.*.33|false,08.41.*.34|false,08.41.*.35|false,08.41.*.36|false,08.41.*.37|false,08' \
b'.41.*.38|false,08.41.*.39|false,08.42.*.*|true,08.42.*.30|false,08.42.*.31|false,08.42.*.32|false,08.42.*.33|false,08.42.*.34|false,08' \
b'.42.*.35|false,08.42.*.36|false,08.42.*.37|false,08.42.*.38|false,08.42.*.39|false\nversion=08.42.50.00.000\ntype=tar\nversionTag=comb' \
b'ined_content\n'
NVSRAM_HEADER = b'nvsram \x00\x00\x00\x01\x00\x00\x00\xa0\x00\x00\x00\x04280X\x00\x00\x00\x00\x00\x00\x00\x032801 2804 2806 \x00\x00' \
b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\x00\x00\x00\x1bArapaho controller, 8.52 FW\x00\x00\x001dual controller configuration, with cac' \
b'he battery\x07\x81A\x08Config\x00\x00\x0008.52.00.00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\x00\x00\x00\x00\x00\xdc\xaf\x00\x00' \
b'\x94\xc1\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xff\x00\x00\x00\x00 2801 2804 2806 \x00\x00\x00\x00\x00' \
b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00' \
b'\x00\x00\x00\x00\x00\x00Board\n .Board Name = "NetApp RAID Controller"\n .NVSRAM Configuration Number' \
b' = "N280X-852834-D02"\n\nUserCfg\n .Enable Synchronous Negotiation = 0x00 \n'
def _set_args(self, args=None):
module_args = self.REQUIRED_PARAMS.copy()
if args is not None:
module_args.update(args)
set_module_args(module_args)
def test_is_firmware_bundled_pass(self):
"""Determine whether firmware file is bundled."""
self._set_args({"firmware": "test.dlp", "nvsram": "test.dlp"})
with patch(builtin_path, mock_open(read_data=b"firmwarexxxxxxxx")) as mock_file:
firmware = NetAppESeriesFirmware()
self.assertEqual(firmware.is_firmware_bundled(), False)
self._set_args({"firmware": "test.dlp", "nvsram": "test.dlp"})
with patch(builtin_path, mock_open(read_data=self.BUNDLE_HEADER[:16])) as mock_file:
firmware = NetAppESeriesFirmware()
self.assertEqual(firmware.is_firmware_bundled(), True)
def test_is_firmware_bundles_fail(self):
"""Verify non-firmware fails."""
self._set_args({"firmware": "test.dlp", "nvsram": "test.dlp"})
with patch(builtin_path, mock_open(read_data=b"xxxxxxxxxxxxxxxx")) as mock_file:
firmware = NetAppESeriesFirmware()
with self.assertRaisesRegexp(AnsibleFailJson, "Firmware file is invalid."):
firmware.is_firmware_bundled()
def test_firmware_version(self):
"""Verify correct firmware version is returned."""
self._set_args({"firmware": "test.dlp", "nvsram": "test.dlp"})
firmware = NetAppESeriesFirmware()
firmware.is_firmware_bundled = lambda: True
with patch(builtin_path, mock_open_with_iter(read_data=self.BUNDLE_HEADER)) as mock_file:
self.assertEqual(firmware.firmware_version(), b"11.40.5")
def test_nvsram_version(self):
"""Verify correct nvsram version is returned."""
self._set_args({"firmware": "test.dlp", "nvsram": "test.dlp"})
firmware = NetAppESeriesFirmware()
with patch(builtin_path, mock_open_with_iter(read_data=self.NVSRAM_HEADER)) as mock_file:
self.assertEqual(firmware.nvsram_version(), b"N280X-852834-D02")
def test_check_system_health_pass(self):
"""Validate check_system_health method."""
self._set_args({"firmware": "test.dlp", "nvsram": "test.dlp"})
firmware = NetAppESeriesFirmware()
with patch(self.REQUEST_FUNC, side_effect=[(200, {"requestId": "1"}),
(200, {"healthCheckRunning": True,
"results": [{"processingTimeMS": 0}]}),
(200, {"healthCheckRunning": False,
"results": [{"successful": True}]})]):
firmware.check_system_health()
def test_check_system_health_fail(self):
"""Validate check_system_health method throws proper exceptions."""
self._set_args({"firmware": "test.dlp", "nvsram": "test.dlp"})
firmware = NetAppESeriesFirmware()
with patch("time.sleep", return_value=None):
with self.assertRaisesRegexp(AnsibleFailJson, "Failed to initiate health check."):
with patch(self.REQUEST_FUNC, return_value=(404, Exception())):
firmware.check_system_health()
with self.assertRaisesRegexp(AnsibleFailJson, "Failed to retrieve health check status."):
with patch(self.REQUEST_FUNC, side_effect=[(200, {"requestId": "1"}),
(404, Exception())]):
firmware.check_system_health()
with self.assertRaisesRegexp(AnsibleFailJson, "Health check failed to complete."):
with patch(self.REQUEST_FUNC, side_effect=[(200, {"requestId": "1"}),
(200, {"healthCheckRunning": True,
"results": [{"processingTimeMS": 120001}]})]):
firmware.check_system_health()
def test_embedded_check_nvsram_compatibility_pass(self):
"""Verify embedded nvsram compatibility."""
self._set_args({"firmware": "test.dlp", "nvsram": "test.dlp"})
firmware = NetAppESeriesFirmware()
with patch(self.CREATE_MULTIPART_FORMDATA_FUNC, return_value=("", {})):
with patch(self.REQUEST_FUNC, return_value=(200, {"signatureTestingPassed": True,
"fileCompatible": True,
"versionContents": [{"module": "nvsram",
"bundledVersion": "N280X-842834-D02",
"onboardVersion": "N280X-842834-D02"}]})):
firmware.embedded_check_nvsram_compatibility()
def test_embedded_check_nvsram_compatibility_fail(self):
"""Verify embedded nvsram compatibility fails with expected exceptions."""
self._set_args({"firmware": "test.dlp", "nvsram": "test.dlp"})
firmware = NetAppESeriesFirmware()
with patch(self.CREATE_MULTIPART_FORMDATA_FUNC, return_value=("", {})):
with self.assertRaisesRegexp(AnsibleFailJson, "Failed to retrieve NVSRAM compatibility results."):
with patch(self.REQUEST_FUNC, return_value=Exception()):
firmware.embedded_check_nvsram_compatibility()
with self.assertRaisesRegexp(AnsibleFailJson, "Invalid NVSRAM file."):
with patch(self.REQUEST_FUNC, return_value=(200, {"signatureTestingPassed": False,
"fileCompatible": False,
"versionContents": [{"module": "nvsram",
"bundledVersion": "N280X-842834-D02",
"onboardVersion": "N280X-842834-D02"}]})):
firmware.embedded_check_nvsram_compatibility()
with self.assertRaisesRegexp(AnsibleFailJson, "Incompatible NVSRAM file."):
with patch(self.REQUEST_FUNC, return_value=(200, {"signatureTestingPassed": True,
"fileCompatible": False,
"versionContents": [{"module": "nvsram",
"bundledVersion": "N280X-842834-D02",
"onboardVersion": "N280X-842834-D02"}]})):
firmware.embedded_check_nvsram_compatibility()
def test_embedded_check_firmware_compatibility_pass(self):
"""Verify embedded firmware compatibility."""
self._set_args({"firmware": "test.dlp", "nvsram": "test.dlp"})
firmware = NetAppESeriesFirmware()
with patch(self.CREATE_MULTIPART_FORMDATA_FUNC, return_value=("", {})):
with patch(self.REQUEST_FUNC, return_value=(200, {
"signatureTestingPassed": True,
"fileCompatible": True,
"versionContents": [
{"module": "bundle", "bundledVersion": "08.42.50.00.000", "onboardVersion": "08.42.30.05"},
{"module": "bundleDisplay", "bundledVersion": "11.40.5", "onboardVersion": "11.40.3R2"},
{"module": "hypervisor", "bundledVersion": "08.42.50.00", "onboardVersion": "08.42.30.05"},
{"module": "raid", "bundledVersion": "08.42.50.00", "onboardVersion": "08.42.30.05"},
{"module": "management", "bundledVersion": "11.42.0000.0028", "onboardVersion": "11.42.0000.0026"},
{"module": "iom", "bundledVersion": "11.42.0G00.0003", "onboardVersion": "11.42.0G00.0001"}]})):
firmware.embedded_check_bundle_compatibility()
def test_embedded_check_firmware_compatibility_fail(self):
"""Verify embedded firmware compatibility fails with expected exceptions."""
self._set_args({"firmware": "test.dlp", "nvsram": "test.dlp"})
firmware = NetAppESeriesFirmware()
with self.assertRaisesRegexp(AnsibleFailJson, "Failed to retrieve bundle compatibility results."):
with patch(self.CREATE_MULTIPART_FORMDATA_FUNC, return_value=("", {})):
with patch(self.REQUEST_FUNC, return_value=Exception()):
firmware.embedded_check_bundle_compatibility()
with self.assertRaisesRegexp(AnsibleFailJson, "Invalid firmware bundle file."):
with patch(self.CREATE_MULTIPART_FORMDATA_FUNC, return_value=("", {})):
with patch(self.REQUEST_FUNC, return_value=(200, {
"signatureTestingPassed": False,
"fileCompatible": True,
"versionContents": [
{"module": "bundle", "bundledVersion": "08.42.50.00.000", "onboardVersion": "08.42.30.05"},
{"module": "bundleDisplay", "bundledVersion": "11.40.5", "onboardVersion": "11.40.3R2"},
{"module": "hypervisor", "bundledVersion": "08.42.50.00", "onboardVersion": "08.42.30.05"},
{"module": "raid", "bundledVersion": "08.42.50.00", "onboardVersion": "08.42.30.05"},
{"module": "management", "bundledVersion": "11.42.0000.0028", "onboardVersion": "11.42.0000.0026"},
{"module": "iom", "bundledVersion": "11.42.0G00.0003", "onboardVersion": "11.42.0G00.0001"}]})):
firmware.embedded_check_bundle_compatibility()
with self.assertRaisesRegexp(AnsibleFailJson, "Incompatible firmware bundle file."):
with patch(self.CREATE_MULTIPART_FORMDATA_FUNC, return_value=("", {})):
with patch(self.REQUEST_FUNC, return_value=(200, {
"signatureTestingPassed": True,
"fileCompatible": False,
"versionContents": [
{"module": "bundle", "bundledVersion": "08.42.50.00.000", "onboardVersion": "08.42.30.05"},
{"module": "bundleDisplay", "bundledVersion": "11.40.5", "onboardVersion": "11.40.3R2"},
{"module": "hypervisor", "bundledVersion": "08.42.50.00", "onboardVersion": "08.42.30.05"},
{"module": "raid", "bundledVersion": "08.42.50.00", "onboardVersion": "08.42.30.05"},
{"module": "management", "bundledVersion": "11.42.0000.0028", "onboardVersion": "11.42.0000.0026"},
{"module": "iom", "bundledVersion": "11.42.0G00.0003", "onboardVersion": "11.42.0G00.0001"}]})):
firmware.embedded_check_bundle_compatibility()
with self.assertRaisesRegexp(AnsibleFailJson, "Downgrades are not permitted."):
with patch(self.CREATE_MULTIPART_FORMDATA_FUNC, return_value=("", {})):
with patch(self.REQUEST_FUNC, return_value=(200, {
"signatureTestingPassed": True,
"fileCompatible": True,
"versionContents": [
{"module": "bundle", "bundledVersion": "08.42.00.00.000", "onboardVersion": "08.50.30.05"},
{"module": "bundleDisplay", "bundledVersion": "11.40.5", "onboardVersion": "11.40.3R2"},
{"module": "hypervisor", "bundledVersion": "08.42.50.00", "onboardVersion": "08.42.30.05"},
{"module": "raid", "bundledVersion": "08.42.50.00", "onboardVersion": "08.42.30.05"},
{"module": "management", "bundledVersion": "11.42.0000.0028", "onboardVersion": "11.42.0000.0026"},
{"module": "iom", "bundledVersion": "11.42.0G00.0003", "onboardVersion": "11.42.0G00.0001"}]})):
firmware.embedded_check_bundle_compatibility()
with self.assertRaisesRegexp(AnsibleFailJson, "Downgrades are not permitted."):
with patch(self.CREATE_MULTIPART_FORMDATA_FUNC, return_value=("", {})):
with patch(self.REQUEST_FUNC, return_value=(200, {
"signatureTestingPassed": True,
"fileCompatible": True,
"versionContents": [
{"module": "bundle", "bundledVersion": "08.42.00.00.000", "onboardVersion": "09.20.30.05"},
{"module": "bundleDisplay", "bundledVersion": "11.40.5", "onboardVersion": "11.40.3R2"},
{"module": "hypervisor", "bundledVersion": "08.42.50.00", "onboardVersion": "08.42.30.05"},
{"module": "raid", "bundledVersion": "08.42.50.00", "onboardVersion": "08.42.30.05"},
{"module": "management", "bundledVersion": "11.42.0000.0028", "onboardVersion": "11.42.0000.0026"},
{"module": "iom", "bundledVersion": "11.42.0G00.0003", "onboardVersion": "11.42.0G00.0001"}]})):
firmware.embedded_check_bundle_compatibility()
def test_embedded_wait_for_upgrade_pass(self):
"""Verify controller reboot wait succeeds."""
self._set_args({"firmware": "test.dlp", "nvsram": "test.dlp"})
firmware = NetAppESeriesFirmware()
firmware.firmware_version = lambda: b"11.40.3R2"
firmware.nvsram_version = lambda: b"N280X-842834-D02"
with patch(self.SLEEP_FUNC, return_value=None):
with patch(self.REQUEST_FUNC, return_value=(200, [{"fwVersion": "08.42.30.05", "nvsramVersion": "N280X-842834-D02",
"extendedSAData": {"codeVersions": [{"codeModule": "bundleDisplay",
"versionString": "11.40.3R2"}]}}])):
firmware.embedded_wait_for_upgrade()
def test_embedded_wait_for_upgrade_fail(self):
"""Verify controller reboot wait throws expected exceptions"""
self._set_args({"firmware": "test.dlp", "nvsram": "test.dlp"})
firmware = NetAppESeriesFirmware()
with self.assertRaisesRegexp(AnsibleFailJson, "Timeout waiting for Santricity Web Services Embedded."):
with patch(self.SLEEP_FUNC, return_value=None):
with patch(self.BASE_REQUEST_FUNC, return_value=Exception()):
firmware.embedded_wait_for_upgrade()
def test_embedded_upgrade_pass(self):
"""Verify embedded upgrade function."""
with patch(self.CREATE_MULTIPART_FORMDATA_FUNC, return_value=("", {})):
with patch(self.SLEEP_FUNC, return_value=None):
self._set_args({"firmware": "test.dlp", "nvsram": "test.dlp"})
firmware = NetAppESeriesFirmware()
with patch(self.REQUEST_FUNC, return_value=(200, "")):
with patch(self.BASE_REQUEST_FUNC, side_effect=[Exception(), Exception(), (200, "")]):
firmware.embedded_upgrade()
self.assertTrue(firmware.upgrade_in_progress)
self._set_args({"firmware": "test.dlp", "nvsram": "test.dlp", "wait_for_completion": True})
firmware = NetAppESeriesFirmware()
firmware.firmware_version = lambda: b"11.40.3R2"
firmware.nvsram_version = lambda: b"N280X-842834-D02"
with patch(self.REQUEST_FUNC, return_value=(200, [{"fwVersion": "08.42.30.05", "nvsramVersion": "N280X-842834-D02",
"extendedSAData": {"codeVersions": [{"codeModule": "bundleDisplay",
"versionString": "11.40.3R2"}]}}])):
firmware.embedded_upgrade()
self.assertFalse(firmware.upgrade_in_progress)
def test_embedded_upgrade_fail(self):
"""Verify embedded upgrade throws expected exception."""
self._set_args({"firmware": "test.dlp", "nvsram": "test.dlp"})
firmware = NetAppESeriesFirmware()
with self.assertRaisesRegexp(AnsibleFailJson, "Failed to upload and activate firmware."):
with patch(self.CREATE_MULTIPART_FORMDATA_FUNC, return_value=("", {})):
with patch(self.REQUEST_FUNC, return_value=Exception()):
firmware.embedded_upgrade()
def test_check_nvsram_compatibility_pass(self):
"""Verify proxy nvsram compatibility."""
self._set_args({"firmware": "test.dlp", "nvsram": "test_nvsram.dlp"})
firmware = NetAppESeriesFirmware()
with patch(self.SLEEP_FUNC, return_value=None):
with patch(self.REQUEST_FUNC, side_effect=[(200, {"requestId": 1}),
(200, {"checkRunning": True}),
(200, {"checkRunning": False,
"results": [{"nvsramFiles": [{"filename": "test_nvsram.dlp"}]}]})]):
firmware.proxy_check_nvsram_compatibility()
def test_check_nvsram_compatibility_fail(self):
"""Verify proxy nvsram compatibility throws expected exceptions."""
self._set_args({"firmware": "test.dlp", "nvsram": "test_nvsram.dlp"})
firmware = NetAppESeriesFirmware()
with patch(self.SLEEP_FUNC, return_value=None):
with self.assertRaisesRegexp(AnsibleFailJson, "Failed to receive NVSRAM compatibility information."):
with patch(self.REQUEST_FUNC, return_value=Exception()):
firmware.proxy_check_nvsram_compatibility()
with self.assertRaisesRegexp(AnsibleFailJson, "Failed to retrieve NVSRAM status update from proxy."):
with patch(self.REQUEST_FUNC, side_effect=[(200, {"requestId": 1}), Exception()]):
firmware.proxy_check_nvsram_compatibility()
with self.assertRaisesRegexp(AnsibleFailJson, "NVSRAM is not compatible."):
with patch(self.REQUEST_FUNC, side_effect=[(200, {"requestId": 1}),
(200, {"checkRunning": True}),
(200, {"checkRunning": False,
"results": [{"nvsramFiles": [{"filename": "not_test_nvsram.dlp"}]}]})]):
firmware.proxy_check_nvsram_compatibility()
def test_check_firmware_compatibility_pass(self):
"""Verify proxy firmware compatibility."""
self._set_args({"firmware": "test_firmware.dlp", "nvsram": "test_nvsram.dlp"})
firmware = NetAppESeriesFirmware()
with patch(self.SLEEP_FUNC, return_value=None):
with patch(self.REQUEST_FUNC, side_effect=[(200, {"requestId": 1}),
(200, {"checkRunning": True}),
(200, {"checkRunning": False,
"results": [{"cfwFiles": [{"filename": "test_firmware.dlp"}]}]})]):
firmware.proxy_check_firmware_compatibility()
def test_check_firmware_compatibility_fail(self):
"""Verify proxy firmware compatibility throws expected exceptions."""
self._set_args({"firmware": "test_firmware.dlp", "nvsram": "test_nvsram.dlp"})
firmware = NetAppESeriesFirmware()
with patch(self.SLEEP_FUNC, return_value=None):
with self.assertRaisesRegexp(AnsibleFailJson, "Failed to receive firmware compatibility information."):
with patch(self.REQUEST_FUNC, return_value=Exception()):
firmware.proxy_check_firmware_compatibility()
with self.assertRaisesRegexp(AnsibleFailJson, "Failed to retrieve firmware status update from proxy."):
with patch(self.REQUEST_FUNC, side_effect=[(200, {"requestId": 1}), Exception()]):
firmware.proxy_check_firmware_compatibility()
with self.assertRaisesRegexp(AnsibleFailJson, "Firmware bundle is not compatible."):
with patch(self.REQUEST_FUNC, side_effect=[(200, {"requestId": 1}),
(200, {"checkRunning": True}),
(200, {"checkRunning": False,
"results": [{"cfwFiles": [{"filename": "not_test_firmware.dlp"}]}]})]):
firmware.proxy_check_firmware_compatibility()
def test_proxy_upload_and_check_compatibility_pass(self):
"""Verify proxy_upload_and_check_compatibility"""
self._set_args({"firmware": "test_firmware.dlp", "nvsram": "test_nvsram.dlp"})
firmware = NetAppESeriesFirmware()
firmware.proxy_check_nvsram_compatibility = lambda: None
firmware.proxy_check_firmware_compatibility = lambda: None
with patch(self.CREATE_MULTIPART_FORMDATA_FUNC, return_value=("headers", "data")):
with patch(self.REQUEST_FUNC, side_effect=[(200, [{"version": "XX.XX.XX.XX", "filename": "test"},
{"version": "XXXXXXXXXX", "filename": "test.dlp"}]),
(200, None), (200, None)]):
firmware.proxy_upload_and_check_compatibility()
with patch(self.REQUEST_FUNC, return_value=(200, [{"version": "XX.XX.XX.XX", "filename": "test"},
{"version": "test_nvsram", "filename": "test_nvsram.dlp"},
{"version": "test", "filename": "test.dlp"},
{"filename": "test_firmware.dlp", "version": "test_firmware"}])):
firmware.proxy_upload_and_check_compatibility()
def test_proxy_upload_and_check_compatibility_fail(self):
"""Verify proxy_upload_and_check_compatibility throws expected exceptions."""
self._set_args({"firmware": "test_firmware.dlp", "nvsram": "test_nvsram.dlp"})
firmware = NetAppESeriesFirmware()
firmware.proxy_check_nvsram_compatibility = lambda: None
with self.assertRaisesRegexp(AnsibleFailJson, "Failed to retrieve existing existing firmware files."):
with patch(self.CREATE_MULTIPART_FORMDATA_FUNC, return_value=("headers", "data")):
with patch(self.REQUEST_FUNC, return_value=Exception()):
firmware.proxy_upload_and_check_compatibility()
with self.assertRaisesRegexp(AnsibleFailJson, "Failed to upload NVSRAM file."):
with patch(self.CREATE_MULTIPART_FORMDATA_FUNC, return_value=("headers", "data")):
with patch(self.REQUEST_FUNC, side_effect=[(200, [{"version": "XX.XX.XX.XX", "filename": "test"},
{"version": "XXXXXXXXXX", "filename": "test.dlp"}]),
Exception()]):
firmware.proxy_upload_and_check_compatibility()
with self.assertRaisesRegexp(AnsibleFailJson, "Failed to upload firmware bundle file."):
with patch(self.CREATE_MULTIPART_FORMDATA_FUNC, return_value=("headers", "data")):
with patch(self.REQUEST_FUNC, side_effect=[(200, [{"version": "XX.XX.XX.XX", "filename": "test"},
{"version": "XXXXXXXXXX", "filename": "test.dlp"}]),
(200, None), Exception()]):
firmware.proxy_upload_and_check_compatibility()
def test_proxy_check_upgrade_required_pass(self):
"""Verify proxy_check_upgrade_required."""
self._set_args({"firmware": "test_firmware.dlp", "nvsram": "test_nvsram.dlp"})
firmware = NetAppESeriesFirmware()
firmware.firmware_version = lambda: b"08.42.50.00"
firmware.nvsram_version = lambda: b"nvsram_version"
with patch(self.REQUEST_FUNC, side_effect=[(200, [{"versionString": "08.42.50.00"}]), (200, ["nvsram_version"])]):
firmware.is_firmware_bundled = lambda: True
firmware.proxy_check_upgrade_required()
self.assertFalse(firmware.upgrade_required)
with patch(self.REQUEST_FUNC, side_effect=[(200, ["08.42.50.00"]), (200, ["nvsram_version"])]):
firmware.is_firmware_bundled = lambda: False
firmware.proxy_check_upgrade_required()
self.assertFalse(firmware.upgrade_required)
firmware.firmware_version = lambda: b"08.42.50.00"
firmware.nvsram_version = lambda: b"not_nvsram_version"
with patch(self.REQUEST_FUNC, side_effect=[(200, [{"versionString": "08.42.50.00"}]), (200, ["nvsram_version"])]):
firmware.is_firmware_bundled = lambda: True
firmware.proxy_check_upgrade_required()
self.assertTrue(firmware.upgrade_required)
with patch(self.REQUEST_FUNC, side_effect=[(200, ["08.42.50.00"]), (200, ["nvsram_version"])]):
firmware.is_firmware_bundled = lambda: False
firmware.proxy_check_upgrade_required()
self.assertTrue(firmware.upgrade_required)
firmware.firmware_version = lambda: b"08.52.00.00"
firmware.nvsram_version = lambda: b"nvsram_version"
with patch(self.REQUEST_FUNC, side_effect=[(200, [{"versionString": "08.42.50.00"}]), (200, ["nvsram_version"])]):
firmware.is_firmware_bundled = lambda: True
firmware.proxy_check_upgrade_required()
self.assertTrue(firmware.upgrade_required)
with patch(self.REQUEST_FUNC, side_effect=[(200, ["08.42.50.00"]), (200, ["nvsram_version"])]):
firmware.is_firmware_bundled = lambda: False
firmware.proxy_check_upgrade_required()
self.assertTrue(firmware.upgrade_required)
firmware.firmware_version = lambda: b"08.52.00.00"
firmware.nvsram_version = lambda: b"not_nvsram_version"
with patch(self.REQUEST_FUNC, side_effect=[(200, [{"versionString": "08.42.50.00"}]), (200, ["nvsram_version"])]):
firmware.is_firmware_bundled = lambda: True
firmware.proxy_check_upgrade_required()
self.assertTrue(firmware.upgrade_required)
with patch(self.REQUEST_FUNC, side_effect=[(200, ["08.42.50.00"]), (200, ["nvsram_version"])]):
firmware.is_firmware_bundled = lambda: False
firmware.proxy_check_upgrade_required()
self.assertTrue(firmware.upgrade_required)
def test_proxy_check_upgrade_required_fail(self):
"""Verify proxy_check_upgrade_required throws expected exceptions."""
self._set_args({"firmware": "test_firmware.dlp", "nvsram": "test_nvsram.dlp"})
firmware = NetAppESeriesFirmware()
firmware.firmware_version = lambda: b"08.42.50.00"
firmware.nvsram_version = lambda: b"not_nvsram_version"
with self.assertRaisesRegexp(AnsibleFailJson, "Failed to retrieve controller firmware information."):
with patch(self.REQUEST_FUNC, return_value=Exception()):
firmware.proxy_check_upgrade_required()
with self.assertRaisesRegexp(AnsibleFailJson, "Failed to retrieve storage system's NVSRAM version."):
with patch(self.REQUEST_FUNC, side_effect=[(200, [{"versionString": "08.42.50.00"}]), Exception()]):
firmware.is_firmware_bundled = lambda: True
firmware.proxy_check_upgrade_required()
with self.assertRaisesRegexp(AnsibleFailJson, "Failed to retrieve storage system's NVSRAM version."):
with patch(self.REQUEST_FUNC, side_effect=[(200, ["08.42.50.00"]), Exception()]):
firmware.is_firmware_bundled = lambda: False
firmware.proxy_check_upgrade_required()
with self.assertRaisesRegexp(AnsibleFailJson, "Downgrades are not permitted."):
with patch(self.REQUEST_FUNC, side_effect=[(200, [{"versionString": "08.42.50.00"}]), (200, ["nvsram_version"])]):
firmware.firmware_version = lambda: b"08.40.00.00"
firmware.nvsram_version = lambda: "nvsram_version"
firmware.is_firmware_bundled = lambda: True
firmware.proxy_check_upgrade_required()
with self.assertRaisesRegexp(AnsibleFailJson, "Downgrades are not permitted."):
with patch(self.REQUEST_FUNC, side_effect=[(200, ["08.42.50.00"]), (200, ["nvsram_version"])]):
firmware.is_firmware_bundled = lambda: False
firmware.proxy_check_upgrade_required()
def test_proxy_wait_for_upgrade_pass(self):
"""Verify proxy_wait_for_upgrade."""
with patch(self.SLEEP_FUNC, return_value=None):
self._set_args({"firmware": "test_firmware.dlp", "nvsram": "expected_nvsram.dlp"})
firmware = NetAppESeriesFirmware()
firmware.is_firmware_bundled = lambda: True
with patch(self.REQUEST_FUNC, side_effect=[(200, {"status": "not_done"}), (200, {"status": "complete"})]):
firmware.proxy_wait_for_upgrade("1")
firmware.is_firmware_bundled = lambda: False
firmware.firmware_version = lambda: b"08.50.00.00"
firmware.nvsram_version = lambda: b"expected_nvsram"
with patch(self.REQUEST_FUNC, side_effect=[(200, ["08.40.00.00"]), (200, ["not_expected_nvsram"]),
(200, ["08.50.00.00"]), (200, ["expected_nvsram"])]):
firmware.proxy_wait_for_upgrade("1")
def test_proxy_wait_for_upgrade_fail(self):
"""Verify proxy_wait_for_upgrade throws expected exceptions."""
with patch(self.SLEEP_FUNC, return_value=None):
self._set_args({"firmware": "test_firmware.dlp", "nvsram": "test_nvsram.dlp"})
firmware = NetAppESeriesFirmware()
firmware.is_firmware_bundled = lambda: True
with self.assertRaisesRegexp(AnsibleFailJson, "Failed to retrieve firmware upgrade status."):
with patch(self.REQUEST_FUNC, return_value=Exception()):
firmware.proxy_wait_for_upgrade("1")
firmware.is_firmware_bundled = lambda: False
with self.assertRaisesRegexp(AnsibleFailJson, "Timed out waiting for firmware upgrade to complete."):
with patch(self.REQUEST_FUNC, return_value=Exception()):
firmware.proxy_wait_for_upgrade("1")
firmware.is_firmware_bundled = lambda: True
with self.assertRaisesRegexp(AnsibleFailJson, "Firmware upgrade failed to complete."):
with patch(self.REQUEST_FUNC, side_effect=[(200, {"status": "not_done"}), (200, {"status": "failed"})]):
firmware.proxy_wait_for_upgrade("1")
def test_proxy_upgrade_fail(self):
"""Verify proxy_upgrade throws expected exceptions."""
self._set_args({"firmware": "test_firmware.dlp", "nvsram": "test_nvsram.dlp"})
firmware = NetAppESeriesFirmware()
firmware.is_firmware_bundled = lambda: True
with self.assertRaisesRegexp(AnsibleFailJson, "Failed to initiate firmware upgrade."):
with patch(self.REQUEST_FUNC, return_value=Exception()):
firmware.proxy_upgrade()
firmware.is_firmware_bundled = lambda: False
with self.assertRaisesRegexp(AnsibleFailJson, "Failed to initiate firmware upgrade."):
with patch(self.REQUEST_FUNC, return_value=Exception()):
firmware.proxy_upgrade()
Write Preview
Loading…
Cancel
Save