summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorZuul <zuul@review.opendev.org>2022-02-20 19:58:33 +0000
committerGerrit Code Review <review@openstack.org>2022-02-20 19:58:33 +0000
commitf165de3e4f913d4e4c7d7c7bcd81a8801294363f (patch)
tree35cb2aa4903ab30d0ecb54c21b3673e9bf85a4cc
parent5abb12ad926420b53f7304973b71a467a58741a3 (diff)
parent196de449b6f7be3707cfd08fd1099c4ed792fba0 (diff)
downloadhorizon-f165de3e4f913d4e4c7d7c7bcd81a8801294363f.tar.gz
Merge "Implement pagination in admin/proj network tab"
-rw-r--r--openstack_dashboard/api/neutron.py410
-rw-r--r--openstack_dashboard/dashboards/admin/networks/tests.py16
-rw-r--r--openstack_dashboard/dashboards/admin/networks/views.py17
-rw-r--r--openstack_dashboard/dashboards/project/networks/tests.py47
-rw-r--r--openstack_dashboard/dashboards/project/networks/views.py23
-rw-r--r--openstack_dashboard/dashboards/project/routers/tests.py8
-rw-r--r--openstack_dashboard/test/integration_tests/pages/admin/network/networkspage.py22
-rw-r--r--openstack_dashboard/test/integration_tests/pages/project/network/networkspage.py14
-rw-r--r--openstack_dashboard/test/integration_tests/tests/test_networks.py100
-rw-r--r--openstack_dashboard/test/test_data/neutron_data.py247
-rw-r--r--openstack_dashboard/test/unit/api/test_neutron.py795
-rw-r--r--openstack_dashboard/test/unit/usage/test_quotas.py2
-rw-r--r--releasenotes/notes/add-networks-pagination-4c05d784998fafb2.yaml5
13 files changed, 1572 insertions, 134 deletions
diff --git a/openstack_dashboard/api/neutron.py b/openstack_dashboard/api/neutron.py
index 0f06a1b22..d1a55f767 100644
--- a/openstack_dashboard/api/neutron.py
+++ b/openstack_dashboard/api/neutron.py
@@ -1029,9 +1029,32 @@ def trunk_update(request, trunk_id, old_trunk, new_trunk):
@profiler.trace
-def network_list(request, **params):
+def network_list_paged(request, page_data, **params):
+ page_data, marker_net = _configure_pagination(request, params, page_data)
+ query_kwargs = {
+ 'request': request,
+ 'page_data': page_data,
+ 'params': params,
+ }
+ return _perform_query(_network_list_paged, query_kwargs, marker_net)
+
+
+def _network_list_paged(request, page_data, params):
+ nets = network_list(
+ request, single_page=page_data['single_page'], **params)
+ return update_pagination(nets, page_data)
+
+
+@profiler.trace
+def network_list(request, single_page=False, **params):
LOG.debug("network_list(): params=%s", params)
- networks = neutronclient(request).list_networks(**params).get('networks')
+ if single_page is True:
+ params['retrieve_all'] = False
+ result = neutronclient(request).list_networks(**params)
+ if single_page is True:
+ result = result.next()
+ networks = result.get('networks')
+
# Get subnet list to expand subnet info in network list.
subnets = subnet_list(request)
subnet_dict = dict((s['id'], s) for s in subnets)
@@ -1070,54 +1093,366 @@ def _is_auto_allocated_network_supported(request):
return nova_auto_supported
+# TODO(ganso): consolidate this function with cinder's and nova's
+@profiler.trace
+def update_pagination(entities, page_data):
+
+ has_more_data, has_prev_data = False, False
+
+ # single_page=True is actually to have pagination enabled
+ if page_data.get('single_page') is not True:
+ return entities, has_more_data, has_prev_data
+
+ if len(entities) > page_data['page_size']:
+ has_more_data = True
+ entities.pop()
+ if page_data.get('marker_id') is not None:
+ has_prev_data = True
+
+ # first page condition when reached via prev back
+ elif (page_data.get('sort_dir') == 'desc' and
+ page_data.get('marker_id') is not None):
+ has_more_data = True
+
+ # last page condition
+ elif page_data.get('marker_id') is not None:
+ has_prev_data = True
+
+ # reverse to maintain same order when going backwards
+ if page_data.get('sort_dir') == 'desc':
+ entities.reverse()
+
+ return entities, has_more_data, has_prev_data
+
+
+def _add_to_nets_and_return(
+ nets, obtained_nets, page_data, filter_tenant_id=None):
+ # remove project non-shared external nets that should
+ # be retrieved by project query
+ if filter_tenant_id:
+ obtained_nets = [net for net in obtained_nets
+ if net['tenant_id'] != filter_tenant_id]
+
+ if (page_data['single_page'] is True and
+ len(obtained_nets) + len(nets) > page_data['limit']):
+ # we need to trim results if we already surpassed the limit
+ # we use limit so we can call update_pagination
+ cut = page_data['limit'] - (len(obtained_nets) + len(nets))
+ nets += obtained_nets[0:cut]
+ return True
+ nets += obtained_nets
+ # we don't need to perform more queries if we already have enough nets
+ if page_data['single_page'] is True and len(nets) == page_data['limit']:
+ return True
+ return False
+
+
+def _query_external_nets(request, include_external, page_data, **params):
+
+ # If the external filter is set to False we don't need to perform this
+ # query
+ # If the shared filter is set to True we don't need to perform this
+ # query (already retrieved)
+ # We are either paginating external nets or not pending more data
+ if (page_data['filter_external'] is not False and include_external and
+ page_data['filter_shared'] is not True and
+ page_data.get('marker_type') in (None, 'ext')):
+
+ # Grab only all external non-shared networks
+ params['router:external'] = True
+ params['shared'] = False
+
+ return _perform_net_query(request, {}, page_data, 'ext', **params)
+
+ return []
+
+
+def _query_shared_nets(request, page_data, **params):
+
+ # If the shared filter is set to False we don't need to perform this query
+ # We are either paginating shared nets or not pending more data
+ if (page_data['filter_shared'] is not False and
+ page_data.get('marker_type') in (None, 'shr')):
+
+ if page_data['filter_external'] is None:
+ params.pop('router:external', None)
+ else:
+ params['router:external'] = page_data['filter_external']
+
+ # Grab only all shared networks
+ # May include shared external nets based on external filter
+ params['shared'] = True
+
+ return _perform_net_query(request, {}, page_data, 'shr', **params)
+
+ return []
+
+
+def _query_project_nets(request, tenant_id, page_data, **params):
+
+ # We don't need to run this query if shared filter is True, as the networks
+ # will be retrieved by another query
+ # We are either paginating project nets or not pending more data
+ if (page_data['filter_shared'] is not True and
+ page_data.get('marker_type') in (None, 'proj')):
+
+ # Grab only non-shared project networks
+ # May include non-shared project external nets based on external filter
+ if page_data['filter_external'] is None:
+ params.pop('router:external', None)
+ else:
+ params['router:external'] = page_data['filter_external']
+ params['shared'] = False
+
+ return _perform_net_query(
+ request, {'tenant_id': tenant_id}, page_data, 'proj', **params)
+
+ return []
+
+
+def _perform_net_query(
+ request, extra_param, page_data, query_marker_type, **params):
+ copy_req_params = copy.deepcopy(params)
+ copy_req_params.update(extra_param)
+ if page_data.get('marker_type') == query_marker_type:
+ copy_req_params['marker'] = page_data['marker_id']
+ # We clear the marker type to allow for other queries if
+ # this one does not fill up the page
+ page_data['marker_type'] = None
+ return network_list(
+ request, single_page=page_data['single_page'], **copy_req_params)
+
+
+def _query_nets_for_tenant(request, include_external, tenant_id, page_data,
+ **params):
+
+ # Save variables
+ page_data['filter_external'] = params.get('router:external')
+ page_data['filter_shared'] = params.get('shared')
+
+ nets = []
+
+ # inverted direction (for prev page)
+ if (page_data.get('single_page') is True and
+ page_data.get('sort_dir') == 'desc'):
+
+ ext_nets = _query_external_nets(
+ request, include_external, page_data, **params)
+ if _add_to_nets_and_return(
+ nets, ext_nets, page_data, filter_tenant_id=tenant_id):
+ return update_pagination(nets, page_data)
+
+ proj_nets = _query_project_nets(
+ request, tenant_id, page_data, **params)
+ if _add_to_nets_and_return(nets, proj_nets, page_data):
+ return update_pagination(nets, page_data)
+
+ shr_nets = _query_shared_nets(
+ request, page_data, **params)
+ if _add_to_nets_and_return(nets, shr_nets, page_data):
+ return update_pagination(nets, page_data)
+
+ # normal direction (for next page)
+ else:
+ shr_nets = _query_shared_nets(
+ request, page_data, **params)
+ if _add_to_nets_and_return(nets, shr_nets, page_data):
+ return update_pagination(nets, page_data)
+
+ proj_nets = _query_project_nets(
+ request, tenant_id, page_data, **params)
+ if _add_to_nets_and_return(nets, proj_nets, page_data):
+ return update_pagination(nets, page_data)
+
+ ext_nets = _query_external_nets(
+ request, include_external, page_data, **params)
+ if _add_to_nets_and_return(
+ nets, ext_nets, page_data, filter_tenant_id=tenant_id):
+ return update_pagination(nets, page_data)
+
+ return update_pagination(nets, page_data)
+
+
+def _configure_marker_type(marker_net, tenant_id=None):
+ if marker_net:
+ if marker_net['shared'] is True:
+ return 'shr'
+ if (marker_net['router:external'] is True and
+ marker_net['tenant_id'] != tenant_id):
+ return 'ext'
+ return 'proj'
+ return None
+
+
+def _reverse_page_order(sort_dir):
+ if sort_dir == 'asc':
+ return 'desc'
+ return 'asc'
+
+
+def _configure_pagination(request, params, page_data=None, tenant_id=None):
+
+ marker_net = None
+ # "single_page" is a neutron API parameter to disable automatic
+ # pagination done by the API. If it is False, it returns all the
+ # results. If page_data param is not present, the method is being
+ # called by someone that does not want/expect pagination.
+ if page_data is None:
+ page_data = {'single_page': False}
+ else:
+ page_data['single_page'] = True
+ if page_data['marker_id']:
+ # this next request is inefficient, but the alternative is for
+ # the UI to send the extra parameters in the request,
+ # maybe a future optimization
+ marker_net = network_get(request, page_data['marker_id'])
+ page_data['marker_type'] = _configure_marker_type(
+ marker_net, tenant_id=tenant_id)
+ else:
+ page_data['marker_type'] = None
+
+ # we query one more than we are actually displaying due to
+ # consistent pagination hack logic used in other services
+ page_data['page_size'] = setting_utils.get_page_size(request)
+ page_data['limit'] = page_data['page_size'] + 1
+ params['limit'] = page_data['limit']
+
+ # Neutron API sort direction is inverted compared to other services
+ page_data['sort_dir'] = page_data.get('sort_dir', "desc")
+ page_data['sort_dir'] = _reverse_page_order(page_data['sort_dir'])
+
+ # params are included in the request to the neutron API
+ params['sort_dir'] = page_data['sort_dir']
+ params['sort_key'] = 'id'
+
+ return page_data, marker_net
+
+
+def _perform_query(
+ query_func, query_kwargs, marker_net, include_pre_auto_allocate=False):
+ networks, has_more_data, has_prev_data = query_func(**query_kwargs)
+
+ # Hack for auto allocated network
+ if include_pre_auto_allocate and not networks:
+ if _is_auto_allocated_network_supported(query_kwargs['request']):
+ networks.append(PreAutoAllocateNetwork(query_kwargs['request']))
+
+ # no pagination case, single_page=True means pagination is enabled
+ if query_kwargs['page_data'].get('single_page') is not True:
+ return networks
+
+ # handle case of full page deletes
+ deleted = query_kwargs['request'].session.pop('network_deleted', None)
+ if deleted and marker_net:
+
+ # contents of last page deleted, invert order, load previous page
+ # based on marker (which ends up not included), remove head and add
+ # marker at the end. Since it is the last page, also force
+ # has_more_data to False because the marker item would always be
+ # the "more_data" of the request.
+ # we do this only if there are no elements to be displayed
+ if ((networks is None or len(networks) == 0) and
+ has_prev_data and not has_more_data and
+ query_kwargs['page_data']['sort_dir'] == 'asc'):
+ # admin section params
+ if 'params' in query_kwargs:
+ query_kwargs['params']['sort_dir'] = 'desc'
+ else:
+ query_kwargs['page_data']['marker_type'] = (
+ _configure_marker_type(marker_net,
+ query_kwargs.get('tenant_id')))
+ query_kwargs['sort_dir'] = 'desc'
+ query_kwargs['page_data']['sort_dir'] = 'desc'
+ networks, has_more_data, has_prev_data = (
+ query_func(**query_kwargs))
+ if networks:
+ if has_prev_data:
+ # if we are back in the first page, we don't remove head
+ networks.pop(0)
+ networks.append(marker_net)
+ has_more_data = False
+
+ # contents of first page deleted (loaded by prev), invert order
+ # and remove marker as if the section was loaded for the first time
+ # we do this regardless of number of elements in the first page
+ elif (has_more_data and not has_prev_data and
+ query_kwargs['page_data']['sort_dir'] == 'desc'):
+ query_kwargs['page_data']['sort_dir'] = 'asc'
+ query_kwargs['page_data']['marker_id'] = None
+ query_kwargs['page_data']['marker_type'] = None
+ # admin section params
+ if 'params' in query_kwargs:
+ if 'marker' in query_kwargs['params']:
+ del query_kwargs['params']['marker']
+ query_kwargs['params']['sort_dir'] = 'asc'
+ else:
+ query_kwargs['sort_dir'] = 'asc'
+ networks, has_more_data, has_prev_data = (
+ query_func(**query_kwargs))
+
+ return networks, has_more_data, has_prev_data
+
+
@profiler.trace
def network_list_for_tenant(request, tenant_id, include_external=False,
- include_pre_auto_allocate=False,
+ include_pre_auto_allocate=False, page_data=None,
**params):
"""Return a network list available for the tenant.
The list contains networks owned by the tenant and public networks.
If requested_networks specified, it searches requested_networks only.
- """
- LOG.debug("network_list_for_tenant(): tenant_id=%(tenant_id)s, "
- "params=%(params)s", {'tenant_id': tenant_id, 'params': params})
- networks = []
- shared = params.get('shared')
- if shared is not None:
- del params['shared']
+ page_data parameter format:
- if shared in (None, False):
- # If a user has admin role, network list returned by Neutron API
- # contains networks that do not belong to that tenant.
- # So we need to specify tenant_id when calling network_list().
- networks += network_list(request, tenant_id=tenant_id,
- shared=False, **params)
+ page_data = {
+ 'marker_id': '<id>',
+ 'sort_dir': '<desc(next)|asc(prev)>'
+ }
- if shared in (None, True):
- # In the current Neutron API, there is no way to retrieve
- # both owner networks and public networks in a single API call.
- networks += network_list(request, shared=True, **params)
+ """
- # Hack for auto allocated network
- if include_pre_auto_allocate and not networks:
- if _is_auto_allocated_network_supported(request):
- networks.append(PreAutoAllocateNetwork(request))
+ # Pagination is implemented consistently with nova and cinder views,
+ # which means it is a bit hacky:
+ # - it requests X units but displays X-1 units
+ # - it ignores the marker metadata from the API response and uses its own
+ # Here we have extra hacks on top of that, because we have to merge the
+ # results of 3 different queries, and decide which one of them we are
+ # actually paginating.
+ # The 3 queries consist of:
+ # 1. Shared=True networks
+ # 2. Project non-shared networks
+ # 3. External non-shared non-project networks
+ # The main reason behind that order is to maintain the current behavior
+ # for how external networks are retrieved and displayed.
+ # The include_external assumption of whether external networks should be
+ # displayed is "overridden" whenever the external network is shared or is
+ # the tenant's. Therefore it refers to only non-shared non-tenant external
+ # networks.
+ # To accomplish pagination, we check the type of network the provided
+ # marker is, to determine which query we have last run and whether we
+ # need to paginate it.
- params['router:external'] = params.get('router:external', True)
- if params['router:external'] and include_external:
- if shared is not None:
- params['shared'] = shared
- fetched_net_ids = [n.id for n in networks]
- # Retrieves external networks when router:external is not specified
- # in (filtering) params or router:external=True filter is specified.
- # When router:external=False is specified there is no need to query
- # networking API because apparently nothing will match the filter.
- ext_nets = network_list(request, **params)
- networks += [n for n in ext_nets if
- n.id not in fetched_net_ids]
+ LOG.debug("network_list_for_tenant(): tenant_id=%(tenant_id)s, "
+ "params=%(params)s, page_data=%(page_data)s", {
+ 'tenant_id': tenant_id,
+ 'params': params,
+ 'page_data': page_data,
+ })
+
+ page_data, marker_net = _configure_pagination(
+ request, params, page_data, tenant_id=tenant_id)
+
+ query_kwargs = {
+ 'request': request,
+ 'include_external': include_external,
+ 'tenant_id': tenant_id,
+ 'page_data': page_data,
+ **params,
+ }
- return networks
+ return _perform_query(
+ _query_nets_for_tenant, query_kwargs, marker_net,
+ include_pre_auto_allocate)
@profiler.trace
@@ -1178,6 +1513,7 @@ def network_update(request, network_id, **kwargs):
def network_delete(request, network_id):
LOG.debug("network_delete(): netid=%s", network_id)
neutronclient(request).delete_network(network_id)
+ request.session['network_deleted'] = network_id
@profiler.trace
diff --git a/openstack_dashboard/dashboards/admin/networks/tests.py b/openstack_dashboard/dashboards/admin/networks/tests.py
index 3c34ad8fa..f2ac0b2ce 100644
--- a/openstack_dashboard/dashboards/admin/networks/tests.py
+++ b/openstack_dashboard/dashboards/admin/networks/tests.py
@@ -68,7 +68,9 @@ class NetworkTests(test.BaseAdminViewTests):
networks = res.context['networks_table'].data
self.assertCountEqual(networks, self.networks.list())
- self.mock_network_list.assert_called_once_with(test.IsHttpRequest())
+ self.mock_network_list.assert_called_once_with(
+ test.IsHttpRequest(), single_page=True,
+ limit=21, sort_dir='asc', sort_key='id')
self.mock_tenant_list.assert_called_once_with(test.IsHttpRequest())
self._check_is_extension_supported(
{'network_availability_zone': 1,
@@ -99,7 +101,9 @@ class NetworkTests(test.BaseAdminViewTests):
self.assertEqual(len(res.context['networks_table'].data), 0)
self.assertMessageCount(res, error=1)
- self.mock_network_list.assert_called_once_with(test.IsHttpRequest())
+ self.mock_network_list.assert_called_once_with(
+ test.IsHttpRequest(), single_page=True,
+ limit=21, sort_dir='asc', sort_key='id')
self._check_is_extension_supported(
{'network_availability_zone': 1,
'dhcp_agent_scheduler': 1})
@@ -964,7 +968,9 @@ class NetworkTests(test.BaseAdminViewTests):
{'network_availability_zone': 1,
'dhcp_agent_scheduler': 2})
self.mock_tenant_list.assert_called_once_with(test.IsHttpRequest())
- self.mock_network_list.assert_called_once_with(test.IsHttpRequest())
+ self.mock_network_list.assert_called_once_with(
+ test.IsHttpRequest(), single_page=True,
+ limit=21, sort_dir='asc', sort_key='id')
self.mock_network_delete.assert_called_once_with(test.IsHttpRequest(),
network.id)
@@ -997,7 +1003,9 @@ class NetworkTests(test.BaseAdminViewTests):
{'network_availability_zone': 1,
'dhcp_agent_scheduler': 2})
self.mock_tenant_list.assert_called_once_with(test.IsHttpRequest())
- self.mock_network_list.assert_called_once_with(test.IsHttpRequest())
+ self.mock_network_list.assert_called_once_with(
+ test.IsHttpRequest(), single_page=True,
+ limit=21, sort_dir='asc', sort_key='id')
self.mock_network_delete.assert_called_once_with(test.IsHttpRequest(),
network.id)
diff --git a/openstack_dashboard/dashboards/admin/networks/views.py b/openstack_dashboard/dashboards/admin/networks/views.py
index df0d1d67e..cba3bc601 100644
--- a/openstack_dashboard/dashboards/admin/networks/views.py
+++ b/openstack_dashboard/dashboards/admin/networks/views.py
@@ -41,7 +41,7 @@ from openstack_dashboard.dashboards.admin.networks \
from openstack_dashboard.dashboards.admin.networks import workflows
-class IndexView(tables.DataTableView):
+class IndexView(tables.PagedTableMixin, tables.DataTableView):
table_class = networks_tables.NetworksTable
page_title = _("Networks")
FILTERS_MAPPING = {'shared': {_("yes"): True, _("no"): False},
@@ -84,8 +84,18 @@ class IndexView(tables.DataTableView):
def get_data(self):
try:
+ marker, sort_dir = self._get_marker()
+
+ page_data = {
+ 'marker_id': marker,
+ 'sort_dir': sort_dir
+ }
+
search_opts = self.get_filters(filters_map=self.FILTERS_MAPPING)
+ if marker:
+ search_opts['marker'] = marker
+
# If the tenant filter selected and the tenant does not exist.
# We do not need to retrieve the list from neutron,just return
# an empty list.
@@ -102,8 +112,11 @@ class IndexView(tables.DataTableView):
return []
self._needs_filter_first = False
- networks = api.neutron.network_list(self.request, **search_opts)
+ networks, self._has_more_data, self._has_prev_data = (
+ api.neutron.network_list_paged(
+ self.request, page_data, **search_opts))
except Exception:
+ self._has_more_data = self._has_prev_data = False
networks = []
msg = _('Network list can not be retrieved.')
exceptions.handle(self.request, msg)
diff --git a/openstack_dashboard/dashboards/project/networks/tests.py b/openstack_dashboard/dashboards/project/networks/tests.py
index dc53c9b99..77dc74902 100644
--- a/openstack_dashboard/dashboards/project/networks/tests.py
+++ b/openstack_dashboard/dashboards/project/networks/tests.py
@@ -103,19 +103,25 @@ class NetworkStubMixin(object):
all_networks = self.networks.list()
self.mock_network_list.side_effect = [
[network for network in all_networks
- if network['tenant_id'] == self.tenant.id],
+ if network.get('shared') is True],
[network for network in all_networks
- if network.get('shared')],
+ if network['tenant_id'] == self.tenant.id and
+ network.get('shared') is False],
[network for network in all_networks
- if network.get('router:external')],
+ if network.get('router:external') is True and
+ network.get('shared') is False],
]
def _check_net_list(self):
self.mock_network_list.assert_has_calls([
- mock.call(test.IsHttpRequest(), tenant_id=self.tenant.id,
- shared=False),
- mock.call(test.IsHttpRequest(), shared=True),
- mock.call(test.IsHttpRequest(), **{'router:external': True}),
+ mock.call(test.IsHttpRequest(), single_page=True, limit=21,
+ sort_dir='asc', sort_key='id', shared=True),
+ mock.call(test.IsHttpRequest(), single_page=True, limit=21,
+ sort_dir='asc', sort_key='id',
+ shared=False, tenant_id=self.tenant.id),
+ mock.call(test.IsHttpRequest(), single_page=True, limit=21,
+ sort_dir='asc', sort_key='id',
+ **{'router:external': True}, shared=False),
])
def _stub_is_extension_supported(self, features):
@@ -148,16 +154,19 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
res = self.client.get(INDEX_URL)
self.assertTemplateUsed(res, INDEX_TEMPLATE)
networks = res.context['networks_table'].data
- self.assertCountEqual(networks, self.networks.list())
self.mock_tenant_quota_usages.assert_has_calls([
mock.call(test.IsHttpRequest(), targets=('network', )),
mock.call(test.IsHttpRequest(), targets=('subnet', )),
])
- self.assertEqual(7, self.mock_tenant_quota_usages.call_count)
+ self.assertEqual(11, self.mock_tenant_quota_usages.call_count)
self.mock_is_extension_supported.assert_called_once_with(
test.IsHttpRequest(), 'network_availability_zone')
self._check_net_list()
+ self.assertCountEqual(networks, [net for net in self.networks.list()
+ if net['tenant_id'] == '1' or
+ net['router:external'] is True or
+ net['shared'] is True])
@test.create_mocks({api.neutron: ('network_list',
'is_extension_supported'),
@@ -175,8 +184,8 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
self.assertMessageCount(res, error=1)
self.mock_network_list.assert_called_once_with(
- test.IsHttpRequest(), tenant_id=self.tenant.id,
- shared=False)
+ test.IsHttpRequest(), single_page=True, limit=21, sort_dir='asc',
+ sort_key='id', shared=True)
self.assert_mock_multiple_calls_with_same_arguments(
self.mock_tenant_quota_usages, 2,
mock.call(test.IsHttpRequest(), targets=('network', )))
@@ -787,7 +796,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
def test_network_create_post_with_subnet_cidr_invalid_v6_range(
self, test_with_subnetpool=False):
network = self.networks.first()
- subnet_v6 = self.subnets.list()[4]
+ subnet_v6 = self.subnets.list()[9]
self._stub_is_extension_supported({'network_availability_zone': False,
'subnet_allocation': True})
@@ -1144,7 +1153,6 @@ class NetworkViewTests(test.TestCase, NetworkStubMixin):
self.assertTemplateUsed(res, INDEX_TEMPLATE)
networks = res.context['networks_table'].data
- self.assertCountEqual(networks, self.networks.list())
button = find_button_fn(res)
self.assertFalse('disabled' in button.classes,
@@ -1155,9 +1163,13 @@ class NetworkViewTests(test.TestCase, NetworkStubMixin):
mock.call(test.IsHttpRequest(), targets=('network', )),
mock.call(test.IsHttpRequest(), targets=('subnet', )),
])
- self.assertEqual(8, self.mock_tenant_quota_usages.call_count)
+ self.assertEqual(12, self.mock_tenant_quota_usages.call_count)
self.mock_is_extension_supported.assert_called_once_with(
test.IsHttpRequest(), 'network_availability_zone')
+ self.assertCountEqual(networks, [net for net in self.networks.list()
+ if net['tenant_id'] == '1' or
+ net['router:external'] is True or
+ net['shared'] is True])
return button
@@ -1181,7 +1193,6 @@ class NetworkViewTests(test.TestCase, NetworkStubMixin):
self.assertTemplateUsed(res, INDEX_TEMPLATE)
networks = res.context['networks_table'].data
- self.assertCountEqual(networks, self.networks.list())
button = find_button_fn(res)
self.assertIn('disabled', button.classes,
@@ -1192,9 +1203,13 @@ class NetworkViewTests(test.TestCase, NetworkStubMixin):
mock.call(test.IsHttpRequest(), targets=('network', )),
mock.call(test.IsHttpRequest(), targets=('subnet', )),
])
- self.assertEqual(8, self.mock_tenant_quota_usages.call_count)
+ self.assertEqual(12, self.mock_tenant_quota_usages.call_count)
self.mock_is_extension_supported.assert_called_once_with(
test.IsHttpRequest(), 'network_availability_zone')
+ self.assertCountEqual(networks, [net for net in self.networks.list()
+ if net['tenant_id'] == '1' or
+ net['router:external'] is True or
+ net['shared'] is True])
return button
diff --git a/openstack_dashboard/dashboards/project/networks/views.py b/openstack_dashboard/dashboards/project/networks/views.py
index fde0fbc9a..a783cdb6a 100644
--- a/openstack_dashboard/dashboards/project/networks/views.py
+++ b/openstack_dashboard/dashboards/project/networks/views.py
@@ -40,7 +40,7 @@ from openstack_dashboard.dashboards.project.networks \
import workflows as project_workflows
-class IndexView(tables.DataTableView):
+class IndexView(tables.PagedTableMixin, tables.DataTableView):
table_class = project_tables.NetworksTable
page_title = _("Networks")
FILTERS_MAPPING = {'shared': {_("yes"): True, _("no"): False},
@@ -51,12 +51,23 @@ class IndexView(tables.DataTableView):
try:
tenant_id = self.request.user.tenant_id
search_opts = self.get_filters(filters_map=self.FILTERS_MAPPING)
- networks = api.neutron.network_list_for_tenant(
- self.request, tenant_id,
- include_external=True,
- include_pre_auto_allocate=True,
- **search_opts)
+
+ marker, sort_dir = self._get_marker()
+ page_data = {
+ 'marker_id': marker,
+ 'sort_dir': sort_dir,
+ }
+
+ networks, self._has_more_data, self._has_prev_data = (
+ api.neutron.network_list_for_tenant(
+ self.request, tenant_id,
+ include_external=True,
+ include_pre_auto_allocate=True,
+ page_data=page_data,
+ **search_opts))
+
except Exception:
+ self._has_more_data = self._has_prev_data = False
networks = []
msg = _('Network list can not be retrieved.')
exceptions.handle(self.request, msg)
diff --git a/openstack_dashboard/dashboards/project/routers/tests.py b/openstack_dashboard/dashboards/project/routers/tests.py
index 4a8b90c39..3e3878738 100644
--- a/openstack_dashboard/dashboards/project/routers/tests.py
+++ b/openstack_dashboard/dashboards/project/routers/tests.py
@@ -830,11 +830,9 @@ class RouterActionTests(test.TestCase):
test.IsHttpRequest(), device_id=router.id)
self.assertEqual(2, self.mock_network_list.call_count)
self.mock_network_list.assert_has_calls([
- mock.call(test.IsHttpRequest(),
- shared=False,
- tenant_id=router['tenant_id']),
- mock.call(test.IsHttpRequest(),
- shared=True),
+ mock.call(test.IsHttpRequest(), single_page=False, shared=True),
+ mock.call(test.IsHttpRequest(), single_page=False,
+ shared=False, tenant_id=router['tenant_id']),
])
@test.create_mocks({api.neutron: ('router_get',
diff --git a/openstack_dashboard/test/integration_tests/pages/admin/network/networkspage.py b/openstack_dashboard/test/integration_tests/pages/admin/network/networkspage.py
index 738eb2ea4..f3a20dfc8 100644
--- a/openstack_dashboard/test/integration_tests/pages/admin/network/networkspage.py
+++ b/openstack_dashboard/test/integration_tests/pages/admin/network/networkspage.py
@@ -15,4 +15,24 @@ from openstack_dashboard.test.integration_tests.pages.project.network \
class NetworksPage(networkspage.NetworksPage):
- pass
+
+ NETWORKS_TABLE_NAME_COLUMN = 'Network Name'
+
+ @property
+ def is_admin(self):
+ return True
+
+ @property
+ def networks_table(self):
+ return NetworksTable(self.driver, self.conf)
+
+
+class NetworksTable(networkspage.NetworksTable):
+
+ CREATE_NETWORK_FORM_FIELDS = (("name", "admin_state",
+ "with_subnet", "az_hints", "tenant_id",
+ "network_type"),
+ ("subnet_name", "cidr", "ip_version",
+ "gateway_ip", "no_gateway"),
+ ("enable_dhcp", "allocation_pools",
+ "dns_nameservers", "host_routes"))
diff --git a/openstack_dashboard/test/integration_tests/pages/project/network/networkspage.py b/openstack_dashboard/test/integration_tests/pages/project/network/networkspage.py
index 00bcb211e..b79bd3cd3 100644
--- a/openstack_dashboard/test/integration_tests/pages/project/network/networkspage.py
+++ b/openstack_dashboard/test/integration_tests/pages/project/network/networkspage.py
@@ -56,6 +56,10 @@ class NetworksPage(basepage.BaseNavigationPage):
self.NETWORKS_TABLE_NAME_COLUMN, name)
@property
+ def is_admin(self):
+ return False
+
+ @property
def networks_table(self):
return NetworksTable(self.driver, self.conf)
@@ -66,9 +70,15 @@ class NetworksPage(basepage.BaseNavigationPage):
gateway_ip=None,
disable_gateway=DEFAULT_DISABLE_GATEWAY,
enable_dhcp=DEFAULT_ENABLE_DHCP, allocation_pools=None,
- dns_name_servers=None, host_routes=None):
+ dns_name_servers=None, host_routes=None,
+ project='admin', net_type='Local'):
create_network_form = self.networks_table.create_network()
- create_network_form.net_name.text = network_name
+ if self.is_admin:
+ create_network_form.network_type.text = net_type
+ create_network_form.tenant_id.text = project
+ create_network_form.name.text = network_name
+ else:
+ create_network_form.net_name.text = network_name
create_network_form.admin_state.value = admin_state
if not create_subnet:
create_network_form.with_subnet.unmark()
diff --git a/openstack_dashboard/test/integration_tests/tests/test_networks.py b/openstack_dashboard/test/integration_tests/tests/test_networks.py
index ed60bd976..80be2ed1e 100644
--- a/openstack_dashboard/test/integration_tests/tests/test_networks.py
+++ b/openstack_dashboard/test/integration_tests/tests/test_networks.py
@@ -21,6 +21,10 @@ class TestNetworks(helpers.TestCase):
NETWORK_NAME = helpers.gen_random_resource_name("network")
SUBNET_NAME = helpers.gen_random_resource_name("subnet")
+ @property
+ def networks_page(self):
+ return self.home_pg.go_to_project_network_networkspage()
+
def test_private_network_create(self):
"""tests the network creation and deletion functionalities:
@@ -30,7 +34,7 @@ class TestNetworks(helpers.TestCase):
* verifies the network does not appear in the table after deletion
"""
- networks_page = self.home_pg.go_to_project_network_networkspage()
+ networks_page = self.networks_page
networks_page.create_network(self.NETWORK_NAME, self.SUBNET_NAME)
self.assertTrue(
@@ -46,3 +50,97 @@ class TestNetworks(helpers.TestCase):
self.assertFalse(
networks_page.find_message_and_dismiss(messages.ERROR))
self.assertFalse(networks_page.is_network_present(self.NETWORK_NAME))
+
+ def test_networks_pagination(self):
+ """This test checks networks pagination
+
+ Steps:
+ 1) Login to Horizon Dashboard
+ 2) Go to Project -> Network -> Networks tab and create
+ three networks
+ 3) Navigate to user settings page
+ 4) Change 'Items Per Page' value to 2
+ 5) Go to Project -> Network -> Networks tab or
+ Admin -> Network -> Networks tab (depends on user)
+ 6) Check that only 'Next' link is available, only one network is
+ available (and it has correct name)
+ 7) Click 'Next' and check that both 'Prev' and 'Next' links are
+ available, only one network is available (and it has correct name)
+ 8) Click 'Next' and check that only 'Prev' link is available,
+ only one network is visible (and it has correct name)
+ 9) Click 'Prev' and check result (should be the same as for step7)
+ 10) Click 'Prev' and check result (should be the same as for step6)
+ 11) Go to user settings page and restore 'Items Per Page'
+ 12) Delete created networks
+ """
+ networks_page = self.networks_page
+ count = 6
+ items_per_page = 2
+ networks_names = ["{0}_{1}".format(self.NETWORK_NAME, i)
+ for i in range(count)]
+ for network_name in networks_names:
+ networks_page.create_network(network_name, self.SUBNET_NAME)
+ self.assertTrue(
+ networks_page.find_message_and_dismiss(messages.SUCCESS))
+ self.assertFalse(
+ networks_page.find_message_and_dismiss(messages.ERROR))
+ self.assertTrue(networks_page.is_network_present(network_name))
+ self.assertTrue(networks_page.is_network_active(network_name))
+
+ networks_page = self.networks_page
+ rows = networks_page.networks_table.get_column_data(
+ name_column=networks_page.NETWORKS_TABLE_NAME_COLUMN)
+ self._change_page_size_setting(items_per_page)
+ networks_page = self.networks_page
+ definitions = []
+ i = 0
+ while i < len(rows):
+ prev = i >= items_per_page
+ next = i < (len(rows) - items_per_page)
+ definition = {'Next': next, 'Prev': prev,
+ 'Count': items_per_page,
+ 'Names': rows[i:i + items_per_page]}
+ definitions.append(definition)
+ networks_page.networks_table.assert_definition(
+ definition,
+ name_column=networks_page.NETWORKS_TABLE_NAME_COLUMN)
+ if next:
+ networks_page.networks_table.turn_next_page()
+ i = i + items_per_page
+
+ definitions.reverse()
+ for definition in definitions:
+ networks_page.networks_table.assert_definition(
+ definition,
+ name_column=networks_page.NETWORKS_TABLE_NAME_COLUMN)
+ if definition['Prev']:
+ networks_page.networks_table.turn_prev_page()
+
+ self._change_page_size_setting()
+
+ networks_page = self.networks_page
+ for network_name in networks_names:
+ networks_page.delete_network(network_name)
+ self.assertTrue(
+ networks_page.find_message_and_dismiss(messages.SUCCESS))
+ self.assertFalse(
+ networks_page.find_message_and_dismiss(messages.ERROR))
+ self.assertFalse(networks_page.is_network_present(network_name))
+
+ def _change_page_size_setting(self, items_per_page=None):
+ settings_page = self.home_pg.go_to_settings_usersettingspage()
+ if items_per_page:
+ settings_page.change_pagesize(items_per_page)
+ else:
+ settings_page.change_pagesize()
+ settings_page.find_message_and_dismiss(messages.SUCCESS)
+
+
+@decorators.services_required("neutron")
+class TestAdminNetworks(helpers.AdminTestCase, TestNetworks):
+ NETWORK_NAME = helpers.gen_random_resource_name("network")
+ SUBNET_NAME = helpers.gen_random_resource_name("subnet")
+
+ @property
+ def networks_page(self):
+ return self.home_pg.go_to_admin_network_networkspage()
diff --git a/openstack_dashboard/test/test_data/neutron_data.py b/openstack_dashboard/test/test_data/neutron_data.py
index ef2ec07e0..df885dee6 100644
--- a/openstack_dashboard/test/test_data/neutron_data.py
+++ b/openstack_dashboard/test/test_data/neutron_data.py
@@ -271,7 +271,7 @@ def data(TEST):
TEST.api_ports.add(port_dict)
TEST.ports.add(neutron.Port(port_dict))
- # External network.
+ # External not shared network.
network_dict = {'admin_state_up': True,
'id': '9b466b94-213a-4cda-badf-72c102a874da',
'name': 'ext_net',
@@ -303,6 +303,162 @@ def data(TEST):
TEST.networks.add(neutron.Network(network))
TEST.subnets.add(subnet)
+ # External shared network.
+
+ network_dict = {'admin_state_up': True,
+ 'id': 'ed351877-4f7b-4672-8164-20a09e4873d3',
+ 'name': 'ext_net_shared',
+ 'status': 'ACTIVE',
+ 'subnets': ['5c59f875-f242-4df2-96e6-7dcc09d6dfc8'],
+ 'tenant_id': '4',
+ 'router:external': True,
+ 'shared': True}
+ subnet_dict = {'allocation_pools': [{'start': '172.24.14.226.',
+ 'end': '172.24.14.238'}],
+ 'dns_nameservers': [],
+ 'host_routes': [],
+ 'cidr': '172.24.14.0/28',
+ 'enable_dhcp': False,
+ 'gateway_ip': '172.24.14.225',
+ 'id': '5c59f875-f242-4df2-96e6-7dcc09d6dfc8',
+ 'ip_version': 4,
+ 'name': 'ext_shr_subnet',
+ 'network_id': network_dict['id'],
+ 'tenant_id': network_dict['tenant_id']}
+
+ TEST.api_networks.add(network_dict)
+ TEST.api_subnets.add(subnet_dict)
+
+ network = copy.deepcopy(network_dict)
+ subnet = neutron.Subnet(subnet_dict)
+ network['subnets'] = [subnet]
+ TEST.networks.add(neutron.Network(network))
+ TEST.subnets.add(subnet)
+
+ # tenant external shared network
+ network_dict = {'admin_state_up': True,
+ 'id': '650de90f-d77f-4863-ae98-39e97ad3ea7a',
+ 'name': 'ext_net_shared_tenant1',
+ 'status': 'ACTIVE',
+ 'subnets': ['d0a5bc19-16f0-45cc-a187-0d1bb36de4c6'],
+ 'tenant_id': '1',
+ 'router:external': True,
+ 'shared': True}
+ subnet_dict = {'allocation_pools': [{'start': '172.34.14.226.',
+ 'end': '172.34.14.238'}],
+ 'dns_nameservers': [],
+ 'host_routes': [],
+ 'cidr': '172.34.14.0/28',
+ 'enable_dhcp': False,
+ 'gateway_ip': '172.34.14.225',
+ 'id': 'd0a5bc19-16f0-45cc-a187-0d1bb36de4c6',
+ 'ip_version': 4,
+ 'name': 'ext_shr_tenant1_subnet',
+ 'network_id': network_dict['id'],
+ 'tenant_id': network_dict['tenant_id']}
+
+ TEST.api_networks.add(network_dict)
+ TEST.api_subnets.add(subnet_dict)
+
+ network = copy.deepcopy(network_dict)
+ subnet = neutron.Subnet(subnet_dict)
+ network['subnets'] = [subnet]
+ TEST.networks.add(neutron.Network(network))
+ TEST.subnets.add(subnet)
+
+ # tenant external non-shared network
+ network_dict = {'admin_state_up': True,
+ 'id': '19c3e662-1635-4876-be41-dbfdef0edd17',
+ 'name': 'ext_net_tenant1',
+ 'status': 'ACTIVE',
+ 'subnets': ['5ba8895c-0b3b-482d-9e42-ce389e1e1fa6'],
+ 'tenant_id': '1',
+ 'router:external': True,
+ 'shared': False}
+ subnet_dict = {'allocation_pools': [{'start': '172.44.14.226.',
+ 'end': '172.44.14.238'}],
+ 'dns_nameservers': [],
+ 'host_routes': [],
+ 'cidr': '172.44.14.0/28',
+ 'enable_dhcp': False,
+ 'gateway_ip': '172.44.14.225',
+ 'id': '5ba8895c-0b3b-482d-9e42-ce389e1e1fa6',
+ 'ip_version': 4,
+ 'name': 'ext_tenant1_subnet',
+ 'network_id': network_dict['id'],
+ 'tenant_id': network_dict['tenant_id']}
+
+ TEST.api_networks.add(network_dict)
+ TEST.api_subnets.add(subnet_dict)
+
+ network = copy.deepcopy(network_dict)
+ subnet = neutron.Subnet(subnet_dict)
+ network['subnets'] = [subnet]
+ TEST.networks.add(neutron.Network(network))
+ TEST.subnets.add(subnet)
+
+ # tenant non-external shared network
+ network_dict = {'admin_state_up': True,
+ 'id': 'fd581273-2601-4057-9c22-1be38f44884e',
+ 'name': 'shr_net_tenant1',
+ 'status': 'ACTIVE',
+ 'subnets': ['d2668892-bc32-4c89-9c63-961920a831d3'],
+ 'tenant_id': '1',
+ 'router:external': False,
+ 'shared': True}
+ subnet_dict = {'allocation_pools': [{'start': '172.54.14.226.',
+ 'end': '172.54.14.238'}],
+ 'dns_nameservers': [],
+ 'host_routes': [],
+ 'cidr': '172.54.14.0/28',
+ 'enable_dhcp': False,
+ 'gateway_ip': '172.54.14.225',
+ 'id': 'd2668892-bc32-4c89-9c63-961920a831d3',
+ 'ip_version': 4,
+ 'name': 'shr_tenant1_subnet',
+ 'network_id': network_dict['id'],
+ 'tenant_id': network_dict['tenant_id']}
+
+ TEST.api_networks.add(network_dict)
+ TEST.api_subnets.add(subnet_dict)
+
+ network = copy.deepcopy(network_dict)
+ subnet = neutron.Subnet(subnet_dict)
+ network['subnets'] = [subnet]
+ TEST.networks.add(neutron.Network(network))
+ TEST.subnets.add(subnet)
+
+ # non-tenant non-external non-shared network
+ network_dict = {'admin_state_up': True,
+ 'id': '7377e545-1527-4ce1-869e-caca192bc049',
+ 'name': 'net_tenant20',
+ 'status': 'ACTIVE',
+ 'subnets': ['c2bbd65e-0c0f-4ab9-8723-2dd102104f3d'],
+ 'tenant_id': '20',
+ 'router:external': False,
+ 'shared': False}
+ subnet_dict = {'allocation_pools': [{'start': '172.64.14.226.',
+ 'end': '172.64.14.238'}],
+ 'dns_nameservers': [],
+ 'host_routes': [],
+ 'cidr': '172.54.14.0/28',
+ 'enable_dhcp': False,
+ 'gateway_ip': '172.64.14.225',
+ 'id': 'c2bbd65e-0c0f-4ab9-8723-2dd102104f3d',
+ 'ip_version': 4,
+ 'name': 'tenant20_subnet',
+ 'network_id': network_dict['id'],
+ 'tenant_id': network_dict['tenant_id']}
+
+ TEST.api_networks.add(network_dict)
+ TEST.api_subnets.add(subnet_dict)
+
+ network = copy.deepcopy(network_dict)
+ subnet = neutron.Subnet(subnet_dict)
+ network['subnets'] = [subnet]
+ TEST.networks.add(neutron.Network(network))
+ TEST.subnets.add(subnet)
+
# 1st v6 network.
network_dict = {'admin_state_up': True,
'id': '96688ea1-ffa5-78ec-22ca-33aaabfaf775',
@@ -1002,3 +1158,92 @@ def data(TEST):
'name': 'nova'
}
)
+
+
+def list_nets_in_query_order(source_list):
+ return ([n for n in source_list if n['shared'] is True] +
+ [n for n in source_list if (n['tenant_id'] == '1' and
+ n['shared'] is False)] +
+ [n for n in source_list if n['router:external'] is True and
+ n['shared'] is False])
+
+
+source_nets_pagination1 = sorted([
+ neutron.Network({
+ 'admin_state_up': True,
+ 'id': uuidutils.generate_uuid(),
+ 'name': 'net{}'.format(i),
+ 'status': 'ACTIVE',
+ 'subnets': [],
+ 'tenant_id': '1',
+ 'router:external': False,
+ 'shared': False}) for i in range(0, 58)
+] + [
+ neutron.Network({
+ 'admin_state_up': True,
+ 'id': uuidutils.generate_uuid(),
+ 'name': 'net_ext',
+ 'status': 'ACTIVE',
+ 'subnets': [],
+ 'tenant_id': '2',
+ 'router:external': True,
+ 'shared': False})
+] + [
+ neutron.Network({
+ 'admin_state_up': True,
+ 'id': uuidutils.generate_uuid(),
+ 'name': 'net_shr',
+ 'status': 'ACTIVE',
+ 'subnets': [],
+ 'tenant_id': '3',
+ 'router:external': False,
+ 'shared': True})
+], key=lambda net: net['id'])
+
+all_nets_pagination1 = list_nets_in_query_order(source_nets_pagination1)
+
+source_nets_pagination2 = sorted([
+ neutron.Network({
+ 'admin_state_up': True,
+ 'id': uuidutils.generate_uuid(),
+ 'name': 'net{}'.format(i),
+ 'status': 'ACTIVE',
+ 'subnets': [],
+ 'tenant_id': '2',
+ 'router:external': True,
+ 'shared': False}) for i in range(0, 25)
+] + [
+ neutron.Network({
+ 'admin_state_up': True,
+ 'id': uuidutils.generate_uuid(),
+ 'name': 'net{}'.format(i),
+ 'status': 'ACTIVE',
+ 'subnets': [],
+ 'tenant_id': '3',
+ 'router:external': False,
+ 'shared': True}) for i in range(0, 25)
+] + [
+ neutron.Network({
+ 'admin_state_up': True,
+ 'id': uuidutils.generate_uuid(),
+ 'name': 'net{}'.format(i),
+ 'status': 'ACTIVE',
+ 'subnets': [],
+ 'tenant_id': '1',
+ 'router:external': False,
+ 'shared': False}) for i in range(0, 10)
+], key=lambda net: net['id'])
+
+all_nets_pagination2 = list_nets_in_query_order(source_nets_pagination2)
+
+source_nets_pagination3 = sorted([
+ neutron.Network({
+ 'admin_state_up': True,
+ 'id': uuidutils.generate_uuid(),
+ 'name': 'net{}'.format(i),
+ 'status': 'ACTIVE',
+ 'subnets': [],
+ 'tenant_id': '1',
+ 'router:external': False,
+ 'shared': False}) for i in range(0, 5)
+], key=lambda net: net['id'])
diff --git a/openstack_dashboard/test/unit/api/test_neutron.py b/openstack_dashboard/test/unit/api/test_neutron.py
index 9affe7f8b..af520abb0 100644
--- a/openstack_dashboard/test/unit/api/test_neutron.py
+++ b/openstack_dashboard/test/unit/api/test_neutron.py
@@ -23,9 +23,11 @@ from django.test.utils import override_settings
from openstack_dashboard import api
from openstack_dashboard import policy
from openstack_dashboard.test import helpers as test
+from openstack_dashboard.test.test_data import neutron_data
class NeutronApiTests(test.APIMockTestCase):
+
@mock.patch.object(api.neutron, 'neutronclient')
def test_network_list(self, mock_neutronclient):
networks = {'networks': self.api_networks.list()}
@@ -46,8 +48,8 @@ class NeutronApiTests(test.APIMockTestCase):
@test.create_mocks({api.neutron: ('network_list',
'subnet_list')})
def _test_network_list_for_tenant(
- self, include_external,
- filter_params, should_called, **extra_kwargs):
+ self, include_external, filter_params, should_called,
+ expected_networks, source_networks=None, **extra_kwargs):
"""Convenient method to test network_list_for_tenant.
:param include_external: Passed to network_list_for_tenant.
@@ -55,38 +57,74 @@ class NeutronApiTests(test.APIMockTestCase):
:param should_called: this argument specifies which methods
should be called. Methods in this list should be called.
Valid values are non_shared, shared, and external.
+ :param expected_networks: the networks to be compared with the result.
+ :param source_networks: networks to override the mocks.
"""
+ has_more_data = None
+ has_prev_data = None
+ marker_calls = []
filter_params = filter_params or {}
- all_networks = self.networks.list()
- tenant_id = '1'
- tenant_networks = [n for n in all_networks
- if n['tenant_id'] == tenant_id]
- shared_networks = [n for n in all_networks if n['shared']]
- external_networks = [n for n in all_networks if n['router:external']]
+ if 'page_data' not in extra_kwargs:
+ call_args = {'single_page': False}
+ else:
+ sort_dir = extra_kwargs['page_data']['sort_dir']
+ # invert sort_dir for calls
+ sort_dir = 'asc' if sort_dir == 'desc' else 'desc'
+ call_args = {'single_page': True, 'limit': 21, 'sort_key': 'id',
+ 'sort_dir': sort_dir}
+ marker_id = extra_kwargs['page_data'].get('marker_id')
+ if extra_kwargs.get('marker_calls') is not None:
+ marker_calls = extra_kwargs.pop('marker_calls')
+ tenant_id = '1'
return_values = []
+ all_networks = (self.networks.list() if source_networks is None
+ else source_networks)
+
expected_calls = []
- if 'non_shared' in should_called:
- params = filter_params.copy()
- params['shared'] = False
- return_values.append(tenant_networks)
- expected_calls.append(
- mock.call(test.IsHttpRequest(), tenant_id=tenant_id, **params),
- )
- if 'shared' in should_called:
- params = filter_params.copy()
- params['shared'] = True
- return_values.append(shared_networks)
- expected_calls.append(
- mock.call(test.IsHttpRequest(), **params),
- )
- if 'external' in should_called:
- params = filter_params.copy()
- params['router:external'] = True
- return_values.append(external_networks)
- expected_calls.append(
- mock.call(test.IsHttpRequest(), **params),
- )
+ call_order = ['shared', 'non_shared', 'external']
+ if call_args.get('sort_dir') == 'desc':
+ call_order.reverse()
+
+ for call in call_order:
+ if call in should_called:
+ params = filter_params.copy()
+ params.update(call_args)
+ if call in marker_calls:
+ params.update({'marker': marker_id})
+ if call == 'external':
+ params['router:external'] = True
+ params['shared'] = False
+ return_values.append(
+ [n for n in all_networks
+ if n['router:external'] is True and
+ n['shared'] is False])
+ expected_calls.append(
+ mock.call(test.IsHttpRequest(), **params))
+ elif call == 'shared':
+ params['shared'] = True
+ external = params.get('router:external')
+ return_values.append(
+ [n for n in all_networks
+ if (n['shared'] is True and
+ n['router:external'] == (
+ external if external is not None
+ else n['router:external']))])
+ expected_calls.append(
+ mock.call(test.IsHttpRequest(), **params))
+ elif call == 'non_shared':
+ params['shared'] = False
+ external = params.get('router:external')
+ return_values.append(
+ [n for n in all_networks
+ if (n['tenant_id'] == '1' and
+ n['shared'] is False and
+ n['router:external'] == (
+ external if external is not None
+ else n['router:external']))])
+ expected_calls.append(
+ mock.call(test.IsHttpRequest(),
+ tenant_id=tenant_id, **params))
self.mock_network_list.side_effect = return_values
extra_kwargs.update(filter_params)
@@ -94,88 +132,186 @@ class NeutronApiTests(test.APIMockTestCase):
self.request, tenant_id,
include_external=include_external,
**extra_kwargs)
-
- expected = []
- if 'non_shared' in should_called:
- expected += tenant_networks
- if 'shared' in should_called:
- expected += shared_networks
- if 'external' in should_called and include_external:
- expected += external_networks
- self.assertEqual(set(n.id for n in expected),
+ if 'page_data' in extra_kwargs:
+ has_more_data = ret_val[1]
+ has_prev_data = ret_val[2]
+ ret_val = ret_val[0]
+ self.mock_network_list.assert_has_calls(expected_calls)
+ self.assertEqual(set(n.id for n in expected_networks),
set(n.id for n in ret_val))
+ self.assertNotIn(api.neutron.AUTO_ALLOCATE_ID,
+ [n.id for n in ret_val])
+ return ret_val, has_more_data, has_prev_data
+
+ @override_settings(OPENSTACK_NEUTRON_NETWORK={
+ 'enable_auto_allocated_network': True})
+ @test.create_mocks({api.neutron: ('network_list',
+ 'subnet_list')})
+ def _test_network_list_paged(
+ self, filter_params, expected_networks, page_data,
+ source_networks=None, **extra_kwargs):
+ """Convenient method to test network_list_paged.
+
+ :param filter_params: Filters passed to network_list_for_tenant
+ :param expected_networks: the networks to be compared with the result.
+ :param page_data: dict provided by UI with pagination info
+ :param source_networks: networks to override the mocks.
+ """
+ filter_params = filter_params or {}
+ sort_dir = page_data['sort_dir']
+ # invert sort_dir for calls
+ sort_dir = 'asc' if sort_dir == 'desc' else 'desc'
+ call_args = {'single_page': True, 'limit': 21, 'sort_key': 'id',
+ 'sort_dir': sort_dir}
+
+ return_values = []
+ all_networks = (self.networks.list() if source_networks is None
+ else source_networks)
+
+ expected_calls = []
+
+ params = filter_params.copy()
+ params.update(call_args)
+ if page_data.get('marker_id'):
+ params.update({'marker': page_data.get('marker_id')})
+ extra_kwargs.update({'marker': page_data.get('marker_id')})
+ return_values.append(all_networks[0:21])
+ expected_calls.append(
+ mock.call(test.IsHttpRequest(), **params))
+
+ self.mock_network_list.side_effect = return_values
+
+ extra_kwargs.update(filter_params)
+ ret_val, has_more_data, has_prev_data = api.neutron.network_list_paged(
+ self.request, page_data, **extra_kwargs)
self.mock_network_list.assert_has_calls(expected_calls)
+ self.assertEqual(set(n.id for n in expected_networks),
+ set(n.id for n in ret_val))
+ self.assertNotIn(api.neutron.AUTO_ALLOCATE_ID,
+ [n.id for n in ret_val])
+ return ret_val, has_more_data, has_prev_data
+ def test_no_pre_auto_allocate_network(self):
# Ensure all three types of networks are not empty. This is required
# to check 'pre_auto_allocate' network is not included.
+ tenant_id = '1'
+ all_networks = self.networks.list()
+ tenant_networks = [n for n in all_networks
+ if n['tenant_id'] == tenant_id]
+ shared_networks = [n for n in all_networks if n['shared']]
+ external_networks = [n for n in all_networks if n['router:external']]
self.assertTrue(tenant_networks)
self.assertTrue(shared_networks)
self.assertTrue(external_networks)
- self.assertNotIn(api.neutron.AUTO_ALLOCATE_ID,
- [n.id for n in ret_val])
def test_network_list_for_tenant(self):
+ expected_networks = [n for n in self.networks.list()
+ if (n['tenant_id'] == '1' or n['shared'] is True)]
self._test_network_list_for_tenant(
include_external=False, filter_params=None,
- should_called=['non_shared', 'shared'])
+ should_called=['non_shared', 'shared'],
+ expected_networks=expected_networks)
def test_network_list_for_tenant_with_external(self):
+ expected_networks = [n for n in self.networks.list()
+ if (n['tenant_id'] == '1' or
+ n['shared'] is True or
+ n['router:external'] is True)]
self._test_network_list_for_tenant(
include_external=True, filter_params=None,
- should_called=['non_shared', 'shared', 'external'])
+ should_called=['non_shared', 'shared', 'external'],
+ expected_networks=expected_networks)
def test_network_list_for_tenant_with_filters_shared_false_wo_incext(self):
+ expected_networks = [n for n in self.networks.list()
+ if (n['tenant_id'] == '1' and
+ n['shared'] is False)]
self._test_network_list_for_tenant(
- include_external=False, filter_params={'shared': True},
- should_called=['shared'])
+ include_external=False, filter_params={'shared': False},
+ should_called=['non_shared'],
+ expected_networks=expected_networks)
def test_network_list_for_tenant_with_filters_shared_true_w_incext(self):
+ expected_networks = [n for n in self.networks.list()
+ if n['shared'] is True]
self._test_network_list_for_tenant(
include_external=True, filter_params={'shared': True},
- should_called=['shared', 'external'])
+ should_called=['shared'],
+ expected_networks=expected_networks)
def test_network_list_for_tenant_with_filters_ext_false_wo_incext(self):
+ expected_networks = [n for n in self.networks.list()
+ if ((n['tenant_id'] == '1' or
+ n['shared'] is True) and
+ n['router:external'] is False)]
self._test_network_list_for_tenant(
include_external=False, filter_params={'router:external': False},
- should_called=['non_shared', 'shared'])
+ should_called=['non_shared', 'shared'],
+ expected_networks=expected_networks)
def test_network_list_for_tenant_with_filters_ext_true_wo_incext(self):
+ expected_networks = [n for n in self.networks.list()
+ if ((n['tenant_id'] == '1' or
+ n['shared'] is True) and
+ n['router:external'] is True)]
self._test_network_list_for_tenant(
include_external=False, filter_params={'router:external': True},
- should_called=['non_shared', 'shared'])
+ should_called=['non_shared', 'shared'],
+ expected_networks=expected_networks)
def test_network_list_for_tenant_with_filters_ext_false_w_incext(self):
+ expected_networks = [n for n in self.networks.list()
+ if ((n['tenant_id'] == '1' or
+ n['shared'] is True) and
+ n['router:external'] is False)]
self._test_network_list_for_tenant(
include_external=True, filter_params={'router:external': False},
- should_called=['non_shared', 'shared'])
+ should_called=['non_shared', 'shared'],
+ expected_networks=expected_networks)
def test_network_list_for_tenant_with_filters_ext_true_w_incext(self):
+ expected_networks = [n for n in self.networks.list()
+ if n['router:external'] is True]
self._test_network_list_for_tenant(
include_external=True, filter_params={'router:external': True},
- should_called=['non_shared', 'shared', 'external'])
+ should_called=['external', 'shared', 'non_shared'],
+ expected_networks=expected_networks)
def test_network_list_for_tenant_with_filters_both_shared_ext(self):
# To check 'shared' filter is specified in network_list
# to look up external networks.
+ expected_networks = [n for n in self.networks.list()
+ if (n['shared'] is True and
+ n['router:external'] is True)]
self._test_network_list_for_tenant(
include_external=True,
filter_params={'router:external': True, 'shared': True},
- should_called=['shared', 'external'])
+ should_called=['shared'],
+ expected_networks=expected_networks)
def test_network_list_for_tenant_with_other_filters(self):
# To check filter parameters other than shared and
# router:external are passed as expected.
+ expected_networks = [n for n in self.networks.list()
+ if (n['router:external'] is True and
+ n['shared'] is False)]
self._test_network_list_for_tenant(
include_external=True,
filter_params={'router:external': True, 'shared': False,
'foo': 'bar'},
- should_called=['non_shared', 'external'])
+ should_called=['external', 'non_shared'],
+ expected_networks=expected_networks)
def test_network_list_for_tenant_no_pre_auto_allocate_if_net_exists(self):
+ expected_networks = [n for n in self.networks.list()
+ if (n['tenant_id'] == '1' or
+ n['shared'] is True or
+ n['router:external'] is True)]
self._test_network_list_for_tenant(
include_external=True, filter_params=None,
should_called=['non_shared', 'shared', 'external'],
- include_pre_auto_allocate=True)
+ include_pre_auto_allocate=True,
+ expected_networks=expected_networks)
@override_settings(OPENSTACK_NEUTRON_NETWORK={
'enable_auto_allocated_network': True})
@@ -197,9 +333,9 @@ class NeutronApiTests(test.APIMockTestCase):
self.assertEqual(2, self.mock_network_list.call_count)
self.mock_network_list.assert_has_calls([
- mock.call(test.IsHttpRequest(), tenant_id=tenant_id,
- shared=False),
- mock.call(test.IsHttpRequest(), shared=True),
+ mock.call(test.IsHttpRequest(), single_page=False, shared=True),
+ mock.call(test.IsHttpRequest(), single_page=False,
+ shared=False, tenant_id=tenant_id)
])
self.mock_is_extension_supported.assert_called_once_with(
test.IsHttpRequest(), 'auto-allocated-topology')
@@ -219,11 +355,554 @@ class NeutronApiTests(test.APIMockTestCase):
self.assertEqual(2, self.mock_network_list.call_count)
self.mock_network_list.assert_has_calls([
- mock.call(test.IsHttpRequest(), tenant_id=tenant_id,
- shared=False),
- mock.call(test.IsHttpRequest(), shared=True),
+ mock.call(test.IsHttpRequest(), single_page=False, shared=True),
+ mock.call(test.IsHttpRequest(), single_page=False,
+ shared=False, tenant_id=tenant_id),
])
+ def test_network_list_for_tenant_first_page_has_more(self):
+ source_networks = neutron_data.source_nets_pagination1
+ all_nets = neutron_data.all_nets_pagination1
+ page1 = all_nets[0:20]
+ page_data = {
+ 'sort_dir': 'desc',
+ 'marker_id': None,
+ }
+ result, more, prev = self._test_network_list_for_tenant(
+ include_external=True, filter_params=None,
+ should_called=['non_shared', 'shared'],
+ expected_networks=page1,
+ page_data=page_data,
+ source_networks=source_networks)
+
+ self.assertEqual(20, len(result))
+ self.assertTrue(more)
+ self.assertFalse(prev)
+ self.assertEqual('net_shr', result[0]['name'])
+ self.assertFalse(result[1]['shared'])
+ self.assertEqual(page1, result)
+
+ @mock.patch.object(api.neutron, 'network_get')
+ def test_network_list_for_tenant_second_page_has_more(self, mock_net_get):
+ all_nets = neutron_data.all_nets_pagination1
+ mock_net_get.return_value = all_nets[19]
+ page2 = all_nets[20:40]
+ page_data = {
+ 'sort_dir': 'desc',
+ 'marker_id': all_nets[19]['id'],
+ }
+ result, more, prev = self._test_network_list_for_tenant(
+ include_external=True, filter_params=None,
+ should_called=['non_shared'],
+ expected_networks=page2,
+ page_data=page_data,
+ source_networks=all_nets[20:41],
+ marker_calls=['non_shared'])
+
+ self.assertEqual(20, len(result))
+ self.assertFalse(result[0]['shared'])
+ self.assertEqual(page2[0]['name'], result[0]['name'])
+ self.assertTrue(more)
+ self.assertTrue(prev)
+ self.assertEqual(page2, result)
+
+ @mock.patch.object(api.neutron, 'network_get')
+ def test_network_list_for_tenant_last_page(self, mock_net_get):
+ all_nets = neutron_data.all_nets_pagination1
+ mock_net_get.return_value = all_nets[39]
+ page3 = all_nets[40:60]
+ page_data = {
+ 'sort_dir': 'desc',
+ 'marker_id': all_nets[39]['id'],
+ }
+ result, more, prev = self._test_network_list_for_tenant(
+ include_external=True, filter_params=None,
+ should_called=['non_shared', 'external'],
+ expected_networks=page3,
+ page_data=page_data,
+ source_networks=page3,
+ marker_calls=['non_shared'])
+
+ self.assertEqual(20, len(result))
+ self.assertFalse(result[0]['router:external'])
+ self.assertEqual('net_ext', result[-1]['name'])
+ self.assertEqual(page3[0]['name'], result[0]['name'])
+ self.assertFalse(more)
+ self.assertTrue(prev)
+ self.assertEqual(page3, result)
+
+ @mock.patch.object(api.neutron, 'network_get')
+ def test_network_list_for_tenant_second_page_by_prev(self, mock_net_get):
+ all_nets = list(neutron_data.all_nets_pagination1)
+ all_nets.reverse()
+ mock_net_get.return_value = all_nets[19]
+ page2 = all_nets[20:40]
+ page_data = {
+ 'sort_dir': 'asc',
+ 'marker_id': all_nets[19]['id'],
+ }
+ result, more, prev = self._test_network_list_for_tenant(
+ include_external=True, filter_params=None,
+ should_called=['non_shared'],
+ expected_networks=page2,
+ page_data=page_data,
+ source_networks=all_nets[20:41],
+ marker_calls=['non_shared'])
+
+ self.assertEqual(20, len(result))
+ self.assertFalse(result[0]['router:external'])
+ self.assertFalse(result[0]['shared'])
+ self.assertFalse(result[-1]['router:external'])
+ self.assertFalse(result[-1]['shared'])
+ self.assertTrue(more)
+ self.assertTrue(prev)
+ page2.reverse()
+ self.assertEqual(page2, result)
+
+ @mock.patch.object(api.neutron, 'network_get')
+ def test_network_list_for_tenant_first_page_by_prev(self, mock_net_get):
+ all_nets = list(neutron_data.all_nets_pagination1)
+ all_nets.reverse()
+ mock_net_get.return_value = all_nets[39]
+ page1 = all_nets[40:60]
+ page_data = {
+ 'sort_dir': 'asc',
+ 'marker_id': all_nets[39]['id'],
+ }
+ result, more, prev = self._test_network_list_for_tenant(
+ include_external=True, filter_params=None,
+ should_called=['non_shared', 'shared'],
+ expected_networks=page1,
+ page_data=page_data,
+ source_networks=page1,
+ marker_calls=['non_shared'])
+
+ self.assertEqual(20, len(result))
+ self.assertTrue(more)
+ self.assertFalse(prev)
+ self.assertFalse(result[1]['shared'])
+ self.assertEqual('net_shr', result[0]['name'])
+ page1.reverse()
+ self.assertEqual(page1, result)
+
+ def test_network_list_for_tenant_first_page_has_more2(self):
+ source_networks = neutron_data.source_nets_pagination2
+ all_nets = neutron_data.all_nets_pagination2
+ page1 = all_nets[0:20]
+ page_data = {
+ 'sort_dir': 'desc',
+ 'marker_id': None,
+ }
+ result, more, prev = self._test_network_list_for_tenant(
+ include_external=True, filter_params=None,
+ should_called=['shared'],
+ expected_networks=page1,
+ page_data=page_data,
+ source_networks=source_networks)
+
+ self.assertEqual(20, len(result))
+ self.assertTrue(more)
+ self.assertFalse(prev)
+ self.assertTrue(result[0]['shared'])
+ self.assertTrue(result[-1]['shared'])
+ self.assertFalse(result[0]['router:external'])
+ self.assertFalse(result[-1]['router:external'])
+ self.assertEqual(page1, result)
+
+ @mock.patch.object(api.neutron, 'network_get')
+ def test_network_list_for_tenant_second_page_has_more2(self, mock_net_get):
+ all_nets = neutron_data.all_nets_pagination2
+ mock_net_get.return_value = all_nets[19]
+ page2 = all_nets[20:40]
+ page_data = {
+ 'sort_dir': 'desc',
+ 'marker_id': all_nets[19]['id'],
+ }
+ result, more, prev = self._test_network_list_for_tenant(
+ include_external=True, filter_params=None,
+ should_called=['shared', 'non_shared', 'external'],
+ expected_networks=page2,
+ page_data=page_data,
+ source_networks=all_nets[20:41],
+ marker_calls=['shared'])
+
+ self.assertEqual(20, len(result))
+ self.assertTrue(result[0]['shared'])
+ self.assertFalse(result[-1]['shared'])
+ self.assertTrue(result[-1]['router:external'])
+ self.assertFalse(result[0]['router:external'])
+ self.assertTrue(more)
+ self.assertTrue(prev)
+ self.assertEqual(page2, result)
+
+ @mock.patch.object(api.neutron, 'network_get')
+ def test_network_list_for_tenant_last_page2(self, mock_net_get):
+ all_nets = neutron_data.all_nets_pagination2
+ mock_net_get.return_value = all_nets[39]
+ page3 = all_nets[40:60]
+ page_data = {
+ 'sort_dir': 'desc',
+ 'marker_id': all_nets[39]['id'],
+ }
+ result, more, prev = self._test_network_list_for_tenant(
+ include_external=True, filter_params=None,
+ should_called=['external'],
+ expected_networks=page3,
+ page_data=page_data,
+ source_networks=page3,
+ marker_calls=['external'])
+
+ self.assertEqual(20, len(result))
+ self.assertTrue(result[0]['router:external'])
+ self.assertFalse(result[0]['shared'])
+ self.assertFalse(result[-1]['shared'])
+ self.assertTrue(result[-1]['router:external'])
+ self.assertFalse(more)
+ self.assertTrue(prev)
+ self.assertEqual(page3, result)
+
+ @mock.patch.object(api.neutron, 'network_get')
+ def test_network_list_for_tenant_second_page_by_prev2(self, mock_net_get):
+ all_nets = list(neutron_data.all_nets_pagination2)
+ all_nets.reverse()
+ mock_net_get.return_value = all_nets[19]
+ page2 = all_nets[20:40]
+ page_data = {
+ 'sort_dir': 'asc',
+ 'marker_id': all_nets[19]['id'],
+ }
+ result, more, prev = self._test_network_list_for_tenant(
+ include_external=True, filter_params=None,
+ should_called=['shared', 'external', 'non_shared'],
+ expected_networks=page2,
+ page_data=page_data,
+ source_networks=all_nets[20:41],
+ marker_calls=['external'])
+
+ self.assertEqual(20, len(result))
+ self.assertTrue(result[0]['shared'])
+ self.assertFalse(result[-1]['shared'])
+ self.assertTrue(result[-1]['router:external'])
+ self.assertFalse(result[0]['router:external'])
+ self.assertTrue(more)
+ self.assertTrue(prev)
+ page2.reverse()
+ self.assertEqual(page2, result)
+
+ @mock.patch.object(api.neutron, 'network_get')
+ def test_network_list_for_tenant_first_page_by_prev2(self, mock_net_get):
+ all_nets = list(neutron_data.all_nets_pagination2)
+ all_nets.reverse()
+ mock_net_get.return_value = all_nets[39]
+ page1 = all_nets[40:60]
+ page_data = {
+ 'sort_dir': 'asc',
+ 'marker_id': all_nets[39]['id'],
+ }
+ result, more, prev = self._test_network_list_for_tenant(
+ include_external=True, filter_params=None,
+ should_called=['shared'],
+ expected_networks=page1,
+ page_data=page_data,
+ source_networks=page1,
+ marker_calls=['shared'])
+
+ self.assertEqual(20, len(result))
+ self.assertTrue(more)
+ self.assertFalse(prev)
+ self.assertTrue(result[0]['shared'])
+ self.assertTrue(result[-1]['shared'])
+ self.assertFalse(result[0]['router:external'])
+ self.assertFalse(result[-1]['router:external'])
+ page1.reverse()
+ self.assertEqual(page1, result)
+
+ def test_network_list_paged_first_page_has_more(self):
+ source_networks = neutron_data.source_nets_pagination1
+ page1 = source_networks[0:20]
+ page_data = {
+ 'sort_dir': 'desc',
+ 'marker_id': None,
+ }
+ result, more, prev = self._test_network_list_paged(
+ filter_params=None,
+ expected_networks=page1,
+ page_data=page_data,
+ source_networks=source_networks)
+
+ self.assertEqual(20, len(result))
+ self.assertTrue(more)
+ self.assertFalse(prev)
+ self.assertEqual(page1, result)
+
+ @mock.patch.object(api.neutron, 'network_get')
+ def test_network_list_paged_second_page_has_more(self, mock_net_get):
+ source_networks = neutron_data.source_nets_pagination1
+ mock_net_get.return_value = source_networks[19]
+ page2 = source_networks[20:40]
+ page_data = {
+ 'sort_dir': 'desc',
+ 'marker_id': source_networks[19]['id'],
+ }
+ result, more, prev = self._test_network_list_paged(
+ filter_params=None,
+ expected_networks=page2,
+ page_data=page_data,
+ source_networks=source_networks[20:41])
+
+ self.assertEqual(20, len(result))
+ self.assertTrue(more)
+ self.assertTrue(prev)
+ self.assertEqual(page2, result)
+
+ @mock.patch.object(api.neutron, 'network_get')
+ def test_network_list_paged_last_page(self, mock_net_get):
+ source_networks = neutron_data.source_nets_pagination1
+ mock_net_get.return_value = source_networks[39]
+ page3 = source_networks[40:60]
+ page_data = {
+ 'sort_dir': 'desc',
+ 'marker_id': source_networks[39]['id'],
+ }
+ result, more, prev = self._test_network_list_paged(
+ filter_params=None,
+ expected_networks=page3,
+ page_data=page_data,
+ source_networks=page3)
+
+ self.assertEqual(20, len(result))
+ self.assertFalse(more)
+ self.assertTrue(prev)
+ self.assertEqual(page3, result)
+
+ @mock.patch.object(api.neutron, 'network_get')
+ def test_network_list_paged_second_page_by_prev(self, mock_net_get):
+ source_networks = neutron_data.source_nets_pagination1
+ source_networks.reverse()
+ mock_net_get.return_value = source_networks[19]
+ page2 = source_networks[20:40]
+ page_data = {
+ 'sort_dir': 'asc',
+ 'marker_id': source_networks[19]['id'],
+ }
+ result, more, prev = self._test_network_list_paged(
+ filter_params=None,
+ expected_networks=page2,
+ page_data=page_data,
+ source_networks=source_networks[20:41])
+
+ self.assertEqual(20, len(result))
+ self.assertTrue(more)
+ self.assertTrue(prev)
+ page2.reverse()
+ self.assertEqual(page2, result)
+
+ @mock.patch.object(api.neutron, 'network_get')
+ def test_network_list_paged_first_page_by_prev(self, mock_net_get):
+ source_networks = neutron_data.source_nets_pagination1
+ source_networks.reverse()
+ mock_net_get.return_value = source_networks[39]
+ page1 = source_networks[40:60]
+ page_data = {
+ 'sort_dir': 'asc',
+ 'marker_id': source_networks[39]['id'],
+ }
+ result, more, prev = self._test_network_list_paged(
+ filter_params=None,
+ expected_networks=page1,
+ page_data=page_data,
+ source_networks=page1)
+
+ self.assertEqual(20, len(result))
+ self.assertTrue(more)
+ self.assertFalse(prev)
+ page1.reverse()
+ self.assertEqual(page1, result)
+
+ def test__perform_query_delete_last_project_without_marker(self):
+ marker_net = neutron_data.source_nets_pagination3[3]
+ query_result = (neutron_data.source_nets_pagination3[0:3], False, True)
+ query_func = mock.Mock(side_effect=[([], False, True), query_result])
+ self.request.session['network_deleted'] = \
+ neutron_data.source_nets_pagination3[4]
+ query_kwargs = {
+ 'request': self.request,
+ 'page_data': {'single_page': True,
+ 'marker_type': 'proj',
+ 'sort_dir': 'asc'},
+ 'sort_dir': 'asc',
+ }
+ modified_query_kwargs = {
+ 'request': self.request,
+ 'page_data': {'single_page': True,
+ 'marker_type': 'proj',
+ 'sort_dir': 'desc'},
+ 'sort_dir': 'desc',
+ }
+ result = api.neutron._perform_query(
+ query_func, dict(query_kwargs), marker_net)
+ self.assertEqual(query_result, result)
+ query_func.assert_has_calls([
+ mock.call(**query_kwargs), mock.call(**modified_query_kwargs)
+ ])
+
+ def test__perform_query_delete_last_project_with_marker(self):
+ marker_net = neutron_data.source_nets_pagination3[3]
+ query_result = (neutron_data.source_nets_pagination3[0:4], False, False)
+ query_func = mock.Mock(side_effect=[([], False, True), query_result])
+ self.request.session['network_deleted'] = \
+ neutron_data.source_nets_pagination3[4]
+ query_kwargs = {
+ 'request': self.request,
+ 'page_data': {'single_page': True,
+ 'marker_type': 'proj',
+ 'sort_dir': 'asc'},
+ 'sort_dir': 'asc',
+ }
+ modified_query_kwargs = {
+ 'request': self.request,
+ 'page_data': {'single_page': True,
+ 'marker_type': 'proj',
+ 'sort_dir': 'desc'},
+ 'sort_dir': 'desc',
+ }
+ result = api.neutron._perform_query(
+ query_func, dict(query_kwargs), marker_net)
+ self.assertEqual(query_result, result)
+ query_func.assert_has_calls([
+ mock.call(**query_kwargs), mock.call(**modified_query_kwargs)
+ ])
+
+ def test__perform_query_delete_last_admin_with_marker(self):
+ marker_net = neutron_data.source_nets_pagination3[3]
+ query_result = (neutron_data.source_nets_pagination3[0:4], False, False)
+ query_func = mock.Mock(side_effect=[([], False, True), query_result])
+ self.request.session['network_deleted'] = \
+ neutron_data.source_nets_pagination3[4]
+ query_kwargs = {
+ 'request': self.request,
+ 'page_data': {'single_page': True,
+ 'marker_type': 'proj',
+ 'sort_dir': 'asc'},
+ 'params': {'sort_dir': 'asc'},
+ }
+ modified_query_kwargs = {
+ 'request': self.request,
+ 'page_data': {'single_page': True,
+ 'marker_type': 'proj',
+ 'sort_dir': 'desc'},
+ 'params': {'sort_dir': 'desc'},
+ }
+ result = api.neutron._perform_query(
+ query_func, dict(query_kwargs), marker_net)
+ self.assertEqual(query_result, result)
+ query_func.assert_has_calls([
+ mock.call(**query_kwargs), mock.call(**modified_query_kwargs)
+ ])
+
+ def test__perform_query_delete_first_admin(self):
+ marker_net = neutron_data.source_nets_pagination3[3]
+ query_result = (neutron_data.source_nets_pagination3[0:3], True, False)
+ query_func = mock.Mock(side_effect=[([], True, False), query_result])
+ self.request.session['network_deleted'] = \
+ neutron_data.source_nets_pagination3[0]
+ query_kwargs = {
+ 'request': self.request,
+ 'page_data': {'single_page': True,
+ 'marker_type': 'proj',
+ 'sort_dir': 'desc',
+ 'marker_id': marker_net['id']},
+ 'params': {'sort_dir': 'desc',
+ 'marker': marker_net['id']},
+ }
+ modified_query_kwargs = {
+ 'request': self.request,
+ 'page_data': {'single_page': True,
+ 'marker_type': None,
+ 'sort_dir': 'asc',
+ 'marker_id': None},
+ 'params': {'sort_dir': 'asc'},
+ }
+ result = api.neutron._perform_query(
+ query_func, dict(query_kwargs), marker_net)
+ self.assertEqual(query_result, result)
+ query_func.assert_has_calls([
+ mock.call(**query_kwargs), mock.call(**modified_query_kwargs)
+ ])
+
+ def test__perform_query_delete_first_proj(self):
+ marker_net = neutron_data.source_nets_pagination3[3]
+ query_result = (neutron_data.source_nets_pagination3[0:3], True, False)
+ query_func = mock.Mock(side_effect=[([], True, False), query_result])
+ self.request.session['network_deleted'] = \
+ neutron_data.source_nets_pagination3[0]
+ query_kwargs = {
+ 'request': self.request,
+ 'page_data': {'single_page': True,
+ 'marker_type': 'proj',
+ 'sort_dir': 'desc',
+ 'marker_id': marker_net['id']},
+ 'sort_dir': 'desc',
+ }
+ modified_query_kwargs = {
+ 'request': self.request,
+ 'page_data': {'single_page': True,
+ 'marker_type': None,
+ 'sort_dir': 'asc',
+ 'marker_id': None},
+ 'sort_dir': 'asc',
+ }
+ result = api.neutron._perform_query(
+ query_func, dict(query_kwargs), marker_net)
+ self.assertEqual(query_result, result)
+ query_func.assert_has_calls([
+ mock.call(**query_kwargs), mock.call(**modified_query_kwargs)
+ ])
+
+ def test__perform_query_normal_paginated(self):
+ query_result = (self.networks.list(), True, True)
+ query_func = mock.Mock(return_value=query_result)
+ query_kwargs = {'request': self.request,
+ 'page_data': {'single_page': True}}
+
+ result = api.neutron._perform_query(query_func, query_kwargs, None)
+ self.assertEqual(query_result, result)
+ query_func.assert_called_once_with(**query_kwargs)
+
+ @override_settings(OPENSTACK_NEUTRON_NETWORK={
+ 'enable_auto_allocated_network': True})
+ @test.create_mocks({api.neutron: ['is_extension_supported'],
+ api.nova: ['is_feature_available']})
+ def test__perform_query_with_preallocated(self):
+ self.mock_is_extension_supported.return_value = True
+ self.mock_is_feature_available.return_value = True
+ query_func = mock.Mock(return_value=([], False, False))
+ query_kwargs = {'request': self.request,
+ 'page_data': {'single_page': True}}
+
+ result = api.neutron._perform_query(
+ query_func, query_kwargs, None, include_pre_auto_allocate=True)
+ self.assertIsInstance(result[0][0], api.neutron.PreAutoAllocateNetwork)
+ self.assertEqual(False, result[1])
+ self.assertEqual(False, result[2])
+ query_func.assert_called_once_with(**query_kwargs)
+
+ def test__perform_query_not_paginated(self):
+ query_result = self.networks.list()
+ query_func = mock.Mock(return_value=(query_result, False, False))
+ query_kwargs1 = {'page_data': {'single_page': False}}
+ query_kwargs2 = {'page_data': {}}
+
+ result = api.neutron._perform_query(query_func, query_kwargs1, None)
+ self.assertEqual(query_result, result)
+ query_func.assert_called_once_with(**query_kwargs1)
+
+ query_func.reset_mock()
+
+ result = api.neutron._perform_query(query_func, query_kwargs2, None)
+ self.assertEqual(query_result, result)
+ query_func.assert_called_once_with(**query_kwargs2)
+
@mock.patch.object(api.neutron, 'neutronclient')
def test_network_get(self, mock_neutronclient):
network = {'network': self.api_networks.first()}
diff --git a/openstack_dashboard/test/unit/usage/test_quotas.py b/openstack_dashboard/test/unit/usage/test_quotas.py
index 3538b8a8d..561715c3d 100644
--- a/openstack_dashboard/test/unit/usage/test_quotas.py
+++ b/openstack_dashboard/test/unit/usage/test_quotas.py
@@ -471,7 +471,7 @@ class QuotaTests(test.APITestCase):
target: {
'used': used,
'quota': limit,
- 'available': limit - used
+ 'available': max(limit - used, 0)
}
}
diff --git a/releasenotes/notes/add-networks-pagination-4c05d784998fafb2.yaml b/releasenotes/notes/add-networks-pagination-4c05d784998fafb2.yaml
new file mode 100644
index 000000000..aca0e64ef
--- /dev/null
+++ b/releasenotes/notes/add-networks-pagination-4c05d784998fafb2.yaml
@@ -0,0 +1,5 @@
+---
+fixes:
+ - |
+ Fixed lack of pagination for the networks page under Project and
+ Admin Dashboard.