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

Redfish modules for Western Digital UltraStar Data102 storage enclosures (#4885) (#4958)

Browse source 
* WDC Redfish Info / Command modules for Western Digital Ultrastar Data102 storage enclosures.
Initial commands include:
* FWActivate
* UpdateAndActivate
* SimpleUpdateStatus

* delete unnecessary __init__.py modules

* PR Feedback

Notes list not guaranteed to be sorted
Use EXAMPLES tos how specifying ioms/basuri
Import missing_required_lib

* Apply suggestions from code review

Suggestions that could be auto-committed.

Co-authored-by: Felix Fontein <felix@fontein.de>

* Remove DNSCacheBypass

It is now the caller's responsibility to deal with stale IP addresses.

* Remove dnspython dependency.

Fix bug that this uncovered.

* Apply suggestions from code review

Co-authored-by: Felix Fontein <felix@fontein.de>

* PR Feedback

* Documentation, simple update status output format, unit tests.

Add docs showing how to use SimpleUpdateStatus
Change the format of SimpleUpateStatus format, put the results in a sub-object.
Fix unit tests whose asserts weren't actually running.

* PR Feedback

register: result on the 2nd example

* Final adjustments for merging for 5.4.0

Co-authored-by: Felix Fontein <felix@fontein.de>
(cherry picked from commit be70d18e3f)

Co-authored-by: Mike <mike@moerk.org>
This commit is contained in:
patchback[bot] 2022-07-17 11:03:55 +02:00 committed by GitHub
parent fbf3b85d6b
commit d7ecd40118
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 1827 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
.github/BOTMETA.yml vendored
View file

@ -304,6 +304,9 @@ files:
$module_utils/utm_utils.py:
maintainers: $team_e_spirit
labels: utm_utils
$module_utils/wdc_redfish_utils.py:
maintainers: $team_wdc
labels: wdc_redfish_utils
$module_utils/xenserver.py:
maintainers: bvitnik
labels: xenserver
@ -968,6 +971,10 @@ files:
$modules/remote_management/redfish/:
maintainers: $team_redfish
ignore: jose-delarosa
$modules/remote_management/redfish/wdc_redfish_command.py:
maintainers: $team_wdc
$modules/remote_management/redfish/wdc_redfish_info.py:
maintainers: $team_wdc
$modules/remote_management/stacki/stacki_host.py:
maintainers: bsanders bbyhuy
labels: stacki_host
@ -1298,3 +1305,4 @@ macros:
team_solaris: bcoca fishman jasperla jpdasma mator scathatheworm troy2914 xen0l
team_suse: commel evrardjp lrupp toabctl AnderEnder alxgu andytom sealor
team_virt: joshainglis karmab tleguern Thulium-Drake Ajpantuso
team_wdc: mikemoerk

4
meta/runtime.yml
View file

@ -1605,6 +1605,10 @@ plugin_routing:
redirect: community.general.cloud.smartos.vmadm
wakeonlan:
redirect: community.general.remote_management.wakeonlan
wdc_redfish_command:
redirect: community.general.remote_management.redfish.wdc_redfish_command
wdc_redfish_info:
redirect: community.general.remote_management.redfish.wdc_redfish_info
webfaction_app:
redirect: community.general.cloud.webfaction.webfaction_app
webfaction_db:

402
plugins/module_utils/wdc_redfish_utils.py Normal file
View file

@ -0,0 +1,402 @@
# -*- coding: utf-8 -*-
# Copyright (c) 2022 Western Digital Corporation
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import datetime
import re
import time
import tarfile
from ansible.module_utils.urls import fetch_file
from ansible_collections.community.general.plugins.module_utils.redfish_utils import RedfishUtils
from ansible.module_utils.six.moves.urllib.parse import urlparse, urlunparse
class WdcRedfishUtils(RedfishUtils):
"""Extension to RedfishUtils to support WDC enclosures."""
# Status codes returned by WDC FW Update Status
UPDATE_STATUS_CODE_READY_FOR_FW_UPDATE = 0
UPDATE_STATUS_CODE_FW_UPDATE_IN_PROGRESS = 1
UPDATE_STATUS_CODE_FW_UPDATE_COMPLETED_WAITING_FOR_ACTIVATION = 2
UPDATE_STATUS_CODE_FW_UPDATE_FAILED = 3
# Status messages returned by WDC FW Update Status
UPDATE_STATUS_MESSAGE_READY_FOR_FW_UDPATE = "Ready for FW update"
UDPATE_STATUS_MESSAGE_FW_UPDATE_IN_PROGRESS = "FW update in progress"
UPDATE_STATUS_MESSAGE_FW_UPDATE_COMPLETED_WAITING_FOR_ACTIVATION = "FW update completed. Waiting for activation."
UPDATE_STATUS_MESSAGE_FW_UPDATE_FAILED = "FW update failed."
def __init__(self,
creds,
root_uris,
timeout,
module,
resource_id,
data_modification):
super(WdcRedfishUtils, self).__init__(creds=creds,
root_uri=root_uris[0],
timeout=timeout,
module=module,
resource_id=resource_id,
data_modification=data_modification)
# Update the root URI if we cannot perform a Redfish GET to the first one
self._set_root_uri(root_uris)
def _set_root_uri(self, root_uris):
"""Set the root URI from a list of options.
If the current root URI is good, just keep it. Else cycle through our options until we find a good one.
A URI is considered good if we can GET uri/redfish/v1.
"""
for root_uri in root_uris:
uri = root_uri + "/redfish/v1"
response = self.get_request(uri)
if response['ret']:
self.root_uri = root_uri
def _find_updateservice_resource(self):
"""Find the update service resource as well as additional WDC-specific resources."""
response = super(WdcRedfishUtils, self)._find_updateservice_resource()
if not response['ret']:
return response
return self._find_updateservice_additional_uris()
def _is_enclosure_multi_tenant(self):
"""Determine if the enclosure is multi-tenant.
The serial number of a multi-tenant enclosure will end in "-A" or "-B".
:return: True/False if the enclosure is multi-tenant or not; None if unable to determine.
"""
response = self.get_request(self.root_uri + self.service_root + "Chassis/Enclosure")
if response['ret'] is False:
return None
pattern = r".*-[A,B]"
data = response['data']
return re.match(pattern, data['SerialNumber']) is not None
def _find_updateservice_additional_uris(self):
"""Find & set WDC-specific update service URIs"""
response = self.get_request(self.root_uri + self._update_uri())
if response['ret'] is False:
return response
data = response['data']
if 'Actions' not in data:
return {'ret': False, 'msg': 'Service does not support SimpleUpdate'}
if '#UpdateService.SimpleUpdate' not in data['Actions']:
return {'ret': False, 'msg': 'Service does not support SimpleUpdate'}
action = data['Actions']['#UpdateService.SimpleUpdate']
if 'target' not in action:
return {'ret': False, 'msg': 'Service does not support SimpleUpdate'}
self.simple_update_uri = action['target']
# Simple update status URI is not provided via GET /redfish/v1/UpdateService
# So we have to hard code it.
self.simple_update_status_uri = "{0}/Status".format(self.simple_update_uri)
# FWActivate URI
if 'Oem' not in data['Actions']:
return {'ret': False, 'msg': 'Service does not support OEM operations'}
if 'WDC' not in data['Actions']['Oem']:
return {'ret': False, 'msg': 'Service does not support WDC operations'}
if '#UpdateService.FWActivate' not in data['Actions']['Oem']['WDC']:
return {'ret': False, 'msg': 'Service does not support FWActivate'}
action = data['Actions']['Oem']['WDC']['#UpdateService.FWActivate']
if 'target' not in action:
return {'ret': False, 'msg': 'Service does not support FWActivate'}
self.firmware_activate_uri = action['target']
return {'ret': True}
def _simple_update_status_uri(self):
return self.simple_update_status_uri
def _firmware_activate_uri(self):
return self.firmware_activate_uri
def _update_uri(self):
return self.update_uri
def get_simple_update_status(self):
"""Issue Redfish HTTP GET to return the simple update status"""
result = {}
response = self.get_request(self.root_uri + self._simple_update_status_uri())
if response['ret'] is False:
return response
result['ret'] = True
data = response['data']
result['entries'] = data
return result
def firmware_activate(self, update_opts):
"""Perform FWActivate using Redfish HTTP API."""
creds = update_opts.get('update_creds')
payload = {}
if creds:
if creds.get('username'):
payload["Username"] = creds.get('username')
if creds.get('password'):
payload["Password"] = creds.get('password')
# Make sure the service supports FWActivate
response = self.get_request(self.root_uri + self._update_uri())
if response['ret'] is False:
return response
data = response['data']
if 'Actions' not in data:
return {'ret': False, 'msg': 'Service does not support FWActivate'}
response = self.post_request(self.root_uri + self._firmware_activate_uri(), payload)
if response['ret'] is False:
return response
return {'ret': True, 'changed': True,
'msg': "FWActivate requested"}
def _get_bundle_version(self,
bundle_uri,
update_creds):
"""Get the firmware version from a bundle file, and whether or not it is multi-tenant.
Only supports HTTP at this time. Assumes URI exists and is a tarfile.
Looks for a file oobm-[version].pkg, such as 'oobm-4.0.13.pkg`. Extracts the version number
from that filename (in the above example, the version number is "4.0.13".
To determine if the bundle is multi-tenant or not, it looks inside the .bin file within the tarfile,
and checks the appropriate byte in the file.
:param str bundle_uri: HTTP URI of the firmware bundle.
:param dict or None update_creds: Dict containing username and password to access the bundle.
:return: Firmware version number contained in the bundle, and whether or not the bundle is multi-tenant.
Either value will be None if unable to deterine.
:rtype: str or None, bool or None
"""
parsed_url = urlparse(bundle_uri)
if update_creds:
original_netloc = parsed_url.netloc
parsed_url._replace(netloc="{0}:{1}{2}".format(update_creds.get("username"),
update_creds.get("password"),
original_netloc))
bundle_temp_filename = fetch_file(module=self.module,
url=urlunparse(parsed_url))
if not tarfile.is_tarfile(bundle_temp_filename):
return None, None
tf = tarfile.open(bundle_temp_filename)
pattern_pkg = r"oobm-(.+)\.pkg"
pattern_bin = r"(.*\.bin)"
bundle_version = None
is_multi_tenant = None
for filename in tf.getnames():
match_pkg = re.match(pattern_pkg, filename)
if match_pkg is not None:
bundle_version = match_pkg.group(1)
match_bin = re.match(pattern_bin, filename)
if match_bin is not None:
bin_filename = match_bin.group(1)
bin_file = tf.extractfile(bin_filename)
bin_file.seek(11)
byte_11 = bin_file.read(1)
is_multi_tenant = byte_11 == b'\x80'
return bundle_version, is_multi_tenant
@staticmethod
def uri_is_http(uri):
"""Return True if the specified URI is http or https.
:param str uri: A URI.
:return: True if the URI is http or https, else False
:rtype: bool
"""
parsed_bundle_uri = urlparse(uri)
return parsed_bundle_uri.scheme.lower() in ['http', 'https']
def update_and_activate(self, update_opts):
"""Update and activate the firmware in a single action.
Orchestrates the firmware update so that everything can be done in a single command.
Compares the update version with the already-installed version -- skips update if they are the same.
Performs retries, handles timeouts as needed.
"""
# Make sure bundle URI is HTTP(s)
bundle_uri = update_opts["update_image_uri"]
if not self.uri_is_http(bundle_uri):
return {
'ret': False,
'msg': 'Bundle URI must be HTTP or HTTPS'
}
# Make sure IOM is ready for update
result = self.get_simple_update_status()
if result['ret'] is False:
return result
update_status = result['entries']
status_code = update_status['StatusCode']
status_description = update_status['Description']
if status_code not in [
self.UPDATE_STATUS_CODE_READY_FOR_FW_UPDATE,
self.UPDATE_STATUS_CODE_FW_UPDATE_FAILED
]:
return {
'ret': False,
'msg': 'Target is not ready for FW update. Current status: {0} ({1})'.format(
status_code, status_description
)}
# Check the FW version in the bundle file, and compare it to what is already on the IOMs
# Bundle version number
update_creds = update_opts.get("update_creds")
bundle_firmware_version, is_bundle_multi_tenant = self._get_bundle_version(bundle_uri,
update_creds)
if bundle_firmware_version is None or is_bundle_multi_tenant is None:
return {
'ret': False,
'msg': 'Unable to extract bundle version or multi-tenant status from update image tarfile'
}
# Verify that the bundle is correctly multi-tenant or not
is_enclosure_multi_tenant = self._is_enclosure_multi_tenant()
if is_enclosure_multi_tenant != is_bundle_multi_tenant:
return {
'ret': False,
'msg': 'Enclosure multi-tenant is {0} but bundle multi-tenant is {1}'.format(
is_enclosure_multi_tenant,
is_bundle_multi_tenant,
)
}
# Version number installed on IOMs
firmware_inventory = self.get_firmware_inventory()
if not firmware_inventory["ret"]:
return firmware_inventory
firmware_inventory_dict = {}
for entry in firmware_inventory["entries"]:
firmware_inventory_dict[entry["Id"]] = entry
iom_a_firmware_version = firmware_inventory_dict.get("IOModuleA_OOBM", {}).get("Version")
iom_b_firmware_version = firmware_inventory_dict.get("IOModuleB_OOBM", {}).get("Version")
# If version is None, we will proceed with the update, because we cannot tell
# for sure that we have a full version match.
if is_enclosure_multi_tenant:
# For multi-tenant, only one of the IOMs will be affected by the firmware update,
# so see if that IOM already has the same firmware version as the bundle.
firmware_already_installed = bundle_firmware_version == self._get_installed_firmware_version_of_multi_tenant_system(
iom_a_firmware_version,
iom_b_firmware_version)
else:
# For single-tenant, see if both IOMs already have the same firmware version as the bundle.
firmware_already_installed = bundle_firmware_version == iom_a_firmware_version == iom_b_firmware_version
# If this FW already installed, return changed: False, and do not update the firmware.
if firmware_already_installed:
return {
'ret': True,
'changed': False,
'msg': 'Version {0} already installed'.format(bundle_firmware_version)
}
# Version numbers don't match the bundle -- proceed with update (unless we are in check mode)
if self.module.check_mode:
return {
'ret': True,
'changed': True,
'msg': 'Update not performed in check mode.'
}
update_successful = False
retry_interval_seconds = 5
max_number_of_retries = 5
retry_number = 0
while retry_number < max_number_of_retries and not update_successful:
if retry_number != 0:
time.sleep(retry_interval_seconds)
retry_number += 1
result = self.simple_update(update_opts)
if result['ret'] is not True:
# Sometimes a timeout error is returned even though the update actually was requested.
# Check the update status to see if the update is in progress.
status_result = self.get_simple_update_status()
if status_result['ret'] is False:
continue
update_status = status_result['entries']
status_code = update_status['StatusCode']
if status_code != self.UPDATE_STATUS_CODE_FW_UPDATE_IN_PROGRESS:
# Update is not in progress -- retry until max number of retries
continue
else:
update_successful = True
else:
update_successful = True
if not update_successful:
# Unable to get SimpleUpdate to work. Return the failure from the SimpleUpdate
return result
# Wait for "ready to activate"
max_wait_minutes = 30
polling_interval_seconds = 30
status_code = self.UPDATE_STATUS_CODE_READY_FOR_FW_UPDATE
start_time = datetime.datetime.now()
# For a short time, target will still say "ready for firmware update" before it transitions
# to "update in progress"
status_codes_for_update_incomplete = [
self.UPDATE_STATUS_CODE_FW_UPDATE_IN_PROGRESS,
self.UPDATE_STATUS_CODE_READY_FOR_FW_UPDATE
]
iteration = 0
while status_code in status_codes_for_update_incomplete \
and datetime.datetime.now() - start_time < datetime.timedelta(minutes=max_wait_minutes):
if iteration != 0:
time.sleep(polling_interval_seconds)
iteration += 1
result = self.get_simple_update_status()
if result['ret'] is False:
continue # We may get timeouts, just keep trying until we give up
update_status = result['entries']
status_code = update_status['StatusCode']
status_description = update_status['Description']
if status_code == self.UPDATE_STATUS_CODE_FW_UPDATE_IN_PROGRESS:
# Once it says update in progress, "ready for update" is no longer a valid status code
status_codes_for_update_incomplete = [self.UPDATE_STATUS_CODE_FW_UPDATE_IN_PROGRESS]
# Update no longer in progress -- verify that it finished
if status_code != self.UPDATE_STATUS_CODE_FW_UPDATE_COMPLETED_WAITING_FOR_ACTIVATION:
return {
'ret': False,
'msg': 'Target is not ready for FW activation after update. Current status: {0} ({1})'.format(
status_code, status_description
)}
self.firmware_activate(update_opts)
return {'ret': True, 'changed': True,
'msg': "Firmware updated and activation initiated."}
def _get_installed_firmware_version_of_multi_tenant_system(self,
iom_a_firmware_version,
iom_b_firmware_version):
"""Return the version for the active IOM on a multi-tenant system.
Only call this on a multi-tenant system.
Given the installed firmware versions for IOM A, B, this method will determine which IOM is active
for this tenanat, and return that IOM's firmware version.
"""
# To determine which IOM we are on, try to GET each IOM resource
# The one we are on will return valid data.
# The other will return an error with message "IOM Module A/B cannot be read"
which_iom_is_this = None
for iom_letter in ['A', 'B']:
iom_uri = "Chassis/IOModule{0}FRU".format(iom_letter)
response = self.get_request(self.root_uri + self.service_root + iom_uri)
if response['ret'] is False:
continue
data = response['data']
if "Id" in data: # Assume if there is an "Id", it is valid
which_iom_is_this = iom_letter
break
if which_iom_is_this == 'A':
return iom_a_firmware_version
elif which_iom_is_this == 'B':
return iom_b_firmware_version
else:
return None

252
plugins/modules/remote_management/redfish/wdc_redfish_command.py Normal file
View file

@ -0,0 +1,252 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# Copyright (c) 2022 Western Digital Corporation
# 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
DOCUMENTATION = '''
---
module: wdc_redfish_command
short_description: Manages WDC UltraStar Data102 Out-Of-Band controllers using Redfish APIs
version_added: 5.4.0
description:
- Builds Redfish URIs locally and sends them to remote OOB controllers to
perform an action.
- Manages OOB controller firmware. For example, Firmware Activate, Update and Activate.
options:
category:
required: true
description:
- Category to execute on OOB controller.
type: str
command:
required: true
description:
- List of commands to execute on OOB controller.
type: list
elements: str
baseuri:
description:
- Base URI of OOB controller. Must include this or I(ioms).
type: str
ioms:
description:
- List of IOM FQDNs for the enclosure. Must include this or I(baseuri).
type: list
elements: str
username:
description:
- User for authentication with OOB controller.
type: str
password:
description:
- Password for authentication with OOB controller.
type: str
auth_token:
description:
- Security token for authentication with OOB controller.
type: str
timeout:
description:
- Timeout in seconds for URL requests to OOB controller.
default: 10
type: int
update_image_uri:
required: false
description:
- The URI of the image for the update.
type: str
update_creds:
required: false
description:
- The credentials for retrieving the update image.
type: dict
suboptions:
username:
required: false
description:
- The username for retrieving the update image.
type: str
password:
required: false
description:
- The password for retrieving the update image.
type: str
requirements:
- dnspython (2.1.0 for Python 3, 1.16.0 for Python 2)
notes:
- In the inventory, you can specify baseuri or ioms. See the EXAMPLES section.
- ioms is a list of FQDNs for the enclosure's IOMs.
author: Mike Moerk (@mikemoerk)
'''
EXAMPLES = '''
- name: Firmware Activate (required after SimpleUpdate to apply the new firmware)
community.general.wdc_redfish_command:
category: Update
command: FWActivate
ioms: "{{ ioms }}"
username: "{{ username }}"
password: "{{ password }}"
- name: Firmware Activate with individual IOMs specified
community.general.wdc_redfish_command:
category: Update
command: FWActivate
ioms:
- iom1.wdc.com
- iom2.wdc.com
username: "{{ username }}"
password: "{{ password }}"
- name: Firmware Activate with baseuri specified
community.general.wdc_redfish_command:
category: Update
command: FWActivate
baseuri: "iom1.wdc.com"
username: "{{ username }}"
password: "{{ password }}"
- name: Update and Activate (orchestrates firmware update and activation with a single command)
community.general.wdc_redfish_command:
category: Update
command: UpdateAndActivate
ioms: "{{ ioms }}"
username: "{{ username }}"
password: "{{ password }}"
update_image_uri: "{{ update_image_uri }}"
update_creds:
username: operator
password: supersecretpwd
'''
RETURN = '''
msg:
description: Message with action result or error description
returned: always
type: str
sample: "Action was successful"
'''
from ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils import WdcRedfishUtils
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.common.text.converters import to_native
CATEGORY_COMMANDS_ALL = {
"Update": [
"FWActivate",
"UpdateAndActivate"
]
}
def main():
module = AnsibleModule(
argument_spec=dict(
category=dict(required=True),
command=dict(required=True, type='list', elements='str'),
ioms=dict(type='list', elements='str'),
baseuri=dict(),
username=dict(),
password=dict(no_log=True),
auth_token=dict(no_log=True),
update_creds=dict(
type='dict',
options=dict(
username=dict(),
password=dict(no_log=True)
)
),
update_image_uri=dict(),
timeout=dict(type='int', default=10)
),
required_together=[
('username', 'password'),
],
required_one_of=[
('username', 'auth_token'),
('baseuri', 'ioms')
],
mutually_exclusive=[
('username', 'auth_token'),
],
supports_check_mode=True
)
category = module.params['category']
command_list = module.params['command']
# admin credentials used for authentication
creds = {'user': module.params['username'],
'pswd': module.params['password'],
'token': module.params['auth_token']}
# timeout
timeout = module.params['timeout']
# Check that Category is valid
if category not in CATEGORY_COMMANDS_ALL:
module.fail_json(msg=to_native("Invalid Category '%s'. Valid Categories = %s" % (category, sorted(CATEGORY_COMMANDS_ALL.keys()))))
# Check that all commands are valid
for cmd in command_list:
# Fail if even one command given is invalid
if cmd not in CATEGORY_COMMANDS_ALL[category]:
module.fail_json(msg=to_native("Invalid Command '%s'. Valid Commands = %s" % (cmd, CATEGORY_COMMANDS_ALL[category])))
# Build root URI(s)
if module.params.get("baseuri") is not None:
root_uris = ["https://" + module.params['baseuri']]
else:
root_uris = [
"https://" + iom for iom in module.params['ioms']
]
rf_utils = WdcRedfishUtils(creds, root_uris, timeout, module,
resource_id=None, data_modification=True)
# Organize by Categories / Commands
if category == "Update":
# execute only if we find UpdateService resources
resource = rf_utils._find_updateservice_resource()
if resource['ret'] is False:
module.fail_json(msg=resource['msg'])
# update options
update_opts = {
'update_creds': module.params['update_creds']
}
for command in command_list:
if command == "FWActivate":
if module.check_mode:
result = {
'ret': True,
'changed': True,
'msg': 'FWActivate not performed in check mode.'
}
else:
result = rf_utils.firmware_activate(update_opts)
elif command == "UpdateAndActivate":
update_opts["update_image_uri"] = module.params['update_image_uri']
result = rf_utils.update_and_activate(update_opts)
if result['ret'] is False:
module.fail_json(msg=to_native(result['msg']))
else:
del result['ret']
changed = result.get('changed', True)
session = result.get('session', dict())
module.exit_json(changed=changed,
session=session,
msg='Action was successful' if not module.check_mode else result.get(
'msg', "No action performed in check mode."
))
if __name__ == '__main__':
main()

214
plugins/modules/remote_management/redfish/wdc_redfish_info.py Normal file
View file

@ -0,0 +1,214 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# Copyright (c) 2022 Western Digital Corporation
# 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
DOCUMENTATION = '''
---
module: wdc_redfish_info
short_description: Manages WDC UltraStar Data102 Out-Of-Band controllers using Redfish APIs
version_added: 5.4.0
description:
- Builds Redfish URIs locally and sends them to remote OOB controllers to
get information back.
options:
category:
required: true
description:
- Category to execute on OOB controller.
type: str
command:
required: true
description:
- List of commands to execute on OOB controller.
type: list
elements: str
baseuri:
description:
- Base URI of OOB controller. Must include this or I(ioms).
type: str
ioms:
description:
- List of IOM FQDNs for the enclosure. Must include this or I(baseuri).
type: list
elements: str
username:
description:
- User for authentication with OOB controller.
type: str
password:
description:
- Password for authentication with OOB controller.
type: str
auth_token:
description:
- Security token for authentication with OOB controller.
type: str
timeout:
description:
- Timeout in seconds for URL requests to OOB controller.
default: 10
type: int
notes:
- In the inventory, you can specify baseuri or ioms. See the EXAMPLES section.
- ioms is a list of FQDNs for the enclosure's IOMs.
author: Mike Moerk (@mikemoerk)
'''
EXAMPLES = '''
- name: Get Simple Update Status with individual IOMs specified
community.general.wdc_redfish_info:
category: Update
command: SimpleUpdateStatus
ioms:
- iom1.wdc.com
- iom2.wdc.com
username: "{{ username }}"
password: "{{ password }}"
register: result
- name: Print fetched information
ansible.builtin.debug:
msg: "{{ result.redfish_facts.simple_update_status.entries | to_nice_json }}"
- name: Get Simple Update Status with baseuri specified
community.general.wdc_redfish_info:
category: Update
command: SimpleUpdateStatus
baseuri: "iom1.wdc.com"
username: "{{ username }}"
password: "{{ password }}"
register: result
- name: Print fetched information
ansible.builtin.debug:
msg: "{{ result.redfish_facts.simple_update_status.entries | to_nice_json }}"
'''
RETURN = '''
Description:
description: Firmware update status description.
returned: always
type: str
sample:
- Ready for FW update
- FW update in progress
- FW update completed. Waiting for activation.
ErrorCode:
description: Numeric error code for firmware update status. Non-zero indicates an error condition.
returned: always
type: int
sample:
- 0
EstimatedRemainingMinutes:
description: Estimated number of minutes remaining in firmware update operation.
returned: always
type: int
sample:
- 0
- 20
StatusCode:
description: Firmware update status code.
returned: always
type: int
sample:
- 0 (Ready for FW update)
- 1 (FW update in progress)
- 2 (FW update completed. Waiting for activation.)
- 3 (FW update failed.)
'''
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.common.text.converters import to_native
from ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils import WdcRedfishUtils
CATEGORY_COMMANDS_ALL = {
"Update": ["SimpleUpdateStatus"]
}
def main():
result = {}
module = AnsibleModule(
argument_spec=dict(
category=dict(required=True),
command=dict(required=True, type='list', elements='str'),
ioms=dict(type='list', elements='str'),
baseuri=dict(),
username=dict(),
password=dict(no_log=True),
auth_token=dict(no_log=True),
timeout=dict(type='int', default=10)
),
required_together=[
('username', 'password'),
],
required_one_of=[
('username', 'auth_token'),
('baseuri', 'ioms')
],
mutually_exclusive=[
('username', 'auth_token'),
],
supports_check_mode=True
)
category = module.params['category']
command_list = module.params['command']
# admin credentials used for authentication
creds = {'user': module.params['username'],
'pswd': module.params['password'],
'token': module.params['auth_token']}
# timeout
timeout = module.params['timeout']
# Check that Category is valid
if category not in CATEGORY_COMMANDS_ALL:
module.fail_json(msg=to_native("Invalid Category '%s'. Valid Categories = %s" % (category, sorted(CATEGORY_COMMANDS_ALL.keys()))))
# Check that all commands are valid
for cmd in command_list:
# Fail if even one command given is invalid
if cmd not in CATEGORY_COMMANDS_ALL[category]:
module.fail_json(msg=to_native("Invalid Command '%s'. Valid Commands = %s" % (cmd, CATEGORY_COMMANDS_ALL[category])))
# Build root URI(s)
if module.params.get("baseuri") is not None:
root_uris = ["https://" + module.params['baseuri']]
else:
root_uris = [
"https://" + iom for iom in module.params['ioms']
]
rf_utils = WdcRedfishUtils(creds, root_uris, timeout, module,
resource_id=None,
data_modification=False
)
# Organize by Categories / Commands
if category == "Update":
# execute only if we find UpdateService resources
resource = rf_utils._find_updateservice_resource()
if resource['ret'] is False:
module.fail_json(msg=resource['msg'])
for command in command_list:
if command == "SimpleUpdateStatus":
simple_update_status_result = rf_utils.get_simple_update_status()
if simple_update_status_result['ret'] is False:
module.fail_json(msg=to_native(result['msg']))
else:
del simple_update_status_result['ret']
result["simple_update_status"] = simple_update_status_result
module.exit_json(changed=False, redfish_facts=result)
if __name__ == '__main__':
main()

733
tests/unit/plugins/modules/remote_management/wdc/test_wdc_redfish_command.py Normal file
View file

@ -0,0 +1,733 @@
# -*- coding: utf-8 -*-
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import shutil
import uuid
import tarfile
import tempfile
import os
from ansible_collections.community.general.tests.unit.compat.mock import patch
from ansible_collections.community.general.tests.unit.compat import unittest
from ansible.module_utils import basic
import ansible_collections.community.general.plugins.modules.remote_management.redfish.wdc_redfish_command as module
from ansible_collections.community.general.tests.unit.plugins.modules.utils import AnsibleExitJson, AnsibleFailJson
from ansible_collections.community.general.tests.unit.plugins.modules.utils import set_module_args, exit_json, fail_json
MOCK_SUCCESSFUL_HTTP_EMPTY_RESPONSE = {
"ret": True,
"data": {
}
}
MOCK_GET_ENCLOSURE_RESPONSE_SINGLE_TENANT = {
"ret": True,
"data": {
"SerialNumber": "12345"
}
}
MOCK_GET_ENCLOSURE_RESPONSE_MULTI_TENANT = {
"ret": True,
"data": {
"SerialNumber": "12345-A"
}
}
MOCK_URL_ERROR = {
"ret": False,
"msg": "This is a mock URL error",
"status": 500
}
MOCK_SUCCESSFUL_RESPONSE_WITH_UPDATE_SERVICE_RESOURCE = {
"ret": True,
"data": {
"UpdateService": {
"@odata.id": "/UpdateService"
}
}
}
MOCK_SUCCESSFUL_RESPONSE_WITH_SIMPLE_UPDATE_AND_FW_ACTIVATE = {
"ret": True,
"data": {
"Actions": {
"#UpdateService.SimpleUpdate": {
"target": "mocked value"
},
"Oem": {
"WDC": {
"#UpdateService.FWActivate": {
"title": "Activate the downloaded firmware.",
"target": "/redfish/v1/UpdateService/Actions/UpdateService.FWActivate"
}
}
}
}
}
}
MOCK_SUCCESSFUL_RESPONSE_WITH_ACTIONS = {
"ret": True,
"data": {
"Actions": {}
}
}
MOCK_GET_IOM_A_MULTI_TENANT = {
"ret": True,
"data": {
"Id": "IOModuleAFRU"
}
}
MOCK_GET_IOM_B_MULTI_TENANAT = {
"ret": True,
"data": {
"error": {
"message": "IOM Module B cannot be read"
}
}
}
MOCK_READY_FOR_FW_UPDATE = {
"ret": True,
"entries": {
"Description": "Ready for FW update",
"StatusCode": 0
}
}
MOCK_FW_UPDATE_IN_PROGRESS = {
"ret": True,
"entries": {
"Description": "FW update in progress",
"StatusCode": 1
}
}
MOCK_WAITING_FOR_ACTIVATION = {
"ret": True,
"entries": {
"Description": "FW update completed. Waiting for activation.",
"StatusCode": 2
}
}
MOCK_SIMPLE_UPDATE_STATUS_LIST = [
MOCK_READY_FOR_FW_UPDATE,
MOCK_FW_UPDATE_IN_PROGRESS,
MOCK_WAITING_FOR_ACTIVATION
]
def get_bin_path(self, arg, required=False):
"""Mock AnsibleModule.get_bin_path"""
return arg
def get_exception_message(ansible_exit_json):
"""From an AnsibleExitJson exception, get the message string."""
return ansible_exit_json.exception.args[0]["msg"]
def is_changed(ansible_exit_json):
"""From an AnsibleExitJson exception, return the value of the changed flag"""
return ansible_exit_json.exception.args[0]["changed"]
def mock_simple_update(*args, **kwargs):
return {
"ret": True
}
def mocked_url_response(*args, **kwargs):
"""Mock to just return a generic string."""
return "/mockedUrl"
def mock_update_url(*args, **kwargs):
"""Mock of the update url"""
return "/UpdateService"
def mock_fw_activate_url(*args, **kwargs):
"""Mock of the FW Activate URL"""
return "/UpdateService.FWActivate"
def empty_return(*args, **kwargs):
"""Mock to just return an empty successful return."""
return {"ret": True}
def mock_get_simple_update_status_ready_for_fw_update(*args, **kwargs):
"""Mock to return simple update status Ready for FW update"""
return MOCK_READY_FOR_FW_UPDATE
def mock_get_request_enclosure_single_tenant(*args, **kwargs):
"""Mock for get_request for single-tenant enclosure."""
if args[1].endswith("/redfish/v1") or args[1].endswith("/redfish/v1/"):
return MOCK_SUCCESSFUL_RESPONSE_WITH_UPDATE_SERVICE_RESOURCE
elif args[1].endswith("/mockedUrl"):
return MOCK_SUCCESSFUL_HTTP_EMPTY_RESPONSE
elif args[1].endswith("Chassis/Enclosure"):
return MOCK_GET_ENCLOSURE_RESPONSE_SINGLE_TENANT
elif args[1].endswith("/UpdateService"):
return MOCK_SUCCESSFUL_RESPONSE_WITH_SIMPLE_UPDATE_AND_FW_ACTIVATE
else:
raise RuntimeError("Illegal call to get_request in test: " + args[1])
def mock_get_request_enclosure_multi_tenant(*args, **kwargs):
"""Mock for get_request with multi-tenant enclosure."""
if args[1].endswith("/redfish/v1") or args[1].endswith("/redfish/v1/"):
return MOCK_SUCCESSFUL_RESPONSE_WITH_UPDATE_SERVICE_RESOURCE
elif args[1].endswith("/mockedUrl"):
return MOCK_SUCCESSFUL_HTTP_EMPTY_RESPONSE
elif args[1].endswith("Chassis/Enclosure"):
return MOCK_GET_ENCLOSURE_RESPONSE_MULTI_TENANT
elif args[1].endswith("/UpdateService"):
return MOCK_SUCCESSFUL_RESPONSE_WITH_SIMPLE_UPDATE_AND_FW_ACTIVATE
elif args[1].endswith("/IOModuleAFRU"):
return MOCK_GET_IOM_A_MULTI_TENANT
elif args[1].endswith("/IOModuleBFRU"):
return MOCK_GET_IOM_B_MULTI_TENANAT
else:
raise RuntimeError("Illegal call to get_request in test: " + args[1])
def mock_post_request(*args, **kwargs):
"""Mock post_request with successful response."""
if args[1].endswith("/UpdateService.FWActivate"):
return {
"ret": True,
"data": ACTION_WAS_SUCCESSFUL_MESSAGE
}
else:
raise RuntimeError("Illegal POST call to: " + args[1])
def mock_get_firmware_inventory_version_1_2_3(*args, **kwargs):
return {
"ret": True,
"entries": [
{
"Id": "IOModuleA_OOBM",
"Version": "1.2.3"
},
{
"Id": "IOModuleB_OOBM",
"Version": "1.2.3"
}
]
}
ERROR_MESSAGE_UNABLE_TO_EXTRACT_BUNDLE_VERSION = "Unable to extract bundle version or multi-tenant status from update image tarfile"
ACTION_WAS_SUCCESSFUL_MESSAGE = "Action was successful"
class TestWdcRedfishCommand(unittest.TestCase):
def setUp(self):
self.mock_module_helper = patch.multiple(basic.AnsibleModule,
exit_json=exit_json,
fail_json=fail_json,
get_bin_path=get_bin_path)
self.mock_module_helper.start()
self.addCleanup(self.mock_module_helper.stop)
self.tempdir = tempfile.mkdtemp()
def tearDown(self):
shutil.rmtree(self.tempdir)
def test_module_fail_when_required_args_missing(self):
with self.assertRaises(AnsibleFailJson):
set_module_args({})
module.main()
def test_module_fail_when_unknown_category(self):
with self.assertRaises(AnsibleFailJson):
set_module_args({
'category': 'unknown',
'command': 'FWActivate',
'username': 'USERID',
'password': 'PASSW0RD=21',
'ioms': [],
})
module.main()
def test_module_fail_when_unknown_command(self):
with self.assertRaises(AnsibleFailJson):
set_module_args({
'category': 'Update',
'command': 'unknown',
'username': 'USERID',
'password': 'PASSW0RD=21',
'ioms': [],
})
module.main()
def test_module_fw_activate_first_iom_unavailable(self):
"""Test that if the first IOM is not available, the 2nd one is used."""
ioms = [
"bad.example.com",
"good.example.com"
]
module_args = {
'category': 'Update',
'command': 'FWActivate',
'username': 'USERID',
'password': 'PASSW0RD=21',
'ioms': ioms
}
set_module_args(module_args)
def mock_get_request(*args, **kwargs):
"""Mock for get_request that will fail on the 'bad' IOM."""
if "bad.example.com" in args[1]:
return MOCK_URL_ERROR
else:
return mock_get_request_enclosure_single_tenant(*args, **kwargs)
with patch.multiple(module.WdcRedfishUtils,
_firmware_activate_uri=mock_fw_activate_url,
_update_uri=mock_update_url,
_find_updateservice_resource=empty_return,
_find_updateservice_additional_uris=empty_return,
get_request=mock_get_request,
post_request=mock_post_request):
with self.assertRaises(AnsibleExitJson) as cm:
module.main()
self.assertEqual(ACTION_WAS_SUCCESSFUL_MESSAGE,
get_exception_message(cm))
def test_module_fw_activate_pass(self):
"""Test the FW Activate command in a passing scenario."""
# Run the same test twice -- once specifying ioms, and once specifying baseuri.
# Both should work the same way.
uri_specifiers = [
{
"ioms": ["example1.example.com"]
},
{
"baseuri": "example1.example.com"
}
]
for uri_specifier in uri_specifiers:
module_args = {
'category': 'Update',
'command': 'FWActivate',
'username': 'USERID',
'password': 'PASSW0RD=21',
}
module_args.update(uri_specifier)
set_module_args(module_args)
with patch.multiple("ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.WdcRedfishUtils",
_firmware_activate_uri=mock_fw_activate_url,
_update_uri=mock_update_url,
_find_updateservice_resource=empty_return,
_find_updateservice_additional_uris=empty_return,
get_request=mock_get_request_enclosure_single_tenant,
post_request=mock_post_request):
with self.assertRaises(AnsibleExitJson) as ansible_exit_json:
module.main()
self.assertEqual(ACTION_WAS_SUCCESSFUL_MESSAGE,
get_exception_message(ansible_exit_json))
self.assertTrue(is_changed(ansible_exit_json))
def test_module_fw_activate_service_does_not_support_fw_activate(self):
"""Test FW Activate when it is not supported."""
expected_error_message = "Service does not support FWActivate"
set_module_args({
'category': 'Update',
'command': 'FWActivate',
'username': 'USERID',
'password': 'PASSW0RD=21',
'ioms': ["example1.example.com"]
})
def mock_update_uri_response(*args, **kwargs):
return {
"ret": True,
"data": {} # No Actions
}
with patch.multiple(module.WdcRedfishUtils,
_firmware_activate_uri=mocked_url_response,
_update_uri=mock_update_url,
_find_updateservice_resource=empty_return,
_find_updateservice_additional_uris=empty_return,
get_request=mock_update_uri_response):
with self.assertRaises(AnsibleFailJson) as cm:
module.main()
self.assertEqual(expected_error_message,
get_exception_message(cm))
def test_module_update_and_activate_image_uri_not_http(self):
"""Test Update and Activate when URI is not http(s)"""
expected_error_message = "Bundle URI must be HTTP or HTTPS"
set_module_args({
'category': 'Update',
'command': 'UpdateAndActivate',
'username': 'USERID',
'password': 'PASSW0RD=21',
'ioms': ["example1.example.com"],
'update_image_uri': "ftp://example.com/image"
})
with patch.multiple(module.WdcRedfishUtils,
_firmware_activate_uri=mocked_url_response,
_update_uri=mock_update_url,
_find_updateservice_resource=empty_return,
_find_updateservice_additional_uris=empty_return):
with self.assertRaises(AnsibleFailJson) as cm:
module.main()
self.assertEqual(expected_error_message,
get_exception_message(cm))
def test_module_update_and_activate_target_not_ready_for_fw_update(self):
"""Test Update and Activate when target is not in the correct state."""
mock_status_code = 999
mock_status_description = "mock status description"
expected_error_message = "Target is not ready for FW update. Current status: {0} ({1})".format(
mock_status_code,
mock_status_description
)
set_module_args({
'category': 'Update',
'command': 'UpdateAndActivate',
'username': 'USERID',
'password': 'PASSW0RD=21',
'ioms': ["example1.example.com"],
'update_image_uri': "http://example.com/image"
})
with patch.object(module.WdcRedfishUtils, "get_simple_update_status") as mock_get_simple_update_status:
mock_get_simple_update_status.return_value = {
"ret": True,
"entries": {
"StatusCode": mock_status_code,
"Description": mock_status_description
}
}
with patch.multiple(module.WdcRedfishUtils,
_firmware_activate_uri=mocked_url_response,
_update_uri=mock_update_url,
_find_updateservice_resource=empty_return,
_find_updateservice_additional_uris=empty_return):
with self.assertRaises(AnsibleFailJson) as cm:
module.main()
self.assertEqual(expected_error_message,
get_exception_message(cm))
def test_module_update_and_activate_bundle_not_a_tarfile(self):
"""Test Update and Activate when bundle is not a tarfile"""
mock_filename = os.path.abspath(__file__)
expected_error_message = ERROR_MESSAGE_UNABLE_TO_EXTRACT_BUNDLE_VERSION
set_module_args({
'category': 'Update',
'command': 'UpdateAndActivate',
'username': 'USERID',
'password': 'PASSW0RD=21',
'ioms': ["example1.example.com"],
'update_image_uri': "http://example.com/image",
"update_creds": {
"username": "image_user",
"password": "image_password"
}
})
with patch('ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.fetch_file') as mock_fetch_file:
mock_fetch_file.return_value = mock_filename
with patch.multiple(module.WdcRedfishUtils,
get_simple_update_status=mock_get_simple_update_status_ready_for_fw_update,
_firmware_activate_uri=mocked_url_response,
_update_uri=mock_update_url,
_find_updateservice_resource=empty_return,
_find_updateservice_additional_uris=empty_return):
with self.assertRaises(AnsibleFailJson) as cm:
module.main()
self.assertEqual(expected_error_message,
get_exception_message(cm))
def test_module_update_and_activate_bundle_contains_no_firmware_version(self):
"""Test Update and Activate when bundle contains no firmware version"""
expected_error_message = ERROR_MESSAGE_UNABLE_TO_EXTRACT_BUNDLE_VERSION
set_module_args({
'category': 'Update',
'command': 'UpdateAndActivate',
'username': 'USERID',
'password': 'PASSW0RD=21',
'ioms': ["example1.example.com"],
'update_image_uri': "http://example.com/image",
"update_creds": {
"username": "image_user",
"password": "image_password"
}
})
tar_name = "empty_tarfile{0}.tar".format(uuid.uuid4())
empty_tarfile = tarfile.open(os.path.join(self.tempdir, tar_name), "w")
empty_tarfile.close()
with patch('ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.fetch_file') as mock_fetch_file:
mock_fetch_file.return_value = os.path.join(self.tempdir, tar_name)
with patch.multiple(module.WdcRedfishUtils,
get_simple_update_status=mock_get_simple_update_status_ready_for_fw_update,
_firmware_activate_uri=mocked_url_response,
_update_uri=mock_update_url,
_find_updateservice_resource=empty_return,
_find_updateservice_additional_uris=empty_return):
with self.assertRaises(AnsibleFailJson) as cm:
module.main()
self.assertEqual(expected_error_message,
get_exception_message(cm))
def test_module_update_and_activate_version_already_installed(self):
"""Test Update and Activate when the bundle version is already installed"""
mock_firmware_version = "1.2.3"
expected_error_message = ACTION_WAS_SUCCESSFUL_MESSAGE
set_module_args({
'category': 'Update',
'command': 'UpdateAndActivate',
'username': 'USERID',
'password': 'PASSW0RD=21',
'ioms': ["example1.example.com"],
'update_image_uri': "http://example.com/image",
"update_creds": {
"username": "image_user",
"password": "image_password"
}
})
tar_name = self.generate_temp_bundlefile(mock_firmware_version=mock_firmware_version,
is_multi_tenant=False)
with patch('ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.fetch_file') as mock_fetch_file:
mock_fetch_file.return_value = os.path.join(self.tempdir, tar_name)
with patch.multiple(module.WdcRedfishUtils,
get_firmware_inventory=mock_get_firmware_inventory_version_1_2_3,
get_simple_update_status=mock_get_simple_update_status_ready_for_fw_update,
_firmware_activate_uri=mocked_url_response,
_update_uri=mock_update_url,
_find_updateservice_resource=empty_return,
_find_updateservice_additional_uris=empty_return,
get_request=mock_get_request_enclosure_single_tenant):
with self.assertRaises(AnsibleExitJson) as result:
module.main()
self.assertEqual(expected_error_message,
get_exception_message(result))
self.assertFalse(is_changed(result))
def test_module_update_and_activate_version_already_installed_multi_tenant(self):
"""Test Update and Activate on multi-tenant when version is already installed"""
mock_firmware_version = "1.2.3"
expected_error_message = ACTION_WAS_SUCCESSFUL_MESSAGE
set_module_args({
'category': 'Update',
'command': 'UpdateAndActivate',
'username': 'USERID',
'password': 'PASSW0RD=21',
'ioms': ["example1.example.com"],
'update_image_uri': "http://example.com/image",
"update_creds": {
"username": "image_user",
"password": "image_password"
}
})
tar_name = self.generate_temp_bundlefile(mock_firmware_version=mock_firmware_version,
is_multi_tenant=True)
with patch('ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.fetch_file') as mock_fetch_file:
mock_fetch_file.return_value = os.path.join(self.tempdir, tar_name)
with patch.multiple(module.WdcRedfishUtils,
get_firmware_inventory=mock_get_firmware_inventory_version_1_2_3,
get_simple_update_status=mock_get_simple_update_status_ready_for_fw_update,
_firmware_activate_uri=mocked_url_response,
_update_uri=mock_update_url,
_find_updateservice_resource=empty_return,
_find_updateservice_additional_uris=empty_return,
get_request=mock_get_request_enclosure_multi_tenant):
with self.assertRaises(AnsibleExitJson) as result:
module.main()
self.assertEqual(expected_error_message,
get_exception_message(result))
self.assertFalse(is_changed(result))
def test_module_update_and_activate_pass(self):
"""Test Update and Activate (happy path)"""
mock_firmware_version = "1.2.2"
set_module_args({
'category': 'Update',
'command': 'UpdateAndActivate',
'username': 'USERID',
'password': 'PASSW0RD=21',
'ioms': ["example1.example.com"],
'update_image_uri': "http://example.com/image",
"update_creds": {
"username": "image_user",
"password": "image_password"
}
})
tar_name = self.generate_temp_bundlefile(mock_firmware_version=mock_firmware_version,
is_multi_tenant=False)
with patch('ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.fetch_file') as mock_fetch_file:
mock_fetch_file.return_value = os.path.join(self.tempdir, tar_name)
with patch.multiple("ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.WdcRedfishUtils",
get_firmware_inventory=mock_get_firmware_inventory_version_1_2_3,
simple_update=mock_simple_update,
_simple_update_status_uri=mocked_url_response,
# _find_updateservice_resource=empty_return,
# _find_updateservice_additional_uris=empty_return,
get_request=mock_get_request_enclosure_single_tenant,
post_request=mock_post_request):
with patch("ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.WdcRedfishUtils.get_simple_update_status"
) as mock_get_simple_update_status:
mock_get_simple_update_status.side_effect = MOCK_SIMPLE_UPDATE_STATUS_LIST
with self.assertRaises(AnsibleExitJson) as ansible_exit_json:
module.main()
self.assertTrue(is_changed(ansible_exit_json))
self.assertEqual(ACTION_WAS_SUCCESSFUL_MESSAGE, get_exception_message(ansible_exit_json))
def test_module_update_and_activate_pass_multi_tenant(self):
"""Test Update and Activate with multi-tenant (happy path)"""
mock_firmware_version = "1.2.2"
set_module_args({
'category': 'Update',
'command': 'UpdateAndActivate',
'username': 'USERID',
'password': 'PASSW0RD=21',
'ioms': ["example1.example.com"],
'update_image_uri': "http://example.com/image",
"update_creds": {
"username": "image_user",
"password": "image_password"
}
})
tar_name = self.generate_temp_bundlefile(mock_firmware_version=mock_firmware_version,
is_multi_tenant=True)
with patch('ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.fetch_file') as mock_fetch_file:
mock_fetch_file.return_value = os.path.join(self.tempdir, tar_name)
with patch.multiple(module.WdcRedfishUtils,
get_firmware_inventory=mock_get_firmware_inventory_version_1_2_3,
simple_update=mock_simple_update,
_simple_update_status_uri=mocked_url_response,
# _find_updateservice_resource=empty_return,
# _find_updateservice_additional_uris=empty_return,
get_request=mock_get_request_enclosure_multi_tenant,
post_request=mock_post_request):
with patch("ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.WdcRedfishUtils.get_simple_update_status"
) as mock_get_simple_update_status:
mock_get_simple_update_status.side_effect = MOCK_SIMPLE_UPDATE_STATUS_LIST
with self.assertRaises(AnsibleExitJson) as ansible_exit_json:
module.main()
self.assertTrue(is_changed(ansible_exit_json))
self.assertEqual(ACTION_WAS_SUCCESSFUL_MESSAGE, get_exception_message(ansible_exit_json))
def test_module_fw_update_multi_tenant_firmware_single_tenant_enclosure(self):
"""Test Update and Activate using multi-tenant bundle on single-tenant enclosure"""
mock_firmware_version = "1.1.1"
expected_error_message = "Enclosure multi-tenant is False but bundle multi-tenant is True"
set_module_args({
'category': 'Update',
'command': 'UpdateAndActivate',
'username': 'USERID',
'password': 'PASSW0RD=21',
'ioms': ["example1.example.com"],
'update_image_uri': "http://example.com/image",
"update_creds": {
"username": "image_user",
"password": "image_password"
}
})
tar_name = self.generate_temp_bundlefile(mock_firmware_version=mock_firmware_version,
is_multi_tenant=True)
with patch('ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.fetch_file') as mock_fetch_file:
mock_fetch_file.return_value = os.path.join(self.tempdir, tar_name)
with patch.multiple(module.WdcRedfishUtils,
get_firmware_inventory=mock_get_firmware_inventory_version_1_2_3(),
get_simple_update_status=mock_get_simple_update_status_ready_for_fw_update,
_firmware_activate_uri=mocked_url_response,
_update_uri=mock_update_url,
_find_updateservice_resource=empty_return,
_find_updateservice_additional_uris=empty_return,
get_request=mock_get_request_enclosure_single_tenant):
with self.assertRaises(AnsibleFailJson) as result:
module.main()
self.assertEqual(expected_error_message,
get_exception_message(result))
def test_module_fw_update_single_tentant_firmware_multi_tenant_enclosure(self):
"""Test Update and Activate using singe-tenant bundle on multi-tenant enclosure"""
mock_firmware_version = "1.1.1"
expected_error_message = "Enclosure multi-tenant is True but bundle multi-tenant is False"
set_module_args({
'category': 'Update',
'command': 'UpdateAndActivate',
'username': 'USERID',
'password': 'PASSW0RD=21',
'ioms': ["example1.example.com"],
'update_image_uri': "http://example.com/image",
"update_creds": {
"username": "image_user",
"password": "image_password"
}
})
tar_name = self.generate_temp_bundlefile(mock_firmware_version=mock_firmware_version,
is_multi_tenant=False)
with patch('ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.fetch_file') as mock_fetch_file:
mock_fetch_file.return_value = os.path.join(self.tempdir, tar_name)
with patch.multiple(module.WdcRedfishUtils,
get_firmware_inventory=mock_get_firmware_inventory_version_1_2_3(),
get_simple_update_status=mock_get_simple_update_status_ready_for_fw_update,
_firmware_activate_uri=mocked_url_response,
_update_uri=mock_update_url,
_find_updateservice_resource=empty_return,
_find_updateservice_additional_uris=empty_return,
get_request=mock_get_request_enclosure_multi_tenant):
with self.assertRaises(AnsibleFailJson) as result:
module.main()
self.assertEqual(expected_error_message,
get_exception_message(result))
def generate_temp_bundlefile(self,
mock_firmware_version,
is_multi_tenant):
"""Generate a temporary fake bundle file.
:param str mock_firmware_version: The simulated firmware version for the bundle.
:param bool is_multi_tenant: Is the simulated bundle multi-tenant?
This can be used for a mock FW update.
"""
tar_name = "tarfile{0}.tar".format(uuid.uuid4())
bundle_tarfile = tarfile.open(os.path.join(self.tempdir, tar_name), "w")
package_filename = "oobm-{0}.pkg".format(mock_firmware_version)
package_filename_path = os.path.join(self.tempdir, package_filename)
package_file = open(package_filename_path, "w")
package_file.close()
bundle_tarfile.add(os.path.join(self.tempdir, package_filename), arcname=package_filename)
bin_filename = "firmware.bin"
bin_filename_path = os.path.join(self.tempdir, bin_filename)
bin_file = open(bin_filename_path, "wb")
byte_to_write = b'\x80' if is_multi_tenant else b'\xFF'
bin_file.write(byte_to_write * 12)
bin_file.close()
for filename in [package_filename, bin_filename]:
bundle_tarfile.add(os.path.join(self.tempdir, filename), arcname=filename)
bundle_tarfile.close()
return tar_name

214
tests/unit/plugins/modules/remote_management/wdc/test_wdc_redfish_info.py Normal file
View file

@ -0,0 +1,214 @@
# -*- coding: utf-8 -*-
# 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
from ansible_collections.community.general.tests.unit.compat.mock import patch
from ansible_collections.community.general.tests.unit.compat import unittest
from ansible.module_utils import basic
import ansible_collections.community.general.plugins.modules.remote_management.redfish.wdc_redfish_info as module
from ansible_collections.community.general.tests.unit.plugins.modules.utils import AnsibleExitJson, AnsibleFailJson
from ansible_collections.community.general.tests.unit.plugins.modules.utils import set_module_args, exit_json, fail_json
MOCK_SUCCESSFUL_RESPONSE_WITH_ACTIONS = {
"ret": True,
"data": {
"Actions": {}
}
}
MOCK_SUCCESSFUL_HTTP_EMPTY_RESPONSE = {
"ret": True,
"data": {
}
}
MOCK_SUCCESSFUL_RESPONSE_WITH_UPDATE_SERVICE_RESOURCE = {
"ret": True,
"data": {
"UpdateService": {
"@odata.id": "/UpdateService"
}
}
}
MOCK_SUCCESSFUL_RESPONSE_WITH_SIMPLE_UPDATE_BUT_NO_FW_ACTIVATE = {
"ret": True,
"data": {
"Actions": {
"#UpdateService.SimpleUpdate": {
"target": "mocked value"
},
"Oem": {
"WDC": {} # No #UpdateService.FWActivate
}
}
}
}
def get_bin_path(self, arg, required=False):
"""Mock AnsibleModule.get_bin_path"""
return arg
def get_redfish_facts(ansible_exit_json):
"""From an AnsibleExitJson exception, get the redfish facts dict."""
return ansible_exit_json.exception.args[0]["redfish_facts"]
def get_exception_message(ansible_exit_json):
"""From an AnsibleExitJson exception, get the message string."""
return ansible_exit_json.exception.args[0]["msg"]
class TestWdcRedfishInfo(unittest.TestCase):
def setUp(self):
self.mock_module_helper = patch.multiple(basic.AnsibleModule,
exit_json=exit_json,
fail_json=fail_json,
get_bin_path=get_bin_path)
self.mock_module_helper.start()
self.addCleanup(self.mock_module_helper.stop)
def test_module_fail_when_required_args_missing(self):
with self.assertRaises(AnsibleFailJson):
set_module_args({})
module.main()
def test_module_fail_when_unknown_category(self):
with self.assertRaises(AnsibleFailJson):
set_module_args({
'category': 'unknown',
'command': 'SimpleUpdateStatus',
'username': 'USERID',
'password': 'PASSW0RD=21',
'ioms': [],
})
module.main()
def test_module_fail_when_unknown_command(self):
with self.assertRaises(AnsibleFailJson):
set_module_args({
'category': 'Update',
'command': 'unknown',
'username': 'USERID',
'password': 'PASSW0RD=21',
'ioms': [],
})
module.main()
def test_module_simple_update_status_pass(self):
set_module_args({
'category': 'Update',
'command': 'SimpleUpdateStatus',
'username': 'USERID',
'password': 'PASSW0RD=21',
'ioms': ["example1.example.com"],
})
def mock_simple_update_status(*args, **kwargs):
return {
"ret": True,
"data": {
"Description": "Ready for FW update",
"ErrorCode": 0,
"EstimatedRemainingMinutes": 0,
"StatusCode": 0
}
}
def mocked_string_response(*args, **kwargs):
return "mockedUrl"
def empty_return(*args, **kwargs):
return {"ret": True}
with patch.multiple(module.WdcRedfishUtils,
_simple_update_status_uri=mocked_string_response,
_find_updateservice_resource=empty_return,
_find_updateservice_additional_uris=empty_return,
get_request=mock_simple_update_status):
with self.assertRaises(AnsibleExitJson) as ansible_exit_json:
module.main()
redfish_facts = get_redfish_facts(ansible_exit_json)
self.assertEqual(mock_simple_update_status()["data"],
redfish_facts["simple_update_status"]["entries"])
def test_module_simple_update_status_updateservice_resource_not_found(self):
set_module_args({
'category': 'Update',
'command': 'SimpleUpdateStatus',
'username': 'USERID',
'password': 'PASSW0RD=21',
'ioms': ["example1.example.com"],
})
with patch.object(module.WdcRedfishUtils, 'get_request') as mock_get_request:
mock_get_request.return_value = {
"ret": True,
"data": {} # Missing UpdateService property
}
with self.assertRaises(AnsibleFailJson) as ansible_exit_json:
module.main()
self.assertEqual("UpdateService resource not found",
get_exception_message(ansible_exit_json))
def test_module_simple_update_status_service_does_not_support_simple_update(self):
set_module_args({
'category': 'Update',
'command': 'SimpleUpdateStatus',
'username': 'USERID',
'password': 'PASSW0RD=21',
'ioms': ["example1.example.com"],
})
def mock_get_request_function(uri):
mock_url_string = "mockURL"
if mock_url_string in uri:
return {
"ret": True,
"data": {
"Actions": { # No #UpdateService.SimpleUpdate
}
}
}
else:
return {
"ret": True,
"data": mock_url_string
}
with patch.object(module.WdcRedfishUtils, 'get_request') as mock_get_request:
mock_get_request.side_effect = mock_get_request_function
with self.assertRaises(AnsibleFailJson) as ansible_exit_json:
module.main()
self.assertEqual("UpdateService resource not found",
get_exception_message(ansible_exit_json))
def test_module_simple_update_status_service_does_not_support_fw_activate(self):
set_module_args({
'category': 'Update',
'command': 'SimpleUpdateStatus',
'username': 'USERID',
'password': 'PASSW0RD=21',
'ioms': ["example1.example.com"],
})
def mock_get_request_function(uri):
if uri.endswith("/redfish/v1") or uri.endswith("/redfish/v1/"):
return MOCK_SUCCESSFUL_RESPONSE_WITH_UPDATE_SERVICE_RESOURCE
elif uri.endswith("/mockedUrl"):
return MOCK_SUCCESSFUL_HTTP_EMPTY_RESPONSE
elif uri.endswith("/UpdateService"):
return MOCK_SUCCESSFUL_RESPONSE_WITH_SIMPLE_UPDATE_BUT_NO_FW_ACTIVATE
else:
raise RuntimeError("Illegal call to get_request in test: " + uri)
with patch("ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.WdcRedfishUtils.get_request") as mock_get_request:
mock_get_request.side_effect = mock_get_request_function
with self.assertRaises(AnsibleFailJson) as ansible_exit_json:
module.main()
self.assertEqual("Service does not support FWActivate",
get_exception_message(ansible_exit_json))