openstack/openstacksdk
Code Issues Proposed changes

Merge "Reduce list_router_interfaces() to necessary API calls"

Browse Source
This commit is contained in:
Zuul 2022-07-16 08:00:24 +00:00 committed by Gerrit Code Review
commit 9d6d8415af
2 changed files with 41 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
openstack/cloud/_network.py
View File

@ -1933,26 +1933,22 @@ class NetworkCloudMixin:
:returns: A list of network ``Port`` objects.
"""
# Find only router interface and gateway ports, ignore L3 HA ports etc.
router_interfaces = self.search_ports(filters={
'device_id': router['id'],
'device_owner': 'network:router_interface'}
) + self.search_ports(filters={
'device_id': router['id'],
'device_owner': 'network:router_interface_distributed'}
) + self.search_ports(filters={
'device_id': router['id'],
'device_owner': 'network:ha_router_replicated_interface'})
router_gateways = self.search_ports(filters={
'device_id': router['id'],
'device_owner': 'network:router_gateway'})
ports = router_interfaces + router_gateways
ports = list(self.network.ports(device_id=router['id']))
if interface_type:
if interface_type == 'internal':
return router_interfaces
if interface_type == 'external':
return router_gateways
return ports
router_interfaces = (
[port for port in ports
if (port['device_owner'] in
['network:router_interface',
'network:router_interface_distributed',
'network:ha_router_replicated_interface'])
] if not interface_type or interface_type == 'internal' else [])
router_gateways = (
[port for port in ports
if port['device_owner'] == 'network:router_gateway'
] if not interface_type or interface_type == 'external' else [])
return router_interfaces + router_gateways
def create_router(
self,

75
openstack/tests/unit/cloud/test_router.py
View File

@ -366,33 +366,23 @@ class TestRouter(base.TestCase):
'mickey')
self.assert_calls()
def _get_mock_dict(self, owner, json):
return dict(method='GET',
uri=self.get_mock_url(
'network', 'public', append=['v2.0', 'ports'],
qs_elements=["device_id=%s" % self.router_id,
"device_owner=network:%s" % owner]),
json=json)
def _test_list_router_interfaces(self, router, interface_type,
router_type="normal",
expected_result=None):
if router_type == "normal":
device_owner = 'router_interface'
elif router_type == "ha":
device_owner = 'ha_router_replicated_interface'
elif router_type == "dvr":
device_owner = 'router_interface_distributed'
internal_port = {
'id': 'internal_port_id',
'fixed_ips': [{
'subnet_id': 'internal_subnet_id',
'ip_address': "10.0.0.1"
}],
'device_id': self.router_id,
'device_owner': 'network:%s' % device_owner
}
external_port = {
internal_ports = [
{
'id': 'internal_port_id',
'fixed_ips': [{
'subnet_id': 'internal_subnet_id',
'ip_address': "10.0.0.1"
}],
'device_id': self.router_id,
'device_owner': device_owner
}
for device_owner in ['network:router_interface',
'network:ha_router_replicated_interface',
'network:router_interface_distributed']]
external_ports = [{
'id': 'external_port_id',
'fixed_ips': [{
'subnet_id': 'external_subnet_id',
@ -400,28 +390,23 @@ class TestRouter(base.TestCase):
}],
'device_id': self.router_id,
'device_owner': 'network:router_gateway'
}
}]
if expected_result is None:
if interface_type == "internal":
expected_result = [internal_port]
expected_result = internal_ports
elif interface_type == "external":
expected_result = [external_port]
expected_result = external_ports
else:
expected_result = [internal_port, external_port]
expected_result = internal_ports + external_ports
mock_uris = []
for port_type in ['router_interface',
'router_interface_distributed',
'ha_router_replicated_interface']:
if port_type == device_owner:
ports = {'ports': [internal_port]}
else:
ports = {'ports': []}
mock_uris.append(self._get_mock_dict(port_type, ports))
mock_uris.append(self._get_mock_dict('router_gateway',
{'ports': [external_port]}))
mock_uri = dict(method='GET',
uri=self.get_mock_url(
'network', 'public', append=['v2.0', 'ports'],
qs_elements=["device_id=%s" % self.router_id]),
json={'ports': (internal_ports + external_ports)})
self.register_uris(mock_uris)
self.register_uris([mock_uri])
ret = self.cloud.list_router_interfaces(router, interface_type)
self.assertEqual(
[_port.Port(**i).to_dict(computed=False) for i in expected_result],
@ -449,11 +434,3 @@ class TestRouter(base.TestCase):
def test_list_router_interfaces_external(self):
self._test_list_router_interfaces(self.router,
interface_type="external")
def test_list_router_interfaces_internal_ha(self):
self._test_list_router_interfaces(self.router, router_type="ha",
interface_type="internal")
def test_list_router_interfaces_internal_dvr(self):
self._test_list_router_interfaces(self.router, router_type="dvr",
interface_type="internal")