x/python-scciclient
Code Issues Proposed changes

Merge "Support firmware upgrade for iRMC and BIOS"

Browse Source
This commit is contained in:
Zuul
2019-03-11 08:48:21 +00:00
committed by Gerrit Code Review
parent d751993972 1d5668953f
commit eb3c658fc4
2 changed files with 393 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

132
scciclient/irmc/scci.py
View File

@@ -68,6 +68,11 @@ class SCCIRAIDNotReady(SCCIError):
super(SCCIRAIDNotReady, self).__init__(message)
class SCCISessionTimeout(SCCIError):
def __init__(self, message):
super(SCCISessionTimeout, self).__init__(message)
"""
List of iRMC S4 supported SCCI commands
@@ -275,13 +280,25 @@ def scci_cmd(host, userid, password, cmd, port=443, auth_method='basic',
try:
header = {'Content-type': 'application/x-www-form-urlencoded'}
r = requests.post(protocol + '://' + host + '/config',
data=cmd,
if kwargs.get('upgrade_type') == 'irmc':
with open(cmd, 'rb') as file:
data = file.read()
config_type = '/irmcupdate?flashSelect=255'
elif kwargs.get('upgrade_type') == 'bios':
with open(cmd, 'rb') as file:
data = file.read()
config_type = '/biosupdate'
else:
data = cmd
config_type = '/config'
r = requests.post(protocol + '://' + host + config_type,
data=data,
headers=header,
verify=False,
timeout=client_timeout,
allow_redirects=False,
auth=auth_obj)
if not do_async:
time.sleep(5)
if DEBUG:
@@ -306,6 +323,9 @@ def scci_cmd(host, userid, password, cmd, port=443, auth_method='basic',
else:
return r
except IOError as input_error:
raise SCCIClientError(input_error)
except ET.ParseError as parse_error:
raise SCCIClientError(parse_error)
@@ -560,6 +580,54 @@ def get_capabilities_properties(d_info,
raise SCCIClientError('Capabilities inspection failed: %s' % err)
def process_session_status(irmc_info, session_timeout, upgrade_type):
"""process session for Bios config backup/restore or RAID config operation
:param irmc_info: node info
:param session_timeout: session timeout
:param upgrade_type: flag to check upgrade with bios or irmc
:return: a dict with following values:
{
'upgrade_message': <Message of firmware upgrade mechanism>,
'upgrade_status'
}
"""
session_expiration = time.time() + session_timeout
while time.time() < session_expiration:
try:
# Get session status to check
session = get_firmware_upgrade_status(irmc_info, upgrade_type)
except SCCIClientError:
# Ignore checking during rebooted server
time.sleep(10)
continue
status = session.find("./Value").text
severity = session.find("./Severity").text
message = session.find("./Message").text
result = {}
if severity == 'Information' and status != '0':
if 'FLASH successful' in message:
result['upgrade_status'] = 'Complete'
return result
# Sleep a bit
time.sleep(5)
elif severity == 'Error':
result['upgrade_status'] = 'Error'
return result
else:
# Error occurred, get session log to see what happened
session_log = message
raise SCCIClientError('Failed to set firmware upgrade. '
'Session log is %s.' % session_log)
else:
raise SCCISessionTimeout('Failed to time out mechanism with %s.'
% session_expiration)
def get_raid_fgi_status(report):
"""Gather fgi(foreground initialization) information of raid configuration
@@ -590,3 +658,63 @@ def get_raid_fgi_status(report):
fgi_status.update({name: status})
return fgi_status
def get_firmware_upgrade_status(irmc_info, upgrade_type):
"""get firmware upgrade status of bios or irmc
:param irmc_info: dict of iRMC params to access the server node
{
'irmc_address': host,
'irmc_username': user_id,
'irmc_password': password,
'irmc_port': 80 or 443, default is 443,
'irmc_auth_method': 'basic' or 'digest', default is 'digest',
'irmc_client_timeout': timeout, default is 60,
...
}
:param upgrade_type: flag to check upgrade with bios or irmc
:raises: ISCCIInvalidInputError if port and/or auth_method params
are invalid
:raises: SCCIClientError if SCCI failed
"""
host = irmc_info.get('irmc_address')
userid = irmc_info.get('irmc_username')
password = irmc_info.get('irmc_password')
port = irmc_info.get('irmc_port', 443)
auth_method = irmc_info.get('irmc_auth_method', 'digest')
client_timeout = irmc_info.get('irmc_client_timeout', 60)
auth_obj = None
try:
protocol = {80: 'http', 443: 'https'}[port]
auth_obj = {
'basic': requests.auth.HTTPBasicAuth(userid, password),
'digest': requests.auth.HTTPDigestAuth(userid, password)
}[auth_method.lower()]
except KeyError:
raise SCCIInvalidInputError(
("Invalid port %(port)d or " +
"auth_method for method %(auth_method)s") %
{'port': port, 'auth_method': auth_method})
try:
if upgrade_type == 'bios':
upgrade_type = '/biosprogress'
elif upgrade_type == 'irmc':
upgrade_type = '/irmcprogress'
r = requests.get(protocol + '://' + host + upgrade_type,
verify=False,
timeout=(10, client_timeout),
allow_redirects=False,
auth=auth_obj)
if r.status_code not in (200, 201):
raise SCCIClientError(
('HTTP PROTOCOL ERROR, STATUS CODE = %s' %
str(r.status_code)))
upgrade_status_xml = ET.fromstring(r.text)
return upgrade_status_xml
except ET.ParseError as parse_error:
raise SCCIClientError(parse_error)

263
scciclient/tests/irmc/test_scci.py
View File

@@ -21,12 +21,18 @@ import xml.etree.ElementTree as ET
import mock
from requests_mock.contrib import fixture as rm_fixture
import six
import six.moves.builtins as __builtin__
import testtools
from scciclient.irmc import ipmi
from scciclient.irmc import scci
from scciclient.irmc import snmp
if six.PY3:
import io
file = io.BytesIO
class SCCITestCase(testtools.TestCase):
"""Tests for SCCI
@@ -912,3 +918,260 @@ class SCCITestCase(testtools.TestCase):
fgi_status_expect = {'0': 'Idle'}
result = scci.get_raid_fgi_status(report_fake)
self.assertEqual(result, fgi_status_expect)
@mock.patch('scciclient.irmc.scci.requests')
def test_fail_get_bios_firmware_status(self, mock_requests):
mock_requests.get.return_value = mock.Mock(
status_code=404,
text="""</head>
<body>
<div id="main">
<div class="content message">
<div id="alert_icon">
</div>
<div id="msg_title" class="title">
File not found <!-- This tag is OPTIONAL and is the
title eg. 404 - File not found -->
</div>
</div>
</div>
</body>
</html>""")
upgrade_type = 'biosss'
self.assertRaises(scci.SCCIClientError,
scci.get_firmware_upgrade_status, self.irmc_info,
upgrade_type=upgrade_type)
@mock.patch('scciclient.irmc.scci.requests')
def test_success_get_bios_firmware_status(self, mock_requests):
mock_requests.get.return_value = mock.Mock(
return_value='ok',
status_code=200,
text="""<?xml version="1.0" encoding="UTF-8"?>
<Status>
<Value>0</Value>
<Severity>Information</Severity>
<Message>No Error</Message>
</Status>""")
expected_status = "0"
expected_severity = "Information"
expected_message = "No Error"
upgrade_type = 'bios'
result = scci.get_firmware_upgrade_status(self.irmc_info, upgrade_type)
self.assertEqual(expected_status, result.find("./Value").text)
self.assertEqual(expected_severity, result.find("./Severity").text)
self.assertEqual(expected_message, result.find("./Message").text)
@mock.patch('scciclient.irmc.scci.requests')
def test_fail_get_irmc_firmware_status(self, mock_requests):
mock_requests.get.return_value = mock.Mock(
status_code=404,
text="""</head>
<body>
<div id="main">
<div class="content message">
<div id="alert_icon">
</div>
<div id="msg_title" class="title">
File not found <!-- This tag is OPTIONAL and is the
title eg. 404 - File not found -->
</div>
</div>
</div>
</body>
</html>""")
upgrade_type = 'irmcccc'
self.assertRaises(scci.SCCIClientError,
scci.get_firmware_upgrade_status, self.irmc_info,
upgrade_type=upgrade_type)
@mock.patch('scciclient.irmc.scci.requests')
def test_success_get_irmc_firmware_status(self, mock_requests):
mock_requests.get.return_value = mock.Mock(
return_value='ok',
status_code=200,
text="""<?xml version="1.0" encoding="UTF-8"?>
<Status>
<Value>0</Value>
<Severity>Information</Severity>
<Message>No Error</Message>
</Status>""")
expected_status = "0"
expected_severity = "Information"
expected_message = "No Error"
upgrade_type = 'irmc'
result = scci.get_firmware_upgrade_status(self.irmc_info, upgrade_type)
self.assertEqual(expected_status, result.find("./Value").text)
self.assertEqual(expected_severity, result.find("./Severity").text)
self.assertEqual(expected_message, result.find("./Message").text)
@mock.patch('scciclient.irmc.scci.get_firmware_upgrade_status')
def test_failed_process_session_bios_status(self, mock_status):
session_timeout = 180
upgrade_type = 'bios'
# Fake status from server
status_fake = ET.Element(self)
status_fake.append(ET.Element("Value", name="Value"))
status_fake.append(ET.Element("Severity", name="Severity"))
status_fake.append(ET.Element("Message", name="Message"))
status_fake.find("./Value").text = '0'
status_fake.find("./Severity").text = 'Error'
status_fake.find("./Message").text = 'File not provided'
mock_status.return_value = status_fake
expected_status = 'Error'
result = scci.process_session_status(self.irmc_info, session_timeout,
upgrade_type)
self.assertEqual(expected_status, result['upgrade_status'])
@mock.patch('scciclient.irmc.scci.get_firmware_upgrade_status')
def test_success_process_session_bios_status(self, mock_status):
session_timeout = 180
upgrade_type = 'bios'
# Fake status from server
status_fake = ET.Element(self)
status_fake.append(ET.Element("Value", name="Value"))
status_fake.append(ET.Element("Severity", name="Severity"))
status_fake.append(ET.Element("Message", name="Message"))
status_fake.find("./Value").text = '9'
status_fake.find("./Severity").text = 'Information'
status_fake.find("./Message").text = 'FLASH successful'
mock_status.return_value = status_fake
expected_status = 'Complete'
result = scci.process_session_status(self.irmc_info, session_timeout,
upgrade_type)
self.assertEqual(expected_status, result['upgrade_status'])
@mock.patch('scciclient.irmc.scci.get_firmware_upgrade_status')
def test_failed_process_session_irmc_status(self, mock_status):
session_timeout = 180
upgrade_type = 'irmc'
# Fake status from server
status_fake = ET.Element(self)
status_fake.append(ET.Element("Value", name="Value"))
status_fake.append(ET.Element("Severity", name="Severity"))
status_fake.append(ET.Element("Message", name="Message"))
status_fake.find("./Value").text = '0'
status_fake.find("./Severity").text = 'Error'
status_fake.find("./Message").text = 'File not provided'
mock_status.return_value = status_fake
expected_status = 'Error'
result = scci.process_session_status(self.irmc_info, session_timeout,
upgrade_type)
self.assertEqual(expected_status, result['upgrade_status'])
@mock.patch('scciclient.irmc.scci.get_firmware_upgrade_status')
def test_success_process_session_irmc_status(self, mock_status):
session_timeout = 180
upgrade_type = 'irmc'
# Fake status from server
status_fake = ET.Element(self)
status_fake.append(ET.Element("Value", name="Value"))
status_fake.append(ET.Element("Severity", name="Severity"))
status_fake.append(ET.Element("Message", name="Message"))
status_fake.find("./Value").text = '9'
status_fake.find("./Severity").text = 'Information'
status_fake.find("./Message").text = 'FLASH successful'
mock_status.return_value = status_fake
expected_status = 'Complete'
result = scci.process_session_status(self.irmc_info, session_timeout,
upgrade_type)
self.assertEqual(expected_status, result['upgrade_status'])
@mock.patch.object(__builtin__, 'open', autospec=True)
def test_create_bios_firmware_upgrade(self, open_mock):
upgrade_type = 'bios'
self.requests_mock.post("http://" + self.irmc_address + "/biosupdate",
text="""<?xml version="1.0" encoding="UTF-8"?>
<Status>
<Value>0</Value>
<Severity>Information</Severity>
<Message>No Error</Message>
</Status>""")
bios_input = '/media/DATA/D3099-B1.UPC'
open_mock.return_value = mock.mock_open(read_data="file").return_value
client = scci.get_client(self.irmc_address,
self.irmc_username,
self.irmc_password,
port=self.irmc_port,
auth_method=self.irmc_auth_method,
client_timeout=self.irmc_client_timeout,
upgrade_type=upgrade_type)
r = client(bios_input)
self.assertEqual(r.status_code, 200)
@mock.patch.object(__builtin__, 'open', side_effect=IOError, autospec=True)
def test_create_fail_bios_firmware_upgrade(self, open_mock):
upgrade_type = 'bios'
self.requests_mock.post("http://" + self.irmc_address + "/biosupdate",
text="""<?xml version="1.0" encoding="UTF-8"?>
<Status>
<Value>6</Value>
<Severity>Error</Severity>
<Message>File not provided</Message>
</Status>
""")
# Fake wrong file directory
bios_input = '/media/DATA/D3099-B101.UPC'
open_mock.return_value = mock.mock_open(read_data="file").return_value
client = scci.get_client(self.irmc_address,
self.irmc_username,
self.irmc_password,
port=self.irmc_port,
auth_method=self.irmc_auth_method,
client_timeout=self.irmc_client_timeout,
upgrade_type=upgrade_type)
self.assertRaises(scci.SCCIClientError, client, bios_input)
@mock.patch.object(__builtin__, 'open', autospec=True)
def test_create_irmc_firmware_upgrade(self, open_mock):
upgrade_type = 'irmc'
self.requests_mock.post("http://" + self.irmc_address +
"/irmcupdate?flashSelect=255",
text="""<?xml version="1.0" encoding="UTF-8"?>
<Status>
<Value>0</Value>
<Severity>Information</Severity>
<Message>No Error</Message>
</Status>""")
irmc_input = '/media/DATA/TX2540M1.bin'
open_mock.return_value = mock.mock_open(read_data="file").return_value
client = scci.get_client(self.irmc_address,
self.irmc_username,
self.irmc_password,
port=self.irmc_port,
auth_method=self.irmc_auth_method,
client_timeout=self.irmc_client_timeout,
upgrade_type=upgrade_type)
r = client(irmc_input)
self.assertEqual(r.status_code, 200)
@mock.patch.object(__builtin__, 'open', side_effect=IOError, autospec=True)
def test_create_fail_irmc_firmware_upgrade(self, open_mock):
upgrade_type = 'irmc'
self.requests_mock.post("http://" + self.irmc_address +
"/irmcupdate?flashSelect=255",
text="""<?xml version="1.0" encoding="UTF-8"?>
<Status>
<Value>6</Value>
<Severity>Error</Severity>
<Message>File not provided</Message>
</Status>
""")
# Fake wrong file directory
irmc_input = '/media/DATA/TX2540M1111.bin'
mock_file_handle = mock.MagicMock(spec=file)
open_mock.return_value = mock_file_handle
client = scci.get_client(self.irmc_address,
self.irmc_username,
self.irmc_password,
port=self.irmc_port,
auth_method=self.irmc_auth_method,
client_timeout=self.irmc_client_timeout,
upgrade_type=upgrade_type)
self.assertRaises(scci.SCCIClientError, client, irmc_input)