454 lines
18 KiB
Python
454 lines
18 KiB
Python
# Copyright 2015 Hewlett Packard Development Company, L.P.
|
|
#
|
|
# 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.
|
|
|
|
import sys
|
|
|
|
import mock
|
|
if 'hp3parclient' not in sys.modules:
|
|
sys.modules['hp3parclient'] = mock.Mock()
|
|
|
|
from manila import exception
|
|
from manila.share.drivers.hp import hp_3par_driver as hp3pardriver
|
|
from manila.share.drivers.hp import hp_3par_mediator as hp3parmediator
|
|
from manila import test
|
|
from manila.tests.share.drivers.hp import test_hp_3par_constants as constants
|
|
|
|
|
|
class HP3ParDriverTestCase(test.TestCase):
|
|
|
|
def setUp(self):
|
|
super(HP3ParDriverTestCase, self).setUp()
|
|
|
|
# Create a mock configuration with attributes and a safe_get()
|
|
self.conf = mock.Mock()
|
|
self.conf.driver_handles_share_servers = False
|
|
self.conf.hp3par_debug = constants.EXPECTED_HP_DEBUG
|
|
self.conf.hp3par_username = constants.USERNAME
|
|
self.conf.hp3par_password = constants.PASSWORD
|
|
self.conf.hp3par_api_url = constants.API_URL
|
|
self.conf.hp3par_san_login = constants.SAN_LOGIN
|
|
self.conf.hp3par_san_password = constants.SAN_PASSWORD
|
|
self.conf.hp3par_san_ip = constants.EXPECTED_IP_1234
|
|
self.conf.hp3par_fpg = constants.EXPECTED_FPG
|
|
self.conf.hp3par_san_ssh_port = constants.PORT
|
|
self.conf.ssh_conn_timeout = constants.TIMEOUT
|
|
self.conf.hp3par_share_ip_address = constants.EXPECTED_IP_10203040
|
|
self.conf.hp3par_fstore_per_share = False
|
|
self.conf.network_config_group = 'test_network_config_group'
|
|
|
|
def safe_get(attr):
|
|
try:
|
|
return self.conf.__getattribute__(attr)
|
|
except AttributeError:
|
|
return None
|
|
self.conf.safe_get = safe_get
|
|
|
|
self.real_hp_3par_mediator = hp3parmediator.HP3ParMediator
|
|
self.mock_object(hp3parmediator, 'HP3ParMediator')
|
|
self.mock_mediator_constructor = hp3parmediator.HP3ParMediator
|
|
self.mock_mediator = self.mock_mediator_constructor()
|
|
|
|
self.driver = hp3pardriver.HP3ParShareDriver(
|
|
configuration=self.conf)
|
|
|
|
def test_driver_setup_success(self):
|
|
"""Driver do_setup without any errors."""
|
|
|
|
self.mock_mediator.get_vfs_name.return_value = constants.EXPECTED_VFS
|
|
|
|
self.driver.do_setup(None)
|
|
conf = self.conf
|
|
self.mock_mediator_constructor.assert_has_calls([
|
|
mock.call(hp3par_san_ssh_port=conf.hp3par_san_ssh_port,
|
|
hp3par_san_password=conf.hp3par_san_password,
|
|
hp3par_username=conf.hp3par_username,
|
|
hp3par_san_login=conf.hp3par_san_login,
|
|
hp3par_debug=conf.hp3par_debug,
|
|
hp3par_api_url=conf.hp3par_api_url,
|
|
hp3par_password=conf.hp3par_password,
|
|
hp3par_san_ip=conf.hp3par_san_ip,
|
|
hp3par_fstore_per_share=conf.hp3par_fstore_per_share,
|
|
ssh_conn_timeout=conf.ssh_conn_timeout)])
|
|
|
|
self.mock_mediator.assert_has_calls([
|
|
mock.call.do_setup(),
|
|
mock.call.get_vfs_name(conf.hp3par_fpg)])
|
|
|
|
self.assertEqual(constants.EXPECTED_VFS, self.driver.vfs)
|
|
|
|
def test_driver_with_setup_error(self):
|
|
"""Driver do_setup when the mediator setup fails."""
|
|
|
|
self.mock_mediator.do_setup.side_effect = (
|
|
exception.ShareBackendException('fail'))
|
|
|
|
self.assertRaises(exception.ShareBackendException,
|
|
self.driver.do_setup, None)
|
|
|
|
conf = self.conf
|
|
self.mock_mediator_constructor.assert_has_calls([
|
|
mock.call(hp3par_san_ssh_port=conf.hp3par_san_ssh_port,
|
|
hp3par_san_password=conf.hp3par_san_password,
|
|
hp3par_username=conf.hp3par_username,
|
|
hp3par_san_login=conf.hp3par_san_login,
|
|
hp3par_debug=conf.hp3par_debug,
|
|
hp3par_api_url=conf.hp3par_api_url,
|
|
hp3par_password=conf.hp3par_password,
|
|
hp3par_san_ip=conf.hp3par_san_ip,
|
|
hp3par_fstore_per_share=conf.hp3par_fstore_per_share,
|
|
ssh_conn_timeout=conf.ssh_conn_timeout)])
|
|
|
|
self.mock_mediator.assert_has_calls([mock.call.do_setup()])
|
|
|
|
def test_driver_with_vfs_error(self):
|
|
"""Driver do_setup when the get_vfs_name fails."""
|
|
|
|
self.mock_mediator.get_vfs_name.side_effect = (
|
|
exception.ShareBackendException('fail'))
|
|
|
|
self.assertRaises(exception.ShareBackendException,
|
|
self.driver.do_setup, None)
|
|
|
|
conf = self.conf
|
|
self.mock_mediator_constructor.assert_has_calls([
|
|
mock.call(hp3par_san_ssh_port=conf.hp3par_san_ssh_port,
|
|
hp3par_san_password=conf.hp3par_san_password,
|
|
hp3par_username=conf.hp3par_username,
|
|
hp3par_san_login=conf.hp3par_san_login,
|
|
hp3par_debug=conf.hp3par_debug,
|
|
hp3par_api_url=conf.hp3par_api_url,
|
|
hp3par_password=conf.hp3par_password,
|
|
hp3par_san_ip=conf.hp3par_san_ip,
|
|
hp3par_fstore_per_share=conf.hp3par_fstore_per_share,
|
|
ssh_conn_timeout=conf.ssh_conn_timeout)])
|
|
|
|
self.mock_mediator.assert_has_calls([
|
|
mock.call.do_setup(),
|
|
mock.call.get_vfs_name(conf.hp3par_fpg)])
|
|
|
|
def init_driver(self):
|
|
"""Simple driver setup for re-use with tests that need one."""
|
|
|
|
self.driver._hp3par = self.mock_mediator
|
|
self.driver.vfs = constants.EXPECTED_VFS
|
|
self.driver.fpg = constants.EXPECTED_FPG
|
|
self.driver.share_ip_address = self.conf.hp3par_share_ip_address
|
|
|
|
def do_create_share(self, protocol, expected_project_id, expected_share_id,
|
|
expected_size):
|
|
"""Re-usable code for create share."""
|
|
context = None
|
|
share_server = None
|
|
share = {
|
|
'project_id': expected_project_id,
|
|
'id': expected_share_id,
|
|
'share_proto': protocol,
|
|
'size': expected_size,
|
|
}
|
|
location = self.driver.create_share(context, share, share_server)
|
|
return location
|
|
|
|
def do_create_share_from_snapshot(self,
|
|
protocol,
|
|
snapshot_id,
|
|
expected_share_id,
|
|
expected_size):
|
|
"""Re-usable code for create share from snapshot."""
|
|
context = None
|
|
share_server = None
|
|
share = {
|
|
'id': expected_share_id,
|
|
'share_proto': protocol,
|
|
'size': expected_size,
|
|
}
|
|
location = self.driver.create_share_from_snapshot(context,
|
|
share,
|
|
snapshot_id,
|
|
share_server)
|
|
return location
|
|
|
|
def test_driver_check_for_setup_error_success(self):
|
|
"""check_for_setup_error when things go well."""
|
|
|
|
# Generally this is always mocked, but here we reference the class.
|
|
hp3parmediator.HP3ParMediator = self.real_hp_3par_mediator
|
|
|
|
self.mock_object(hp3pardriver, 'LOG')
|
|
self.init_driver()
|
|
self.driver.check_for_setup_error()
|
|
expected_calls = [
|
|
mock.call.debug('HP3ParShareDriver SHA1: %s', mock.ANY),
|
|
mock.call.debug('HP3ParMediator SHA1: %s', mock.ANY)
|
|
]
|
|
hp3pardriver.LOG.assert_has_calls(expected_calls)
|
|
|
|
def test_driver_check_for_setup_error_exception(self):
|
|
"""check_for_setup_error catch and log any exceptions."""
|
|
|
|
# Since HP3ParMediator is mocked, we'll hit the except/log.
|
|
self.mock_object(hp3pardriver, 'LOG')
|
|
self.init_driver()
|
|
self.driver.check_for_setup_error()
|
|
expected_calls = [
|
|
mock.call.debug('HP3ParShareDriver SHA1: %s', mock.ANY),
|
|
mock.call.debug('Source code SHA1 not logged due to: %s', mock.ANY)
|
|
]
|
|
hp3pardriver.LOG.assert_has_calls(expected_calls)
|
|
|
|
def test_driver_create_cifs_share(self):
|
|
self.init_driver()
|
|
|
|
expected_location = '\\\\%s\%s' % (constants.EXPECTED_IP_10203040,
|
|
constants.EXPECTED_SHARE_NAME)
|
|
|
|
self.mock_mediator.create_share.return_value = (
|
|
constants.EXPECTED_SHARE_NAME)
|
|
|
|
location = self.do_create_share(constants.CIFS,
|
|
constants.EXPECTED_PROJECT_ID,
|
|
constants.EXPECTED_SHARE_ID,
|
|
constants.EXPECTED_SIZE_2)
|
|
|
|
self.assertEqual(expected_location, location)
|
|
expected_calls = [mock.call.create_share(
|
|
constants.EXPECTED_PROJECT_ID,
|
|
constants.EXPECTED_SHARE_ID,
|
|
constants.CIFS,
|
|
constants.EXPECTED_FPG,
|
|
constants.EXPECTED_VFS,
|
|
size=constants.EXPECTED_SIZE_2)]
|
|
self.mock_mediator.assert_has_calls(expected_calls)
|
|
|
|
def test_driver_create_nfs_share(self):
|
|
self.init_driver()
|
|
|
|
expected_location = ':'.join((constants.EXPECTED_IP_10203040,
|
|
constants.EXPECTED_SHARE_PATH))
|
|
|
|
self.mock_mediator.create_share.return_value = (
|
|
constants.EXPECTED_SHARE_PATH)
|
|
|
|
location = self.do_create_share(constants.NFS,
|
|
constants.EXPECTED_PROJECT_ID,
|
|
constants.EXPECTED_SHARE_ID,
|
|
constants.EXPECTED_SIZE_1)
|
|
|
|
self.assertEqual(expected_location, location)
|
|
expected_calls = [
|
|
mock.call.create_share(constants.EXPECTED_PROJECT_ID,
|
|
constants.EXPECTED_SHARE_ID,
|
|
constants.NFS,
|
|
constants.EXPECTED_FPG,
|
|
constants.EXPECTED_VFS,
|
|
size=constants.EXPECTED_SIZE_1)]
|
|
|
|
self.mock_mediator.assert_has_calls(expected_calls)
|
|
|
|
def test_driver_create_cifs_share_from_snapshot(self):
|
|
self.init_driver()
|
|
|
|
expected_location = '\\\\%s\%s' % (constants.EXPECTED_IP_10203040,
|
|
constants.EXPECTED_SHARE_NAME)
|
|
|
|
self.mock_mediator.create_share_from_snapshot.return_value = (
|
|
constants.EXPECTED_SHARE_NAME)
|
|
|
|
location = self.do_create_share_from_snapshot(
|
|
constants.CIFS,
|
|
constants.SNAPSHOT_INFO,
|
|
constants.EXPECTED_SHARE_ID,
|
|
constants.EXPECTED_SIZE_2)
|
|
|
|
self.assertEqual(expected_location, location)
|
|
expected_calls = [
|
|
mock.call.create_share_from_snapshot(constants.EXPECTED_SHARE_ID,
|
|
constants.CIFS,
|
|
constants.EXPECTED_FSTORE,
|
|
constants.EXPECTED_SHARE_ID,
|
|
constants.NFS,
|
|
constants.EXPECTED_SNAP_ID,
|
|
constants.EXPECTED_FPG,
|
|
constants.EXPECTED_VFS)]
|
|
self.mock_mediator.assert_has_calls(expected_calls)
|
|
|
|
def test_driver_create_nfs_share_from_snapshot(self):
|
|
self.init_driver()
|
|
|
|
expected_location = ':'.join((constants.EXPECTED_IP_10203040,
|
|
constants.EXPECTED_SHARE_PATH))
|
|
|
|
self.mock_mediator.create_share_from_snapshot.return_value = (
|
|
constants.EXPECTED_SHARE_PATH)
|
|
|
|
location = self.do_create_share_from_snapshot(
|
|
constants.NFS,
|
|
constants.SNAPSHOT_INFO,
|
|
constants.EXPECTED_SHARE_ID,
|
|
constants.EXPECTED_SIZE_1)
|
|
|
|
self.assertEqual(expected_location, location)
|
|
expected_calls = [
|
|
mock.call.create_share_from_snapshot(constants.EXPECTED_SHARE_ID,
|
|
constants.NFS,
|
|
constants.EXPECTED_PROJECT_ID,
|
|
constants.EXPECTED_SHARE_ID,
|
|
constants.NFS,
|
|
constants.EXPECTED_SNAP_ID,
|
|
constants.EXPECTED_FPG,
|
|
constants.EXPECTED_VFS)]
|
|
|
|
self.mock_mediator.assert_has_calls(expected_calls)
|
|
|
|
def test_driver_delete_share(self):
|
|
self.init_driver()
|
|
|
|
context = None
|
|
share_server = None
|
|
share = {
|
|
'project_id': constants.EXPECTED_PROJECT_ID,
|
|
'id': constants.EXPECTED_SHARE_ID,
|
|
'share_proto': constants.CIFS,
|
|
'size': constants.EXPECTED_SIZE_1,
|
|
}
|
|
|
|
self.driver.delete_share(context, share, share_server)
|
|
|
|
expected_calls = [
|
|
mock.call.delete_share(constants.EXPECTED_PROJECT_ID,
|
|
constants.EXPECTED_SHARE_ID,
|
|
constants.CIFS,
|
|
constants.EXPECTED_FPG,
|
|
constants.EXPECTED_VFS)]
|
|
|
|
self.mock_mediator.assert_has_calls(expected_calls)
|
|
|
|
def test_driver_create_snapshot(self):
|
|
self.init_driver()
|
|
|
|
context = None
|
|
share_server = None
|
|
self.driver.create_snapshot(context,
|
|
constants.SNAPSHOT_INFO,
|
|
share_server)
|
|
|
|
expected_calls = [
|
|
mock.call.create_snapshot(constants.EXPECTED_PROJECT_ID,
|
|
constants.EXPECTED_SHARE_ID,
|
|
constants.NFS,
|
|
constants.EXPECTED_SNAP_ID,
|
|
constants.EXPECTED_FPG,
|
|
constants.EXPECTED_VFS)]
|
|
self.mock_mediator.assert_has_calls(expected_calls)
|
|
|
|
def test_driver_delete_snapshot(self):
|
|
self.init_driver()
|
|
|
|
context = None
|
|
share_server = None
|
|
self.driver.delete_snapshot(context,
|
|
constants.SNAPSHOT_INFO,
|
|
share_server)
|
|
|
|
expected_calls = [
|
|
mock.call.delete_snapshot(constants.EXPECTED_PROJECT_ID,
|
|
constants.EXPECTED_SHARE_ID,
|
|
constants.NFS,
|
|
constants.EXPECTED_SNAP_ID,
|
|
constants.EXPECTED_FPG,
|
|
constants.EXPECTED_VFS)
|
|
]
|
|
self.mock_mediator.assert_has_calls(expected_calls)
|
|
|
|
def test_driver_allow_access(self):
|
|
self.init_driver()
|
|
|
|
context = None
|
|
self.driver.allow_access(context,
|
|
constants.NFS_SHARE_INFO,
|
|
constants.ACCESS_INFO)
|
|
|
|
expected_calls = [
|
|
mock.call.allow_access(constants.EXPECTED_PROJECT_ID,
|
|
constants.EXPECTED_SHARE_ID,
|
|
constants.NFS,
|
|
constants.IP,
|
|
constants.EXPECTED_IP_1234,
|
|
constants.EXPECTED_FPG,
|
|
constants.EXPECTED_VFS)
|
|
]
|
|
self.mock_mediator.assert_has_calls(expected_calls)
|
|
|
|
def test_driver_deny_access(self):
|
|
self.init_driver()
|
|
|
|
context = None
|
|
self.driver.deny_access(context,
|
|
constants.NFS_SHARE_INFO,
|
|
constants.ACCESS_INFO)
|
|
|
|
expected_calls = [
|
|
mock.call.deny_access(constants.EXPECTED_PROJECT_ID,
|
|
constants.EXPECTED_SHARE_ID,
|
|
constants.NFS,
|
|
constants.IP,
|
|
constants.EXPECTED_IP_1234,
|
|
constants.EXPECTED_FPG,
|
|
constants.EXPECTED_VFS)
|
|
]
|
|
self.mock_mediator.assert_has_calls(expected_calls)
|
|
|
|
def test_driver_get_share_stats_no_refresh(self):
|
|
"""Driver does not call mediator when refresh=False."""
|
|
|
|
self.init_driver()
|
|
self.driver._stats = constants.EXPECTED_STATS
|
|
|
|
result = self.driver.get_share_stats(refresh=False)
|
|
|
|
self.assertEqual(constants.EXPECTED_STATS, result)
|
|
self.assertEqual([], self.mock_mediator.mock_calls)
|
|
|
|
def test_driver_get_share_stats_with_refresh(self):
|
|
"""Driver adds stats from mediator to expected structure."""
|
|
|
|
self.init_driver()
|
|
expected_free = constants.EXPECTED_SIZE_1
|
|
expected_capacity = constants.EXPECTED_SIZE_2
|
|
|
|
self.mock_mediator.get_capacity.return_value = {
|
|
'free_capacity_gb': expected_free,
|
|
'total_capacity_gb': expected_capacity
|
|
}
|
|
|
|
expected_result = {
|
|
'driver_handles_share_servers': False,
|
|
'QoS_support': False,
|
|
'driver_version': '1.0',
|
|
'free_capacity_gb': expected_free,
|
|
'reserved_percentage': 0,
|
|
'share_backend_name': 'HP_3PAR',
|
|
'storage_protocol': 'NFS_CIFS',
|
|
'total_capacity_gb': expected_capacity,
|
|
'vendor_name': 'HP',
|
|
'pools': None,
|
|
}
|
|
|
|
result = self.driver.get_share_stats(refresh=True)
|
|
self.assertEqual(expected_result, result)
|
|
|
|
expected_calls = [
|
|
mock.call.get_capacity(constants.EXPECTED_FPG)
|
|
]
|
|
self.mock_mediator.assert_has_calls(expected_calls)
|