openstack/octavia
Code Issues Proposed changes

Merge "Add unit tests for neutron utils, add model/util for floating_ip"

Browse Source
This commit is contained in:
Zuul 2018-01-11 23:06:30 +00:00 committed by Gerrit Code Review
commit ab45b15c27
4 changed files with 188 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
octavia/network/data_models.py
View File

@ -104,6 +104,26 @@ class FixedIP(data_models.BaseDataModel):
self.subnet = subnet
class FloatingIP(data_models.BaseDataModel):
def __init__(self, id=None, description=None, project_id=None,
status=None, router_id=None, port_id=None,
floating_network_id=None, floating_ip_address=None,
fixed_ip_address=None, fixed_port_id=None):
self.id = id
self.description = description
self.project_id = project_id
self.status = status
self.router_id = router_id
self.port_id = port_id
self.floating_network_id = floating_network_id
self.floating_ip_address = floating_ip_address
self.fixed_ip_address = fixed_ip_address
self.fixed_port_id = fixed_port_id
# Need to provide this for compatibility in case it is used as a VIP
self.network_id = floating_network_id
class AmphoraNetworkConfig(data_models.BaseDataModel):
def __init__(self, amphora=None, vip_subnet=None, vip_port=None,

17
octavia/network/drivers/neutron/utils.py
View File

@ -77,3 +77,20 @@ def convert_fixed_ip_dict_to_model(fixed_ip_dict):
def convert_qos_policy_dict_to_model(qos_policy_dict):
qos_policy = qos_policy_dict.get('policy', qos_policy_dict)
return network_models.QosPolicy(id=qos_policy.get('id'))
# We can't use "floating_ip" because we need to match the neutron client method
def convert_floatingip_dict_to_model(floating_ip_dict):
floating_ip = floating_ip_dict.get('floatingip', floating_ip_dict)
return network_models.FloatingIP(
id=floating_ip.get('id'),
description=floating_ip.get('description'),
project_id=floating_ip.get('project_id', floating_ip.get('tenant_id')),
status=floating_ip.get('status'),
router_id=floating_ip.get('router_id'),
port_id=floating_ip.get('port_id'),
floating_network_id=floating_ip.get('floating_network_id'),
floating_ip_address=floating_ip.get('floating_ip_address'),
fixed_ip_address=floating_ip.get('fixed_ip_address'),
fixed_port_id=floating_ip.get('fixed_port_id')
)

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

@ -37,8 +37,15 @@ 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_HOST_ROUTES = []
MOCK_SUBNET = {'subnet': {'id': MOCK_SUBNET_ID,
'network_id': MOCK_NETWORK_ID}}
'network_id': MOCK_NETWORK_ID,
'name': MOCK_SUBNET_NAME,
'tenant_id': MOCK_PROJECT_ID,
'gateway_ip': MOCK_GATEWAY_IP,
'cidr': MOCK_CIDR,
'ip_version': MOCK_IP_VERSION,
'host_routes': MOCK_HOST_ROUTES}}
MOCK_SUBNET2 = {'subnet': {'id': MOCK_SUBNET_ID2,
'network_id': MOCK_NETWORK_ID2}}
MOCK_HOST_ROUTES = []
@ -91,6 +98,32 @@ MOCK_NEUTRON_PORT2 = {'port': {'network_id': MOCK_NETWORK_ID2,
'fixed_ips': [{'ip_address': MOCK_IP_ADDRESS2,
'subnet_id': MOCK_SUBNET_ID2}]}}
MOCK_NETWORK = {'network': {'id': MOCK_NETWORK_ID,
'name': MOCK_NETWORK_NAME,
'tenant_id': MOCK_PROJECT_ID,
'admin_state_up': MOCK_ADMIN_STATE_UP,
'subnets': [MOCK_SUBNET_ID],
'mtu': MOCK_MTU,
'provider:network_type': 'flat',
'provider:physical_network': MOCK_NETWORK_NAME,
'provider:segmentation_id': MOCK_SEGMENTATION_ID,
'router:external': MOCK_ROUTER_EXTERNAL}}
MOCK_FIXED_IP = {'fixed_ip': {'subnet_id': MOCK_SUBNET_ID,
'ip_address': MOCK_IP_ADDRESS}}
MOCK_FLOATING_IP_ID = 'floating-ip-1'
MOCK_FLOATING_IP_DESC = 'TestFloatingIP1'
MOCK_ROUTER_ID = 'mock-router-1'
MOCK_FLOATING_IP = {'floatingip': {'id': MOCK_FLOATING_IP_ID,
'description': MOCK_FLOATING_IP_DESC,
'tenant_id': MOCK_PROJECT_ID,
'status': MOCK_STATUS,
'port_id': MOCK_PORT_ID,
'router_id': MOCK_ROUTER_ID,
'floating_network_id': MOCK_NETWORK_ID,
'floating_ip_address': MOCK_IP_ADDRESS,
'fixed_ip_address': MOCK_IP_ADDRESS2,
'fixed_port_id': MOCK_PORT_ID2}}
MOCK_AMP_ID1 = 'amp1-id'
MOCK_AMP_ID2 = 'amp2-id'
MOCK_AMP_COMPUTE_ID1 = 'amp1-compute-id'

117
octavia/tests/unit/network/drivers/neutron/test_utils.py Normal file
View File

@ -0,0 +1,117 @@
# Copyright 2017 GoDaddy
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from octavia.network.drivers.neutron import utils
from octavia.tests.common import constants as t_constants
from octavia.tests.unit import base
class TestNeutronUtils(base.TestCase):
def setUp(self):
super(TestNeutronUtils, self).setUp()
def _compare_ignore_value_none(self, obj1_in, obj2_in):
obj1 = {key: obj1_in[key] for key in obj1_in
if obj1_in[key] is not None}
obj2 = {key: obj2_in[key] for key in obj2_in
if obj2_in[key] is not None}
self.assertEqual(obj1, obj2)
def _in_ignore_value_none(self, needle, haystack):
newneedle = {key: needle[key] for key in needle
if needle[key] is not None}
newhaystack = []
for hay in haystack:
newhaystack.append({key: hay[key] for key in hay
if hay[key] is not None})
self.assertIn(newneedle, newhaystack)
def test_convert_subnet_dict_to_model(self):
model_obj = utils.convert_subnet_dict_to_model(
t_constants.MOCK_SUBNET)
assert_dict = dict(
id=t_constants.MOCK_SUBNET_ID,
name=t_constants.MOCK_SUBNET_NAME,
network_id=t_constants.MOCK_NETWORK_ID,
project_id=t_constants.MOCK_PROJECT_ID,
gateway_ip=t_constants.MOCK_GATEWAY_IP,
cidr=t_constants.MOCK_CIDR,
ip_version=t_constants.MOCK_IP_VERSION
)
self._compare_ignore_value_none(model_obj.to_dict(), assert_dict)
def test_convert_port_dict_to_model(self):
model_obj = utils.convert_port_dict_to_model(
t_constants.MOCK_NEUTRON_PORT)
assert_dict = dict(
id=t_constants.MOCK_PORT_ID,
name=t_constants.MOCK_PORT_NAME,
device_id=t_constants.MOCK_DEVICE_ID,
device_owner=t_constants.MOCK_DEVICE_OWNER,
mac_address=t_constants.MOCK_MAC_ADDR,
network_id=t_constants.MOCK_NETWORK_ID,
status=t_constants.MOCK_STATUS,
project_id=t_constants.MOCK_PROJECT_ID,
admin_state_up=t_constants.MOCK_ADMIN_STATE_UP,
)
self._compare_ignore_value_none(model_obj.to_dict(), assert_dict)
fixed_ips = t_constants.MOCK_NEUTRON_PORT['port']['fixed_ips']
for ip in model_obj.fixed_ips:
self._in_ignore_value_none(ip.to_dict(), fixed_ips)
def test_convert_network_dict_to_model(self):
model_obj = utils.convert_network_dict_to_model(
t_constants.MOCK_NETWORK)
assert_dict = dict(
id=t_constants.MOCK_NETWORK_ID,
name=t_constants.MOCK_NETWORK_NAME,
subnets=[t_constants.MOCK_SUBNET_ID],
project_id=t_constants.MOCK_PROJECT_ID,
admin_state_up=t_constants.MOCK_ADMIN_STATE_UP,
mtu=t_constants.MOCK_MTU,
provider_network_type=t_constants.MOCK_NETWORK_TYPE,
provider_physical_network=t_constants.MOCK_NETWORK_NAME,
provider_segmentation_id=t_constants.MOCK_SEGMENTATION_ID,
router_external=t_constants.MOCK_ROUTER_EXTERNAL
)
model_dict = model_obj.to_dict()
model_dict['subnets'] = model_obj.subnets
self._compare_ignore_value_none(model_dict, assert_dict)
def test_convert_fixed_ip_dict_to_model(self):
model_obj = utils.convert_fixed_ip_dict_to_model(
t_constants.MOCK_FIXED_IP)
assert_dict = dict(
subnet_id=t_constants.MOCK_SUBNET_ID,
ip_address=t_constants.MOCK_IP_ADDRESS
)
self._compare_ignore_value_none(model_obj.to_dict(), assert_dict)
def test_convert_floatingip_dict_to_model(self):
model_obj = utils.convert_floatingip_dict_to_model(
t_constants.MOCK_FLOATING_IP)
assert_dict = dict(
id=t_constants.MOCK_FLOATING_IP_ID,
description=t_constants.MOCK_FLOATING_IP_DESC,
project_id=t_constants.MOCK_PROJECT_ID,
status=t_constants.MOCK_STATUS,
router_id=t_constants.MOCK_ROUTER_ID,
port_id=t_constants.MOCK_PORT_ID,
floating_network_id=t_constants.MOCK_NETWORK_ID,
network_id=t_constants.MOCK_NETWORK_ID,
floating_ip_address=t_constants.MOCK_IP_ADDRESS,
fixed_ip_address=t_constants.MOCK_IP_ADDRESS2,
fixed_port_id=t_constants.MOCK_PORT_ID2
)
self._compare_ignore_value_none(model_obj.to_dict(), assert_dict)