QNAP Manila driver: Access rule setting is override by the later rule setting.

When user try to add access rule1 and access rule2, share will first set to
the rule1 and then override by the rule2.

We expect that share should apply both access rule1 and access rule2.

Change-Id: Id77cffb5efe4388e3b66aa85fc89cf6f51d5bd98
Closes-Bug: #1736370
This commit is contained in:
Chris Yang 2017-12-05 18:53:10 +08:00
parent 2c67dc9f4e
commit 205ef44283
6 changed files with 413 additions and 116 deletions

View File

@ -177,6 +177,9 @@ class QnapAPIExecutor(object):
for key in params: for key in params:
value = params[key] value = params[key]
if value is not None: if value is not None:
if isinstance(value, list):
sanitized_params[key] = [six.text_type(v) for v in value]
else:
sanitized_params[key] = six.text_type(value) sanitized_params[key] = six.text_type(value)
return sanitized_params return sanitized_params
@ -580,6 +583,52 @@ class QnapAPIExecutor(object):
if root.find('result').text < '0': if root.find('result').text < '0':
raise exception.ShareBackendException(msg=MSG_UNEXPECT_RESP) raise exception.ShareBackendException(msg=MSG_UNEXPECT_RESP)
@_connection_checker
def edit_host(self, hostname, ipv4_list):
"""Execute edit_host API."""
params = {
'module': 'hosts',
'func': 'apply_sethost',
'name': hostname,
'ipaddr_v4': ipv4_list,
'sid': self.sid,
}
sanitized_params = self._sanitize_params(params)
# urlencode with True parameter to parse ipv4_list
sanitized_params = urllib.parse.urlencode(sanitized_params, True)
url = ('/cgi-bin/accessrights/accessrightsRequest.cgi?%s' %
sanitized_params)
res_details = self._execute_and_get_response_details(self.ip, url)
root = ET.fromstring(res_details['data'])
if root.find('authPassed').text == '0':
raise exception.ShareBackendException(msg=MSG_SESSION_EXPIRED)
if root.find('result').text < '0':
raise exception.ShareBackendException(msg=MSG_UNEXPECT_RESP)
@_connection_checker
def delete_host(self, hostname):
"""Execute delete_host API."""
params = {
'module': 'hosts',
'func': 'apply_delhost',
'host_name': hostname,
'sid': self.sid,
}
sanitized_params = self._sanitize_params(params)
sanitized_params = urllib.parse.urlencode(sanitized_params)
url = ('/cgi-bin/accessrights/accessrightsRequest.cgi?%s' %
sanitized_params)
res_details = self._execute_and_get_response_details(self.ip, url)
root = ET.fromstring(res_details['data'])
if root.find('authPassed').text == '0':
raise exception.ShareBackendException(msg=MSG_SESSION_EXPIRED)
if root.find('result').text < '0':
raise exception.ShareBackendException(msg=MSG_UNEXPECT_RESP)
@_connection_checker @_connection_checker
def set_nfs_access(self, sharename, access, host_name): def set_nfs_access(self, sharename, access, host_name):
"""Execute set_nfs_access API.""" """Execute set_nfs_access API."""

View File

@ -16,7 +16,9 @@
Share driver for QNAP Storage. Share driver for QNAP Storage.
This driver supports QNAP Storage for NFS. This driver supports QNAP Storage for NFS.
""" """
import datetime
import re import re
import time
from oslo_config import cfg from oslo_config import cfg
from oslo_log import log as logging from oslo_log import log as logging
@ -62,9 +64,11 @@ class QnapShareDriver(driver.ShareDriver):
Version history: Version history:
1.0.0 - Initial driver (Only NFS) 1.0.0 - Initial driver (Only NFS)
1.0.1 - Add support for QES fw 1.1.4. 1.0.1 - Add support for QES fw 1.1.4.
1.0.2 - Fix bug #1736370, QNAP Manila driver: Access rule setting is
override by the another access rule.
""" """
DRIVER_VERSION = '1.0.1' DRIVER_VERSION = '1.0.2'
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
"""Initialize QnapShareDriver.""" """Initialize QnapShareDriver."""
@ -193,6 +197,16 @@ class QnapShareDriver(driver.ShareDriver):
{'ifx': infix, {'ifx': infix,
'time': timeutils.utcnow().strftime('%Y%m%d%H%M%S%f')}) 'time': timeutils.utcnow().strftime('%Y%m%d%H%M%S%f')})
def _gen_host_name(self, vol_name_timestamp, access_level):
# host_name will be manila-{vol_name_timestamp}-ro or
# manila-{vol_name_timestamp}-rw
return 'manila-{}-{}'.format(vol_name_timestamp, access_level)
def _get_timestamp_from_vol_name(self, vol_name):
vol_name_split = vol_name.split('-')
dt = datetime.datetime.strptime(vol_name_split[2], '%Y%m%d%H%M%S%f')
return int(time.mktime(dt.timetuple()))
def _get_location_path(self, share_name, share_proto, ip, vol_id): def _get_location_path(self, share_name, share_proto, ip, vol_id):
if share_proto == 'NFS': if share_proto == 'NFS':
vol = self.api_executor.get_specific_volinfo(vol_id) vol = self.api_executor.get_specific_volinfo(vol_id)
@ -503,30 +517,31 @@ class QnapShareDriver(driver.ShareDriver):
self.configuration.qnap_share_ip, self.configuration.qnap_share_ip,
create_volID) create_volID)
def _get_manila_hostIPv4s(self, hostlist): def _get_vol_host(self, host_list, vol_name_timestamp):
host_dict_IPs = [] vol_host_list = []
if hostlist is None: if host_list is None:
return host_dict_IPs return vol_host_list
for host in hostlist: for host in host_list:
# Check host alias name with prefix "manila-hst-" to verify this # Check host alias name with prefix "manila-{vol_name_timestamp}"
# host is created/managed by Manila or not. # to find the host of this manila share.
if (re.match("^manila-hst-[0-9]+$", host.find('name').text) LOG.debug('_get_vol_host name:%s', host.find('name').text)
is not None): # Because driver supports only IPv4 now, check "netaddrs"
LOG.debug('host netaddrs text: %s', host.find('netaddrs').text) # have "ipv4" tag to get address.
if host.find('netaddrs').text is not None: if re.match("^manila-{}".format(vol_name_timestamp),
# Because Manila supports only IPv4 now, check "netaddrs" host.find('name').text):
# have "ipv4" tag to verify this host is created/managed
# by Manila or not.
if host.find('netaddrs/ipv4').text is not None:
host_dict = { host_dict = {
'index': host.find('index').text, 'index': host.find('index').text,
'hostid': host.find('hostid').text, 'hostid': host.find('hostid').text,
'name': host.find('name').text, 'name': host.find('name').text,
'netaddrs': host.find('netaddrs').find('ipv4').text 'ipv4': [],
} }
host_dict_IPs.append(host_dict) for ipv4 in host.findall('netaddrs/ipv4'):
return host_dict_IPs host_dict['ipv4'].append(ipv4.text)
vol_host_list.append(host_dict)
LOG.debug('_get_vol_host vol_host_list:%s', vol_host_list)
return vol_host_list
@utils.synchronized('qnap-update_access')
def update_access(self, context, share, access_rules, add_rules, def update_access(self, context, share, access_rules, add_rules,
delete_rules, share_server=None): delete_rules, share_server=None):
if not (add_rules or delete_rules): if not (add_rules or delete_rules):
@ -540,6 +555,15 @@ class QnapShareDriver(driver.ShareDriver):
# Clear all current ACLs # Clear all current ACLs
self.api_executor.set_nfs_access(volName, 2, "all") self.api_executor.set_nfs_access(volName, 2, "all")
vol_name_timestamp = self._get_timestamp_from_vol_name(volName)
host_list = self.api_executor.get_host_list()
LOG.debug('host_list:%s', host_list)
vol_host_list = self._get_vol_host(host_list, vol_name_timestamp)
# If host already exist, delete the host
if len(vol_host_list) > 0:
for vol_host in vol_host_list:
self.api_executor.delete_host(vol_host['name'])
# Add each one through all rules. # Add each one through all rules.
for access in access_rules: for access in access_rules:
self._allow_access(context, share, access, share_server) self._allow_access(context, share, access, share_server)
@ -556,45 +580,71 @@ class QnapShareDriver(driver.ShareDriver):
access_type = access['access_type'] access_type = access['access_type']
access_level = access['access_level'] access_level = access['access_level']
access_to = access['access_to'] access_to = access['access_to']
LOG.debug('share_proto: %(share_proto)s '
'access_type: %(access_type)s'
'access_level: %(access_level)s'
'access_to: %(access_to)s',
{'share_proto': share_proto,
'access_type': access_type,
'access_level': access_level,
'access_to': access_to})
self._check_share_access(share_proto, access_type) self._check_share_access(share_proto, access_type)
hostlist = self.api_executor.get_host_list() vol_name = self.private_storage.get(share['id'], 'volName')
host_dict_IPs = self._get_manila_hostIPv4s(hostlist) vol_name_timestamp = self._get_timestamp_from_vol_name(vol_name)
LOG.debug('host_dict_IPs: %s', host_dict_IPs) host_name = self._gen_host_name(vol_name_timestamp, access_level)
if len(host_dict_IPs) == 0:
host_name = self._gen_random_name("host") host_list = self.api_executor.get_host_list()
LOG.debug('vol_name: %(vol_name)s '
'access_level: %(access_level)s '
'host_name: %(host_name)s '
'host_list: %(host_list)s ',
{'vol_name': vol_name,
'access_level': access_level,
'host_name': host_name,
'host_list': host_list})
filter_host_list = self._get_vol_host(host_list, vol_name_timestamp)
if len(filter_host_list) == 0:
# if host does not exist, create a host for the share
self.api_executor.add_host(host_name, access_to) self.api_executor.add_host(host_name, access_to)
elif (len(filter_host_list) == 1 and
filter_host_list[0]['name'] == host_name):
# if the host exist, and this host is for the same access right,
# add ip to the host.
ipv4_list = filter_host_list[0]['ipv4']
if access_to not in ipv4_list:
ipv4_list.append(access_to)
LOG.debug('vol_host["ipv4"]: %s', filter_host_list[0]['ipv4'])
LOG.debug('ipv4_list: %s', ipv4_list)
self.api_executor.edit_host(host_name, ipv4_list)
else: else:
for host in host_dict_IPs: # Until now, share of QNAP NAS can only apply one access level for
LOG.debug('host[netaddrs]: %s', host['netaddrs']) # all ips. "rw" for some ips and "ro" for else is not allowed.
LOG.debug('access_to: %s', access_to) support_level = (constants.ACCESS_LEVEL_RW if
if host['netaddrs'] == access_to: access_level == constants.ACCESS_LEVEL_RO
LOG.debug('in match ip') else constants.ACCESS_LEVEL_RO)
host_name = host['name'] reason = _('Share only supports one access '
break 'level: %s') % support_level
if host is host_dict_IPs[-1]: LOG.error(reason)
host_name = self._gen_random_name("host") raise exception.InvalidShareAccess(reason=reason)
self.api_executor.add_host(host_name, access_to) access = 1 if access_level == constants.ACCESS_LEVEL_RO else 0
self.api_executor.set_nfs_access(vol_name, access, host_name)
volName = self.private_storage.get(share['id'], 'volName')
LOG.debug('volName: %(volName)s for share: %(share)s',
{'volName': volName, 'share': share['id']})
LOG.debug('access_level: %(access)s for share: %(share)s',
{'access': access_level, 'share': share['id']})
LOG.debug('host_name: %(host)s for share: %(share)s',
{'host': host_name, 'share': share['id']})
if access_level == constants.ACCESS_LEVEL_RO:
self.api_executor.set_nfs_access(volName, 1, host_name)
elif access_level == constants.ACCESS_LEVEL_RW:
self.api_executor.set_nfs_access(volName, 0, host_name)
def _deny_access(self, context, share, access, share_server=None): def _deny_access(self, context, share, access, share_server=None):
"""Deny access to the share.""" """Deny access to the share."""
share_proto = share['share_proto'] share_proto = share['share_proto']
access_type = access['access_type'] access_type = access['access_type']
access_level = access['access_level']
access_to = access['access_to'] access_to = access['access_to']
LOG.debug('share_proto: %(share_proto)s '
'access_type: %(access_type)s'
'access_level: %(access_level)s'
'access_to: %(access_to)s',
{'share_proto': share_proto,
'access_type': access_type,
'access_level': access_level,
'access_to': access_to})
try: try:
self._check_share_access(share_proto, access_type) self._check_share_access(share_proto, access_type)
@ -602,23 +652,34 @@ class QnapShareDriver(driver.ShareDriver):
LOG.warning('The denied rule is invalid and does not exist.') LOG.warning('The denied rule is invalid and does not exist.')
return return
hostlist = self.api_executor.get_host_list() vol_name = self.private_storage.get(share['id'], 'volName')
host_dict_IPs = self._get_manila_hostIPv4s(hostlist) vol_name_timestamp = self._get_timestamp_from_vol_name(vol_name)
LOG.debug('host_dict_IPs: %s', host_dict_IPs) host_name = self._gen_host_name(vol_name_timestamp, access_level)
if len(host_dict_IPs) == 0: host_list = self.api_executor.get_host_list()
return LOG.debug('vol_name: %(vol_name)s '
'access_level: %(access_level)s '
'host_name: %(host_name)s '
'host_list: %(host_list)s ',
{'vol_name': vol_name,
'access_level': access_level,
'host_name': host_name,
'host_list': host_list})
filter_host_list = self._get_vol_host(host_list, vol_name_timestamp)
# if share already have host, remove ip from host
for vol_host in filter_host_list:
if vol_host['name'] == host_name:
ipv4_list = vol_host['ipv4']
if access_to in ipv4_list:
ipv4_list.remove(access_to)
LOG.debug('vol_host["ipv4"]: %s', vol_host['ipv4'])
LOG.debug('ipv4_list: %s', ipv4_list)
if len(ipv4_list) == 0: # if list empty, remove the host
self.api_executor.set_nfs_access(
vol_name, 2, host_name)
self.api_executor.delete_host(host_name)
else: else:
for host in host_dict_IPs: self.api_executor.edit_host(host_name, ipv4_list)
if (host['netaddrs'] == access_to):
host_name = host['name']
break break
if (host is host_dict_IPs[-1]):
return
volName = self.private_storage.get(share['id'], 'volName')
LOG.debug('volName: %s', volName)
self.api_executor.set_nfs_access(volName, 2, host_name)
def _check_share_access(self, share_proto, access_type): def _check_share_access(self, share_proto, access_type):
if share_proto == 'NFS' and access_type != 'ip': if share_proto == 'NFS' and access_type != 'ip':

View File

@ -192,7 +192,7 @@ FAKE_RES_DETAIL_DATA_GET_HOST_LIST = """
<host> <host>
<index><![CDATA[fakeHostIndex]]></index> <index><![CDATA[fakeHostIndex]]></index>
<hostid><![CDATA[fakeHostId]]></hostid> <hostid><![CDATA[fakeHostId]]></hostid>
<name><![CDATA[manila-hst-123]]></name> <name><![CDATA[manila-fakeHostName]]></name>
<netaddrs> <netaddrs>
<ipv4> <ipv4>
<![CDATA[fakeIp]]> <![CDATA[fakeIp]]>
@ -287,6 +287,15 @@ FAKE_RES_DETAIL_DATA_GET_HOST_LIST_API = """
<result><![CDATA[0]]></result> <result><![CDATA[0]]></result>
</QDocRoot>""" </QDocRoot>"""
FAKE_RES_DETAIL_DATA_GET_NO_HOST_LIST_API = """
<QDocRoot version="1.0">
<authPassed><![CDATA[1]]></authPassed>
<ES_RET_CODE><![CDATA[1]]></ES_RET_CODE>
<content>
</content>
<result><![CDATA[0]]></result>
</QDocRoot>"""
FAKE_RES_DETAIL_DATA_CREATE_SNAPSHOT = """ FAKE_RES_DETAIL_DATA_CREATE_SNAPSHOT = """
<QDocRoot version="1.0"> <QDocRoot version="1.0">
<authPassed><![CDATA[1]]></authPassed> <authPassed><![CDATA[1]]></authPassed>
@ -546,7 +555,7 @@ class FakeDeleteSnapshotResponseShareNotExist(object):
class FakeGetHostListResponse(object): class FakeGetHostListResponse(object):
"""Fake pool info response.""" """Fake host info response."""
status = 'fackStatus' status = 'fackStatus'
@ -555,6 +564,16 @@ class FakeGetHostListResponse(object):
return FAKE_RES_DETAIL_DATA_GET_HOST_LIST_API return FAKE_RES_DETAIL_DATA_GET_HOST_LIST_API
class FakeGetNoHostListResponse(object):
"""Fake host info response."""
status = 'fackStatus'
def read(self):
"""Mock response.read."""
return FAKE_RES_DETAIL_DATA_GET_NO_HOST_LIST_API
class FakeAuthPassFailResponse(object): class FakeAuthPassFailResponse(object):
"""Fake pool info response.""" """Fake pool info response."""

View File

@ -91,14 +91,17 @@ class QnapAPITestCase(QnapShareDriverBaseTestCase):
host='QnapShareDriver', host='QnapShareDriver',
size=10) size=10)
def _sanitize_params(self, params): def _sanitize_params(self, params, doseq=False):
sanitized_params = {} sanitized_params = {}
for key in params: for key in params:
value = params[key] value = params[key]
if value is not None: if value is not None:
if isinstance(value, list):
sanitized_params[key] = [six.text_type(v) for v in value]
else:
sanitized_params[key] = six.text_type(value) sanitized_params[key] = six.text_type(value)
sanitized_params = urllib.parse.urlencode(sanitized_params) sanitized_params = urllib.parse.urlencode(sanitized_params, doseq)
return sanitized_params return sanitized_params
@ddt.data('fake_share_name', 'fakeLabel') @ddt.data('fake_share_name', 'fakeLabel')
@ -493,14 +496,16 @@ class QnapAPITestCase(QnapShareDriverBaseTestCase):
expected_call_list, expected_call_list,
mock_http_connection.return_value.request.call_args_list) mock_http_connection.return_value.request.call_args_list)
def test_get_host_list(self): @ddt.data(fakes.FakeGetHostListResponse(),
fakes.FakeGetNoHostListResponse())
def test_get_host_list(self, fakeGetHostListResponse):
"""Test get host list api.""" """Test get host list api."""
mock_http_connection = six.moves.http_client.HTTPConnection mock_http_connection = six.moves.http_client.HTTPConnection
mock_http_connection.return_value.getresponse.side_effect = [ mock_http_connection.return_value.getresponse.side_effect = [
fakes.FakeLoginResponse(), fakes.FakeLoginResponse(),
fakes.FakeGetBasicInfoResponseEs_1_1_3(), fakes.FakeGetBasicInfoResponseEs_1_1_3(),
fakes.FakeLoginResponse(), fakes.FakeLoginResponse(),
fakes.FakeGetHostListResponse()] fakeGetHostListResponse]
self._do_setup('http://1.2.3.4:8080', '1.2.3.4', 'admin', self._do_setup('http://1.2.3.4:8080', '1.2.3.4', 'admin',
'qnapadmin', 'Storage Pool 1') 'qnapadmin', 'Storage Pool 1')
@ -560,8 +565,8 @@ class QnapAPITestCase(QnapShareDriverBaseTestCase):
expected_call_list, expected_call_list,
mock_http_connection.return_value.request.call_args_list) mock_http_connection.return_value.request.call_args_list)
def test_set_nfs_access(self): def test_edit_host(self):
"""Test get host list api.""" """Test edit host api."""
mock_http_connection = six.moves.http_client.HTTPConnection mock_http_connection = six.moves.http_client.HTTPConnection
mock_http_connection.return_value.getresponse.side_effect = [ mock_http_connection.return_value.getresponse.side_effect = [
fakes.FakeLoginResponse(), fakes.FakeLoginResponse(),
@ -569,6 +574,75 @@ class QnapAPITestCase(QnapShareDriverBaseTestCase):
fakes.FakeLoginResponse(), fakes.FakeLoginResponse(),
fakes.FakeGetHostListResponse()] fakes.FakeGetHostListResponse()]
self._do_setup('http://1.2.3.4:8080', '1.2.3.4', 'admin',
'qnapadmin', 'Storage Pool 1')
self.driver.api_executor.edit_host(
'fakeHostName', ['fakeIpV4'])
fake_params = {
'module': 'hosts',
'func': 'apply_sethost',
'name': 'fakeHostName',
'ipaddr_v4': ['fakeIpV4'],
'sid': 'fakeSid',
}
sanitized_params = self._sanitize_params(fake_params, doseq=True)
fake_url = (
('/cgi-bin/accessrights/accessrightsRequest.cgi?%s') %
sanitized_params)
expected_call_list = [
mock.call('GET', self.login_url),
mock.call('GET', self.get_basic_info_url),
mock.call('GET', self.login_url),
mock.call('GET', fake_url)]
self.assertEqual(
expected_call_list,
mock_http_connection.return_value.request.call_args_list)
def test_delete_host(self):
"""Test delete host api."""
mock_http_connection = six.moves.http_client.HTTPConnection
mock_http_connection.return_value.getresponse.side_effect = [
fakes.FakeLoginResponse(),
fakes.FakeGetBasicInfoResponseEs_1_1_3(),
fakes.FakeLoginResponse(),
fakes.FakeGetHostListResponse()]
self._do_setup('http://1.2.3.4:8080', '1.2.3.4', 'admin',
'qnapadmin', 'Storage Pool 1')
self.driver.api_executor.delete_host('fakeHostName')
fake_params = {
'module': 'hosts',
'func': 'apply_delhost',
'host_name': 'fakeHostName',
'sid': 'fakeSid',
}
sanitized_params = self._sanitize_params(fake_params)
fake_url = (
('/cgi-bin/accessrights/accessrightsRequest.cgi?%s') %
sanitized_params)
expected_call_list = [
mock.call('GET', self.login_url),
mock.call('GET', self.get_basic_info_url),
mock.call('GET', self.login_url),
mock.call('GET', fake_url)]
self.assertEqual(
expected_call_list,
mock_http_connection.return_value.request.call_args_list)
@ddt.data(fakes.FakeGetHostListResponse())
def test_set_nfs_access(self, fakeGetHostListResponse):
"""Test get host list api."""
mock_http_connection = six.moves.http_client.HTTPConnection
mock_http_connection.return_value.getresponse.side_effect = [
fakes.FakeLoginResponse(),
fakes.FakeGetBasicInfoResponseEs_1_1_3(),
fakes.FakeLoginResponse(),
fakeGetHostListResponse]
self._do_setup('http://1.2.3.4:8080', '1.2.3.4', 'admin', self._do_setup('http://1.2.3.4:8080', '1.2.3.4', 'admin',
'qnapadmin', 'Storage Pool 1') 'qnapadmin', 'Storage Pool 1')
self.driver.api_executor.set_nfs_access( self.driver.api_executor.set_nfs_access(
@ -749,6 +823,24 @@ class QnapAPITestCase(QnapShareDriverBaseTestCase):
'ipv4': 'fakeIpV4'}, 'ipv4': 'fakeIpV4'},
fakes.FakeAuthPassFailResponse(), fakes.FakeAuthPassFailResponse(),
fakes.FakeGetBasicInfoResponseEs_1_1_3()], fakes.FakeGetBasicInfoResponseEs_1_1_3()],
['self.driver.api_executor.edit_host',
{'hostname': 'fakeHostName',
'ipv4_list': 'fakeIpV4List'},
fakes.FakeResultNegativeResponse(),
fakes.FakeGetBasicInfoResponseEs_1_1_3()],
['self.driver.api_executor.edit_host',
{'hostname': 'fakeHostName',
'ipv4_list': 'fakeIpV4List'},
fakes.FakeAuthPassFailResponse(),
fakes.FakeGetBasicInfoResponseEs_1_1_3()],
['self.driver.api_executor.delete_host',
{'hostname': 'fakeHostName'},
fakes.FakeResultNegativeResponse(),
fakes.FakeGetBasicInfoResponseEs_1_1_3()],
['self.driver.api_executor.delete_host',
{'hostname': 'fakeHostName'},
fakes.FakeAuthPassFailResponse(),
fakes.FakeGetBasicInfoResponseEs_1_1_3()],
['self.driver.api_executor.get_host_list', ['self.driver.api_executor.get_host_list',
{}, {},
fakes.FakeResultNegativeResponse(), fakes.FakeResultNegativeResponse(),

View File

@ -721,15 +721,22 @@ class QnapShareDriverTestCase(QnapShareDriverBaseTestCase):
snapshot=fake_snapshot, snapshot=fake_snapshot,
share_server=None) share_server=None)
@mock.patch.object(qnap.QnapShareDriver, '_get_timestamp_from_vol_name')
@mock.patch.object(qnap.QnapShareDriver, '_allow_access') @mock.patch.object(qnap.QnapShareDriver, '_allow_access')
@ddt.data('fakeHostName', 'fakeHostNameNotMatch')
def test_update_access_allow_access( def test_update_access_allow_access(
self, mock_allow_access): self, fakeHostName, mock_allow_access,
mock_get_timestamp_from_vol_name):
"""Test update access with allow access rules.""" """Test update access with allow access rules."""
mock_private_storage = mock.Mock() mock_private_storage = mock.Mock()
mock_private_storage.get.return_value = 'fakeVolName' mock_private_storage.get.return_value = 'fakeVolName'
mock_api_executor = qnap.QnapShareDriver._create_api_executor mock_api_executor = qnap.QnapShareDriver._create_api_executor
mock_api_executor.return_value.get_host_list.return_value = (
self.get_host_list_return_value())
mock_api_executor.return_value.set_nfs_access.return_value = None mock_api_executor.return_value.set_nfs_access.return_value = None
mock_api_executor.return_value.delete_host.return_value = None
mock_allow_access.return_value = None mock_allow_access.return_value = None
mock_get_timestamp_from_vol_name.return_value = fakeHostName
self._do_setup('http://1.2.3.4:8080', '1.2.3.4', 'admin', self._do_setup('http://1.2.3.4:8080', '1.2.3.4', 'admin',
'qnapadmin', 'Storage Pool 1', 'qnapadmin', 'Storage Pool 1',
@ -1044,7 +1051,7 @@ class QnapShareDriverTestCase(QnapShareDriverBaseTestCase):
mock_api_return.get_specific_poolinfo.assert_called_once_with( mock_api_return.get_specific_poolinfo.assert_called_once_with(
self.driver.configuration.qnap_poolname) self.driver.configuration.qnap_poolname)
def test_get_manila_host_ipv4s(self): def test_get_vol_host(self):
"""Test get manila host IPV4s.""" """Test get manila host IPV4s."""
mock_private_storage = mock.Mock() mock_private_storage = mock.Mock()
@ -1059,28 +1066,31 @@ class QnapShareDriverTestCase(QnapShareDriverBaseTestCase):
'index': host.find('index').text, 'index': host.find('index').text,
'hostid': host.find('hostid').text, 'hostid': host.find('hostid').text,
'name': host.find('name').text, 'name': host.find('name').text,
'netaddrs': host.find('netaddrs').find('ipv4').text 'ipv4': [host.find('netaddrs').find('ipv4').text]
} }
expect_host_dict_ips.append(host_dict) expect_host_dict_ips.append(host_dict)
self.assertEqual( self.assertEqual(
expect_host_dict_ips, self.driver._get_manila_hostIPv4s( expect_host_dict_ips, self.driver._get_vol_host(
host_list)) host_list, 'fakeHostName'))
@mock.patch.object(qnap.QnapShareDriver, '_gen_random_name') @mock.patch.object(qnap.QnapShareDriver, '_gen_host_name')
@mock.patch.object(qnap.QnapShareDriver, '_get_timestamp_from_vol_name')
@mock.patch.object(qnap.QnapShareDriver, '_check_share_access') @mock.patch.object(qnap.QnapShareDriver, '_check_share_access')
def test_allow_access_ro( def test_allow_access_ro(
self, self,
mock_check_share_access, mock_check_share_access,
mock_gen_random_name): mock_get_timestamp_from_vol_name,
mock_gen_host_name):
"""Test allow_access with access type ro.""" """Test allow_access with access type ro."""
fake_access = fakes.AccessClass('fakeAccessType', 'ro', 'fakeIp') fake_access = fakes.AccessClass('fakeAccessType', 'ro', 'fakeIp')
mock_private_storage = mock.Mock() mock_private_storage = mock.Mock()
mock_private_storage.get.return_value = 'fakeVolName'
mock_api_executor = qnap.QnapShareDriver._create_api_executor mock_api_executor = qnap.QnapShareDriver._create_api_executor
mock_api_executor.return_value.get_host_list.return_value = ( mock_api_executor.return_value.get_host_list.return_value = []
self.get_host_list_return_value()) mock_get_timestamp_from_vol_name.return_value = 'fakeHostName'
mock_gen_random_name.return_value = 'fakeHostName' mock_gen_host_name.return_value = 'manila-fakeHostName-ro'
mock_api_executor.return_value.add_host.return_value = None mock_api_executor.return_value.add_host.return_value = None
mock_api_executor.return_value.set_nfs_access.return_value = None mock_api_executor.return_value.set_nfs_access.return_value = None
@ -1092,14 +1102,17 @@ class QnapShareDriverTestCase(QnapShareDriverBaseTestCase):
mock_check_share_access.assert_called_once_with( mock_check_share_access.assert_called_once_with(
'NFS', 'fakeAccessType') 'NFS', 'fakeAccessType')
mock_gen_random_name.assert_called_once_with('host')
mock_api_executor.return_value.add_host.assert_called_once_with( mock_api_executor.return_value.add_host.assert_called_once_with(
'fakeHostName', 'fakeIp') 'manila-fakeHostName-ro', 'fakeIp')
@mock.patch.object(qnap.QnapShareDriver, '_gen_host_name')
@mock.patch.object(qnap.QnapShareDriver, '_get_timestamp_from_vol_name')
@mock.patch.object(qnap.QnapShareDriver, '_check_share_access') @mock.patch.object(qnap.QnapShareDriver, '_check_share_access')
def test_allow_access_ro_with_hostlist( def test_allow_access_ro_with_hostlist(
self, self,
mock_check_share_access): mock_check_share_access,
mock_get_timestamp_from_vol_name,
mock_gen_host_name):
"""Test allow_access_ro_with_hostlist.""" """Test allow_access_ro_with_hostlist."""
host_dict_ips = [] host_dict_ips = []
for host in self.get_host_list_return_value(): for host in self.get_host_list_return_value():
@ -1108,18 +1121,21 @@ class QnapShareDriverTestCase(QnapShareDriverBaseTestCase):
'index': host.find('index').text, 'index': host.find('index').text,
'hostid': host.find('hostid').text, 'hostid': host.find('hostid').text,
'name': host.find('name').text, 'name': host.find('name').text,
'netaddrs': host.find('netaddrs').find('ipv4').text} 'ipv4': [host.find('netaddrs').find('ipv4').text]}
host_dict_ips.append(host_dict) host_dict_ips.append(host_dict)
for host in host_dict_ips: for host in host_dict_ips:
fake_access_to = host['netaddrs'] fake_access_to = host['ipv4']
fake_access = fakes.AccessClass( fake_access = fakes.AccessClass(
'fakeAccessType', 'ro', fake_access_to) 'fakeAccessType', 'ro', fake_access_to)
mock_private_storage = mock.Mock() mock_private_storage = mock.Mock()
mock_private_storage.get.return_value = 'fakeVolName'
mock_api_executor = qnap.QnapShareDriver._create_api_executor mock_api_executor = qnap.QnapShareDriver._create_api_executor
mock_api_executor.return_value.get_host_list.return_value = ( mock_api_executor.return_value.get_host_list.return_value = (
self.get_host_list_return_value()) self.get_host_list_return_value())
mock_get_timestamp_from_vol_name.return_value = 'fakeHostName'
mock_gen_host_name.return_value = 'manila-fakeHostName'
mock_api_executor.return_value.set_nfs_access.return_value = None mock_api_executor.return_value.set_nfs_access.return_value = None
self._do_setup('http://1.2.3.4:8080', '1.2.3.4', 'admin', self._do_setup('http://1.2.3.4:8080', '1.2.3.4', 'admin',
@ -1131,20 +1147,64 @@ class QnapShareDriverTestCase(QnapShareDriverBaseTestCase):
mock_check_share_access.assert_called_once_with( mock_check_share_access.assert_called_once_with(
'NFS', 'fakeAccessType') 'NFS', 'fakeAccessType')
@mock.patch.object(qnap.QnapShareDriver, '_gen_random_name') @mock.patch.object(qnap.QnapShareDriver, '_gen_host_name')
@mock.patch.object(qnap.QnapShareDriver, '_get_timestamp_from_vol_name')
@mock.patch.object(qnap.QnapShareDriver, '_check_share_access')
def test_allow_access_rw_with_hostlist_invalid_access(
self,
mock_check_share_access,
mock_get_timestamp_from_vol_name,
mock_gen_host_name):
"""Test allow_access_rw_invalid_access."""
host_dict_ips = []
for host in self.get_host_list_return_value():
if host.find('netaddrs/ipv4').text is not None:
host_dict = {
'index': host.find('index').text,
'hostid': host.find('hostid').text,
'name': host.find('name').text,
'ipv4': [host.find('netaddrs').find('ipv4').text]}
host_dict_ips.append(host_dict)
for host in host_dict_ips:
fake_access_to = host['ipv4']
fake_access = fakes.AccessClass(
'fakeAccessType', 'rw', fake_access_to)
mock_private_storage = mock.Mock()
mock_private_storage.get.return_value = 'fakeVolName'
mock_api_executor = qnap.QnapShareDriver._create_api_executor
mock_api_executor.return_value.get_host_list.return_value = (
self.get_host_list_return_value())
mock_get_timestamp_from_vol_name.return_value = 'fakeHostName'
mock_gen_host_name.return_value = 'manila-fakeHostName-rw'
self._do_setup('http://1.2.3.4:8080', '1.2.3.4', 'admin',
'qnapadmin', 'Storage Pool 1',
private_storage=mock_private_storage)
self.assertRaises(
exception.InvalidShareAccess,
self.driver._allow_access,
context='context',
share=self.share,
access=fake_access,
share_server=None)
@mock.patch.object(qnap.QnapShareDriver, '_get_timestamp_from_vol_name')
@mock.patch.object(qnap.QnapShareDriver, '_check_share_access') @mock.patch.object(qnap.QnapShareDriver, '_check_share_access')
def test_allow_access_rw( def test_allow_access_rw(
self, self,
mock_check_share_access, mock_check_share_access,
mock_gen_random_name): mock_get_timestamp_from_vol_name):
"""Test allow_access with access type rw.""" """Test allow_access with access type rw."""
fake_access = fakes.AccessClass('fakeAccessType', 'rw', 'fakeIp') fake_access = fakes.AccessClass('fakeAccessType', 'rw', 'fakeIp')
mock_private_storage = mock.Mock() mock_private_storage = mock.Mock()
mock_private_storage.get.return_value = 'fakeVolName'
mock_api_executor = qnap.QnapShareDriver._create_api_executor mock_api_executor = qnap.QnapShareDriver._create_api_executor
mock_api_executor.return_value.get_host_list.return_value = ( mock_api_executor.return_value.get_host_list.return_value = []
self.get_host_list_return_value()) mock_get_timestamp_from_vol_name.return_value = 'fakeHostName'
mock_gen_random_name.return_value = 'fakeHostName'
mock_api_executor.return_value.add_host.return_value = None mock_api_executor.return_value.add_host.return_value = None
mock_api_executor.return_value.set_nfs_access.return_value = None mock_api_executor.return_value.set_nfs_access.return_value = None
@ -1156,44 +1216,50 @@ class QnapShareDriverTestCase(QnapShareDriverBaseTestCase):
mock_check_share_access.assert_called_once_with( mock_check_share_access.assert_called_once_with(
'NFS', 'fakeAccessType') 'NFS', 'fakeAccessType')
mock_gen_random_name.assert_called_once_with('host')
mock_api_executor.return_value.add_host.assert_called_once_with( mock_api_executor.return_value.add_host.assert_called_once_with(
'fakeHostName', 'fakeIp') 'manila-fakeHostName-rw', 'fakeIp')
@mock.patch.object(qnap.QnapShareDriver, '_get_manila_hostIPv4s') @mock.patch.object(qnap.QnapShareDriver, '_gen_host_name')
@mock.patch.object(qnap.QnapShareDriver, '_gen_random_name')
@mock.patch.object(qnap.QnapShareDriver, '_check_share_access') @mock.patch.object(qnap.QnapShareDriver, '_check_share_access')
def test_allow_access_without_hostlist( def test_allow_access_ro_without_hostlist(
self, self,
mock_check_share_access, mock_check_share_access,
mock_gen_random_name, mock_gen_host_name):
mock_get_manila_hostipv4s):
"""Test allow access without host list.""" """Test allow access without host list."""
fake_access = fakes.AccessClass('fakeAccessType', 'ro', 'fakeIp') fake_access = fakes.AccessClass('fakeAccessType', 'ro', 'fakeIp')
mock_private_storage = mock.Mock() mock_private_storage = mock.Mock()
mock_api_executor = qnap.QnapShareDriver._create_api_executor mock_api_executor = qnap.QnapShareDriver._create_api_executor
mock_api_executor.return_value.get_host_list.return_value = None mock_api_executor.return_value.get_host_list.return_value = None
mock_gen_random_name.return_value = 'fakeHostName' mock_gen_host_name.return_value = 'fakeHostName'
mock_api_executor.return_value.add_host.return_value = None mock_api_executor.return_value.add_host.return_value = None
mock_api_executor.return_value.set_nfs_access.return_value = None mock_api_executor.return_value.set_nfs_access.return_value = None
self._do_setup('http://1.2.3.4:8080', '1.2.3.4', 'admin', self._do_setup('http://1.2.3.4:8080', '1.2.3.4', 'admin',
'qnapadmin', 'Storage Pool 1', 'qnapadmin', 'Storage Pool 1',
private_storage=mock_private_storage) private_storage=mock_private_storage)
share_name = self.driver._gen_random_name('share')
mock_private_storage.get.return_value = share_name
self.driver._allow_access( self.driver._allow_access(
'context', self.share, fake_access, share_server=None) 'context', self.share, fake_access, share_server=None)
mock_check_share_access.assert_called_once_with( mock_check_share_access.assert_called_once_with(
'NFS', 'fakeAccessType') 'NFS', 'fakeAccessType')
mock_gen_random_name.assert_called_once_with('host')
mock_api_executor.return_value.add_host.assert_called_once_with( mock_api_executor.return_value.add_host.assert_called_once_with(
'fakeHostName', 'fakeIp') 'fakeHostName', 'fakeIp')
@mock.patch.object(qnap.QnapShareDriver, '_get_vol_host')
@mock.patch.object(qnap.QnapShareDriver, '_gen_host_name')
@mock.patch.object(qnap.QnapShareDriver, '_get_timestamp_from_vol_name')
@mock.patch.object(qnap.QnapShareDriver, '_check_share_access') @mock.patch.object(qnap.QnapShareDriver, '_check_share_access')
def test_deny_access_with_hostlist( def test_deny_access_with_hostlist(
self, self,
mock_check_share_access): mock_check_share_access,
mock_get_timestamp_from_vol_name,
mock_gen_host_name,
mock_get_vol_host):
"""Test deny access.""" """Test deny access."""
host_dict_ips = [] host_dict_ips = []
for host in self.get_host_list_return_value(): for host in self.get_host_list_return_value():
@ -1202,11 +1268,11 @@ class QnapShareDriverTestCase(QnapShareDriverBaseTestCase):
'index': host.find('index').text, 'index': host.find('index').text,
'hostid': host.find('hostid').text, 'hostid': host.find('hostid').text,
'name': host.find('name').text, 'name': host.find('name').text,
'netaddrs': host.find('netaddrs').find('ipv4').text} 'ipv4': [host.find('netaddrs').find('ipv4').text]}
host_dict_ips.append(host_dict) host_dict_ips.append(host_dict)
for host in host_dict_ips: for host in host_dict_ips:
fake_access_to = host['netaddrs'] fake_access_to = host['ipv4'][0]
fake_access = fakes.AccessClass('fakeAccessType', 'ro', fake_access_to) fake_access = fakes.AccessClass('fakeAccessType', 'ro', fake_access_to)
mock_private_storage = mock.Mock() mock_private_storage = mock.Mock()
@ -1216,6 +1282,9 @@ class QnapShareDriverTestCase(QnapShareDriverBaseTestCase):
qnap.QnapShareDriver._create_api_executor.return_value) qnap.QnapShareDriver._create_api_executor.return_value)
mock_api_return.get_host_list.return_value = ( mock_api_return.get_host_list.return_value = (
self.get_host_list_return_value()) self.get_host_list_return_value())
mock_get_timestamp_from_vol_name.return_value = 'fakeTimeStamp'
mock_gen_host_name.return_value = 'manila-fakeHostName'
mock_get_vol_host.return_value = host_dict_ips
mock_api_return.add_host.return_value = None mock_api_return.add_host.return_value = None
mock_api_return.set_nfs_access.return_value = None mock_api_return.set_nfs_access.return_value = None
@ -1227,13 +1296,13 @@ class QnapShareDriverTestCase(QnapShareDriverBaseTestCase):
mock_check_share_access.assert_called_once_with( mock_check_share_access.assert_called_once_with(
'NFS', 'fakeAccessType') 'NFS', 'fakeAccessType')
mock_api_return.set_nfs_access.assert_called_once_with(
'vol_name', 2, 'manila-hst-123')
@mock.patch.object(qnap.QnapShareDriver, '_get_timestamp_from_vol_name')
@mock.patch.object(qnap.QnapShareDriver, '_check_share_access') @mock.patch.object(qnap.QnapShareDriver, '_check_share_access')
def test_deny_access_with_hostlist_not_equel_access_to( def test_deny_access_with_hostlist_not_equel_access_to(
self, self,
mock_check_share_access): mock_check_share_access,
mock_get_timestamp_from_vol_name):
"""Test deny access.""" """Test deny access."""
fake_access = fakes.AccessClass('fakeAccessType', 'ro', 'fakeIp') fake_access = fakes.AccessClass('fakeAccessType', 'ro', 'fakeIp')
@ -1254,18 +1323,20 @@ class QnapShareDriverTestCase(QnapShareDriverBaseTestCase):
mock_check_share_access.assert_called_once_with( mock_check_share_access.assert_called_once_with(
'NFS', 'fakeAccessType') 'NFS', 'fakeAccessType')
@mock.patch.object(qnap.QnapShareDriver, '_get_manila_hostIPv4s') @mock.patch.object(qnap.QnapShareDriver, '_get_timestamp_from_vol_name')
@mock.patch.object(qnap.QnapShareDriver, '_check_share_access') @mock.patch.object(qnap.QnapShareDriver, '_check_share_access')
def test_deny_access_without_hostlist( def test_deny_access_without_hostlist(
self, self,
mock_check_share_access, mock_check_share_access,
mock_get_manila_hostipv4s): mock_get_timestamp_from_vol_name):
"""Test deny access without hostlist.""" """Test deny access without hostlist."""
fake_access = fakes.AccessClass('fakeAccessType', 'ro', 'fakeIp') fake_access = fakes.AccessClass('fakeAccessType', 'ro', 'fakeIp')
mock_private_storage = mock.Mock() mock_private_storage = mock.Mock()
mock_private_storage.get.return_value = 'fakeVolName'
mock_api_executor = qnap.QnapShareDriver._create_api_executor mock_api_executor = qnap.QnapShareDriver._create_api_executor
mock_api_executor.return_value.get_host_list.return_value = None mock_api_executor.return_value.get_host_list.return_value = None
mock_get_timestamp_from_vol_name.return_value = 'fakeHostName'
mock_api_executor.return_value.add_host.return_value = None mock_api_executor.return_value.add_host.return_value = None
mock_api_executor.return_value.set_nfs_access.return_value = None mock_api_executor.return_value.set_nfs_access.return_value = None

View File

@ -0,0 +1,5 @@
---
fixes:
- |
Fixed the QNAP driver that the access rule setting is overridden by the later
access rule setting.