This commit fixes incompatibility imported in4b751fb3f4which changes default value passed to requests module function. Change-Id: I67663e6bd6e060f572e9c26cf504abd31ba49402 (cherry picked from commit56273075aa)
		
			
				
	
	
		
			1185 lines
		
	
	
		
			53 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
			
		
		
	
	
			1185 lines
		
	
	
		
			53 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
#
 | 
						|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
 | 
						|
# not use this file except in compliance with the License. You may obtain
 | 
						|
# a copy of the License at
 | 
						|
#
 | 
						|
#      http://www.apache.org/licenses/LICENSE-2.0
 | 
						|
#
 | 
						|
# Unless required by applicable law or agreed to in writing, software
 | 
						|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 | 
						|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 | 
						|
# License for the specific language governing permissions and limitations
 | 
						|
# under the License.
 | 
						|
 | 
						|
"""
 | 
						|
Test class for iRMC Power Driver
 | 
						|
"""
 | 
						|
 | 
						|
import os
 | 
						|
import time
 | 
						|
from unittest import mock
 | 
						|
import xml.etree.ElementTree as ET
 | 
						|
 | 
						|
import defusedxml.ElementTree as dET
 | 
						|
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
 | 
						|
 | 
						|
    Unit Test Cases for power on/off/reset, and mount cd/fd
 | 
						|
    """
 | 
						|
 | 
						|
    def setUp(self):
 | 
						|
        super(SCCITestCase, self).setUp()
 | 
						|
 | 
						|
        self.requests_mock = self.useFixture(rm_fixture.Fixture())
 | 
						|
 | 
						|
        with open(os.path.join(
 | 
						|
                os.path.dirname(__file__),
 | 
						|
                'fixtures/irmc_report_ok.xml'), "r") as report_ok:
 | 
						|
            self.report_ok_txt = report_ok.read()
 | 
						|
        self.report_ok_xml = dET.fromstring(self.report_ok_txt)
 | 
						|
 | 
						|
        with open(os.path.join(
 | 
						|
                os.path.dirname(__file__),
 | 
						|
                'fixtures/irmc_report_ng.xml'), "r") as report_ng:
 | 
						|
            self.report_ng_txt = report_ng.read()
 | 
						|
        self.report_ng_xml = dET.fromstring(self.report_ng_txt)
 | 
						|
 | 
						|
        self.irmc_address = '10.124.196.159'
 | 
						|
        self.irmc_username = 'admin'
 | 
						|
        self.irmc_password = 'admin0'
 | 
						|
        self.irmc_port = 80
 | 
						|
        self.irmc_auth_method = 'basic'
 | 
						|
        self.irmc_client_timeout = 60
 | 
						|
        self.irmc_info = {'irmc_address': self.irmc_address,
 | 
						|
                          'irmc_username': self.irmc_username,
 | 
						|
                          'irmc_password': self.irmc_password,
 | 
						|
                          'irmc_snmp_port': 161,
 | 
						|
                          'irmc_snmp_version': 'v2c',
 | 
						|
                          'irmc_snmp_community': 'public',
 | 
						|
                          'irmc_snmp_security': None,
 | 
						|
                          'irmc_client_timeout': self.irmc_client_timeout,
 | 
						|
                          'irmc_sensor_method': 'ipmitool',
 | 
						|
                          'irmc_auth_method': self.irmc_auth_method,
 | 
						|
                          'irmc_port': 443,
 | 
						|
                          'irmc_tempdir': "/tmp"
 | 
						|
                          }
 | 
						|
 | 
						|
        self.irmc_remote_image_server = '10.33.110.49'
 | 
						|
        self.irmc_remote_image_user_domain = 'example.local'
 | 
						|
        self.irmc_remote_image_share_type = scci.ShareType.nfs
 | 
						|
        self.irmc_remote_image_share_name = 'share'
 | 
						|
        self.irmc_remote_image_deploy_iso = 'ubuntu-14.04.1-server-amd64.iso'
 | 
						|
        self.irmc_remote_image_username = 'deployer'
 | 
						|
        self.irmc_remote_image_user_password = 'password'
 | 
						|
 | 
						|
    def test_get_share_type_ok(self):
 | 
						|
        nfs_result = scci.get_share_type("nfs")
 | 
						|
        self.assertEqual(scci.ShareType.nfs, nfs_result)
 | 
						|
        cifs_result = scci.get_share_type("cifs")
 | 
						|
        self.assertEqual(scci.ShareType.cifs, cifs_result)
 | 
						|
 | 
						|
        NFS_result = scci.get_share_type("NFS")
 | 
						|
        self.assertEqual(scci.ShareType.nfs, NFS_result)
 | 
						|
        CIFS_result = scci.get_share_type("CIFS")
 | 
						|
        self.assertEqual(scci.ShareType.cifs, CIFS_result)
 | 
						|
 | 
						|
    def test_get_share_type_ng(self):
 | 
						|
        self.assertRaises(KeyError,
 | 
						|
                          scci.get_share_type,
 | 
						|
                          "abc")
 | 
						|
 | 
						|
    @mock.patch('scciclient.irmc.scci.requests')
 | 
						|
    def test_scci_cmd_protocol_https_ok(self, mock_requests):
 | 
						|
        https_port = 443
 | 
						|
        mock_requests.post.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>""")
 | 
						|
        returned_mock_requests_post = scci.scci_cmd(
 | 
						|
            self.irmc_address,
 | 
						|
            self.irmc_username,
 | 
						|
            self.irmc_password,
 | 
						|
            scci.POWER_ON,
 | 
						|
            port=https_port,
 | 
						|
            auth_method=self.irmc_auth_method,
 | 
						|
            client_timeout=self.irmc_client_timeout)
 | 
						|
        mock_requests.post.assert_called_with(
 | 
						|
            'https://' + self.irmc_address + '/config',
 | 
						|
            data=scci.POWER_ON,
 | 
						|
            headers={'Content-type': 'application/x-www-form-urlencoded'},
 | 
						|
            verify=False,
 | 
						|
            timeout=self.irmc_client_timeout,
 | 
						|
            allow_redirects=False,
 | 
						|
            auth=mock_requests.auth.HTTPBasicAuth(self.irmc_username,
 | 
						|
                                                  self.irmc_password))
 | 
						|
        self.assertEqual('ok', returned_mock_requests_post.return_value)
 | 
						|
 | 
						|
    def test_scci_cmd_protocol_http_and_auth_basic_ok(self):
 | 
						|
        self.requests_mock.post("http://" + self.irmc_address + "/config",
 | 
						|
                                text="""<?xml version="1.0" encoding="UTF-8"?>
 | 
						|
                                        <Status>
 | 
						|
                                            <Value>0</Value>
 | 
						|
                                            <Severity>Information</Severity>
 | 
						|
                                            <Message>No Error</Message>
 | 
						|
                                        </Status>""")
 | 
						|
 | 
						|
        r = scci.scci_cmd(self.irmc_address,
 | 
						|
                          self.irmc_username,
 | 
						|
                          self.irmc_password,
 | 
						|
                          scci.POWER_ON,
 | 
						|
                          port=self.irmc_port,
 | 
						|
                          auth_method=self.irmc_auth_method,
 | 
						|
                          client_timeout=self.irmc_client_timeout)
 | 
						|
        self.assertEqual(r.status_code, 200)
 | 
						|
 | 
						|
    def test_scci_cmd_protocol_http_and_auth_digest_ok(self):
 | 
						|
        self.requests_mock.post("http://" + self.irmc_address + "/config",
 | 
						|
                                text="""<?xml version="1.0" encoding="UTF-8"?>
 | 
						|
                                     <Status>
 | 
						|
                                         <Value>0</Value>
 | 
						|
                                         <Severity>Information</Severity>
 | 
						|
                                         <Message>No Error</Message>
 | 
						|
                                     </Status>""")
 | 
						|
 | 
						|
        auth_digest = 'digest'
 | 
						|
        r = scci.scci_cmd(self.irmc_address,
 | 
						|
                          self.irmc_username,
 | 
						|
                          self.irmc_password,
 | 
						|
                          scci.POWER_ON,
 | 
						|
                          port=self.irmc_port,
 | 
						|
                          auth_method=auth_digest,
 | 
						|
                          client_timeout=self.irmc_client_timeout)
 | 
						|
        self.assertEqual(r.status_code, 200)
 | 
						|
 | 
						|
    def test_scci_cmd_authentication_failure(self):
 | 
						|
        self.requests_mock.post("http://" + self.irmc_address + "/config",
 | 
						|
                                text="401 Unauthorized",
 | 
						|
                                status_code=401)
 | 
						|
 | 
						|
        e = self.assertRaises(scci.SCCIClientError,
 | 
						|
                              scci.scci_cmd,
 | 
						|
                              self.irmc_address,
 | 
						|
                              self.irmc_username,
 | 
						|
                              self.irmc_password,
 | 
						|
                              scci.POWER_ON,
 | 
						|
                              port=self.irmc_port,
 | 
						|
                              auth_method=self.irmc_auth_method,
 | 
						|
                              client_timeout=self.irmc_client_timeout)
 | 
						|
 | 
						|
        self.assertEqual(
 | 
						|
            'HTTP PROTOCOL ERROR, STATUS CODE = 401',
 | 
						|
            str(e))
 | 
						|
 | 
						|
    def test_scci_cmd_protocol_ng(self):
 | 
						|
        ssh_port = 22
 | 
						|
        e = self.assertRaises(scci.SCCIInvalidInputError,
 | 
						|
                              scci.scci_cmd,
 | 
						|
                              self.irmc_address,
 | 
						|
                              self.irmc_username,
 | 
						|
                              self.irmc_password,
 | 
						|
                              scci.POWER_ON,
 | 
						|
                              port=ssh_port,
 | 
						|
                              auth_method=self.irmc_auth_method,
 | 
						|
                              client_timeout=self.irmc_client_timeout)
 | 
						|
        self.assertEqual((("Invalid port %(port)d or "
 | 
						|
                           "auth_method for method %(auth_method)s") %
 | 
						|
                          {'port': ssh_port,
 | 
						|
                           'auth_method': self.irmc_auth_method}), str(e))
 | 
						|
 | 
						|
    def test_scci_cmd_auth_method_ng(self):
 | 
						|
        unknown_auth_method = 'unknown'
 | 
						|
        e = self.assertRaises(scci.SCCIInvalidInputError,
 | 
						|
                              scci.scci_cmd,
 | 
						|
                              self.irmc_address,
 | 
						|
                              self.irmc_username,
 | 
						|
                              self.irmc_password,
 | 
						|
                              scci.POWER_ON,
 | 
						|
                              port=self.irmc_port,
 | 
						|
                              auth_method=unknown_auth_method,
 | 
						|
                              client_timeout=self.irmc_client_timeout)
 | 
						|
        self.assertEqual(("Invalid port %(port)d or "
 | 
						|
                          "auth_method for method %(auth_method)s") %
 | 
						|
                         {'port': self.irmc_port,
 | 
						|
                          'auth_method': unknown_auth_method}, str(e))
 | 
						|
 | 
						|
    def test_power_on_scci_xml_parse_failed(self):
 | 
						|
        self.requests_mock.post(
 | 
						|
            "http://" + self.irmc_address + "/config",
 | 
						|
            text="""<?xml version="1.0" encoding="UTF-8"?>
 | 
						|
            <Status>
 | 
						|
            <Value>31</Value>
 | 
						|
            <Severity>Error</Severity>
 | 
						|
            <Message>Error 31 (Import of settings in WinSCU XML format failed)
 | 
						|
            occurred</Message>
 | 
						|
            <Error Context="SCCI" OC="0" OE="0" OI="0">
 | 
						|
            XML parser creation failed (Error parsing:
 | 
						|
            attribute value should start with a quote
 | 
						|
            SEQ>  <CMD Context="SCCI" OC=PowerOnCabinet OE="0" OI="0"
 | 
						|
            -----------------------------^------------------------------).
 | 
						|
            </Error>
 | 
						|
            </Status>""")
 | 
						|
 | 
						|
        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)
 | 
						|
 | 
						|
        e = self.assertRaises(scci.SCCIClientError,
 | 
						|
                              client,
 | 
						|
                              scci.POWER_ON)
 | 
						|
        self.assertEqual(
 | 
						|
            'not well-formed (invalid token): line 10, column 41',
 | 
						|
            str(e))
 | 
						|
 | 
						|
    def test_power_on_http_failed(self):
 | 
						|
        self.requests_mock.post("http://" + self.irmc_address + "/config",
 | 
						|
                                text="anything",
 | 
						|
                                status_code=302)
 | 
						|
 | 
						|
        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)
 | 
						|
 | 
						|
        e = self.assertRaises(scci.SCCIClientError,
 | 
						|
                              client,
 | 
						|
                              scci.POWER_ON)
 | 
						|
        self.assertEqual(
 | 
						|
            'HTTP PROTOCOL ERROR, STATUS CODE = 302',
 | 
						|
            str(e))
 | 
						|
 | 
						|
    @mock.patch.object(time, 'sleep')
 | 
						|
    def test_power_on_ok(self, mock_sleep):
 | 
						|
        self.requests_mock.post("http://" + self.irmc_address + "/config",
 | 
						|
                                text="""<?xml version="1.0" encoding="UTF-8"?>
 | 
						|
                                <Status>
 | 
						|
                                    <Value>0</Value>
 | 
						|
                                    <Severity>Information</Severity>
 | 
						|
                                    <Message>No Error</Message>
 | 
						|
                                </Status>""")
 | 
						|
 | 
						|
        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)
 | 
						|
        r = client(scci.POWER_ON)
 | 
						|
        self.assertEqual(r.status_code, 200)
 | 
						|
        mock_sleep.assert_called_once_with(3)
 | 
						|
 | 
						|
    @mock.patch.object(time, 'sleep')
 | 
						|
    def test_power_off_ok(self, mock_sleep):
 | 
						|
        self.requests_mock.post("http://" + self.irmc_address + "/config",
 | 
						|
                                text="""<?xml version="1.0" encoding="UTF-8"?>
 | 
						|
                                <Status>
 | 
						|
                                    <Value>0</Value>
 | 
						|
                                    <Severity>Information</Severity>
 | 
						|
                                    <Message>No Error</Message>
 | 
						|
                                </Status>""")
 | 
						|
 | 
						|
        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)
 | 
						|
        r = client(scci.POWER_OFF)
 | 
						|
        self.assertEqual(r.status_code, 200)
 | 
						|
        mock_sleep.assert_called_once_with(3)
 | 
						|
 | 
						|
    def test_power_cycle_ok(self):
 | 
						|
        self.requests_mock.post("http://" + self.irmc_address + "/config",
 | 
						|
                                text="""<?xml version="1.0" encoding="UTF-8"?>
 | 
						|
                                <Status>
 | 
						|
                                    <Value>0</Value>
 | 
						|
                                    <Severity>Information</Severity>
 | 
						|
                                    <Message>No Error</Message>
 | 
						|
                                </Status>""")
 | 
						|
 | 
						|
        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)
 | 
						|
        r = client(scci.POWER_CYCLE)
 | 
						|
        self.assertEqual(r.status_code, 200)
 | 
						|
 | 
						|
    @mock.patch.object(time, 'sleep')
 | 
						|
    def test_power_reset_ok(self, mock_sleep):
 | 
						|
        self.requests_mock.post("http://" + self.irmc_address + "/config",
 | 
						|
                                text="""<?xml version="1.0" encoding="UTF-8"?>
 | 
						|
                                <Status>
 | 
						|
                                    <Value>0</Value>
 | 
						|
                                    <Severity>Information</Severity>
 | 
						|
                                    <Message>No Error</Message>
 | 
						|
                                </Status>""")
 | 
						|
 | 
						|
        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)
 | 
						|
        r = client(scci.POWER_RESET)
 | 
						|
        self.assertEqual(r.status_code, 200)
 | 
						|
        mock_sleep.assert_called_once_with(3)
 | 
						|
 | 
						|
    def test_power_raise_nmi_ok(self):
 | 
						|
        self.requests_mock.post("http://" + self.irmc_address + "/config",
 | 
						|
                                text="""<?xml version="1.0" encoding="UTF-8"?>
 | 
						|
                                <Status>
 | 
						|
                                    <Value>0</Value>
 | 
						|
                                    <Severity>Information</Severity>
 | 
						|
                                    <Message>No Error</Message>
 | 
						|
                                </Status>""")
 | 
						|
 | 
						|
        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)
 | 
						|
        r = client(scci.POWER_RAISE_NMI)
 | 
						|
        self.assertEqual(r.status_code, 200)
 | 
						|
 | 
						|
    def test_power_soft_off_ok(self):
 | 
						|
        self.requests_mock.post("http://" + self.irmc_address + "/config",
 | 
						|
                                text="""<?xml version="1.0" encoding="UTF-8"?>
 | 
						|
                                <Status>
 | 
						|
                                    <Value>0</Value>
 | 
						|
                                    <Severity>Information</Severity>
 | 
						|
                                    <Message>No Error</Message>
 | 
						|
                                </Status>""")
 | 
						|
 | 
						|
        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)
 | 
						|
        r = client(scci.POWER_SOFT_OFF)
 | 
						|
        self.assertEqual(r.status_code, 200)
 | 
						|
 | 
						|
    def test_power_soft_off_ng(self):
 | 
						|
        self.requests_mock.post(
 | 
						|
            "http://" + self.irmc_address + "/config",
 | 
						|
            text="""<?xml version="1.0" encoding="UTF-8"?>
 | 
						|
            <Status>
 | 
						|
            <Value>31</Value>
 | 
						|
            <Severity>Error</Severity>
 | 
						|
            <Message>Error 31 (Import of settings in WinSCU"""
 | 
						|
            """ XML format failed) occurred</Message>
 | 
						|
            <Error Context="SCCI" OC="ShutdownRequestCancelled"
 | 
						|
             OE="0" OI="0">ServerView Agent not connected</Error>
 | 
						|
            </Status>""")
 | 
						|
 | 
						|
        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)
 | 
						|
        e = self.assertRaises(scci.SCCIClientError,
 | 
						|
                              client,
 | 
						|
                              scci.POWER_SOFT_OFF)
 | 
						|
        self.assertEqual(
 | 
						|
            'SCCI PROTOCOL ERROR, STATUS CODE = 31,'
 | 
						|
            ' ERROR = ServerView Agent not connected, MESSAGE = Error 31'
 | 
						|
            ' (Import of settings in WinSCU XML format failed) occurred',
 | 
						|
            str(e))
 | 
						|
 | 
						|
    def test_power_soft_cycle_ok(self):
 | 
						|
        self.requests_mock.post("http://" + self.irmc_address + "/config",
 | 
						|
                                text="""<?xml version="1.0" encoding="UTF-8"?>
 | 
						|
                                <Status>
 | 
						|
                                    <Value>0</Value>
 | 
						|
                                    <Severity>Information</Severity>
 | 
						|
                                    <Message>No Error</Message>
 | 
						|
                                </Status>""")
 | 
						|
 | 
						|
        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)
 | 
						|
        r = client(scci.POWER_SOFT_CYCLE)
 | 
						|
        self.assertEqual(r.status_code, 200)
 | 
						|
 | 
						|
    def test_power_soft_cycle_ng(self):
 | 
						|
        self.requests_mock.post(
 | 
						|
            "http://" + self.irmc_address + "/config",
 | 
						|
            text="""<?xml version="1.0" encoding="UTF-8"?>
 | 
						|
            <Status>
 | 
						|
            <Value>31</Value>
 | 
						|
            <Severity>Error</Severity>
 | 
						|
            <Message>Error 31 (Import of settings in WinSCU"""
 | 
						|
            """ XML format failed) occurred</Message>
 | 
						|
            <Error Context="SCCI" OC="ShutdownRequestCancelled"
 | 
						|
             OE="0" OI="0">ServerView Agent not connected</Error>
 | 
						|
            </Status>""")
 | 
						|
 | 
						|
        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)
 | 
						|
        e = self.assertRaises(scci.SCCIClientError,
 | 
						|
                              client,
 | 
						|
                              scci.POWER_SOFT_CYCLE)
 | 
						|
        self.assertEqual(
 | 
						|
            'SCCI PROTOCOL ERROR, STATUS CODE = 31,'
 | 
						|
            ' ERROR = ServerView Agent not connected, MESSAGE = Error 31'
 | 
						|
            ' (Import of settings in WinSCU XML format failed) occurred',
 | 
						|
            str(e))
 | 
						|
 | 
						|
    def test_power_cancel_shutdown_ok(self):
 | 
						|
        self.requests_mock.post("http://" + self.irmc_address + "/config",
 | 
						|
                                text="""<?xml version="1.0" encoding="UTF-8"?>
 | 
						|
                                <Status>
 | 
						|
                                    <Value>0</Value>
 | 
						|
                                    <Severity>Information</Severity>
 | 
						|
                                    <Message>No Error</Message>
 | 
						|
                                </Status>""")
 | 
						|
 | 
						|
        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)
 | 
						|
        r = client(scci.POWER_CANCEL_SHUTDOWN)
 | 
						|
        self.assertEqual(r.status_code, 200)
 | 
						|
 | 
						|
    def test_power_cancel_shutdown_ng(self):
 | 
						|
        self.requests_mock.post(
 | 
						|
            "http://" + self.irmc_address + "/config",
 | 
						|
            text="""<?xml version="1.0" encoding="UTF-8"?>
 | 
						|
            <Status>
 | 
						|
            <Value>31</Value>
 | 
						|
            <Severity>Error</Severity>
 | 
						|
            <Message>Error 31 (Import of settings in WinSCU"""
 | 
						|
            """ XML format failed) occurred</Message>
 | 
						|
            <Error Context="SCCI" OC="ShutdownRequestCancelled"
 | 
						|
             OE="0" OI="0">ServerView Agent not connected</Error>
 | 
						|
            </Status>""")
 | 
						|
 | 
						|
        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)
 | 
						|
        e = self.assertRaises(scci.SCCIClientError,
 | 
						|
                              client,
 | 
						|
                              scci.POWER_CANCEL_SHUTDOWN)
 | 
						|
        self.assertEqual(
 | 
						|
            'SCCI PROTOCOL ERROR, STATUS CODE = 31,'
 | 
						|
            ' ERROR = ServerView Agent not connected, MESSAGE = Error 31'
 | 
						|
            ' (Import of settings in WinSCU XML format failed) occurred',
 | 
						|
            str(e))
 | 
						|
 | 
						|
    def test_get_sensor_data_records_ok(self):
 | 
						|
        sensor = scci.get_sensor_data_records(self.report_ok_xml)
 | 
						|
        self.assertEqual(len(sensor), 10)
 | 
						|
 | 
						|
    def test_get_sensor_data_records_ng(self):
 | 
						|
        sensor = scci.get_sensor_data_records(self.report_ng_xml)
 | 
						|
        self.assertIsNone(sensor)
 | 
						|
 | 
						|
    def test_get_irmc_version_ok(self):
 | 
						|
        version = scci.get_irmc_version(self.report_ok_xml)
 | 
						|
        self.assertEqual(version.attrib['Name'], "iRMC S4")
 | 
						|
 | 
						|
    def test_get_irmc_version_ng(self):
 | 
						|
        version = scci.get_irmc_version(self.report_ng_xml)
 | 
						|
        self.assertIsNone(version)
 | 
						|
 | 
						|
    def test_get_report_ok(self):
 | 
						|
        self.requests_mock.get(
 | 
						|
            "http://" + self.irmc_address + "/report.xml",
 | 
						|
            text=self.report_ok_txt,
 | 
						|
            headers={'Content-Type': "application/x-www-form-urlencoded"})
 | 
						|
 | 
						|
        root = scci.get_report(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)
 | 
						|
 | 
						|
        self.assertEqual(root.tag, 'Root')
 | 
						|
 | 
						|
        sensor = scci.get_sensor_data_records(root)
 | 
						|
        self.assertEqual(sensor.tag, 'SensorDataRecords')
 | 
						|
 | 
						|
    def test_get_report_http_failed(self):
 | 
						|
        self.requests_mock.get(
 | 
						|
            "http://" + self.irmc_address + "/report.xml",
 | 
						|
            text=self.report_ok_txt,
 | 
						|
            headers={'Content-Type': "application/x-www-form-urlencoded"},
 | 
						|
            status_code=302)
 | 
						|
 | 
						|
        e = self.assertRaises(scci.SCCIClientError,
 | 
						|
                              scci.get_report,
 | 
						|
                              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)
 | 
						|
        self.assertEqual(
 | 
						|
            'HTTP PROTOCOL ERROR, STATUS CODE = 302',
 | 
						|
            str(e))
 | 
						|
 | 
						|
    @mock.patch.object(time, 'sleep')
 | 
						|
    def test_virtual_media_cd_setting_ok(self, sleep_mock):
 | 
						|
        self.requests_mock.post("http://" + self.irmc_address + "/config",
 | 
						|
                                text="""<?xml version="1.0" encoding="UTF-8"?>
 | 
						|
                                <Status>
 | 
						|
                                    <Value>0</Value>
 | 
						|
                                    <Severity>Information</Severity>
 | 
						|
                                    <Message>No Error</Message>
 | 
						|
                                </Status>""")
 | 
						|
 | 
						|
        cmd = scci.get_virtual_cd_set_params_cmd(
 | 
						|
            self.irmc_remote_image_server,
 | 
						|
            self.irmc_remote_image_user_domain,
 | 
						|
            self.irmc_remote_image_share_type,
 | 
						|
            self.irmc_remote_image_share_name,
 | 
						|
            self.irmc_remote_image_deploy_iso,
 | 
						|
            self.irmc_remote_image_username,
 | 
						|
            self.irmc_remote_image_user_password)
 | 
						|
        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)
 | 
						|
 | 
						|
        r = client(cmd, do_async=False)
 | 
						|
        self.assertEqual(r.status_code, 200)
 | 
						|
        sleep_mock.assert_called_once_with(5)
 | 
						|
 | 
						|
    @mock.patch.object(time, 'sleep')
 | 
						|
    def test_virtual_media_fd_setting_ok(self, sleep_mock):
 | 
						|
        self.requests_mock.post("http://" + self.irmc_address + "/config",
 | 
						|
                                text="""<?xml version="1.0" encoding="UTF-8"?>
 | 
						|
                                <Status>
 | 
						|
                                    <Value>0</Value>
 | 
						|
                                    <Severity>Information</Severity>
 | 
						|
                                    <Message>No Error</Message>
 | 
						|
                                </Status>""")
 | 
						|
 | 
						|
        cmd = scci.get_virtual_fd_set_params_cmd(
 | 
						|
            self.irmc_remote_image_server,
 | 
						|
            self.irmc_remote_image_user_domain,
 | 
						|
            self.irmc_remote_image_share_type,
 | 
						|
            self.irmc_remote_image_share_name,
 | 
						|
            'floppy1.flp',
 | 
						|
            self.irmc_remote_image_username,
 | 
						|
            self.irmc_remote_image_user_password)
 | 
						|
        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)
 | 
						|
        r = client(cmd, do_async=False)
 | 
						|
        self.assertEqual(r.status_code, 200)
 | 
						|
        sleep_mock.assert_called_once_with(5)
 | 
						|
 | 
						|
    @mock.patch.object(time, 'sleep')
 | 
						|
    def test_mount_cd_ok(self, sleep_mock):
 | 
						|
        self.requests_mock.post("http://" + self.irmc_address + "/config",
 | 
						|
                                text="""<?xml version="1.0" encoding="UTF-8"?>
 | 
						|
                                <Status>
 | 
						|
                                    <Value>0</Value>
 | 
						|
                                    <Severity>Information</Severity>
 | 
						|
                                    <Message>No Error</Message>
 | 
						|
                                </Status>""")
 | 
						|
 | 
						|
        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)
 | 
						|
        r = client(scci.MOUNT_CD)
 | 
						|
        self.assertEqual(r.status_code, 200)
 | 
						|
        self.assertFalse(sleep_mock.called)
 | 
						|
 | 
						|
    @mock.patch.object(time, 'sleep')
 | 
						|
    def test_mount_fd_ok(self, sleep_mock):
 | 
						|
        self.requests_mock.post("http://" + self.irmc_address + "/config",
 | 
						|
                                text="""<?xml version="1.0" encoding="UTF-8"?>
 | 
						|
                                <Status>
 | 
						|
                                    <Value>0</Value>
 | 
						|
                                    <Severity>Information</Severity>
 | 
						|
                                    <Message>No Error</Message>
 | 
						|
                                </Status>""")
 | 
						|
 | 
						|
        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)
 | 
						|
        r = client(scci.MOUNT_FD)
 | 
						|
        self.assertEqual(r.status_code, 200)
 | 
						|
        self.assertFalse(sleep_mock.called)
 | 
						|
 | 
						|
    def test_unmount_cd_ok(self):
 | 
						|
        self.requests_mock.post("http://" + self.irmc_address + "/config",
 | 
						|
                                text="""<?xml version="1.0" encoding="UTF-8"?>
 | 
						|
                                <Status>
 | 
						|
                                <Value>0</Value>
 | 
						|
                                <Severity>Information</Severity>
 | 
						|
                                <Message>No Error</Message>
 | 
						|
                                </Status>""")
 | 
						|
 | 
						|
        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)
 | 
						|
        r = client(scci.UNMOUNT_CD)
 | 
						|
        self.assertEqual(r.status_code, 200)
 | 
						|
 | 
						|
    def test_unmount_fd_ok(self):
 | 
						|
        self.requests_mock.post("http://" + self.irmc_address + "/config",
 | 
						|
                                text="""<?xml version="1.0" encoding="UTF-8"?>
 | 
						|
                                <Status>
 | 
						|
                                    <Value>0</Value>
 | 
						|
                                    <Severity>Information</Severity>
 | 
						|
                                    <Message>No Error</Message>
 | 
						|
                                </Status>""")
 | 
						|
 | 
						|
        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)
 | 
						|
        r = client(scci.MOUNT_FD)
 | 
						|
        self.assertEqual(r.status_code, 200)
 | 
						|
 | 
						|
    def test_get_essential_properties(self):
 | 
						|
        ESSENTIAL_PROPERTIES_KEYS = {
 | 
						|
            'memory_mb', 'local_gb', 'cpus', 'cpu_arch'}
 | 
						|
        expected = {'memory_mb': 8192,
 | 
						|
                    'local_gb': 185,
 | 
						|
                    'cpus': 16,
 | 
						|
                    'cpu_arch': 'x86_64'}
 | 
						|
 | 
						|
        result = scci.get_essential_properties(
 | 
						|
            self.report_ok_xml, ESSENTIAL_PROPERTIES_KEYS)
 | 
						|
 | 
						|
        self.assertEqual(expected, result)
 | 
						|
 | 
						|
    def test_get_essential_properties_empty_cpu_socket(self):
 | 
						|
        ESSENTIAL_PROPERTIES_KEYS = {
 | 
						|
            'memory_mb', 'local_gb', 'cpus', 'cpu_arch'}
 | 
						|
        expected = {'memory_mb': 8192,
 | 
						|
                    'local_gb': 185,
 | 
						|
                    'cpus': 16,
 | 
						|
                    'cpu_arch': 'x86_64'}
 | 
						|
 | 
						|
        result = scci.get_essential_properties(
 | 
						|
            self.report_ng_xml, ESSENTIAL_PROPERTIES_KEYS)
 | 
						|
 | 
						|
        self.assertEqual(expected, result)
 | 
						|
 | 
						|
    @mock.patch.object(ipmi, 'get_pci_device')
 | 
						|
    @mock.patch.object(snmp, 'get_server_model')
 | 
						|
    @mock.patch.object(snmp, 'get_irmc_firmware_version')
 | 
						|
    @mock.patch.object(snmp, 'get_bios_firmware_version')
 | 
						|
    @mock.patch.object(ipmi, 'get_tpm_status')
 | 
						|
    def test_get_capabilities_properties(self,
 | 
						|
                                         tpm_mock,
 | 
						|
                                         bios_mock,
 | 
						|
                                         irmc_mock,
 | 
						|
                                         server_mock,
 | 
						|
                                         pci_device_mock):
 | 
						|
        capabilities_properties = {'trusted_boot', 'irmc_firmware_version',
 | 
						|
                                   'rom_firmware_version', 'server_model',
 | 
						|
                                   'pci_gpu_devices'}
 | 
						|
        gpu_ids = ['0x1000/0x0079', '0x2100/0x0080']
 | 
						|
        kwargs = {}
 | 
						|
        kwargs['sleep_flag'] = True
 | 
						|
 | 
						|
        tpm_mock.return_value = False
 | 
						|
        bios_mock.return_value = 'V4.6.5.4 R1.15.0 for D3099-B1x'
 | 
						|
        irmc_mock.return_value = 'iRMC S4-7.82F'
 | 
						|
        server_mock.return_value = 'TX2540M1F5'
 | 
						|
        pci_device_mock.side_effect = [1]
 | 
						|
 | 
						|
        expected = {'irmc_firmware_version': 'iRMC S4-7.82F',
 | 
						|
                    'pci_gpu_devices': 1,
 | 
						|
                    'rom_firmware_version': 'V4.6.5.4 R1.15.0 for D3099-B1x',
 | 
						|
                    'server_model': 'TX2540M1F5',
 | 
						|
                    'trusted_boot': False}
 | 
						|
 | 
						|
        result = scci.get_capabilities_properties(
 | 
						|
            self.irmc_info,
 | 
						|
            capabilities_properties,
 | 
						|
            gpu_ids,
 | 
						|
            **kwargs)
 | 
						|
 | 
						|
        self.assertEqual(expected, result)
 | 
						|
        tpm_mock.assert_called_once_with(self.irmc_info)
 | 
						|
        bios_mock.assert_called_once_with(mock.ANY)
 | 
						|
        irmc_mock.assert_called_once_with(mock.ANY)
 | 
						|
        server_mock.assert_called_once_with(mock.ANY)
 | 
						|
        pci_device_mock.assert_called_once_with(self.irmc_info, gpu_ids)
 | 
						|
 | 
						|
    @mock.patch.object(ipmi, 'get_pci_device')
 | 
						|
    @mock.patch.object(snmp, 'get_server_model')
 | 
						|
    @mock.patch.object(snmp, 'get_irmc_firmware_version')
 | 
						|
    @mock.patch.object(snmp, 'get_bios_firmware_version')
 | 
						|
    @mock.patch.object(ipmi, 'get_tpm_status')
 | 
						|
    def test_get_capabilities_properties_with_cpu_fpga(self,
 | 
						|
                                                       tpm_mock,
 | 
						|
                                                       bios_mock,
 | 
						|
                                                       irmc_mock,
 | 
						|
                                                       server_mock,
 | 
						|
                                                       pci_device_mock):
 | 
						|
        capabilities_properties = {'trusted_boot', 'irmc_firmware_version',
 | 
						|
                                   'rom_firmware_version', 'server_model',
 | 
						|
                                   'pci_gpu_devices', 'cpu_fpga'}
 | 
						|
        gpu_ids = ['0x1000/0x0079', '0x2100/0x0080']
 | 
						|
        cpu_fpgas = ['0x1000/0x0179', '0x2100/0x0180']
 | 
						|
        kwargs = {}
 | 
						|
        kwargs['sleep_flag'] = True
 | 
						|
 | 
						|
        tpm_mock.return_value = False
 | 
						|
        bios_mock.return_value = 'V4.6.5.4 R1.15.0 for D3099-B1x'
 | 
						|
        irmc_mock.return_value = 'iRMC S4-7.82F'
 | 
						|
        server_mock.return_value = 'TX2540M1F5'
 | 
						|
        pci_device_mock.side_effect = [1, 1]
 | 
						|
 | 
						|
        expected = {'irmc_firmware_version': 'iRMC S4-7.82F',
 | 
						|
                    'pci_gpu_devices': 1,
 | 
						|
                    'cpu_fpga': 1,
 | 
						|
                    'rom_firmware_version': 'V4.6.5.4 R1.15.0 for D3099-B1x',
 | 
						|
                    'server_model': 'TX2540M1F5',
 | 
						|
                    'trusted_boot': False}
 | 
						|
 | 
						|
        result = scci.get_capabilities_properties(
 | 
						|
            self.irmc_info,
 | 
						|
            capabilities_properties,
 | 
						|
            gpu_ids,
 | 
						|
            cpu_fpgas,
 | 
						|
            **kwargs)
 | 
						|
 | 
						|
        self.assertEqual(expected, result)
 | 
						|
        tpm_mock.assert_called_once_with(self.irmc_info)
 | 
						|
        bios_mock.assert_called_once_with(mock.ANY)
 | 
						|
        irmc_mock.assert_called_once_with(mock.ANY)
 | 
						|
        server_mock.assert_called_once_with(mock.ANY)
 | 
						|
        pci_device_mock.assert_has_calls([
 | 
						|
            mock.call(self.irmc_info, gpu_ids),
 | 
						|
            mock.call(self.irmc_info, cpu_fpgas)])
 | 
						|
 | 
						|
    @mock.patch.object(ipmi, 'get_pci_device')
 | 
						|
    @mock.patch.object(snmp, 'get_server_model')
 | 
						|
    @mock.patch.object(snmp, 'get_irmc_firmware_version')
 | 
						|
    @mock.patch.object(snmp, 'get_bios_firmware_version')
 | 
						|
    @mock.patch.object(ipmi, 'get_tpm_status')
 | 
						|
    def test_get_capabilities_properties_blank(self,
 | 
						|
                                               tpm_mock,
 | 
						|
                                               bios_mock,
 | 
						|
                                               irmc_mock,
 | 
						|
                                               server_mock,
 | 
						|
                                               pci_device_mock):
 | 
						|
 | 
						|
        capabilities_properties = {}
 | 
						|
        gpu_ids = ['0x1000/0x0079', '0x2100/0x0080']
 | 
						|
        cpu_fpgas = ['0x1000/0x0179', '0x2100/0x0180']
 | 
						|
        kwargs = {}
 | 
						|
        kwargs['sleep_flag'] = True
 | 
						|
 | 
						|
        tpm_mock.return_value = False
 | 
						|
        bios_mock.return_value = 'V4.6.5.4 R1.15.0 for D3099-B1x'
 | 
						|
        irmc_mock.return_value = 'iRMC S4-7.82F'
 | 
						|
        server_mock.return_value = 'TX2540M1F5'
 | 
						|
        pci_device_mock.side_effect = [1, 1]
 | 
						|
 | 
						|
        expected = {}
 | 
						|
 | 
						|
        result = scci.get_capabilities_properties(
 | 
						|
            self.irmc_info,
 | 
						|
            capabilities_properties,
 | 
						|
            gpu_ids,
 | 
						|
            cpu_fpgas,
 | 
						|
            **kwargs)
 | 
						|
 | 
						|
        self.assertEqual(expected, result)
 | 
						|
 | 
						|
    @mock.patch.object(ipmi, '_send_raw_command')
 | 
						|
    @mock.patch.object(snmp.SNMPClient, 'get')
 | 
						|
    def test_get_capabilities_properties_scci_client_error(self,
 | 
						|
                                                           snmp_mock,
 | 
						|
                                                           ipmiraw_mock):
 | 
						|
        capabilities_properties = {'trusted_boot', 'irmc_firmware_version',
 | 
						|
                                   'rom_firmware_version', 'server_model',
 | 
						|
                                   'pci_gpu_devices', 'cpu_fpga'}
 | 
						|
        gpu_ids = ['0x1000/0x0079', '0x2100/0x0080']
 | 
						|
        cpu_fpgas = ['0x1000/0x0179', '0x2100/0x0180']
 | 
						|
        kwargs = {}
 | 
						|
        kwargs['sleep_flag'] = True
 | 
						|
 | 
						|
        ipmiraw_mock.return_value = None
 | 
						|
        snmp_mock.side_effect = snmp.SNMPFailure("error")
 | 
						|
 | 
						|
        e = self.assertRaises(scci.SCCIClientError,
 | 
						|
                              scci.get_capabilities_properties,
 | 
						|
                              self.irmc_info,
 | 
						|
                              capabilities_properties,
 | 
						|
                              gpu_ids,
 | 
						|
                              cpu_fpgas,
 | 
						|
                              **kwargs)
 | 
						|
        self.assertEqual('Capabilities inspection failed: SNMP operation \''
 | 
						|
                         'GET BIOS FIRMWARE VERSION\' failed: error', str(e))
 | 
						|
 | 
						|
    @mock.patch.object(ipmi, 'get_pci_device')
 | 
						|
    @mock.patch.object(snmp.SNMPClient, 'get')
 | 
						|
    def test_get_capabilities_properties_scci_client_error_ipmi(self,
 | 
						|
                                                                snmp_mock,
 | 
						|
                                                                ipmi_mock):
 | 
						|
        capabilities_properties = {'trusted_boot', 'irmc_firmware_version',
 | 
						|
                                   'rom_firmware_version', 'server_model',
 | 
						|
                                   'pci_gpu_devices', 'cpu_fpga'}
 | 
						|
        gpu_ids = ['0x1000/0x0079', '0x2100/0x0080']
 | 
						|
        cpu_fpgas = ['0x1000/0x0179', '0x2100/0x0180']
 | 
						|
        kwargs = {}
 | 
						|
        kwargs['sleep_flag'] = True
 | 
						|
 | 
						|
        ipmi_mock.side_effect = ipmi.IPMIFailure("IPMI error")
 | 
						|
        snmp_mock.return_value = None
 | 
						|
 | 
						|
        e = self.assertRaises(scci.SCCIClientError,
 | 
						|
                              scci.get_capabilities_properties,
 | 
						|
                              self.irmc_info,
 | 
						|
                              capabilities_properties,
 | 
						|
                              gpu_ids,
 | 
						|
                              cpu_fpgas,
 | 
						|
                              **kwargs)
 | 
						|
        self.assertEqual('Capabilities inspection failed: IPMI error', str(e))
 | 
						|
 | 
						|
    def test_fail_get_raid_fgi_status(self):
 | 
						|
        report_fake = self.report_ok_xml
 | 
						|
        report_fake.find("./Software/ServerView/ServerViewRaid").clear()
 | 
						|
        self.assertRaises(scci.SCCIInvalidInputError,
 | 
						|
                          scci.get_raid_fgi_status, report=report_fake)
 | 
						|
 | 
						|
    def test_fail_get_raid_fgi_status_1(self):
 | 
						|
        report_fake = self.report_ok_xml
 | 
						|
        report_fake.find("./Software/ServerView/ServerViewRaid/amEMSV"
 | 
						|
                         "/System/Adapter").clear()
 | 
						|
        self.assertRaises(scci.SCCIRAIDNotReady,
 | 
						|
                          scci.get_raid_fgi_status, report=report_fake)
 | 
						|
 | 
						|
    def test_get_raid_fgi_status_ok(self):
 | 
						|
 | 
						|
        # Fake activity status of FGI in xml report
 | 
						|
        url = "./Software/ServerView/ServerViewRaid/amEMSV/System/Adapter"
 | 
						|
        report_fake = self.report_ok_xml
 | 
						|
        report_input = report_fake.find(url)
 | 
						|
        element_1 = ET.Element("LogicalDrive", name="LogicalDrive")
 | 
						|
        element_2 = ET.Element("LogDriveNumber", name="LogDriveNumber")
 | 
						|
        element_3 = ET.Element("Activity", name="Activity")
 | 
						|
        report_input.append(element_1)
 | 
						|
        report_input.find("./LogicalDrive").append(element_2)
 | 
						|
        report_fake.find(url + "/LogicalDrive/LogDriveNumber").text = '0'
 | 
						|
        report_input.find("./LogicalDrive").append(element_3)
 | 
						|
        report_fake.find(url + "/LogicalDrive/Activity").text = 'Idle'
 | 
						|
 | 
						|
        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.get')
 | 
						|
    def test_fail_get_bios_firmware_status(self, mock_requests_get):
 | 
						|
        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.get')
 | 
						|
    def test_success_get_bios_firmware_status(self, mock_requests_get):
 | 
						|
        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.get')
 | 
						|
    def test_fail_get_irmc_firmware_status(self, mock_requests_get):
 | 
						|
        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.get')
 | 
						|
    def test_success_get_irmc_firmware_status(self, mock_requests_get):
 | 
						|
        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)
 |