openstack/manila
Code Issues Proposed changes

Improve share-network list API filtering

Browse Source 
Add filtering share-network list by creation date, associated security
service ID, project ID (for admin context only).

Limit share network list using 'offset' and 'limit' options.

Add policy rule for 'get_all_share_networks' action.

Remove 'status' field from returned share nerwork's view,
as share_network model doesn't have such attribute. So, None was always
returned as a status.

Fix DB api function share_network_get_all_by_security_service - second param
should be named 'security_service_id' instead of 'share_network_id'.

Add unit and tempest tests.

Implements bp improve-share-network-list-filtering

Change-Id: I7c3e44068fb39672d1d6c9e06527fde2604672a9
This commit is contained in:
Julia Varlamova 2014-10-21 13:10:24 +04:00
parent 2d811e86cb
commit 6c04272429
10 changed files with 459 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

98
contrib/tempest/tempest/api/share/admin/test_share_networks.py Normal file
View File

@ -0,0 +1,98 @@
# Copyright 2014 Mirantis Inc.
# 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 tempest.api.share import base
from tempest.api.share import test_share_networks
from tempest import test
class ShareNetworkAdminTest(
base.BaseSharesAdminTest,
test_share_networks.ShareNetworkListMixin):
@classmethod
def setUpClass(cls):
super(ShareNetworkAdminTest, cls).setUpClass()
ss_data = cls.generate_security_service_data()
resp, cls.ss_ldap = cls.create_security_service(**ss_data)
cls.data_sn_with_ldap_ss = {
'name': 'sn_with_ldap_ss',
'neutron_net_id': '1111',
'neutron_subnet_id': '2222',
'created_at': '2002-02-02',
'updated_at': None,
'network_type': 'vlan',
'segmentation_id': 1000,
'cidr': '10.0.0.0/24',
'ip_version': 4,
'description': 'fake description',
}
__, cls.sn_with_ldap_ss = cls.create_share_network(
cleanup_in_class=True,
**cls.data_sn_with_ldap_ss)
resp, body = cls.shares_client.add_sec_service_to_share_network(
cls.sn_with_ldap_ss["id"],
cls.ss_ldap["id"])
cls.isolated_client = cls.get_client_with_isolated_creds(
type_of_creds='alt')
cls.data_sn_with_kerberos_ss = {
'name': 'sn_with_kerberos_ss',
'neutron_net_id': '3333',
'neutron_subnet_id': '4444',
'created_at': '2003-03-03',
'updated_at': None,
'neutron_net_id': 'test net id',
'neutron_subnet_id': 'test subnet id',
'network_type': 'local',
'segmentation_id': 2000,
'cidr': '10.0.0.0/13',
'ip_version': 6,
'description': 'fake description',
}
resp, cls.ss_kerberos = cls.isolated_client.create_security_service(
ss_type='kerberos',
**cls.data_sn_with_ldap_ss)
__, cls.sn_with_kerberos_ss = cls.isolated_client.create_share_network(
cleanup_in_class=True,
**cls.data_sn_with_kerberos_ss)
resp, body = cls.isolated_client.add_sec_service_to_share_network(
cls.sn_with_kerberos_ss["id"],
cls.ss_kerberos["id"])
@test.attr(type=["gate", "smoke", ])
def test_list_share_networks_all_tenants(self):
resp, listed = self.shares_client.list_share_networks_with_detail(
{'all_tenants': 1})
self.assertIn(int(resp["status"]), test.HTTP_SUCCESS)
self.assertTrue(any(self.sn_with_ldap_ss['id'] == sn['id']
for sn in listed))
self.assertTrue(any(self.sn_with_kerberos_ss['id'] == sn['id']
for sn in listed))
@test.attr(type=["gate", "smoke", ])
def test_list_share_networks_filter_by_porject_ip(self):
resp, listed = self.shares_client.list_share_networks_with_detail(
{'project_id': self.sn_with_kerberos_ss['project_id']})
self.assertIn(int(resp["status"]), test.HTTP_SUCCESS)
self.assertTrue(any(self.sn_with_kerberos_ss['id'] == sn['id']
for sn in listed))
self.assertTrue(all(self.sn_with_kerberos_ss['project_id'] ==
sn['project_id'] for sn in listed))

162
contrib/tempest/tempest/api/share/test_share_networks.py
View File

@ -13,18 +13,134 @@
# License for the specific language governing permissions and limitations
# under the License.
import six # noqa
from tempest.api.share import base
from tempest import test
class ShareNetworksTest(base.BaseSharesTest):
class ShareNetworkListMixin(object):
@test.attr(type=["gate", "smoke", ])
def test_list_share_networks(self):
resp, listed = self.shares_client.list_share_networks()
self.assertIn(int(resp["status"]), test.HTTP_SUCCESS)
any(self.sn_with_ldap_ss["id"] in sn["id"] for sn in listed)
# verify keys
keys = ["name", "id"]
[self.assertIn(key, sn.keys()) for sn in listed for key in keys]
@test.attr(type=["gate", "smoke", ])
def test_list_share_networks_with_detail(self):
resp, listed = self.shares_client.list_share_networks_with_detail()
self.assertIn(int(resp["status"]), test.HTTP_SUCCESS)
any(self.sn_with_ldap_ss["id"] in sn["id"] for sn in listed)
# verify keys
keys = [
"name", "id", "description", "network_type",
"project_id", "cidr", "ip_version",
"neutron_net_id", "neutron_subnet_id",
"created_at", "updated_at", "segmentation_id",
]
[self.assertIn(key, sn.keys()) for sn in listed for key in keys]
@test.attr(type=["gate", "smoke", ])
def test_list_share_networks_filter_by_ss(self):
resp, listed = self.shares_client.list_share_networks_with_detail(
{'security_service_id': self.ss_ldap['id']})
self.assertIn(int(resp["status"]), test.HTTP_SUCCESS)
self.assertTrue(any(self.sn_with_ldap_ss['id'] == sn['id']
for sn in listed))
for sn in listed:
resp, ss_list = self.shares_client.\
list_sec_services_for_share_network(sn['id'])
self.assertTrue(any(ss['id'] == self.ss_ldap['id']
for ss in ss_list))
@test.attr(type=["gate", "smoke", ])
def test_list_share_networks_all_filter_opts(self):
valid_filter_opts = {
'created_before': '2002-10-10',
'created_since': '2001-01-01',
'neutron_net_id': '1111',
'neutron_subnet_id': '2222',
'network_type': 'vlan',
'segmentation_id': 1000,
'cidr': '10.0.0.0/24',
'ip_version': 4,
'name': 'sn_with_ldap_ss'
}
resp, listed = self.shares_client.list_share_networks_with_detail(
valid_filter_opts)
self.assertIn(int(resp["status"]), test.HTTP_SUCCESS)
self.assertTrue(any(self.sn_with_ldap_ss['id'] == sn['id']
for sn in listed))
created_before = valid_filter_opts.pop('created_before')
created_since = valid_filter_opts.pop('created_since')
for sn in listed:
self.assertTrue(all(sn[key] == value for key, value in
six.iteritems(valid_filter_opts)))
self.assertTrue(sn['created_at'] <= created_before)
self.assertTrue(sn['created_at'] >= created_since)
class ShareNetworksTest(base.BaseSharesTest, ShareNetworkListMixin):
@classmethod
def setUpClass(cls):
super(ShareNetworksTest, cls).setUpClass()
cls.data = cls.generate_share_network_data()
__, cls.sn = cls.create_share_network(cleanup_in_class=True,
**cls.data)
ss_data = cls.generate_security_service_data()
resp, cls.ss_ldap = cls.create_security_service(**ss_data)
cls.data_sn_with_ldap_ss = {
'name': 'sn_with_ldap_ss',
'neutron_net_id': '1111',
'neutron_subnet_id': '2222',
'created_at': '2002-02-02',
'updated_at': None,
'network_type': 'vlan',
'segmentation_id': 1000,
'cidr': '10.0.0.0/24',
'ip_version': 4,
'description': 'fake description',
}
__, cls.sn_with_ldap_ss = cls.create_share_network(
cleanup_in_class=True,
**cls.data_sn_with_ldap_ss)
resp, body = cls.shares_client.add_sec_service_to_share_network(
cls.sn_with_ldap_ss["id"],
cls.ss_ldap["id"])
cls.data_sn_with_kerberos_ss = {
'name': 'sn_with_kerberos_ss',
'neutron_net_id': '3333',
'neutron_subnet_id': '4444',
'created_at': '2003-03-03',
'updated_at': None,
'neutron_net_id': 'test net id',
'neutron_subnet_id': 'test subnet id',
'network_type': 'local',
'segmentation_id': 2000,
'cidr': '10.0.0.0/13',
'ip_version': 6,
'description': 'fake description',
}
resp, cls.ss_kerberos = cls.create_security_service(
ss_type='kerberos',
**cls.data_sn_with_ldap_ss)
__, cls.sn_with_kerberos_ss = cls.create_share_network(
cleanup_in_class=True,
**cls.data_sn_with_kerberos_ss)
resp, body = cls.shares_client.add_sec_service_to_share_network(
cls.sn_with_kerberos_ss["id"],
cls.ss_kerberos["id"])
@test.attr(type=["gate", "smoke", ])
def test_create_delete_share_network(self):
@ -42,15 +158,20 @@ class ShareNetworksTest(base.BaseSharesTest):
@test.attr(type=["gate", "smoke", ])
def test_get_share_network(self):
resp, get = self.shares_client.get_share_network(self.sn["id"])
resp, get = self.shares_client.get_share_network(
self.sn_with_ldap_ss["id"])
self.assertIn(int(resp["status"]), test.HTTP_SUCCESS)
self.assertDictContainsSubset(self.data, get)
self.assertEqual('2002-02-02T00:00:00.000000', get['created_at'])
data = self.data_sn_with_ldap_ss.copy()
del data['created_at']
self.assertDictContainsSubset(data, get)
@test.attr(type=["gate", "smoke", ])
def test_update_share_network(self):
update_data = self.generate_share_network_data()
resp, updated = self.shares_client.update_share_network(self.sn["id"],
**update_data)
resp, updated = self.shares_client.update_share_network(
self.sn_with_ldap_ss["id"],
**update_data)
self.assertIn(int(resp["status"]), test.HTTP_SUCCESS)
self.assertDictContainsSubset(update_data, updated)
@ -67,31 +188,6 @@ class ShareNetworksTest(base.BaseSharesTest):
self.assertIn(int(resp["status"]), test.HTTP_SUCCESS)
self.assertDictContainsSubset(update_dict, updated)
@test.attr(type=["gate", "smoke", ])
def test_list_share_networks(self):
resp, listed = self.shares_client.list_share_networks()
self.assertIn(int(resp["status"]), test.HTTP_SUCCESS)
any(self.sn["id"] in sn["id"] for sn in listed)
# verify keys
keys = ["name", "id", "status"]
[self.assertIn(key, sn.keys()) for sn in listed for key in keys]
@test.attr(type=["gate", "smoke", ])
def test_list_share_networks_with_detail(self):
resp, listed = self.shares_client.list_share_networks_with_detail()
self.assertIn(int(resp["status"]), test.HTTP_SUCCESS)
any(self.sn["id"] in sn["id"] for sn in listed)
# verify keys
keys = [
"name", "id", "status", "description", "network_type",
"project_id", "cidr", "ip_version",
"neutron_net_id", "neutron_subnet_id",
"created_at", "updated_at", "segmentation_id",
]
[self.assertIn(key, sn.keys()) for sn in listed for key in keys]
@test.attr(type=["gate", "smoke", ])
def test_recreate_share_network(self):
# generate data for share network

26
contrib/tempest/tempest/api/share/test_share_networks_negative.py
View File

@ -76,3 +76,29 @@ class ShareNetworksNegativeTest(base.BaseSharesTest):
self.assertRaises(exceptions.NotFound,
self.shares_client.get_security_service,
sn["id"])
@test.attr(type=["gate", "smoke", "negative"])
def test_try_list_share_networks_all_tenants(self):
self.assertRaises(exceptions.Unauthorized,
self.shares_client.list_share_networks_with_detail,
params={'all_tenants': 1})
@test.attr(type=["gate", "smoke", "negative"])
def test_try_list_share_networks_project_id(self):
self.assertRaises(exceptions.Unauthorized,
self.shares_client.list_share_networks_with_detail,
params={'project_id': 'some_project'})
@test.attr(type=["gate", "smoke", "negative"])
def test_try_list_share_networks_wrong_created_since_value(self):
self.assertRaises(
exceptions.BadRequest,
self.shares_client.list_share_networks_with_detail,
params={'created_since': '2014-10-23T08:31:58.000000'})
@test.attr(type=["gate", "smoke", "negative"])
def test_try_list_share_networks_wrong_created_before_value(self):
self.assertRaises(
exceptions.BadRequest,
self.shares_client.list_share_networks_with_detail,
params={'created_before': '2014-10-23T08:31:58.000000'})

3
etc/manila/policy.json
View File

@ -59,5 +59,6 @@
"share_network:detail": [["rule:default"]],
"share_network:show": [["rule:default"]],
"share_network:add_security_service": [["rule:default"]],
"share_network:remove_security_service": [["rule:default"]]
"share_network:remove_security_service": [["rule:default"]],
"share_network:get_all_share_networks": [["rule:admin_api"]]
}

60
manila/api/v1/share_networks.py
View File

@ -16,10 +16,12 @@
"""The shares api."""
from oslo.db import exception as db_exception
from oslo.utils import timeutils
import six
import webob
from webob import exc
from manila.api import common
from manila.api.openstack import wsgi
from manila.api.views import share_networks as share_networks_views
from manila.api import xmlutil
@ -126,32 +128,82 @@ class ShareNetworkController(wsgi.Controller):
def _get_share_networks(self, req, is_detail=True):
"""Returns a list of share networks."""
context = req.environ['manila.context']
policy.check_policy(context, RESOURCE_NAME, 'index')
search_opts = {}
search_opts.update(req.GET)
if search_opts.pop('all_tenants', None):
if ('all_tenants' in search_opts or
('project_id' in search_opts and
search_opts['project_id'] != context.project_id)):
policy.check_policy(context, RESOURCE_NAME,
'get_all_share_networks')
if 'security_service_id' in search_opts:
networks = db_api.share_network_get_all_by_security_service(
context, search_opts['security_service_id'])
elif ('project_id' in search_opts and
search_opts['project_id'] != context.project_id):
networks = db_api.share_network_get_all_by_project(
context, search_opts['project_id'])
elif 'all_tenants' in search_opts:
networks = db_api.share_network_get_all(context)
else:
networks = db_api.share_network_get_all_by_project(
context,
context.project_id)
date_parsing_error_msg = '''%s is not in yyyy-mm-dd format.'''
if 'created_since' in search_opts:
try:
created_since = timeutils.parse_strtime(
search_opts['created_since'],
fmt="%Y-%m-%d")
except ValueError:
msg = date_parsing_error_msg % search_opts['created_since']
raise exc.HTTPBadRequest(explanation=msg)
networks = [network for network in networks
if network['created_at'] >= created_since]
if 'created_before' in search_opts:
try:
created_before = timeutils.parse_strtime(
search_opts['created_before'],
fmt="%Y-%m-%d")
except ValueError:
msg = date_parsing_error_msg % search_opts['created_before']
raise exc.HTTPBadRequest(explanation=msg)
networks = [network for network in networks
if network['created_at'] <= created_before]
opts_to_remove = [
'all_tenants',
'created_since',
'created_before',
'limit',
'offset',
'security_service_id',
]
for opt in opts_to_remove:
search_opts.pop(opt, None)
if search_opts:
for key, value in six.iteritems(search_opts):
if key in ['ip_version', 'segmentation_id']:
value = int(value)
networks = [network for network in networks
if network[key] == value]
return self._view_builder.build_share_networks(networks, is_detail)
limited_list = common.limited(networks, req)
return self._view_builder.build_share_networks(limited_list, is_detail)
@wsgi.serializers(xml=ShareNetworksTemplate)
def index(self, req):
"""Returns a summary list of share networks."""
policy.check_policy(req.environ['manila.context'], RESOURCE_NAME,
'index')
return self._get_share_networks(req, is_detail=False)
@wsgi.serializers(xml=ShareNetworksTemplate)
def detail(self, req):
"""Returns a detailed list of share networks."""
policy.check_policy(req.environ['manila.context'], RESOURCE_NAME,
'detail')
return self._get_share_networks(req)
@wsgi.serializers(xml=ShareNetworkTemplate)

1
manila/api/views/share_networks.py
View File

@ -35,7 +35,6 @@ class ViewBuilder(common.ViewBuilder):
sn = {
'id': share_network.get('id'),
'name': share_network.get('name'),
'status': share_network.get('status'),
}
if is_detail:
sn.update({

4
manila/db/api.py
View File

@ -511,10 +511,10 @@ def share_network_get_all_by_project(context, project_id):
return IMPL.share_network_get_all_by_project(context, project_id)
def share_network_get_all_by_security_service(context, share_network_id):
def share_network_get_all_by_security_service(context, security_service_id):
"""Get all share network DB records for the given project."""
return IMPL.share_network_get_all_by_security_service(
context, share_network_id)
context, security_service_id)
def share_network_add_security_service(context, id, security_service_id):

4
manila/db/sqlalchemy/api.py
View File

@ -1757,13 +1757,13 @@ def share_network_get_all_by_project(context, project_id, user_id=None,
@require_context
def share_network_get_all_by_security_service(context, share_network_id):
def share_network_get_all_by_security_service(context, security_service_id):
session = get_session()
return model_query(context, models.ShareNetwork, session=session).\
join(models.ShareNetworkSecurityServiceAssociation,
models.ShareNetwork.id ==
models.ShareNetworkSecurityServiceAssociation.share_network_id).\
filter_by(security_service_id=share_network_id, deleted=False).\
filter_by(security_service_id=security_service_id, deleted=False).\
options(joinedload('share_servers')).all()

150
manila/tests/api/v1/test_share_networks.py
View File

@ -15,10 +15,11 @@
import mock
from oslo.db import exception as db_exception
from oslo.utils import timeutils
from six.moves.urllib import parse
from webob import exc as webob_exc
from manila.api.v1 import share_networks
from manila.common import constants
from manila.db import api as db_api
from manila import exception
from manila import quota
@ -29,7 +30,7 @@ from manila.tests.api import fakes
fake_share_network = {
'id': 'fake network id',
'project_id': 'fake project',
'created_at': None,
'created_at': timeutils.parse_strtime('2002-02-02', fmt="%Y-%m-%d"),
'updated_at': None,
'neutron_net_id': 'fake net id',
'neutron_subnet_id': 'fake subnet id',
@ -39,14 +40,35 @@ fake_share_network = {
'ip_version': 4,
'name': 'fake name',
'description': 'fake description',
'status': constants.STATUS_INACTIVE,
'share_servers': [],
'security_services': []
}
fake_share_network_shortened = {
'id': 'fake network id',
'name': 'fake name',
'status': constants.STATUS_INACTIVE,
}
fake_share_network_with_ss = {
'id': 'sn-id',
'project_id': 'fake',
'created_at': timeutils.parse_strtime('2001-01-01', fmt="%Y-%m-%d"),
'updated_at': None,
'neutron_net_id': '1111',
'neutron_subnet_id': '2222',
'network_type': 'local',
'segmentation_id': 2000,
'cidr': '8.0.0.0/12',
'ip_version': 6,
'name': 'test-sn',
'description': 'fake description',
'share_servers': [],
'security_services': [{'id': 'fake-ss-id'}]
}
fake_sn_with_ss_shortened = {
'id': 'sn-id',
'name': 'test-sn',
}
QUOTAS = quota.QUOTAS
@ -64,7 +86,6 @@ class ShareNetworkAPITest(test.TestCase):
def _check_share_network_view_shortened(self, view, share_nw):
self.assertEqual(view['id'], share_nw['id'])
self.assertEqual(view['name'], share_nw['name'])
self.assertEqual(view['status'], share_nw['status'])
def _check_share_network_view(self, view, share_nw):
self.assertEqual(view['id'], share_nw['id'])
@ -82,10 +103,9 @@ class ShareNetworkAPITest(test.TestCase):
self.assertEqual(view['ip_version'], share_nw['ip_version'])
self.assertEqual(view['name'], share_nw['name'])
self.assertEqual(view['description'], share_nw['description'])
self.assertEqual(view['status'], share_nw['status'])
self.assertEqual(view['created_at'], None)
self.assertEqual(view['updated_at'], None)
self.assertEqual(view['created_at'], share_nw['created_at'])
self.assertEqual(view['updated_at'], share_nw['updated_at'])
self.assertFalse('shares' in view)
self.assertFalse('network_allocations' in view)
self.assertFalse('security_services' in view)
@ -218,6 +238,120 @@ class ShareNetworkAPITest(test.TestCase):
result[share_networks.RESOURCES_NAME][0],
fake_share_network)
@mock.patch.object(db_api, 'share_network_get_all_by_security_service',
mock.Mock())
def test_index_filter_by_security_service(self):
db_api.share_network_get_all_by_security_service.return_value = [
fake_share_network_with_ss]
req = fakes.HTTPRequest.blank(
'/share_networks?security_service_id=fake-ss-id')
result = self.controller.index(req)
db_api.share_network_get_all_by_security_service.\
assert_called_once_with(req.environ['manila.context'],
'fake-ss-id')
self.assertEqual(1, len(result[share_networks.RESOURCES_NAME]))
self._check_share_network_view_shortened(
result[share_networks.RESOURCES_NAME][0],
fake_sn_with_ss_shortened)
@mock.patch.object(db_api, 'share_network_get_all', mock.Mock())
def test_index_all_tenants_non_admin_context(self):
req = fakes.HTTPRequest.blank(
'/share_networks?all_tenants=1')
self.assertRaises(exception.PolicyNotAuthorized, self.controller.index,
req)
self.assertFalse(db_api.share_network_get_all.called)
@mock.patch.object(db_api, 'share_network_get_all', mock.Mock())
def test_index_all_tenants_admin_context(self):
db_api.share_network_get_all.return_value = [fake_share_network]
req = fakes.HTTPRequest.blank(
'/share_networks?all_tenants=1',
use_admin_context=True)
result = self.controller.index(req)
db_api.share_network_get_all.assert_called_once_with(
req.environ['manila.context'])
self.assertEqual(1, len(result[share_networks.RESOURCES_NAME]))
self._check_share_network_view_shortened(
result[share_networks.RESOURCES_NAME][0],
fake_share_network_shortened)
@mock.patch.object(db_api, 'share_network_get_all_by_project', mock.Mock())
def test_index_filter_by_project_id_non_admin_context(self):
req = fakes.HTTPRequest.blank(
'/share_networks?project_id=fake project')
self.assertRaises(exception.PolicyNotAuthorized, self.controller.index,
req)
self.assertFalse(db_api.share_network_get_all_by_project.called)
@mock.patch.object(db_api, 'share_network_get_all_by_project', mock.Mock())
def test_index_filter_by_project_id_admin_context(self):
db_api.share_network_get_all_by_project.return_value = [
fake_share_network,
fake_share_network_with_ss,
]
req = fakes.HTTPRequest.blank(
'/share_networks?project_id=fake',
use_admin_context=True)
result = self.controller.index(req)
db_api.share_network_get_all_by_project.assert_called_once_with(
req.environ['manila.context'], 'fake')
self.assertEqual(1, len(result[share_networks.RESOURCES_NAME]))
self._check_share_network_view_shortened(
result[share_networks.RESOURCES_NAME][0],
fake_sn_with_ss_shortened)
@mock.patch.object(db_api, 'share_network_get_all_by_security_service',
mock.Mock())
def test_index_filter_by_ss_and_project_id_admin_context(self):
db_api.share_network_get_all_by_security_service.return_value = [
fake_share_network,
fake_share_network_with_ss,
]
req = fakes.HTTPRequest.blank(
'/share_networks?security_service_id=fake-ss-id&project_id=fake',
use_admin_context=True)
result = self.controller.index(req)
db_api.share_network_get_all_by_security_service.\
assert_called_once_with(req.environ['manila.context'],
'fake-ss-id')
self.assertEqual(1, len(result[share_networks.RESOURCES_NAME]))
self._check_share_network_view_shortened(
result[share_networks.RESOURCES_NAME][0],
fake_sn_with_ss_shortened)
@mock.patch.object(db_api, 'share_network_get_all_by_project',
mock.Mock())
def test_index_all_filter_opts(self):
valid_filter_opts = {
'created_before': '2001-02-02',
'created_since': '1999-01-01',
'neutron_net_id': '1111',
'neutron_subnet_id': '2222',
'network_type': 'local',
'segmentation_id': 2000,
'cidr': '8.0.0.0/12',
'ip_version': 6,
'name': 'test-sn'
}
db_api.share_network_get_all_by_project.return_value = [
fake_share_network,
fake_share_network_with_ss]
query_string = '/share-networks?' + parse.urlencode(sorted(
[(k, v) for (k, v) in list(valid_filter_opts.items())]))
for use_admin_context in [True, False]:
req = fakes.HTTPRequest.blank(query_string,
use_admin_context=use_admin_context)
result = self.controller.index(req)
db_api.share_network_get_all_by_project.assert_called_with(
req.environ['manila.context'],
'fake')
self.assertEqual(1, len(result[share_networks.RESOURCES_NAME]))
self._check_share_network_view_shortened(
result[share_networks.RESOURCES_NAME][0],
fake_sn_with_ss_shortened)
@mock.patch.object(db_api, 'share_network_get', mock.Mock())
def test_update_nominal(self):
share_nw = 'fake network id'

2
manila/tests/policy.json
View File

@ -16,9 +16,11 @@
"share_network:create": [],
"share_network:index": [],
"share_network:detail": [],
"share_network:show": [],
"share_network:update": [],
"share_network:delete": [],
"share_network:get_all_share_networks": [["rule:admin_api"]],
"share_server:index": [["rule:admin_api"]],
"share_server:show": [["rule:admin_api"]],