openstack/horizon
Code Issues Proposed changes

SDK for Neutron networks and subnets

Browse Source 
Depends-On: https://review.opendev.org/c/909656
Related-Bug: #1999774
Change-Id: Ic454dcb06b9efe2a4735637bd804d8a337c394cd
This commit is contained in:
elajkat 2023-08-10 14:55:16 +02:00 committed by Lajos Katona
parent c0d558c65f
commit a2d84f10e0
23 changed files with 485 additions and 317 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
horizon/tables/base.py
View File

@ -425,9 +425,15 @@ class Column(html.HTMLElement):
display_value = None
if self.display_choices:
display_value = [display for (value, display) in
self.display_choices
if value.lower() == (data or '').lower()]
display_value = []
for (value, display) in self.display_choices:
data_lower = ''
try:
data_lower = (data or '').lower()
if value.lower() == data_lower:
display_value.append(display)
except AttributeError:
continue
if display_value:
data = display_value[0]

148
openstack_dashboard/api/neutron.py
View File

@ -22,14 +22,23 @@ from collections.abc import Sequence
import copy
import itertools
import logging
import types
import netaddr
from django.conf import settings
from django.utils.translation import gettext_lazy as _
from keystoneauth1 import exceptions as ks_exceptions
from keystoneauth1 import session
from keystoneauth1 import token_endpoint
from neutronclient.common import exceptions as neutron_exc
from neutronclient.v2_0 import client as neutron_client
from novaclient import exceptions as nova_exc
import openstack
from openstack import exceptions as sdk_exceptions
import requests
from openstack_auth import utils as auth_utils
from horizon import exceptions
from horizon import messages
@ -67,6 +76,11 @@ VNIC_TYPES = [
class NeutronAPIDictWrapper(base.APIDictWrapper):
def __init__(self, apidict):
if 'is_admin_state_up' in apidict:
if apidict['is_admin_state_up']:
apidict['admin_state'] = 'UP'
else:
apidict['admin_state'] = 'DOWN'
if 'admin_state_up' in apidict:
if apidict['admin_state_up']:
apidict['admin_state'] = 'UP'
@ -665,6 +679,7 @@ class FloatingIpManager(object):
def __init__(self, request):
self.request = request
self.client = neutronclient(request)
self.net_client = networkclient(request)
@profiler.trace
def list_pools(self):
@ -674,7 +689,7 @@ class FloatingIpManager(object):
"""
search_opts = {'router:external': True}
return [FloatingIpPool(pool) for pool
in self.client.list_networks(**search_opts).get('networks')]
in self.net_client.networks(**search_opts)]
def _get_instance_type_from_device_owner(self, device_owner):
for key, value in self.device_owner_map.items():
@ -817,8 +832,8 @@ class FloatingIpManager(object):
if p.device_id in gw_routers)
# we have to include any shared subnets as well because we may not
# have permission to see the router interface to infer connectivity
shared = set(s.id for n in network_list(self.request, shared=True)
for s in n.subnets)
shared = set(s.id for n in network_list(self.request, is_shared=True)
for s in n['subnets'])
return reachable_subnets | shared
@profiler.trace
@ -942,6 +957,35 @@ def neutronclient(request):
return c
@memoized
def networkclient(request):
token_id, neutron_url, auth_url = get_auth_params_from_request(request)
insecure = settings.OPENSTACK_SSL_NO_VERIFY
cacert = settings.OPENSTACK_SSL_CACERT
verify = cacert if not insecure else False
token_auth = token_endpoint.Token(
endpoint=neutron_url,
token=token_id)
k_session = session.Session(
auth=token_auth,
original_ip=auth_utils.get_client_ip(request),
verify=verify,
# TODO(lajoskatona): cert should be None of a tuple in the form of
# (cert, key).
# In a devstack with enable_service tls-proxy:
# cert=('/path/to/devstack-cert.crt', '/path/to/devstack-cert.key')
# For this new horizon cfg option should be added.
)
c = openstack.connection.Connection(
session=k_session,
region_name=request.user.services_region,
app_name='horizon', app_version='1.0'
)
return c.network
@profiler.trace
def list_resources_with_long_filters(list_method,
filter_attr, filter_values, **params):
@ -1154,23 +1198,57 @@ def _network_list_paged(request, page_data, params):
@profiler.trace
def network_list(request, single_page=False, **params):
LOG.debug("network_list(): params=%s", params)
list_values = []
if 'id' in params and isinstance(params['id'], frozenset):
list_values = list(params['id'])
if 'id' in params and isinstance(params['id'], list):
list_values = params['id']
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')
networks = []
if 'tenant_id' in params:
params['project_id'] = params.pop('tenant_id')
for value in list_values:
params['id'] = value
for net in networkclient(request).networks(**params):
networks.append(net)
if not list_values:
networks = networkclient(request).networks(**params)
# Get subnet list to expand subnet info in network list.
subnets = subnet_list(request)
subnet_dict = dict((s['id'], s) for s in subnets)
if not isinstance(networks, (list, types.GeneratorType)):
networks = [networks]
nets_with_subnet = []
net_ids = set()
# Expand subnet list from subnet_id to values.
for n in networks:
# Due to potential timing issues, we can't assume the subnet_dict data
# is in sync with the network data.
n['subnets'] = [subnet_dict[s] for s in n.get('subnets', []) if
s in subnet_dict]
return [Network(n) for n in networks]
subnet_l_ready = False
runs = 0
max_runs = 3
while not subnet_l_ready and runs < max_runs:
networks, cp_nets = itertools.tee(networks, 2)
try:
for n in cp_nets:
# Due to potential timing issues, we can't assume the
# subnet_dict data is in sync with the network data.
net_dict = n.to_dict()
net_dict['subnets'] = [
subnet_dict[s] for s in net_dict.get('subnet_ids', [])
if s in subnet_dict
]
if net_dict['id'] not in net_ids:
nets_with_subnet.append(net_dict)
net_ids.add(net_dict['id'])
subnet_l_ready = True
except (requests.exceptions.SSLError, ks_exceptions.SSLError):
LOG.warning('Retry due to SSLError')
runs += 1
continue
return [Network(n) for n in nets_with_subnet]
def _is_auto_allocated_network_supported(request):
@ -1381,9 +1459,9 @@ def _query_nets_for_tenant(request, include_external, tenant_id, page_data,
def _configure_marker_type(marker_net, tenant_id=None):
if marker_net:
if marker_net['shared'] is True:
if marker_net['is_shared'] is True:
return 'shr'
if (marker_net['router:external'] is True and
if (marker_net['is_router_external'] is True and
marker_net['tenant_id'] != tenant_id):
return 'ext'
return 'proj'
@ -1565,8 +1643,8 @@ def network_list_for_tenant(request, tenant_id, include_external=False,
def network_get(request, network_id, expand_subnet=True, **params):
LOG.debug("network_get(): netid=%(network_id)s, params=%(params)s",
{'network_id': network_id, 'params': params})
network = neutronclient(request).show_network(network_id,
**params).get('network')
network = networkclient(request).get_network(network_id,
**params).to_dict()
if expand_subnet:
# NOTE(amotoki): There are some cases where a user has no permission
# to get subnet details, but the condition is complicated. We first
@ -1582,8 +1660,8 @@ def network_get(request, network_id, expand_subnet=True, **params):
# call subnet_get() for each subnet instead of calling
# subnet_list() once.
network['subnets'] = [subnet_get(request, sid)
for sid in network['subnets']]
except neutron_exc.NotFound:
for sid in network['subnet_ids']]
except sdk_exceptions.ResourceNotFound:
pass
return Network(network)
@ -1600,8 +1678,7 @@ def network_create(request, **kwargs):
LOG.debug("network_create(): kwargs = %s", kwargs)
if 'tenant_id' not in kwargs:
kwargs['tenant_id'] = request.user.project_id
body = {'network': kwargs}
network = neutronclient(request).create_network(body=body).get('network')
network = networkclient(request).create_network(**kwargs).to_dict()
return Network(network)
@ -1609,16 +1686,15 @@ def network_create(request, **kwargs):
def network_update(request, network_id, **kwargs):
LOG.debug("network_update(): netid=%(network_id)s, params=%(params)s",
{'network_id': network_id, 'params': kwargs})
body = {'network': kwargs}
network = neutronclient(request).update_network(network_id,
body=body).get('network')
network = networkclient(request).update_network(network_id,
**kwargs).to_dict()
return Network(network)
@profiler.trace
def network_delete(request, network_id):
LOG.debug("network_delete(): netid=%s", network_id)
neutronclient(request).delete_network(network_id)
networkclient(request).delete_network(network_id)
request.session['network_deleted'] = network_id
@ -1626,16 +1702,17 @@ def network_delete(request, network_id):
@memoized
def subnet_list(request, **params):
LOG.debug("subnet_list(): params=%s", params)
subnets = neutronclient(request).list_subnets(**params).get('subnets')
return [Subnet(s) for s in subnets]
subnets = networkclient(request).subnets(**params)
ret_val = [Subnet(s.to_dict()) for s in subnets]
return ret_val
@profiler.trace
def subnet_get(request, subnet_id, **params):
LOG.debug("subnet_get(): subnetid=%(subnet_id)s, params=%(params)s",
{'subnet_id': subnet_id, 'params': params})
subnet = neutronclient(request).show_subnet(subnet_id,
**params).get('subnet')
subnet = networkclient(request).get_subnet(subnet_id,
**params).to_dict()
return Subnet(subnet)
@ -1660,11 +1737,11 @@ def subnet_create(request, network_id, **kwargs):
"""
LOG.debug("subnet_create(): netid=%(network_id)s, kwargs=%(kwargs)s",
{'network_id': network_id, 'kwargs': kwargs})
body = {'subnet': {'network_id': network_id}}
body = {'network_id': network_id}
if 'tenant_id' not in kwargs:
kwargs['tenant_id'] = request.user.project_id
body['subnet'].update(kwargs)
subnet = neutronclient(request).create_subnet(body=body).get('subnet')
body.update(kwargs)
subnet = networkclient(request).create_subnet(**body).to_dict()
return Subnet(subnet)
@ -1672,16 +1749,15 @@ def subnet_create(request, network_id, **kwargs):
def subnet_update(request, subnet_id, **kwargs):
LOG.debug("subnet_update(): subnetid=%(subnet_id)s, kwargs=%(kwargs)s",
{'subnet_id': subnet_id, 'kwargs': kwargs})
body = {'subnet': kwargs}
subnet = neutronclient(request).update_subnet(subnet_id,
body=body).get('subnet')
subnet = networkclient(request).update_subnet(subnet_id,
**kwargs).to_dict()
return Subnet(subnet)
@profiler.trace
def subnet_delete(request, subnet_id):
LOG.debug("subnet_delete(): subnetid=%s", subnet_id)
neutronclient(request).delete_subnet(subnet_id)
networkclient(request).delete_subnet(subnet_id)
@profiler.trace

6
openstack_dashboard/dashboards/admin/networks/subnets/tests.py
View File

@ -137,7 +137,7 @@ class NetworkSubnetTests(test.BaseAdminViewTests):
cidr=subnet.cidr,
ip_version=subnet.ip_version,
gateway_ip=subnet.gateway_ip,
enable_dhcp=subnet.enable_dhcp,
enable_dhcp=subnet.is_dhcp_enabled,
allocation_pools=subnet.allocation_pools,
tenant_id=subnet.tenant_id)
@ -194,7 +194,7 @@ class NetworkSubnetTests(test.BaseAdminViewTests):
cidr=subnet.cidr,
ip_version=subnet.ip_version,
gateway_ip=subnet.gateway_ip,
enable_dhcp=subnet.enable_dhcp,
enable_dhcp=subnet.is_dhcp_enabled,
tenant_id=subnet.tenant_id)
@test.create_mocks({api.neutron: ('network_get',
@ -278,7 +278,7 @@ class NetworkSubnetTests(test.BaseAdminViewTests):
self.mock_subnet_update.assert_called_once_with(
test.IsHttpRequest(), subnet.id,
name=subnet.name,
enable_dhcp=subnet.enable_dhcp,
enable_dhcp=subnet.is_dhcp_enabled,
dns_nameservers=[],
host_routes=[])

6
openstack_dashboard/dashboards/admin/networks/tables.py
View File

@ -84,15 +84,15 @@ class NetworksTable(tables.DataTable):
verbose_name=_("Subnets Associated"),)
num_agents = tables.Column("num_agents",
verbose_name=_("DHCP Agents"))
shared = tables.Column("shared", verbose_name=_("Shared"),
shared = tables.Column("is_shared", verbose_name=_("Shared"),
filters=(filters.yesno, filters.capfirst))
external = tables.Column("router:external",
external = tables.Column("is_router_external",
verbose_name=_("External"),
filters=(filters.yesno, filters.capfirst))
status = tables.Column(
"status", verbose_name=_("Status"),
display_choices=project_tables.STATUS_DISPLAY_CHOICES)
admin_state = tables.Column("admin_state",
admin_state = tables.Column("is_admin_state_up",
verbose_name=_("Admin State"),
display_choices=DISPLAY_CHOICES)
availability_zones = tables.Column(get_availability_zones,

36
openstack_dashboard/dashboards/admin/networks/tests.py
View File

@ -409,7 +409,7 @@ class NetworkTests(test.BaseAdminViewTests):
form_data = {'tenant_id': tenant_id,
'name': network.name,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'external': True,
'shared': True,
'network_type': 'local'}
@ -423,7 +423,7 @@ class NetworkTests(test.BaseAdminViewTests):
self.mock_subnetpool_list.assert_called_once_with(test.IsHttpRequest())
params = {'name': network.name,
'tenant_id': tenant_id,
'admin_state_up': network.admin_state_up,
'admin_state_up': network.is_admin_state_up,
'router:external': True,
'shared': True,
'provider:network_type': 'local'}
@ -456,7 +456,7 @@ class NetworkTests(test.BaseAdminViewTests):
form_data = {'tenant_id': tenant_id,
'name': network.name,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'external': True,
'shared': True,
'network_type': 'local',
@ -478,7 +478,7 @@ class NetworkTests(test.BaseAdminViewTests):
self.mock_subnetpool_list.assert_called_once_with(test.IsHttpRequest())
params = {'name': network.name,
'tenant_id': tenant_id,
'admin_state_up': network.admin_state_up,
'admin_state_up': network.is_admin_state_up,
'router:external': True,
'shared': True,
'provider:network_type': 'local',
@ -505,7 +505,7 @@ class NetworkTests(test.BaseAdminViewTests):
form_data = {'tenant_id': tenant_id,
'name': network.name,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'external': True,
'shared': True,
'mtu': 1450,
@ -520,7 +520,7 @@ class NetworkTests(test.BaseAdminViewTests):
self.mock_subnetpool_list.assert_called_once_with(test.IsHttpRequest())
params = {'name': network.name,
'tenant_id': tenant_id,
'admin_state_up': network.admin_state_up,
'admin_state_up': network.is_admin_state_up,
'router:external': True,
'shared': True,
'mtu': 1450,
@ -554,7 +554,7 @@ class NetworkTests(test.BaseAdminViewTests):
form_data = {'tenant_id': tenant_id,
'name': network.name,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'external': True,
'shared': True,
'network_type': 'local',
@ -574,7 +574,7 @@ class NetworkTests(test.BaseAdminViewTests):
self.mock_subnetpool_list.assert_called_once_with(test.IsHttpRequest())
params = {'name': network.name,
'tenant_id': tenant_id,
'admin_state_up': network.admin_state_up,
'admin_state_up': network.is_admin_state_up,
'router:external': True,
'shared': True,
'provider:network_type': 'local'}
@ -584,7 +584,7 @@ class NetworkTests(test.BaseAdminViewTests):
'name': subnet.name,
'network_id': subnet.network_id,
'cidr': subnet.cidr,
'enable_dhcp': subnet.enable_dhcp,
'enable_dhcp': subnet.is_dhcp_enabled,
'gateway_ip': subnet.gateway_ip,
'ip_version': subnet.ip_version}
self.mock_subnet_create.assert_called_once_with(test.IsHttpRequest(),
@ -609,7 +609,7 @@ class NetworkTests(test.BaseAdminViewTests):
form_data = {'tenant_id': tenant_id,
'name': network.name,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'external': True,
'shared': False,
'network_type': 'local'}
@ -627,7 +627,7 @@ class NetworkTests(test.BaseAdminViewTests):
self.mock_subnetpool_list.assert_called_once_with(test.IsHttpRequest())
params = {'name': network.name,
'tenant_id': tenant_id,
'admin_state_up': network.admin_state_up,
'admin_state_up': network.is_admin_state_up,
'router:external': True,
'shared': False,
'provider:network_type': 'local'}
@ -649,7 +649,7 @@ class NetworkTests(test.BaseAdminViewTests):
form_data = {'tenant_id': tenant_id,
'name': network.name,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'external': True,
'shared': False,
'network_type': 'vlan',
@ -682,7 +682,7 @@ class NetworkTests(test.BaseAdminViewTests):
form_data = {'tenant_id': tenant_id,
'name': network.name,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'external': True,
'shared': False,
'network_type': 'gre',
@ -718,7 +718,7 @@ class NetworkTests(test.BaseAdminViewTests):
form_data = {'tenant_id': tenant_id,
'name': network.name,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'external': True,
'shared': False,
'network_type': 'vxlan',
@ -834,7 +834,7 @@ class NetworkTests(test.BaseAdminViewTests):
form_data = {'network_id': network.id,
'name': network.name,
'tenant_id': network.tenant_id,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'shared': True,
'external': True}
url = reverse('horizon:admin:networks:update', args=[network.id])
@ -844,7 +844,7 @@ class NetworkTests(test.BaseAdminViewTests):
params = {'name': network.name,
'shared': True,
'admin_state_up': network.admin_state_up,
'admin_state_up': network.is_admin_state_up,
'router:external': True}
self.mock_network_update.assert_called_once_with(test.IsHttpRequest(),
network.id,
@ -859,7 +859,7 @@ class NetworkTests(test.BaseAdminViewTests):
network = self.networks.first()
params = {'name': network.name,
'shared': False,
'admin_state_up': network.admin_state_up,
'admin_state_up': network.is_admin_state_up,
'router:external': False}
self.mock_network_update.side_effect = self.exceptions.neutron
self.mock_network_get.return_value = network
@ -867,7 +867,7 @@ class NetworkTests(test.BaseAdminViewTests):
form_data = {'network_id': network.id,
'name': network.name,
'tenant_id': network.tenant_id,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'shared': False,
'external': False}
url = reverse('horizon:admin:networks:update', args=[network.id])

8
openstack_dashboard/dashboards/admin/networks/views.py
View File

@ -160,9 +160,9 @@ class UpdateView(user_views.UpdateView):
return {'network_id': network['id'],
'tenant_id': network['tenant_id'],
'name': network['name'],
'admin_state': network['admin_state_up'],
'shared': network['shared'],
'external': network['router__external']}
'admin_state': network['is_admin_state_up'],
'shared': network['is_shared'],
'external': network['is_router_external']}
class NetworkDetailsTabs(tabs.DetailTabsGroup):
@ -204,5 +204,5 @@ class DetailView(tabs.TabbedTableView):
context["actions"] = table.render_row_actions(network)
choices = networks_tables.DISPLAY_CHOICES
network.admin_state_label = (
filters.get_display_label(choices, network.admin_state))
filters.get_display_label(choices, network.is_admin_state_up))
return context

2
openstack_dashboard/dashboards/project/instances/tests.py
View File

@ -3075,7 +3075,7 @@ class ConsoleManagerTests(helpers.ResetImageAPIVersionMixin, helpers.TestCase):
def test_interface_attach_get(self):
server = self.servers.first()
tenant_networks = [net for net in self.networks.list()
if not net['router:external']]
if not net['is_router_external']]
net1 = tenant_networks[0]
self.mock_network_list_for_tenant.return_value = tenant_networks
ports = self.ports.list()

2
openstack_dashboard/dashboards/project/instances/utils.py
View File

@ -107,7 +107,7 @@ def network_field_data(request, include_empty_option=False, with_cidr=False,
_networks = []
for n in networks:
if not n['subnets']:
if not n['subnet_ids']:
continue
v = n.name_or_id
if with_cidr:

8
openstack_dashboard/dashboards/project/network_topology/tests.py
View File

@ -60,9 +60,9 @@ class NetworkTopologyTests(test.TestCase):
self.mock_server_list.return_value = [self.servers.list(), False]
tenant_networks = [net for net in self.networks.list()
if not net['router:external']]
if not net['is_router_external']]
external_networks = [net for net in self.networks.list()
if net['router:external']]
if net['is_router_external']]
self.mock_network_list_for_tenant.return_value = tenant_networks
# router1 : gateway port not in the port list
@ -117,7 +117,7 @@ class NetworkTopologyTests(test.TestCase):
'id': net.id,
'url': '/project/networks/%s/detail' % net.id,
'name': net.name,
'router:external': net.router__external,
'router:external': net.is_router_external,
'status': net.status.title(),
'original_status': net.status,
'subnets': [{
@ -130,7 +130,7 @@ class NetworkTopologyTests(test.TestCase):
'id': net.id,
'url': '/project/networks/%s/detail' % net.id,
'name': net.name,
'router:external': net.router__external,
'router:external': net.is_router_external,
'status': net.status.title(),
'allow_delete_subnet': True,
'original_status': net.status,

4
openstack_dashboard/dashboards/project/network_topology/views.py
View File

@ -284,7 +284,7 @@ class JSONView(View):
'status': self.trans.network[network.status],
'allow_delete_subnet': allow_delete_subnet,
'original_status': network.status,
'router:external': network['router:external']}
'router:external': network['is_router_external']}
self.add_resource_url('horizon:project:networks:subnets:detail',
obj['subnets'])
networks.append(obj)
@ -315,7 +315,7 @@ class JSONView(View):
'subnets': subnets,
'status': self.trans.network[publicnet.status],
'original_status': publicnet.status,
'router:external': publicnet['router:external']})
'router:external': publicnet['is_router_external']})
self.add_resource_url('horizon:project:networks:detail',
networks)

18
openstack_dashboard/dashboards/project/networks/subnets/tests.py
View File

@ -176,7 +176,7 @@ class NetworkSubnetTests(test.TestCase):
cidr=subnet.cidr,
ip_version=subnet.ip_version,
gateway_ip=subnet.gateway_ip,
enable_dhcp=subnet.enable_dhcp,
enable_dhcp=subnet.is_dhcp_enabled,
allocation_pools=subnet.allocation_pools)
@test.create_mocks({api.neutron: ('network_get',
@ -212,7 +212,7 @@ class NetworkSubnetTests(test.TestCase):
cidr=subnet.cidr,
ip_version=subnet.ip_version,
gateway_ip=subnet.gateway_ip,
enable_dhcp=subnet.enable_dhcp,
enable_dhcp=subnet.is_dhcp_enabled,
allocation_pools=subnet.allocation_pools,
dns_nameservers=subnet.dns_nameservers,
host_routes=subnet.host_routes)
@ -250,7 +250,7 @@ class NetworkSubnetTests(test.TestCase):
cidr=subnet.cidr,
ip_version=subnet.ip_version,
gateway_ip=None,
enable_dhcp=subnet.enable_dhcp,
enable_dhcp=subnet.is_dhcp_enabled,
allocation_pools=subnet.allocation_pools)
@test.create_mocks({api.neutron: ('network_get',)})
@ -311,7 +311,7 @@ class NetworkSubnetTests(test.TestCase):
cidr=subnet.cidr,
ip_version=subnet.ip_version,
gateway_ip=subnet.gateway_ip,
enable_dhcp=subnet.enable_dhcp)
enable_dhcp=subnet.is_dhcp_enabled)
@test.create_mocks({api.neutron: ('network_get',
'is_extension_supported',
@ -640,7 +640,7 @@ class NetworkSubnetTests(test.TestCase):
cidr=subnet.cidr,
ip_version=subnet.ip_version,
gateway_ip=subnet.gateway_ip,
enable_dhcp=subnet.enable_dhcp,
enable_dhcp=subnet.is_dhcp_enabled,
allocation_pools=subnet.allocation_pools)
@test.create_mocks({api.neutron: ('network_get',
@ -676,7 +676,7 @@ class NetworkSubnetTests(test.TestCase):
cidr=subnet.cidr,
ip_version=subnet.ip_version,
gateway_ip=subnet.gateway_ip,
enable_dhcp=subnet.enable_dhcp,
enable_dhcp=subnet.is_dhcp_enabled,
allocation_pools=subnet.allocation_pools,
ipv6_address_mode='slaac',
ipv6_ra_mode='slaac')
@ -711,7 +711,7 @@ class NetworkSubnetTests(test.TestCase):
test.IsHttpRequest(),
subnet.id,
name=subnet.name,
enable_dhcp=subnet.enable_dhcp,
enable_dhcp=subnet.is_dhcp_enabled,
dns_nameservers=[],
host_routes=[])
@ -748,7 +748,7 @@ class NetworkSubnetTests(test.TestCase):
subnet.id,
name=subnet.name,
gateway_ip=gateway_ip,
enable_dhcp=subnet.enable_dhcp,
enable_dhcp=subnet.is_dhcp_enabled,
dns_nameservers=[],
host_routes=[])
@ -784,7 +784,7 @@ class NetworkSubnetTests(test.TestCase):
subnet.id,
name=subnet.name,
gateway_ip=None,
enable_dhcp=subnet.enable_dhcp,
enable_dhcp=subnet.is_dhcp_enabled,
dns_nameservers=[],
host_routes=[])

2
openstack_dashboard/dashboards/project/networks/subnets/views.py
View File

@ -78,7 +78,7 @@ class UpdateView(workflows.WorkflowView):
initial['subnet_id'] = subnet['id']
initial['subnet_name'] = subnet['name']
for key in ('cidr', 'ip_version', 'enable_dhcp'):
for key in ('cidr', 'ip_version', 'is_dhcp_enabled'):
initial[key] = subnet[key]
initial['gateway_ip'] = subnet['gateway_ip'] or ''

8
openstack_dashboard/dashboards/project/networks/tables.py
View File

@ -146,7 +146,7 @@ class CreateSubnet(subnet_tables.SubnetPolicyTargetMixin, tables.LinkAction):
def get_subnets(network):
template_name = 'project/networks/_network_ips.html'
context = {"subnets": network.subnets}
context = {"subnets": network.to_dict()['subnets']}
return template.loader.render_to_string(template_name, context)
@ -193,13 +193,13 @@ class NetworksTable(tables.DataTable):
link=get_network_link)
subnets = tables.Column(get_subnets,
verbose_name=_("Subnets Associated"),)
shared = tables.Column("shared", verbose_name=_("Shared"),
shared = tables.Column("is_shared", verbose_name=_("Shared"),
filters=(filters.yesno, filters.capfirst))
external = tables.Column("router:external", verbose_name=_("External"),
external = tables.Column("is_router_external", verbose_name=_("External"),
filters=(filters.yesno, filters.capfirst))
status = tables.Column("status", verbose_name=_("Status"),
display_choices=STATUS_DISPLAY_CHOICES)
admin_state = tables.Column("admin_state",
admin_state = tables.Column("is_admin_state_up",
verbose_name=_("Admin State"),
display_choices=DISPLAY_CHOICES)
availability_zones = tables.Column(get_availability_zones,

2
openstack_dashboard/dashboards/project/networks/tabs.py
View File

@ -49,7 +49,7 @@ class OverviewTab(tabs.Tab):
filters.get_display_label(choices, network.status))
choices = project_tables.DISPLAY_CHOICES
network.admin_state_label = (
filters.get_display_label(choices, network.admin_state))
filters.get_display_label(choices, network.is_admin_state_up))
except Exception:
msg = _('Unable to retrieve details for network "%s".') \
% (network_id)

4
openstack_dashboard/dashboards/project/networks/templates/networks/_detail_overview.html
View File

@ -13,9 +13,9 @@
<dt title="{% trans 'Admin State' %}">{% trans "Admin State" %}</dt>
<dd>{{ network.admin_state_label|default:_("Unknown") }}</dd>
<dt title="{% trans 'Shared' %}">{% trans "Shared" %}</dt>
<dd>{{ network.shared|yesno|capfirst }}</dd>
<dd>{{ network.is_shared|yesno|capfirst }}</dd>
<dt title="{% trans 'External Network' %}">{% trans "External Network" %}</dt>
<dd>{{ network.router__external|yesno|capfirst }}</dd>
<dd>{{ network.is_router_external|yesno|capfirst }}</dd>
<dt title="{% trans 'MTU' %}">{% trans "MTU" %}</dt>
<dd>{{ network.mtu|default:_("Unknown") }}</dd>
{% if network.provider__network_type %}

87
openstack_dashboard/dashboards/project/networks/tests.py
View File

@ -52,7 +52,7 @@ def form_data_subnet(subnet,
data['gateway_ip'] = gateway_ip or ''
data['no_gateway'] = no_gateway or (gateway_ip is None)
data['enable_dhcp'] = get_value(enable_dhcp, subnet.enable_dhcp)
data['enable_dhcp'] = get_value(enable_dhcp, subnet.is_dhcp_enabled)
if data['ip_version'] == 6:
data['ipv6_modes'] = subnet.ipv6_modes
@ -72,7 +72,7 @@ def form_data_no_subnet():
'ip_version': 4,
'gateway_ip': '',
'no_gateway': False,
'enable_dhcp': True,
'is_dhcp_enabled': True,
'allocation_pools': '',
'dns_nameservers': '',
'host_routes': ''}
@ -103,13 +103,13 @@ class NetworkStubMixin(object):
all_networks = self.networks.list()
self.mock_network_list.side_effect = [
[network for network in all_networks
if network.get('shared') is True],
if network.get('is_shared') is True],
[network for network in all_networks
if network['tenant_id'] == self.tenant.id and
network.get('shared') is False],
network.get('is_shared') is False],
[network for network in all_networks
if network.get('router:external') is True and
network.get('shared') is False],
if network.get('is_router_external') is True and
network.get('is_shared') is False],
]
def _check_net_list(self):
@ -398,7 +398,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
def test_network_create_post(self):
network = self.networks.first()
params = {'name': network.name,
'admin_state_up': network.admin_state_up,
'admin_state_up': network.is_admin_state_up,
'shared': False}
self._stub_is_extension_supported({'network_availability_zone': False,
'subnet_allocation': True})
@ -406,7 +406,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
self.mock_network_create.return_value = network
form_data = {'net_name': network.name,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'shared': False,
'with_subnet': False}
form_data.update(form_data_no_subnet())
@ -429,7 +429,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
def test_network_create_post_with_az(self):
network = self.networks.first()
params = {'name': network.name,
'admin_state_up': network.admin_state_up,
'admin_state_up': network.is_admin_state_up,
'shared': False,
'availability_zone_hints': ['nova']}
@ -441,7 +441,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
self.mock_network_create.return_value = network
form_data = {'net_name': network.name,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'shared': False,
'with_subnet': False,
'az_hints': ['nova']}
@ -466,7 +466,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
def test_network_create_post_with_mtu(self):
network = self.networks.first()
params = {'name': network.name,
'admin_state_up': network.admin_state_up,
'admin_state_up': network.is_admin_state_up,
'shared': False,
'mtu': 1450}
self._stub_is_extension_supported({'network_availability_zone': False,
@ -475,7 +475,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
self.mock_network_create.return_value = network
form_data = {'net_name': network.name,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'shared': False,
'with_subnet': False,
'mtu': 1450}
@ -498,7 +498,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
def test_network_create_post_with_shared(self):
network = self.networks.first()
params = {'name': network.name,
'admin_state_up': network.admin_state_up,
'admin_state_up': network.is_admin_state_up,
'shared': True}
self._stub_is_extension_supported({'network_availability_zone': False,
'subnet_allocation': True})
@ -506,7 +506,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
self.mock_network_create.return_value = network
form_data = {'net_name': network.name,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'shared': True,
'with_subnet': False}
form_data.update(form_data_no_subnet())
@ -530,7 +530,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
network = self.networks.first()
subnet = self.subnets.first()
params = {'name': network.name,
'admin_state_up': network.admin_state_up,
'admin_state_up': network.is_admin_state_up,
'shared': False}
subnet_params = {'network_id': network.id,
'tenant_id': network.tenant_id,
@ -538,7 +538,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
'cidr': subnet.cidr,
'ip_version': subnet.ip_version,
'gateway_ip': subnet.gateway_ip,
'enable_dhcp': subnet.enable_dhcp}
'enable_dhcp': subnet.is_dhcp_enabled}
if not test_with_ipv6:
subnet.ip_version = 4
subnet_params['ip_version'] = subnet.ip_version
@ -550,7 +550,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
self.mock_subnet_create.return_value = subnet
form_data = {'net_name': network.name,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'shared': False,
'with_subnet': True}
form_data.update(form_data_subnet(subnet, allocation_pools=[]))
@ -579,14 +579,14 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
network = self.networks.first()
params = {'name': network.name,
'shared': False,
'admin_state_up': network.admin_state_up}
'admin_state_up': network.is_admin_state_up}
self._stub_is_extension_supported({'network_availability_zone': False,
'subnet_allocation': True})
self.mock_subnetpool_list.return_value = self.subnetpools.list()
self.mock_network_create.side_effect = self.exceptions.neutron
form_data = {'net_name': network.name,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'shared': False,
'with_subnet': False}
form_data.update(form_data_no_subnet())
@ -610,14 +610,14 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
subnet = self.subnets.first()
params = {'name': network.name,
'shared': False,
'admin_state_up': network.admin_state_up}
'admin_state_up': network.is_admin_state_up}
self._stub_is_extension_supported({'network_availability_zone': False,
'subnet_allocation': True})
self.mock_subnetpool_list.return_value = self.subnetpools.list()
self.mock_network_create.side_effect = self.exceptions.neutron
form_data = {'net_name': network.name,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'shared': False,
'with_subnet': True}
form_data.update(form_data_subnet(subnet, allocation_pools=[]))
@ -643,7 +643,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
subnet = self.subnets.first()
params = {'name': network.name,
'shared': False,
'admin_state_up': network.admin_state_up}
'admin_state_up': network.is_admin_state_up}
self._stub_is_extension_supported({'network_availability_zone': False,
'subnet_allocation': True})
self.mock_subnetpool_list.return_value = self.subnetpools.list()
@ -652,7 +652,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
self.mock_network_delete.return_value = None
form_data = {'net_name': network.name,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'shared': False,
'with_subnet': True}
form_data.update(form_data_subnet(subnet, allocation_pools=[]))
@ -675,7 +675,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
cidr=subnet.cidr,
ip_version=subnet.ip_version,
gateway_ip=subnet.gateway_ip,
enable_dhcp=subnet.enable_dhcp)
enable_dhcp=subnet.is_dhcp_enabled)
self.mock_network_delete.assert_called_once_with(
test.IsHttpRequest(), network.id)
@ -690,7 +690,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
self.mock_subnetpool_list.side_effect = self.exceptions.neutron
form_data = {'net_name': network.name,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'shared': False,
'with_subnet': True}
if test_with_snpool:
@ -726,7 +726,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
form_data = {'net_name': network.name,
'shared': False,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'with_subnet': True}
if test_with_subnetpool:
subnetpool = self.subnetpools.first()
@ -763,7 +763,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
form_data = {'net_name': network.name,
'shared': False,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'with_subnet': True}
if test_with_subnetpool:
subnetpool = self.subnetpools.first()
@ -804,7 +804,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
form_data = {'net_name': network.name,
'shared': False,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'with_subnet': True}
if test_with_subnetpool:
subnetpool = self.subnetpools.first()
@ -840,7 +840,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
cidr = '30.30.30.0/24'
gateway_ip = '30.30.30.1'
params = {'name': network.name,
'admin_state_up': network.admin_state_up,
'admin_state_up': network.is_admin_state_up,
'shared': False}
subnet_params = {'network_id': network.id,
'tenant_id': network.tenant_id,
@ -848,7 +848,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
'cidr': cidr,
'ip_version': subnet.ip_version,
'gateway_ip': gateway_ip,
'enable_dhcp': subnet.enable_dhcp}
'enable_dhcp': subnet.is_dhcp_enabled}
self._stub_is_extension_supported({'network_availability_zone': False,
'subnet_allocation': True})
@ -857,7 +857,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
self.mock_subnet_create.return_value = subnet
form_data = {'net_name': network.name,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'shared': False,
'with_subnet': True}
@ -893,7 +893,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
cidr = '2001:0DB8:0:CD30:123:4567:89AB:CDEF/60'
form_data = {'net_name': network.name,
'shared': False,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'with_subnet': True}
if test_with_subnetpool:
subnetpool = self.subnetpools.first()
@ -929,7 +929,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
gateway_ip = '2001:0DB8:0:CD30:123:4567:89AB:CDEF'
form_data = {'net_name': network.name,
'shared': False,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'with_subnet': True}
if test_with_subnetpool:
subnetpool = self.subnetpools.first()
@ -959,7 +959,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
network = self.networks.first()
subnet = self.subnets.first()
params = {'name': network.name,
'admin_state_up': network.admin_state_up,
'admin_state_up': network.is_admin_state_up,
'shared': False}
self._stub_is_extension_supported({'network_availability_zone': False,
@ -969,7 +969,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
self.mock_subnet_create.return_value = subnet
form_data = {'net_name': network.name,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'shared': False,
'with_subnet': True}
subnet_params = {'network_id': network.id,
@ -978,7 +978,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
'cidr': subnet.cidr,
'ip_version': subnet.ip_version,
'gateway_ip': None,
'enable_dhcp': subnet.enable_dhcp}
'enable_dhcp': subnet.is_dhcp_enabled}
form_data.update(form_data_subnet(subnet, allocation_pools=[],
no_gateway=True, gateway_ip="."))
url = reverse('horizon:project:networks:create')
@ -1031,7 +1031,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
form_data = {'network_id': network.id,
'shared': False,
'name': network.name,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'tenant_id': network.tenant_id}
url = reverse('horizon:project:networks:update', args=[network.id])
res = self.client.post(url, form_data)
@ -1040,7 +1040,8 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
self.mock_network_update.assert_called_once_with(
test.IsHttpRequest(), network.id, name=network.name,
admin_state_up=network.admin_state_up, shared=network.shared)
admin_state_up=network.is_admin_state_up,
shared=network.is_shared)
self.mock_network_get.assert_called_once_with(
test.IsHttpRequest(), network.id, expand_subnet=False)
@ -1054,7 +1055,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
form_data = {'network_id': network.id,
'shared': False,
'name': network.name,
'admin_state': network.admin_state_up,
'admin_state': network.is_admin_state_up,
'tenant_id': network.tenant_id}
url = reverse('horizon:project:networks:update', args=[network.id])
res = self.client.post(url, form_data)
@ -1065,14 +1066,14 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
test.IsHttpRequest(), network.id, expand_subnet=False)
self.mock_network_update.assert_called_once_with(
test.IsHttpRequest(), network.id, name=network.name,
admin_state_up=network.admin_state_up, shared=False)
admin_state_up=network.is_admin_state_up, shared=False)
@test.create_mocks({api.neutron: ('network_list',
'network_delete',
'is_extension_supported')})
def test_delete_network_no_subnet(self):
network = self.networks.first()
network.subnets = []
network.subnet_ids = []
self.mock_is_extension_supported.return_value = True
self._stub_net_list()
self.mock_network_delete.return_value = None
@ -1092,7 +1093,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
'is_extension_supported')})
def test_delete_network_with_subnet(self):
network = self.networks.first()
network.subnets = [subnet.id for subnet in network.subnets]
network.subnets = [subnet for subnet in network.subnet_ids]
self.mock_is_extension_supported.return_value = True
self._stub_net_list()
@ -1114,7 +1115,7 @@ class NetworkTests(test.TestCase, NetworkStubMixin):
'is_extension_supported')})
def test_delete_network_exception(self):
network = self.networks.first()
network.subnets = [subnet.id for subnet in network.subnets]
network.subnets = [subnet for subnet in network.subnet_ids]
self.mock_is_extension_supported.return_value = True
self._stub_net_list()

12
openstack_dashboard/dashboards/project/networks/views.py
View File

@ -43,9 +43,9 @@ from openstack_dashboard.dashboards.project.networks \
class IndexView(tables.PagedTableMixin, tables.DataTableView):
table_class = project_tables.NetworksTable
page_title = _("Networks")
FILTERS_MAPPING = {'shared': {_("yes"): True, _("no"): False},
'router:external': {_("yes"): True, _("no"): False},
'admin_state_up': {_("up"): True, _("down"): False}}
FILTERS_MAPPING = {'is_shared': {_("yes"): True, _("no"): False},
'is_router_external': {_("yes"): True, _("no"): False},
'is_admin_state_up': {_("up"): True, _("down"): False}}
def get_data(self):
try:
@ -127,8 +127,8 @@ class UpdateView(forms.ModalFormView):
return {'network_id': network['id'],
'tenant_id': network['tenant_id'],
'name': network['name'],
'admin_state': network['admin_state_up'],
'shared': network['shared']}
'admin_state': network['is_admin_state_up'],
'shared': network['is_shared']}
class DetailView(tabs.TabbedTableView):
@ -169,5 +169,5 @@ class DetailView(tabs.TabbedTableView):
filters.get_display_label(choices, network.status))
choices = project_tables.DISPLAY_CHOICES
network.admin_state_label = (
filters.get_display_label(choices, network.admin_state))
filters.get_display_label(choices, network.is_admin_state_up))
return context

2
openstack_dashboard/dashboards/project/routers/tests.py
View File

@ -60,7 +60,7 @@ class RouterMixin(object):
self._check_mock_external_network_get(router)
def _mock_external_network_list(self, count=1, alter_ids=False):
ext_nets = [n for n in self.networks.list() if n['router:external']]
ext_nets = [n for n in self.networks.list() if n['is_router_external']]
if alter_ids:
for ext_net in ext_nets:
ext_net.id += 'some extra garbage'

132
openstack_dashboard/test/test_data/neutron_data.py
View File

@ -14,6 +14,8 @@
import copy
from openstack.network.v2 import network as sdk_net
from openstack.network.v2 import subnet as sdk_subnet
from oslo_utils import uuidutils
from openstack_dashboard.api import base
@ -82,22 +84,30 @@ def data(TEST):
TEST.api_tp_trunks = utils.TestDataContainer()
TEST.api_tp_ports = utils.TestDataContainer()
# Data returned by SDK:
TEST.api_networks_sdk = list()
TEST.api_subnets_sdk = list()
# 1st network.
network_dict = {'admin_state_up': True,
network_dict = {'is_admin_state_up': True,
'id': '82288d84-e0a5-42ac-95be-e6af08727e42',
'name': 'net1',
'status': 'ACTIVE',
'subnets': ['e8abc972-eb0c-41f1-9edd-4bc6e3bcd8c9',
'41e53a49-442b-4307-9e9a-88967a6b6657'],
'subnet_ids': ['e8abc972-eb0c-41f1-9edd-4bc6e3bcd8c9',
'41e53a49-442b-4307-9e9a-88967a6b6657'],
'tenant_id': '1',
'is_router_external': False,
'router:external': False,
'is_shared': False,
'shared': False}
subnet_dict = {'allocation_pools': [{'end': '10.0.0.254',
'start': '10.0.0.2'}],
'dns_nameservers': [],
'host_routes': [],
'cidr': '10.0.0.0/24',
'enable_dhcp': True,
'is_dhcp_enabled': True,
'gateway_ip': '10.0.0.1',
'id': network_dict['subnets'][0],
'ip_version': 4,
@ -110,7 +120,7 @@ def data(TEST):
'dns_nameservers': [],
'host_routes': [],
'cidr': 'fdb6:b88a:488e::/64',
'enable_dhcp': True,
'is_dhcp_enabled': True,
'gateway_ip': 'fdb6:b88a:488e::1',
'id': network_dict['subnets'][1],
'ip_version': 6,
@ -122,6 +132,9 @@ def data(TEST):
}
TEST.api_networks.add(network_dict)
TEST.api_networks_sdk.append(sdk_net.Network(**network_dict))
TEST.api_subnets_sdk.append(sdk_subnet.Subnet(**subnet_dict))
TEST.api_subnets_sdk.append(sdk_subnet.Subnet(**subnetv6_dict))
TEST.api_subnets.add(subnet_dict)
TEST.api_subnets.add(subnetv6_dict)
@ -245,14 +258,17 @@ def data(TEST):
TEST.ports.add(neutron.Port(port_dict))
# 2nd network.
network_dict = {'admin_state_up': True,
network_dict = {'is_admin_state_up': True,
'id': '72c3ab6c-c80f-4341-9dc5-210fa31ac6c2',
'name': 'net2',
'status': 'ACTIVE',
'subnets': ['3f7c5d79-ee55-47b0-9213-8e669fb03009'],
'subnet_ids': ['3f7c5d79-ee55-47b0-9213-8e669fb03009'],
'tenant_id': '2',
'is_router_external': False,
'router:external': False,
'shared': True}
'shared': True,
'is_shared': True}
subnet_dict = {'allocation_pools': [{'end': '172.16.88.254',
'start': '172.16.88.2'}],
'dns_nameservers': ['10.56.1.20', '10.56.1.21'],
@ -261,7 +277,7 @@ def data(TEST):
{'destination': '192.168.21.0/24',
'nexthop': '172.16.88.252'}],
'cidr': '172.16.88.0/24',
'enable_dhcp': True,
'is_dhcp_enabled': True,
'gateway_ip': '172.16.88.1',
'id': '3f7c5d79-ee55-47b0-9213-8e669fb03009',
'ip_version': 4,
@ -269,7 +285,9 @@ def data(TEST):
'network_id': network_dict['id'],
'tenant_id': network_dict['tenant_id']}
TEST.api_networks_sdk.append(sdk_net.Network(**network_dict))
TEST.api_networks.add(network_dict)
TEST.api_subnets_sdk.append(sdk_subnet.Subnet(**subnet_dict))
TEST.api_subnets.add(subnet_dict)
network = copy.deepcopy(network_dict)
@ -302,20 +320,23 @@ def data(TEST):
TEST.ports.add(neutron.Port(port_dict))
# External not shared network.
network_dict = {'admin_state_up': True,
network_dict = {'is_admin_state_up': True,
'id': '9b466b94-213a-4cda-badf-72c102a874da',
'name': 'ext_net',
'status': 'ACTIVE',
'subnets': ['d6bdc71c-7566-4d32-b3ff-36441ce746e8'],
'subnet_ids': ['d6bdc71c-7566-4d32-b3ff-36441ce746e8'],
'tenant_id': '3',
'is_router_external': True,
'router:external': True,
'shared': False}
'shared': False,
'is_shared': False}
subnet_dict = {'allocation_pools': [{'start': '172.24.4.226.',
'end': '172.24.4.238'}],
'dns_nameservers': [],
'host_routes': [],
'cidr': '172.24.4.0/28',
'enable_dhcp': False,
'is_dhcp_enabled': False,
'gateway_ip': '172.24.4.225',
'id': 'd6bdc71c-7566-4d32-b3ff-36441ce746e8',
'ip_version': 4,
@ -324,7 +345,9 @@ def data(TEST):
'tenant_id': network_dict['tenant_id']}
ext_net = network_dict
TEST.api_networks_sdk.append(sdk_net.Network(**network_dict))
TEST.api_networks.add(network_dict)
TEST.api_subnets_sdk.append(sdk_subnet.Subnet(**subnet_dict))
TEST.api_subnets.add(subnet_dict)
network = copy.deepcopy(network_dict)
@ -335,20 +358,23 @@ def data(TEST):
# External shared network.
network_dict = {'admin_state_up': True,
network_dict = {'is_admin_state_up': True,
'id': 'ed351877-4f7b-4672-8164-20a09e4873d3',
'name': 'ext_net_shared',
'status': 'ACTIVE',
'subnets': ['5c59f875-f242-4df2-96e6-7dcc09d6dfc8'],
'subnet_ids': ['5c59f875-f242-4df2-96e6-7dcc09d6dfc8'],
'tenant_id': '4',
'is_router_external': True,
'router:external': True,
'shared': True}
'shared': True,
'is_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,
'is_dhcp_enabled': False,
'gateway_ip': '172.24.14.225',
'id': '5c59f875-f242-4df2-96e6-7dcc09d6dfc8',
'ip_version': 4,
@ -356,7 +382,9 @@ def data(TEST):
'network_id': network_dict['id'],
'tenant_id': network_dict['tenant_id']}
TEST.api_networks_sdk.append(sdk_net.Network(**network_dict))
TEST.api_networks.add(network_dict)
TEST.api_subnets_sdk.append(sdk_subnet.Subnet(**subnet_dict))
TEST.api_subnets.add(subnet_dict)
network = copy.deepcopy(network_dict)
@ -366,20 +394,23 @@ def data(TEST):
TEST.subnets.add(subnet)
# tenant external shared network
network_dict = {'admin_state_up': True,
network_dict = {'is_admin_state_up': True,
'id': '650de90f-d77f-4863-ae98-39e97ad3ea7a',
'name': 'ext_net_shared_tenant1',
'status': 'ACTIVE',
'subnets': ['d0a5bc19-16f0-45cc-a187-0d1bb36de4c6'],
'subnet_ids': ['d0a5bc19-16f0-45cc-a187-0d1bb36de4c6'],
'tenant_id': '1',
'is_router_external': True,
'router:external': True,
'shared': True}
'shared': True,
'is_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,
'is_dhcp_enabled': False,
'gateway_ip': '172.34.14.225',
'id': 'd0a5bc19-16f0-45cc-a187-0d1bb36de4c6',
'ip_version': 4,
@ -387,7 +418,9 @@ def data(TEST):
'network_id': network_dict['id'],
'tenant_id': network_dict['tenant_id']}
TEST.api_networks_sdk.append(sdk_net.Network(**network_dict))
TEST.api_networks.add(network_dict)
TEST.api_subnets_sdk.append(sdk_subnet.Subnet(**subnet_dict))
TEST.api_subnets.add(subnet_dict)
network = copy.deepcopy(network_dict)
@ -397,20 +430,23 @@ def data(TEST):
TEST.subnets.add(subnet)
# tenant external non-shared network
network_dict = {'admin_state_up': True,
network_dict = {'is_admin_state_up': True,
'id': '19c3e662-1635-4876-be41-dbfdef0edd17',
'name': 'ext_net_tenant1',
'status': 'ACTIVE',
'subnets': ['5ba8895c-0b3b-482d-9e42-ce389e1e1fa6'],
'subnet_ids': ['5ba8895c-0b3b-482d-9e42-ce389e1e1fa6'],
'tenant_id': '1',
'is_router_external': True,
'router:external': True,
'shared': False}
'shared': False,
'is_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,
'is_dhcp_enabled': False,
'gateway_ip': '172.44.14.225',
'id': '5ba8895c-0b3b-482d-9e42-ce389e1e1fa6',
'ip_version': 4,
@ -418,7 +454,9 @@ def data(TEST):
'network_id': network_dict['id'],
'tenant_id': network_dict['tenant_id']}
TEST.api_networks_sdk.append(sdk_net.Network(**network_dict))
TEST.api_networks.add(network_dict)
TEST.api_subnets_sdk.append(sdk_subnet.Subnet(**subnet_dict))
TEST.api_subnets.add(subnet_dict)
network = copy.deepcopy(network_dict)
@ -428,20 +466,23 @@ def data(TEST):
TEST.subnets.add(subnet)
# tenant non-external shared network
network_dict = {'admin_state_up': True,
network_dict = {'is_admin_state_up': True,
'id': 'fd581273-2601-4057-9c22-1be38f44884e',
'name': 'shr_net_tenant1',
'status': 'ACTIVE',
'subnets': ['d2668892-bc32-4c89-9c63-961920a831d3'],
'subnet_ids': ['d2668892-bc32-4c89-9c63-961920a831d3'],
'tenant_id': '1',
'is_router_external': False,
'router:external': False,
'shared': True}
'shared': True,
'is_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,
'is_dhcp_enabled': False,
'gateway_ip': '172.54.14.225',
'id': 'd2668892-bc32-4c89-9c63-961920a831d3',
'ip_version': 4,
@ -449,7 +490,9 @@ def data(TEST):
'network_id': network_dict['id'],
'tenant_id': network_dict['tenant_id']}
TEST.api_networks_sdk.append(sdk_net.Network(**network_dict))
TEST.api_networks.add(network_dict)
TEST.api_subnets_sdk.append(sdk_subnet.Subnet(**subnet_dict))
TEST.api_subnets.add(subnet_dict)
network = copy.deepcopy(network_dict)
@ -459,20 +502,23 @@ def data(TEST):
TEST.subnets.add(subnet)
# non-tenant non-external non-shared network
network_dict = {'admin_state_up': True,
network_dict = {'is_admin_state_up': True,
'id': '7377e545-1527-4ce1-869e-caca192bc049',
'name': 'net_tenant20',
'status': 'ACTIVE',
'subnets': ['c2bbd65e-0c0f-4ab9-8723-2dd102104f3d'],
'subnet_ids': ['c2bbd65e-0c0f-4ab9-8723-2dd102104f3d'],
'tenant_id': '20',
'is_router_external': False,
'router:external': False,
'shared': False}
'shared': False,
'is_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,
'is_dhcp_enabled': False,
'gateway_ip': '172.64.14.225',
'id': 'c2bbd65e-0c0f-4ab9-8723-2dd102104f3d',
'ip_version': 4,
@ -480,7 +526,9 @@ def data(TEST):
'network_id': network_dict['id'],
'tenant_id': network_dict['tenant_id']}
TEST.api_networks_sdk.append(sdk_net.Network(**network_dict))
TEST.api_networks.add(network_dict)
TEST.api_subnets_sdk.append(sdk_subnet.Subnet(**subnet_dict))
TEST.api_subnets.add(subnet_dict)
network = copy.deepcopy(network_dict)
@ -490,20 +538,23 @@ def data(TEST):
TEST.subnets.add(subnet)
# 1st v6 network.
network_dict = {'admin_state_up': True,
network_dict = {'is_admin_state_up': True,
'id': '96688ea1-ffa5-78ec-22ca-33aaabfaf775',
'name': 'v6_net1',
'status': 'ACTIVE',
'subnets': ['88ddd443-4377-ab1f-87dd-4bc4a662dbb6'],
'subnet_ids': ['88ddd443-4377-ab1f-87dd-4bc4a662dbb6'],
'tenant_id': '1',
'is_router_external': False,
'router:external': False,
'shared': False}
'shared': False,
'is_shared': False}
subnet_dict = {'allocation_pools': [{'end': 'ff09::ff',
'start': 'ff09::02'}],
'dns_nameservers': [],
'host_routes': [],
'cidr': 'ff09::/64',
'enable_dhcp': True,
'is_dhcp_enabled': True,
'gateway_ip': 'ff09::1',
'id': network_dict['subnets'][0],
'ip_version': 6,
@ -512,7 +563,9 @@ def data(TEST):
'tenant_id': network_dict['tenant_id'],
'ipv6_modes': 'none/none'}
TEST.api_networks_sdk.append(sdk_net.Network(**network_dict))
TEST.api_networks.add(network_dict)
TEST.api_subnets_sdk.append(sdk_subnet.Subnet(**subnet_dict))
TEST.api_subnets.add(subnet_dict)
network = copy.deepcopy(network_dict)
@ -522,20 +575,23 @@ def data(TEST):
TEST.subnets.add(subnet)
# 2nd v6 network - slaac.
network_dict = {'admin_state_up': True,
network_dict = {'is_admin_state_up': True,
'id': 'c62e4bb3-296a-4cd1-8f6b-aaa7a0092326',
'name': 'v6_net2',
'status': 'ACTIVE',
'subnets': ['5d736a21-0036-4779-8f8b-eed5f98077ec'],
'subnet_ids': ['5d736a21-0036-4779-8f8b-eed5f98077ec'],
'tenant_id': '1',
'is_router_external': False,
'router:external': False,
'shared': False}
'shared': False,
'is_shared': False}
subnet_dict = {'allocation_pools': [{'end': 'ff09::ff',
'start': 'ff09::02'}],
'dns_nameservers': [],
'host_routes': [],
'cidr': 'ff09::/64',
'enable_dhcp': True,
'is_dhcp_enabled': True,
'gateway_ip': 'ff09::1',
'id': network_dict['subnets'][0],
'ip_version': 6,
@ -544,7 +600,9 @@ def data(TEST):
'tenant_id': network_dict['tenant_id'],
'ipv6_modes': 'slaac/slaac'}
TEST.api_networks_sdk.append(sdk_net.Network(**network_dict))
TEST.api_networks.add(network_dict)
TEST.api_subnets_sdk.append(sdk_subnet.Subnet(**subnet_dict))
TEST.api_subnets.add(subnet_dict)
network = copy.deepcopy(network_dict)
@ -1292,6 +1350,8 @@ source_nets_pagination1 = sorted([
'subnets': [],
'tenant_id': '1',
'router:external': False,
'is_router_external': False,
'is_shared': False,
'shared': False}) for i in range(0, 58)
] + [
neutron.Network({
@ -1302,6 +1362,8 @@ source_nets_pagination1 = sorted([
'subnets': [],
'tenant_id': '2',
'router:external': True,
'is_router_external': True,
'is_shared': False,
'shared': False})
] + [
neutron.Network({
@ -1312,6 +1374,8 @@ source_nets_pagination1 = sorted([
'subnets': [],
'tenant_id': '3',
'router:external': False,
'is_router_external': False,
'is_shared': True,
'shared': True})
], key=lambda net: net['id'])
@ -1326,6 +1390,8 @@ source_nets_pagination2 = sorted([
'subnets': [],
'tenant_id': '2',
'router:external': True,
'is_router_external': True,
'is_shared': False,
'shared': False}) for i in range(0, 25)
] + [
neutron.Network({
@ -1336,6 +1402,8 @@ source_nets_pagination2 = sorted([
'subnets': [],
'tenant_id': '3',
'router:external': False,
'is_router_external': False,
'is_shared': True,
'shared': True}) for i in range(0, 25)
] + [
neutron.Network({
@ -1346,6 +1414,8 @@ source_nets_pagination2 = sorted([
'subnets': [],
'tenant_id': '1',
'router:external': False,
'is_router_external': False,
'is_shared': False,
'shared': False}) for i in range(0, 10)
], key=lambda net: net['id'])
@ -1360,5 +1430,7 @@ source_nets_pagination3 = sorted([
'subnets': [],
'tenant_id': '1',
'router:external': False,
'is_router_external': False,
'is_shared': False,
'shared': False}) for i in range(0, 5)
], key=lambda net: net['id'])

6
openstack_dashboard/test/unit/api/rest/test_network.py
View File

@ -51,9 +51,9 @@ class RestNetworkApiFloatingIpTests(test.RestAPITestCase):
@test.create_mocks({api.neutron: ['floating_ip_pools_list']})
def test_floating_ip_pool_list(self):
pools = [api.neutron.FloatingIpPool(n)
for n in self.api_networks.list()
if n['router:external']]
pools = [api.neutron.FloatingIpPool(n.to_dict())
for n in self.api_networks_sdk
if n['is_router_external']]
request = self.mock_rest_request()
self.mock_floating_ip_pools_list.return_value = pools

18
openstack_dashboard/test/unit/api/test_network.py
View File

@ -28,6 +28,8 @@ class NetworkApiNeutronTests(test.APIMockTestCase):
super().setUp()
neutronclient = mock.patch.object(api.neutron, 'neutronclient').start()
self.qclient = neutronclient.return_value
network_client = mock.patch.object(api.neutron, 'networkclient').start()
self.sdk_net_client = network_client.return_value
def _get_expected_addresses(self, server, no_fip_expected=True):
server_ports = self.ports.filter(device_id=server.id)
@ -76,7 +78,7 @@ class NetworkApiNeutronTests(test.APIMockTestCase):
assoc_fips = [fip for fip in self.api_floating_ips.list()
if fip['port_id'] in server_port_ids]
server_network_ids = [p['network_id'] for p in server_ports]
server_networks = [net for net in self.api_networks.list()
server_networks = [net for net in self.api_networks_sdk
if net['id'] in server_network_ids]
list_ports_retvals = [{'ports': server_ports}]
@ -85,9 +87,8 @@ class NetworkApiNeutronTests(test.APIMockTestCase):
self.qclient.list_floatingips.return_value = {'floatingips':
assoc_fips}
list_ports_retvals.append({'ports': self.api_ports.list()})
self.qclient.list_networks.return_value = {'networks': server_networks}
self.qclient.list_subnets.return_value = {'subnets':
self.api_subnets.list()}
self.sdk_net_client.networks.return_value = server_networks
self.sdk_net_client.subnets.return_value = self.api_subnets_sdk
api.network.servers_update_addresses(self.request, servers)
@ -131,9 +132,12 @@ class NetworkApiNeutronTests(test.APIMockTestCase):
else:
self.assertEqual(0, self.qclient.list_floatingips.call_count)
self.qclient.list_ports.assert_has_calls(expected_list_ports)
self.qclient.list_networks.assert_called_once_with(
id=frozenset(server_network_ids))
self.qclient.list_subnets.assert_called_once_with()
nets_calls = []
for server_net_id in server_network_ids:
nets_calls.append(mock.call(id=server_net_id))
self.sdk_net_client.networks.assert_has_calls(nets_calls,
any_order=True)
self.sdk_net_client.subnets.assert_called_once_with()
@override_settings(OPENSTACK_NEUTRON_NETWORK={'enable_router': True})
def test_servers_update_addresses(self):

275
openstack_dashboard/test/unit/api/test_neutron.py
View File

@ -16,6 +16,7 @@ from unittest import mock
import netaddr
from neutronclient.common import exceptions as neutron_exc
from openstack import exceptions as sdk_exceptions
from oslo_utils import uuidutils
from django.test.utils import override_settings
@ -28,25 +29,26 @@ from openstack_dashboard.test.test_data import neutron_data
class NeutronApiTests(test.APIMockTestCase):
@mock.patch.object(api.neutron, 'neutronclient')
@mock.patch.object(api.neutron, 'networkclient')
def test_network_list(self, mock_neutronclient):
networks = {'networks': self.api_networks.list()}
subnets = {'subnets': self.api_subnets.list()}
networks = self.api_networks_sdk
subnets = self.api_subnets_sdk
neutronclient = mock_neutronclient.return_value
neutronclient.list_networks.return_value = networks
neutronclient.list_subnets.return_value = subnets
neutronclient.networks.return_value = networks
neutronclient.subnets.return_value = subnets
ret_val = api.neutron.network_list(self.request)
for n in ret_val:
self.assertIsInstance(n, api.neutron.Network)
neutronclient.list_networks.assert_called_once_with()
neutronclient.list_subnets.assert_called_once_with()
neutronclient.networks.assert_called_once_with()
neutronclient.subnets.assert_called_once_with()
@override_settings(OPENSTACK_NEUTRON_NETWORK={
'enable_auto_allocated_network': True})
@test.create_mocks({api.neutron: ('network_list',
'subnet_list')})
'subnet_list',
'list_extensions')})
def _test_network_list_for_tenant(
self, include_external, filter_params, should_called,
expected_networks, source_networks=None, **extra_kwargs):
@ -126,6 +128,7 @@ class NeutronApiTests(test.APIMockTestCase):
mock.call(test.IsHttpRequest(),
tenant_id=tenant_id, **params))
self.mock_network_list.side_effect = return_values
self.mock_list_extensions.side_effect = {'extensions': []}
extra_kwargs.update(filter_params)
ret_val = api.neutron.network_list_for_tenant(
@ -198,52 +201,53 @@ class NeutronApiTests(test.APIMockTestCase):
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']]
shared_networks = [n for n in all_networks if n['is_shared']]
external_networks = [n for n in all_networks if n['is_router_external']]
self.assertTrue(tenant_networks)
self.assertTrue(shared_networks)
self.assertTrue(external_networks)
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)]
expected_networks = [n for n in self.api_networks_sdk
if (n['tenant_id'] == '1' or
n['is_shared'] is True)]
self._test_network_list_for_tenant(
include_external=False, filter_params=None,
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()
expected_networks = [n for n in self.api_networks_sdk
if (n['tenant_id'] == '1' or
n['shared'] is True or
n['router:external'] is True)]
n['is_shared'] is True or
n['is_router_external'] is True)]
self._test_network_list_for_tenant(
include_external=True, filter_params=None,
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()
expected_networks = [n for n in self.api_networks_sdk
if (n['tenant_id'] == '1' and
n['shared'] is False)]
n['is_shared'] is False)]
self._test_network_list_for_tenant(
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]
expected_networks = [n for n in self.api_networks_sdk
if n['is_shared'] is True]
self._test_network_list_for_tenant(
include_external=True, filter_params={'shared': True},
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()
expected_networks = [n for n in self.api_networks_sdk
if ((n['tenant_id'] == '1' or
n['shared'] is True) and
n['router:external'] is False)]
n['is_shared'] is True) and
n['is_router_external'] is False)]
self._test_network_list_for_tenant(
include_external=False, filter_params={'router:external': False},
should_called=['non_shared', 'shared'],
@ -252,8 +256,8 @@ class NeutronApiTests(test.APIMockTestCase):
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)]
n['is_shared'] is True) and
n['is_router_external'] is True)]
self._test_network_list_for_tenant(
include_external=False, filter_params={'router:external': True},
should_called=['non_shared', 'shared'],
@ -262,8 +266,8 @@ class NeutronApiTests(test.APIMockTestCase):
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)]
n['is_shared'] is True) and
n['is_router_external'] is False)]
self._test_network_list_for_tenant(
include_external=True, filter_params={'router:external': False},
should_called=['non_shared', 'shared'],
@ -271,7 +275,7 @@ class NeutronApiTests(test.APIMockTestCase):
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]
if n['is_router_external'] is True]
self._test_network_list_for_tenant(
include_external=True, filter_params={'router:external': True},
should_called=['external', 'shared', 'non_shared'],
@ -281,8 +285,8 @@ class NeutronApiTests(test.APIMockTestCase):
# 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)]
if (n['is_shared'] is True and
n['is_router_external'] is True)]
self._test_network_list_for_tenant(
include_external=True,
filter_params={'router:external': True, 'shared': True},
@ -293,8 +297,8 @@ class NeutronApiTests(test.APIMockTestCase):
# 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)]
if (n['is_router_external'] is True and
n['is_shared'] is False)]
self._test_network_list_for_tenant(
include_external=True,
filter_params={'router:external': True, 'shared': False,
@ -305,8 +309,8 @@ class NeutronApiTests(test.APIMockTestCase):
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)]
n['is_shared'] is True or
n['is_router_external'] is True)]
self._test_network_list_for_tenant(
include_external=True, filter_params=None,
should_called=['non_shared', 'shared', 'external'],
@ -903,68 +907,68 @@ class NeutronApiTests(test.APIMockTestCase):
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()}
subnet = {'subnet': self.api_subnets.first()}
subnetv6 = {'subnet': self.api_subnets.list()[1]}
network_id = self.api_networks.first()['id']
subnet_id = self.api_networks.first()['subnets'][0]
subnetv6_id = self.api_networks.first()['subnets'][1]
@mock.patch.object(api.neutron, 'networkclient')
def test_network_get(self, mock_networkclient):
network = self.api_networks_sdk[0]
subnet = self.api_subnets_sdk[0]
subnetv6 = self.api_subnets_sdk[1]
network_id = self.api_networks_sdk[0]['id']
subnet_id = self.api_networks_sdk[0]['subnets'][0]
subnetv6_id = self.api_networks_sdk[0]['subnets'][1]
neutronclient = mock_neutronclient.return_value
neutronclient.show_network.return_value = network
neutronclient.show_subnet.side_effect = [subnet, subnetv6]
neutronclient = mock_networkclient.return_value
neutronclient.get_network.return_value = network
neutronclient.get_subnet.side_effect = [subnet, subnetv6]
ret_val = api.neutron.network_get(self.request, network_id)
self.assertIsInstance(ret_val, api.neutron.Network)
self.assertEqual(2, len(ret_val['subnets']))
self.assertIsInstance(ret_val['subnets'][0], api.neutron.Subnet)
neutronclient.show_network.assert_called_once_with(network_id)
neutronclient.show_subnet.assert_has_calls([
neutronclient.get_network.assert_called_once_with(network_id)
neutronclient.get_subnet.assert_has_calls([
mock.call(subnet_id),
mock.call(subnetv6_id),
])
@mock.patch.object(api.neutron, 'neutronclient')
def test_network_get_with_subnet_get_notfound(self, mock_neutronclient):
network = {'network': self.api_networks.first()}
network_id = self.api_networks.first()['id']
subnet_id = self.api_networks.first()['subnets'][0]
@mock.patch.object(api.neutron, 'networkclient')
def test_network_get_with_subnet_get_notfound(self, mock_networkclient):
network = self.api_networks_sdk[0]
network_id = self.api_networks_sdk[0]['id']
subnet_id = self.api_networks_sdk[0]['subnet_ids'][0]
neutronclient = mock_neutronclient.return_value
neutronclient.show_network.return_value = network
neutronclient.show_subnet.side_effect = neutron_exc.NotFound
neutronclient = mock_networkclient.return_value
neutronclient.get_network.return_value = network
neutronclient.get_subnet.side_effect = sdk_exceptions.ResourceNotFound
ret_val = api.neutron.network_get(self.request, network_id)
self.assertIsInstance(ret_val, api.neutron.Network)
self.assertEqual(2, len(ret_val['subnets']))
self.assertNotIsInstance(ret_val['subnets'][0], api.neutron.Subnet)
self.assertIsInstance(ret_val['subnets'][0], str)
neutronclient.show_network.assert_called_once_with(network_id)
neutronclient.show_subnet.assert_called_once_with(subnet_id)
self.assertEqual(2, len(ret_val['subnet_ids']))
self.assertNotIsInstance(ret_val['subnet_ids'][0], api.neutron.Subnet)
self.assertIsInstance(ret_val['subnet_ids'][0], str)
neutronclient.get_network.assert_called_once_with(network_id)
neutronclient.get_subnet.assert_called_once_with(subnet_id)
@mock.patch.object(api.neutron, 'neutronclient')
def test_network_create(self, mock_neutronclient):
network = {'network': self.api_networks.first()}
form_data = {'network': {'name': 'net1',
'tenant_id': self.request.user.project_id}}
neutronclient = mock_neutronclient.return_value
@mock.patch.object(api.neutron, 'networkclient')
def test_network_create(self, mock_networkclient):
network = self.api_networks_sdk[0]
form_data = {'name': 'net1',
'tenant_id': self.request.user.project_id}
neutronclient = mock_networkclient.return_value
neutronclient.create_network.return_value = network
ret_val = api.neutron.network_create(self.request, name='net1')
self.assertIsInstance(ret_val, api.neutron.Network)
neutronclient.create_network.assert_called_once_with(body=form_data)
neutronclient.create_network.assert_called_once_with(**form_data)
@mock.patch.object(api.neutron, 'neutronclient')
def test_network_update(self, mock_neutronclient):
network = {'network': self.api_networks.first()}
network_id = self.api_networks.first()['id']
@mock.patch.object(api.neutron, 'networkclient')
def test_network_update(self, mock_networkclient):
network = self.api_networks_sdk[0]
network_id = self.api_networks_sdk[0]['id']
neutronclient = mock_neutronclient.return_value
form_data = {'network': {'name': 'net1'}}
neutronclient = mock_networkclient.return_value
form_data = {'name': 'net1'}
neutronclient.update_network.return_value = network
ret_val = api.neutron.network_update(self.request, network_id,
@ -972,13 +976,13 @@ class NeutronApiTests(test.APIMockTestCase):
self.assertIsInstance(ret_val, api.neutron.Network)
neutronclient.update_network.assert_called_once_with(network_id,
body=form_data)
**form_data)
@mock.patch.object(api.neutron, 'neutronclient')
def test_network_delete(self, mock_neutronclient):
network_id = self.api_networks.first()['id']
@mock.patch.object(api.neutron, 'networkclient')
def test_network_delete(self, mock_networkclient):
network_id = self.api_networks_sdk[0]['id']
neutronclient = mock_neutronclient.return_value
neutronclient = mock_networkclient.return_value
neutronclient.delete_network.return_value = None
api.neutron.network_delete(self.request, network_id)
@ -1018,35 +1022,35 @@ class NeutronApiTests(test.APIMockTestCase):
neutronclient.show_network_ip_availability.assert_called_once_with(
network)
@mock.patch.object(api.neutron, 'neutronclient')
def test_subnet_list(self, mock_neutronclient):
subnets = {'subnets': self.api_subnets.list()}
@mock.patch.object(api.neutron, 'networkclient')
def test_subnet_list(self, mock_networkclient):
subnets = self.api_subnets_sdk
neutronclient = mock_neutronclient.return_value
neutronclient.list_subnets.return_value = subnets
neutronclient = mock_networkclient.return_value
neutronclient.subnets.return_value = subnets
ret_val = api.neutron.subnet_list(self.request)
for n in ret_val:
self.assertIsInstance(n, api.neutron.Subnet)
neutronclient.list_subnets.assert_called_once_with()
neutronclient.subnets.assert_called_once_with()
@mock.patch.object(api.neutron, 'neutronclient')
def test_subnet_get(self, mock_neutronclient):
subnet = {'subnet': self.api_subnets.first()}
subnet_id = self.api_subnets.first()['id']
@mock.patch.object(api.neutron, 'networkclient')
def test_subnet_get(self, mock_networkclient):
subnet = self.api_subnets_sdk[0]
subnet_id = self.api_subnets_sdk[0]['id']
neutronclient = mock_neutronclient.return_value
neutronclient.show_subnet.return_value = subnet
neutronclient = mock_networkclient.return_value
neutronclient.get_subnet.return_value = subnet
ret_val = api.neutron.subnet_get(self.request, subnet_id)
self.assertIsInstance(ret_val, api.neutron.Subnet)
neutronclient.show_subnet.assert_called_once_with(subnet_id)
neutronclient.get_subnet.assert_called_once_with(subnet_id)
@mock.patch.object(api.neutron, 'neutronclient')
def test_subnet_create(self, mock_neutronclient):
subnet_data = self.api_subnets.first()
@mock.patch.object(api.neutron, 'networkclient')
def test_subnet_create(self, mock_networkclient):
subnet_data = self.api_subnets_sdk[0]
params = {'network_id': subnet_data['network_id'],
'tenant_id': subnet_data['tenant_id'],
'name': subnet_data['name'],
@ -1054,36 +1058,35 @@ class NeutronApiTests(test.APIMockTestCase):
'ip_version': subnet_data['ip_version'],
'gateway_ip': subnet_data['gateway_ip']}
neutronclient = mock_neutronclient.return_value
neutronclient.create_subnet.return_value = {'subnet': subnet_data}
neutronclient = mock_networkclient.return_value
neutronclient.create_subnet.return_value = subnet_data
ret_val = api.neutron.subnet_create(self.request, **params)
self.assertIsInstance(ret_val, api.neutron.Subnet)
neutronclient.create_subnet.assert_called_once_with(
body={'subnet': params})
neutronclient.create_subnet.assert_called_once_with(**params)
@mock.patch.object(api.neutron, 'neutronclient')
def test_subnet_update(self, mock_neutronclient):
subnet_data = self.api_subnets.first()
@mock.patch.object(api.neutron, 'networkclient')
def test_subnet_update(self, mock_networkclient):
subnet_data = self.api_subnets_sdk[0]
subnet_id = subnet_data['id']
params = {'name': subnet_data['name'],
'gateway_ip': subnet_data['gateway_ip']}
neutronclient = mock_neutronclient.return_value
neutronclient.update_subnet.return_value = {'subnet': subnet_data}
neutronclient = mock_networkclient.return_value
neutronclient.update_subnet.return_value = subnet_data
ret_val = api.neutron.subnet_update(self.request, subnet_id, **params)
self.assertIsInstance(ret_val, api.neutron.Subnet)
neutronclient.update_subnet.assert_called_once_with(
subnet_id, body={'subnet': params})
subnet_id, **params)
@mock.patch.object(api.neutron, 'neutronclient')
def test_subnet_delete(self, mock_neutronclient):
subnet_id = self.api_subnets.first()['id']
@mock.patch.object(api.neutron, 'networkclient')
def test_subnet_delete(self, mock_networkclient):
subnet_id = self.api_subnets_sdk[0]['id']
neutronclient = mock_neutronclient.return_value
neutronclient = mock_networkclient.return_value
neutronclient.delete_subnet.return_value = None
api.neutron.subnet_delete(self.request, subnet_id)
@ -2401,7 +2404,9 @@ class NeutronApiFloatingIpTests(test.APIMockTestCase):
def setUp(self):
super().setUp()
neutronclient = mock.patch.object(api.neutron, 'neutronclient').start()
networkclient = mock.patch.object(api.neutron, 'networkclient').start()
self.qclient = neutronclient.return_value
self.netclient = networkclient.return_value
@override_settings(OPENSTACK_NEUTRON_NETWORK={'enable_router': True})
def test_floating_ip_supported(self):
@ -2413,15 +2418,15 @@ class NeutronApiFloatingIpTests(test.APIMockTestCase):
def test_floating_ip_pools_list(self):
search_opts = {'router:external': True}
ext_nets = [n for n in self.api_networks.list()
if n['router:external']]
self.qclient.list_networks.return_value = {'networks': ext_nets}
ext_nets = [n for n in self.api_networks_sdk
if n['is_router_external']]
self.netclient.networks.return_value = ext_nets
rets = api.neutron.floating_ip_pools_list(self.request)
for attr in ['id', 'name']:
self.assertEqual([p[attr] for p in ext_nets],
[getattr(p, attr) for p in rets])
self.qclient.list_networks.assert_called_once_with(**search_opts)
self.netclient.networks.assert_called_once_with(**search_opts)
def test_floating_ip_list(self):
fips = self.api_floating_ips.list()
@ -2510,7 +2515,7 @@ class NeutronApiFloatingIpTests(test.APIMockTestCase):
def test_floating_ip_allocate(self):
ext_nets = [n for n in self.api_networks.list()
if n['router:external']]
if n['is_router_external']]
ext_net = ext_nets[0]
fip = self.api_floating_ips.first()
self.qclient.create_floatingip.return_value = {'floatingip': fip}
@ -2580,8 +2585,8 @@ class NeutronApiFloatingIpTests(test.APIMockTestCase):
# Port on the first subnet is connected to a router
# attached to external network in neutron_data.
subnet_id = self.subnets.first().id
shared_nets = [n for n in self.api_networks.list() if n['shared']]
shared_subnet_ids = [s for n in shared_nets for s in n['subnets']]
shared_nets = [n for n in self.api_networks_sdk if n['shared']]
shared_subnet_ids = [s for n in shared_nets for s in n['subnet_ids']]
target_ports = []
for p in ports:
if p['device_owner'].startswith('network:'):
@ -2604,16 +2609,15 @@ class NeutronApiFloatingIpTests(test.APIMockTestCase):
novaclient.versions.get_current.return_value = ver
novaclient.servers.list.return_value = servers
ext_nets = [n for n in self.api_networks.list()
if n['router:external']]
list_networks_retvals = [{'networks': ext_nets},
{'networks': shared_nets}]
self.qclient.list_networks.side_effect = list_networks_retvals
ext_nets = [n for n in self.api_networks_sdk
if n['is_router_external']]
list_networks_retvals = [ext_nets, shared_nets]
self.netclient.networks.side_effect = list_networks_retvals
self.qclient.list_routers.return_value = {'routers':
self.api_routers.list()}
shared_subs = [s for s in self.api_subnets.list()
shared_subs = [s for s in self.api_subnets_sdk
if s['id'] in shared_subnet_ids]
self.qclient.list_subnets.return_value = {'subnets': shared_subs}
self.netclient.subnets.return_value = shared_subs
rets = api.neutron.floating_ip_target_list(self.request)
@ -2628,12 +2632,12 @@ class NeutronApiFloatingIpTests(test.APIMockTestCase):
novaclient.versions.get_current.assert_called_once_with()
novaclient.servers.list.assert_called_once_with(
False, {'project_id': self.request.user.tenant_id})
self.qclient.list_networks.assert_has_calls([
self.netclient.networks.assert_has_calls([
mock.call(**{'router:external': True}),
mock.call(shared=True),
mock.call(is_shared=True),
])
self.qclient.list_routers.assert_called_once_with()
self.qclient.list_subnets.assert_called_once_with()
self.netclient.subnets.assert_called_once_with()
@mock.patch.object(api._nova, 'novaclient')
def _test_target_floating_ip_port_by_instance(self, server, ports,
@ -2643,26 +2647,27 @@ class NeutronApiFloatingIpTests(test.APIMockTestCase):
list_ports_retvals = []
self.qclient.list_ports.side_effect = list_ports_retvals
list_nets_retvals = []
self.qclient.list_networks.side_effect = list_nets_retvals
self.netclient.networks.side_effect = list_nets_retvals
# _target_ports_by_instance()
list_ports_retvals.append({'ports': candidates})
# _get_reachable_subnets()
ext_nets = [n for n in self.api_networks.list()
if n['router:external']]
list_nets_retvals.append({'networks': ext_nets})
ext_nets = [n for n in self.api_networks_sdk
if n['is_router_external']]
list_nets_retvals.append(ext_nets)
self.qclient.list_routers.side_effect = [{'routers':
self.api_routers.list()}]
rinfs = [p for p in ports
if p['device_owner'] in api.neutron.ROUTER_INTERFACE_OWNERS]
list_ports_retvals.append({'ports': rinfs})
shared_nets = [n for n in self.api_networks.list() if n['shared']]
list_nets_retvals.append({'networks': shared_nets})
shared_nets = [n for n in self.api_networks_sdk if n['is_shared']]
list_nets_retvals.append(shared_nets)
shared_subnet_ids = [s for n in shared_nets for s in n['subnets']]
shared_subs = [s for s in self.api_subnets.list()
shared_subs = [s for s in self.api_subnets_sdk
if s['id'] in shared_subnet_ids]
self.qclient.list_subnets.side_effect = [{'subnets': shared_subs}]
self.netclient.subnets.side_effect = [shared_subs]
# _get_server_name()
novaclient = mock_novaclient.return_value
@ -2677,12 +2682,12 @@ class NeutronApiFloatingIpTests(test.APIMockTestCase):
mock.call(device_id=server.id),
mock.call(device_owner=api.neutron.ROUTER_INTERFACE_OWNERS),
])
self.qclient.list_networks.assert_has_calls([
self.netclient.networks.assert_has_calls([
mock.call(**{'router:external': True}),
mock.call(shared=True),
mock.call(is_shared=True),
])
self.qclient.list_routers.assert_called_once_with()
self.qclient.list_subnets.assert_called_once_with()
self.netclient.subnets.assert_called_once_with()
novaclient.versions.get_current.assert_called_once_with()
novaclient.servers.get.assert_called_once_with(server.id)

4
openstack_dashboard/utils/filters.py
View File

@ -38,6 +38,10 @@ def get_display_label(choices, status):
"""
for (value, label) in choices:
try:
(status or '').lower()
except AttributeError:
continue
if value == (status or '').lower():
display_label = label
break