227 lines
7.9 KiB
Python
227 lines
7.9 KiB
Python
# Copyright (c) 2015 EMC Corporation.
|
|
# All Rights Reserved.
|
|
#
|
|
# 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.
|
|
|
|
from eventlet import greenthread
|
|
import mock
|
|
from oslo_concurrency import processutils
|
|
from six.moves.urllib import error as url_error
|
|
from six.moves.urllib import request as url_request
|
|
|
|
from manila import exception
|
|
from manila.share import configuration as conf
|
|
from manila.share.drivers.dell_emc.common.enas import connector
|
|
from manila import test
|
|
from manila.tests.share.drivers.dell_emc.common.enas import fakes
|
|
from manila.tests.share.drivers.dell_emc.common.enas import utils as enas_utils
|
|
from manila import utils
|
|
|
|
|
|
class XMLAPIConnectorTestData(object):
|
|
FAKE_BODY = '<fakebody></fakebody>'
|
|
FAKE_RESP = '<Response></Response>'
|
|
FAKE_METHOD = 'fake_method'
|
|
|
|
FAKE_KEY = 'key'
|
|
FAKE_VALUE = 'value'
|
|
|
|
@staticmethod
|
|
def req_auth_url():
|
|
return 'https://' + fakes.FakeData.emc_nas_server + '/Login'
|
|
|
|
@staticmethod
|
|
def req_credential():
|
|
return (
|
|
'user=' + fakes.FakeData.emc_nas_login
|
|
+ '&password=' + fakes.FakeData.emc_nas_password
|
|
+ '&Login=Login'
|
|
).encode()
|
|
|
|
@staticmethod
|
|
def req_url_encode():
|
|
return {'Content-Type': 'application/x-www-form-urlencoded'}
|
|
|
|
@staticmethod
|
|
def req_url():
|
|
return (
|
|
'https://'
|
|
+ fakes.FakeData.emc_nas_server
|
|
+ '/servlets/CelerraManagementServices'
|
|
)
|
|
|
|
|
|
XML_CONN_TD = XMLAPIConnectorTestData
|
|
|
|
|
|
class XMLAPIConnectorTest(test.TestCase):
|
|
@mock.patch.object(url_request, 'Request', mock.Mock())
|
|
def setUp(self):
|
|
super(XMLAPIConnectorTest, self).setUp()
|
|
|
|
emc_share_driver = fakes.FakeEMCShareDriver()
|
|
|
|
self.configuration = emc_share_driver.configuration
|
|
|
|
xml_socket = mock.Mock()
|
|
xml_socket.read = mock.Mock(return_value=XML_CONN_TD.FAKE_RESP)
|
|
opener = mock.Mock()
|
|
opener.open = mock.Mock(return_value=xml_socket)
|
|
|
|
with mock.patch.object(url_request, 'build_opener',
|
|
mock.Mock(return_value=opener)):
|
|
self.XmlConnector = connector.XMLAPIConnector(
|
|
configuration=self.configuration, debug=False)
|
|
|
|
expected_calls = [
|
|
mock.call(XML_CONN_TD.req_auth_url(),
|
|
XML_CONN_TD.req_credential(),
|
|
XML_CONN_TD.req_url_encode()),
|
|
]
|
|
|
|
url_request.Request.assert_has_calls(expected_calls)
|
|
|
|
def test_request_with_debug(self):
|
|
self.XmlConnector.debug = True
|
|
|
|
request = mock.Mock()
|
|
request.headers = {XML_CONN_TD.FAKE_KEY: XML_CONN_TD.FAKE_VALUE}
|
|
request.get_full_url = mock.Mock(
|
|
return_value=XML_CONN_TD.FAKE_VALUE)
|
|
|
|
with mock.patch.object(url_request, 'Request',
|
|
mock.Mock(return_value=request)):
|
|
rsp = self.XmlConnector.request(XML_CONN_TD.FAKE_BODY,
|
|
XML_CONN_TD.FAKE_METHOD)
|
|
|
|
self.assertEqual(XML_CONN_TD.FAKE_RESP, rsp)
|
|
|
|
def test_request_with_no_authorized_exception(self):
|
|
xml_socket = mock.Mock()
|
|
xml_socket.read = mock.Mock(return_value=XML_CONN_TD.FAKE_RESP)
|
|
|
|
hook = enas_utils.RequestSideEffect()
|
|
hook.append(ex=url_error.HTTPError(XML_CONN_TD.req_url(),
|
|
'403', 'fake_message', None, None))
|
|
hook.append(xml_socket)
|
|
hook.append(xml_socket)
|
|
|
|
self.XmlConnector.url_opener.open = mock.Mock(side_effect=hook)
|
|
|
|
self.XmlConnector.request(XML_CONN_TD.FAKE_BODY)
|
|
|
|
def test_request_with_general_exception(self):
|
|
hook = enas_utils.RequestSideEffect()
|
|
hook.append(ex=url_error.HTTPError(XML_CONN_TD.req_url(),
|
|
'error_code', 'fake_message',
|
|
None, None))
|
|
self.XmlConnector.url_opener.open = mock.Mock(side_effect=hook)
|
|
|
|
self.assertRaises(exception.ManilaException,
|
|
self.XmlConnector.request,
|
|
XML_CONN_TD.FAKE_BODY)
|
|
|
|
|
|
class MockSSH(object):
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, type, value, traceback):
|
|
pass
|
|
|
|
|
|
class MockSSHPool(object):
|
|
def __init__(self):
|
|
self.ssh = MockSSH()
|
|
|
|
def item(self):
|
|
try:
|
|
return self.ssh
|
|
finally:
|
|
pass
|
|
|
|
|
|
class CmdConnectorTest(test.TestCase):
|
|
def setUp(self):
|
|
super(CmdConnectorTest, self).setUp()
|
|
|
|
self.configuration = conf.Configuration(None)
|
|
self.configuration.append_config_values = mock.Mock(return_value=0)
|
|
self.configuration.emc_nas_login = fakes.FakeData.emc_nas_login
|
|
self.configuration.emc_nas_password = fakes.FakeData.emc_nas_password
|
|
self.configuration.emc_nas_server = fakes.FakeData.emc_nas_server
|
|
self.configuration.emc_ssl_cert_verify = False
|
|
self.configuration.emc_ssl_cert_path = None
|
|
|
|
self.sshpool = MockSSHPool()
|
|
with mock.patch.object(utils, "SSHPool",
|
|
mock.Mock(return_value=self.sshpool)):
|
|
self.CmdHelper = connector.SSHConnector(
|
|
configuration=self.configuration, debug=False)
|
|
|
|
utils.SSHPool.assert_called_once_with(
|
|
ip=fakes.FakeData.emc_nas_server,
|
|
port=22,
|
|
conn_timeout=None,
|
|
login=fakes.FakeData.emc_nas_login,
|
|
password=fakes.FakeData.emc_nas_password)
|
|
|
|
def test_run_ssh(self):
|
|
with mock.patch.object(processutils, "ssh_execute",
|
|
mock.Mock(return_value=('fake_output', ''))):
|
|
cmd_list = ['fake', 'cmd']
|
|
self.CmdHelper.run_ssh(cmd_list)
|
|
|
|
processutils.ssh_execute.assert_called_once_with(
|
|
self.sshpool.item(), 'fake cmd', check_exit_code=False)
|
|
|
|
def test_run_ssh_with_debug(self):
|
|
self.CmdHelper.debug = True
|
|
|
|
with mock.patch.object(processutils, "ssh_execute",
|
|
mock.Mock(return_value=('fake_output', ''))):
|
|
cmd_list = ['fake', 'cmd']
|
|
self.CmdHelper.run_ssh(cmd_list)
|
|
|
|
processutils.ssh_execute.assert_called_once_with(
|
|
self.sshpool.item(), 'fake cmd', check_exit_code=False)
|
|
|
|
@mock.patch.object(
|
|
processutils, "ssh_execute",
|
|
mock.Mock(side_effect=processutils.ProcessExecutionError))
|
|
def test_run_ssh_exception(self):
|
|
cmd_list = ['fake', 'cmd']
|
|
|
|
self.mock_object(greenthread, 'sleep', mock.Mock())
|
|
|
|
sshpool = MockSSHPool()
|
|
|
|
with mock.patch.object(utils, "SSHPool",
|
|
mock.Mock(return_value=sshpool)):
|
|
self.CmdHelper = connector.SSHConnector(self.configuration)
|
|
|
|
self.assertRaises(processutils.ProcessExecutionError,
|
|
self.CmdHelper.run_ssh,
|
|
cmd_list,
|
|
True)
|
|
|
|
utils.SSHPool.assert_called_once_with(
|
|
ip=fakes.FakeData.emc_nas_server,
|
|
port=22,
|
|
conn_timeout=None,
|
|
login=fakes.FakeData.emc_nas_login,
|
|
password=fakes.FakeData.emc_nas_password)
|
|
|
|
processutils.ssh_execute.assert_called_once_with(
|
|
sshpool.item(), 'fake cmd', check_exit_code=True)
|