openstack/octavia
Code Issues Proposed changes

Merge "Improve Neutron driver _get_resource()"

Browse Source
This commit is contained in:
Zuul 2017-12-19 23:44:51 +00:00 committed by Gerrit Code Review
commit 4b48af9186
4 changed files with 133 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
octavia/network/drivers/neutron/base.py
View File

@ -156,8 +156,8 @@ class BaseNeutronDriver(base.AbstractNetworkDriver):
message = _('{resource_type} not found '
'({resource_type} id: {resource_id}).').format(
resource_type=resource_type, resource_id=resource_id)
raise getattr(base, '%sNotFound' %
resource_type.capitalize())(message)
raise getattr(base, '%sNotFound' % ''.join(
[w.capitalize() for w in resource_type.split('_')]))(message)
except Exception:
message = _('Error retrieving {resource_type} '
'({resource_type} id: {resource_id}.').format(
@ -165,21 +165,32 @@ class BaseNeutronDriver(base.AbstractNetworkDriver):
LOG.exception(message)
raise base.NetworkException(message)
def _get_resource_by_filters(self, resource_type, **filters):
# Filter items are unique, return the filtered item from the list.
def _get_resources_by_filters(self, resource_type, unique_item=False,
**filters):
"""Retrieves item(s) from filters. By default, a list is returned.
If unique_item set to True, only the first resource is returned.
"""
try:
resource = getattr(self.neutron_client, 'list_%ss' %
resource_type)(**filters)
resource = resource['%ss' % resource_type][0]
return getattr(utils, 'convert_%s_dict_to_model' %
resource_type)(resource)
conversion_function = getattr(
utils,
'convert_%s_dict_to_model' % resource_type)
if unique_item:
return conversion_function(resource['%ss' % resource_type][0])
elif not resource['%ss' % resource_type]:
# no items found
raise neutron_client_exceptions.NotFound()
else:
return list(map(conversion_function,
resource['%ss' % resource_type]))
except neutron_client_exceptions.NotFound:
message = _('{resource_type} not found '
'({resource_type} Filters: {filters}.').format(
resource_type=resource_type, filters=filters)
LOG.exception(message)
raise getattr(base, '%sNotFound' %
resource_type.capitalize())(message)
raise getattr(base, '%sNotFound' % ''.join(
[w.capitalize() for w in resource_type.split('_')]))(message)
except Exception:
message = _('Error retrieving {resource_type} '
'({resource_type} Filters: {filters}.').format(
@ -197,15 +208,18 @@ class BaseNeutronDriver(base.AbstractNetworkDriver):
return self._get_resource('port', port_id)
def get_network_by_name(self, network_name):
return self._get_resource_by_filters('network', name=network_name)
return self._get_resources_by_filters(
'network', unique_item=True, name=network_name)
def get_subnet_by_name(self, subnet_name):
return self._get_resource_by_filters('subnet', name=subnet_name)
return self._get_resources_by_filters(
'subnet', unique_item=True, name=subnet_name)
def get_port_by_name(self, port_name):
return self._get_resource_by_filters('port', name=port_name)
return self._get_resources_by_filters(
'port', unique_item=True, name=port_name)
def get_port_by_net_id_device_id(self, network_id, device_id):
return self._get_resource_by_filters('port',
network_id=network_id,
device_id=device_id)
return self._get_resources_by_filters(
'port', unique_item=True,
network_id=network_id, device_id=device_id)

41
octavia/tests/common/constants.py
View File

@ -27,34 +27,67 @@ MOCK_SUBNET_NAME = 'TestSubnet1'
MOCK_PORT_ID = 'mock-port-1'
MOCK_PORT_ID2 = 'mock-port-2'
MOCK_PORT_NAME = 'TestPort1'
MOCK_PORT_NAME2 = 'TestPort2'
MOCK_COMPUTE_ID = 'mock-compute-1'
MOCK_IP_ADDRESS = '10.0.0.1'
MOCK_IP_ADDRESS2 = '10.0.0.2'
MOCK_GATEWAY_IP = '10.0.0.3'
MOCK_IP_VERSION = 4
MOCK_CIDR = '10.0.0.0/24'
MOCK_MAC_ADDR = 'fe:16:3e:00:95:5c'
MOCK_MAC_ADDR2 = 'fe:16:3e:00:95:5d'
MOCK_PROJECT_ID = 'mock-project-1'
MOCK_SUBNET = {'subnet': {'id': MOCK_SUBNET_ID,
'network_id': MOCK_NETWORK_ID}}
MOCK_SUBNET2 = {'subnet': {'id': MOCK_SUBNET_ID2,
'network_id': MOCK_NETWORK_ID2}}
MOCK_HOST_ROUTES = []
MOCK_NOVA_INTERFACE = MockNovaInterface()
MOCK_SUBNET = {'subnet': {'id': MOCK_SUBNET_ID, 'network_id': MOCK_NETWORK_ID}}
MOCK_NOVA_INTERFACE.net_id = MOCK_NETWORK_ID
MOCK_NOVA_INTERFACE.port_id = MOCK_PORT_ID
MOCK_NOVA_INTERFACE.fixed_ips = [{'ip_address': MOCK_IP_ADDRESS}]
MOCK_NOVA_INTERFACE2 = MockNovaInterface()
MOCK_SUBNET2 = {'subnet': {'id': MOCK_SUBNET_ID2,
'network_id': MOCK_NETWORK_ID2}}
MOCK_NOVA_INTERFACE2.net_id = MOCK_NETWORK_ID2
MOCK_NOVA_INTERFACE2.port_id = MOCK_PORT_ID2
MOCK_NOVA_INTERFACE2.fixed_ips = [{'ip_address': MOCK_IP_ADDRESS2}]
MOCK_DEVICE_OWNER = 'Moctavia'
MOCK_DEVICE_ID = 'Moctavia123'
MOCK_DEVICE_ID2 = 'Moctavia124'
MOCK_SECURITY_GROUP_ID = 'security-group-1'
MOCK_SECURITY_GROUP_NAME = 'SecurityGroup1'
MOCK_ADMIN_STATE_UP = True
MOCK_STATUS = 'ACTIVE'
MOCK_MTU = 1500
MOCK_NETWORK_TYPE = 'flat'
MOCK_SEGMENTATION_ID = 1
MOCK_ROUTER_EXTERNAL = False
MOCK_NEUTRON_PORT = {'port': {'network_id': MOCK_NETWORK_ID,
'device_id': MOCK_COMPUTE_ID,
'device_id': MOCK_DEVICE_ID,
'device_owner': MOCK_DEVICE_OWNER,
'id': MOCK_PORT_ID,
'name': MOCK_PORT_NAME,
'tenant_id': MOCK_PROJECT_ID,
'admin_state_up': MOCK_ADMIN_STATE_UP,
'status': MOCK_STATUS,
'mac_address': MOCK_MAC_ADDR,
'fixed_ips': [{'ip_address': MOCK_IP_ADDRESS,
'subnet_id': MOCK_SUBNET_ID}]}}
MOCK_NEUTRON_PORT2 = {'port': {'network_id': MOCK_NETWORK_ID2,
'device_id': MOCK_DEVICE_ID2,
'device_owner': MOCK_DEVICE_OWNER,
'id': MOCK_PORT_ID2,
'name': MOCK_PORT_NAME2,
'tenant_id': MOCK_PROJECT_ID,
'admin_state_up': MOCK_ADMIN_STATE_UP,
'status': MOCK_STATUS,
'mac_address': MOCK_MAC_ADDR2,
'fixed_ips': [{'ip_address': MOCK_IP_ADDRESS2,
'subnet_id': MOCK_SUBNET_ID2}]}}
MOCK_AMP_ID1 = 'amp1-id'
MOCK_AMP_ID2 = 'amp2-id'
MOCK_AMP_COMPUTE_ID1 = 'amp1-compute-id'

2
octavia/tests/unit/network/drivers/neutron/test_allowed_address_pairs.py
View File

@ -405,7 +405,7 @@ class TestAllowedAddressPairsDriver(base.TestCase):
self.driver.allocate_vip, fake_lb)
def test_allocate_vip_when_no_port_provided(self):
port_create_dict = copy.copy(t_constants.MOCK_NEUTRON_PORT)
port_create_dict = copy.deepcopy(t_constants.MOCK_NEUTRON_PORT)
port_create_dict['port']['device_owner'] = (
allowed_address_pairs.OCTAVIA_OWNER)
port_create_dict['port']['device_id'] = 'lb-1'

67
octavia/tests/unit/network/drivers/neutron/test_base.py
View File

@ -165,7 +165,7 @@ class TestBaseNeutronNetworkDriver(base.TestCase):
list_ports = self.driver.neutron_client.list_ports
list_ports.side_effect = TypeError
o_ifaces = self.driver.get_plugged_networks(
t_constants.MOCK_COMPUTE_ID)
t_constants.MOCK_DEVICE_ID)
self.assertEqual(0, len(o_ifaces))
list_ports.side_effect = None
list_ports.reset_mock()
@ -176,7 +176,7 @@ class TestBaseNeutronNetworkDriver(base.TestCase):
}
list_ports.return_value = {'ports': [port1, port2]}
plugged_networks = self.driver.get_plugged_networks(
t_constants.MOCK_COMPUTE_ID)
t_constants.MOCK_DEVICE_ID)
for pn in plugged_networks:
self.assertIn(pn.port_id, [port1.get('id'), port2.get('id')])
self.assertIn(pn.network_id, [port1.get('network_id'),
@ -347,3 +347,66 @@ class TestBaseNeutronNetworkDriver(base.TestCase):
self.driver.get_port_by_net_id_device_id,
t_constants.MOCK_NETWORK_ID,
t_constants.MOCK_DEVICE_ID)
def test_get_ports_by_net_id_device_id(self):
"""Test get_port_by_net_id_device_id, when port is not unique.
The expected result is: only the first port is returned.
"""
list_port = self.driver.neutron_client.list_ports
list_port.return_value = {
'ports': [t_constants.MOCK_NEUTRON_PORT,
t_constants.MOCK_NEUTRON_PORT2,
],
}
port = self.driver.get_port_by_net_id_device_id(
t_constants.MOCK_NETWORK_ID, t_constants.MOCK_DEVICE_ID)
self.assertIsInstance(port, network_models.Port)
self.assertEqual(t_constants.MOCK_PORT_ID, port.id)
self.assertEqual(t_constants.MOCK_DEVICE_ID, port.device_id)
self.assertEqual(t_constants.MOCK_PORT_NAME, port.name)
self.assertEqual(t_constants.MOCK_MAC_ADDR, port.mac_address)
self.assertEqual(t_constants.MOCK_NETWORK_ID, port.network_id)
self.assertEqual(1, len(port.fixed_ips))
self.assertIsInstance(port.fixed_ips[0], network_models.FixedIP)
self.assertEqual(t_constants.MOCK_SUBNET_ID,
port.fixed_ips[0].subnet_id)
self.assertEqual(t_constants.MOCK_IP_ADDRESS,
port.fixed_ips[0].ip_address)
# Negative
list_port.side_effect = neutron_client_exceptions.NotFound
self.assertRaises(network_base.PortNotFound,
self.driver.get_port_by_net_id_device_id,
t_constants.MOCK_PORT_NAME,
t_constants.MOCK_DEVICE_ID)
list_port.side_effect = Exception
self.assertRaises(network_base.NetworkException,
self.driver.get_port_by_net_id_device_id,
t_constants.MOCK_NETWORK_ID,
t_constants.MOCK_DEVICE_ID)
def test_get_multiple_ports_by_net_id_device_id(self):
"""Test _get_resources_by_filters, when result is not unique"""
list_port = self.driver.neutron_client.list_ports
list_port.return_value = {
'ports': [t_constants.MOCK_NEUTRON_PORT,
t_constants.MOCK_NEUTRON_PORT2,
],
}
ports = self.driver._get_resources_by_filters(
'port',
network_id=t_constants.MOCK_NETWORK_ID,
device_id=t_constants.MOCK_DEVICE_ID,
)
self.assertIsInstance(ports, list)
port1, port2 = ports
self.assertEqual(t_constants.MOCK_PORT_ID, port1.id)
self.assertEqual(t_constants.MOCK_PORT_ID2, port2.id)
self.assertEqual(t_constants.MOCK_IP_ADDRESS,
port1.fixed_ips[0].ip_address)
self.assertEqual(t_constants.MOCK_IP_ADDRESS2,
port2.fixed_ips[0].ip_address)