156 lines
6.9 KiB
Python
156 lines
6.9 KiB
Python
# Copyright (c) 2018 OpenStack Foundation.
|
|
#
|
|
# 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.
|
|
|
|
import mock
|
|
|
|
import netaddr
|
|
|
|
from neutron.objects import port_forwarding
|
|
from neutron.objects import router
|
|
from neutron.tests import tools
|
|
from neutron.tests.unit.objects import test_base as obj_test_base
|
|
from neutron.tests.unit import testlib_api
|
|
|
|
|
|
class PortForwardingObjectTestCase(obj_test_base.BaseObjectIfaceTestCase):
|
|
|
|
_test_class = port_forwarding.PortForwarding
|
|
|
|
def setUp(self):
|
|
super(PortForwardingObjectTestCase, self).setUp()
|
|
self.fip_db_fields = self.get_random_db_fields(router.FloatingIP)
|
|
del self.fip_db_fields['floating_ip_address']
|
|
# 'portforwardings' table will store the 'internal_ip_address' and
|
|
# 'internal_port' as a single 'socket' column.
|
|
# Port forwarding object accepts 'internal_ip_address' and
|
|
# 'internal_port', but can not filter the records in db, so the
|
|
# valid filters can not contain them.
|
|
not_supported_filter_fields = ['internal_ip_address', 'internal_port']
|
|
invalid_fields = set(
|
|
self._test_class.synthetic_fields).union(
|
|
set(not_supported_filter_fields))
|
|
self.valid_field = [f for f in self._test_class.fields
|
|
if f not in invalid_fields][0]
|
|
|
|
def random_generate_fip_obj(db_fields, **floatingip):
|
|
if db_fields.get(
|
|
'id', None) and floatingip.get(
|
|
'id', None) and db_fields.get('id') == floatingip.get('id'):
|
|
return db_fields
|
|
db_fields['id'] = floatingip.get('id', None)
|
|
db_fields['floating_ip_address'] = tools.get_random_ip_address(
|
|
version=4)
|
|
return self.fip_db_fields
|
|
self.mock_fip_obj = mock.patch.object(
|
|
router.FloatingIP, 'get_object',
|
|
side_effect=lambda _, **y: router.FloatingIP.db_model(
|
|
**random_generate_fip_obj(self.fip_db_fields, **y))).start()
|
|
|
|
|
|
class PortForwardingDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
|
|
testlib_api.SqlTestCase):
|
|
|
|
_test_class = port_forwarding.PortForwarding
|
|
|
|
def setUp(self):
|
|
super(PortForwardingDbObjectTestCase, self).setUp()
|
|
self.update_obj_fields(
|
|
{'floatingip_id':
|
|
lambda: self._create_test_fip_id_for_port_forwarding(),
|
|
'internal_port_id': lambda: self._create_test_port_id()})
|
|
# 'portforwardings' table will store the 'internal_ip_address' and
|
|
# 'internal_port' as a single 'socket' column.
|
|
# Port forwarding object accepts 'internal_ip_address' and
|
|
# 'internal_port', but can not filter the records in db, so the
|
|
# valid filters can not contain them.
|
|
not_supported_filter_fields = ['internal_ip_address', 'internal_port']
|
|
invalid_fields = set(
|
|
self._test_class.synthetic_fields).union(
|
|
set(not_supported_filter_fields))
|
|
self.valid_field = [f for f in self._test_class.fields
|
|
if f not in invalid_fields][0]
|
|
self.valid_field_filter = {self.valid_field:
|
|
self.obj_fields[-1][self.valid_field]}
|
|
|
|
def _create_test_fip_id_for_port_forwarding(self):
|
|
fake_fip = '172.23.3.0'
|
|
ext_net_id = self._create_external_network_id()
|
|
router_id = self._create_test_router_id()
|
|
values = {
|
|
'floating_ip_address': netaddr.IPAddress(fake_fip),
|
|
'floating_network_id': ext_net_id,
|
|
'floating_port_id': self._create_test_port_id(
|
|
network_id=ext_net_id),
|
|
'router_id': router_id,
|
|
}
|
|
fip_obj = router.FloatingIP(self.context, **values)
|
|
fip_obj.create()
|
|
return fip_obj.id
|
|
|
|
def test_db_obj(self):
|
|
# The reason for rewriting this test is:
|
|
# 1. Currently, the existing test_db_obj test in
|
|
# obj_test_base.BaseDbObjectTestCase is not suitable for the case,
|
|
# for example, the db model is not the same with obj fields
|
|
# definition.
|
|
# 2. For port forwarding, the db model will store and accept 'socket',
|
|
# but the obj fields just only support accepting the parameters
|
|
# generate 'socket', such as 'internal_ip_address' and
|
|
# 'internal_port'.
|
|
obj = self._make_object(self.obj_fields[0])
|
|
self.assertIsNone(obj.db_obj)
|
|
|
|
obj.create()
|
|
self.assertIsNotNone(obj.db_obj)
|
|
# Make sure the created obj socket field is correct.
|
|
created_socket = obj.db_obj.socket.split(":")
|
|
self.assertEqual(created_socket[0], str(obj.internal_ip_address))
|
|
self.assertEqual(created_socket[1], str(obj.internal_port))
|
|
|
|
fields_to_update = self.get_updatable_fields(self.obj_fields[1])
|
|
if fields_to_update:
|
|
old_fields = {}
|
|
for key, val in fields_to_update.items():
|
|
db_model_attr = (
|
|
obj.fields_need_translation.get(key, key))
|
|
|
|
old_fields[db_model_attr] = obj.db_obj[
|
|
db_model_attr] if hasattr(
|
|
obj.db_obj, db_model_attr) else getattr(
|
|
obj, db_model_attr)
|
|
setattr(obj, key, val)
|
|
obj.update()
|
|
self.assertIsNotNone(obj.db_obj)
|
|
# Make sure the updated obj socket field is correct.
|
|
updated_socket = obj.db_obj.socket.split(":")
|
|
self.assertEqual(updated_socket[0],
|
|
str(self.obj_fields[1]['internal_ip_address']))
|
|
self.assertEqual(updated_socket[1],
|
|
str(self.obj_fields[1]['internal_port']))
|
|
# Then check all update fields had been updated.
|
|
for k, v in obj.modify_fields_to_db(fields_to_update).items():
|
|
self.assertEqual(v, obj.db_obj[k], '%s attribute differs' % k)
|
|
|
|
obj.delete()
|
|
self.assertIsNone(obj.db_obj)
|
|
|
|
def test_get_objects_queries_constant(self):
|
|
# NOTE(bzhao) Port Forwarding uses query FLoatingIP for injecting
|
|
# floating_ip_address and router_id, not depends on relationship,
|
|
# so it will cost extra SQL query each time for finding the
|
|
# associated Floating IP by floatingip_id each time(or each
|
|
# Port Forwarding Object). Rework this if this customized OVO
|
|
# needs to be changed.
|
|
pass
|