openstack/neutron
Code Issues Proposed changes
c511964d70
neutron/neutron/tests/unit/objects/test_subnet.py
Brian Haley 7594bb0627 Remove the dependency on the "mock" package 
Now that we are python3 only, we should move to using the built
in version of mock that supports all of our testing needs and
remove the dependency on the "mock" package.

This patch moves all references to "import mock" to
"from unittest import mock". It also cleans up some new line
inconsistency.

Fixed an inconsistency in the OVSBridge.deferred() definition
as it needs to also have an *args argument.

Fixed an issue where an l3-agent test was mocking
functools.partial, causing a python3.8 failure.

Unit tests only, removing from tests/base.py affects
functional tests which need additional work.

Change-Id: I40e8a8410840c3774c72ae1a8054574445d66ece
2020-04-28 18:05:37 -04:00

310 lines
12 KiB
Python
Raw Blame History

# 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 unittest import mock
from neutron_lib import context
from oslo_utils import uuidutils
from neutron.db import rbac_db_models
from neutron.objects import base as obj_base
from neutron.objects.db import api as obj_db_api
from neutron.objects import network as net_obj
from neutron.objects import rbac_db
from neutron.objects import subnet
from neutron.tests.unit.objects import test_base as obj_test_base
from neutron.tests.unit import testlib_api
class IPAllocationPoolObjectIfaceTestCase(
obj_test_base.BaseObjectIfaceTestCase):
_test_class = subnet.IPAllocationPool
class IPAllocationPoolDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
testlib_api.SqlTestCase):
_test_class = subnet.IPAllocationPool
def setUp(self):
super(IPAllocationPoolDbObjectTestCase, self).setUp()
self.update_obj_fields(
{'subnet_id': lambda: self._create_test_subnet_id()})
class DNSNameServerObjectIfaceTestCase(obj_test_base.BaseObjectIfaceTestCase):
_test_class = subnet.DNSNameServer
def setUp(self):
super(DNSNameServerObjectIfaceTestCase, self).setUp()
self.pager_map[self._test_class.obj_name()] = (
obj_base.Pager(sorts=[('order', True)]))
class DNSNameServerDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
testlib_api.SqlTestCase):
_test_class = subnet.DNSNameServer
def setUp(self):
super(DNSNameServerDbObjectTestCase, self).setUp()
self._subnet_id = self._create_test_subnet_id()
self.update_obj_fields({'subnet_id': self._subnet_id})
def _create_dnsnameservers(self):
for obj in self.obj_fields:
dns = self._make_object(obj)
dns.create()
def test_get_objects_sort_by_order_asc(self):
self._create_dnsnameservers()
objs = self._test_class.get_objects(self.context)
fields_sorted = sorted([obj['order'] for obj in self.obj_fields])
self.assertEqual(fields_sorted, [obj.order for obj in objs])
def test_get_objects_sort_by_order_desc(self):
self._create_dnsnameservers()
pager = obj_base.Pager(sorts=[('order', False)])
objs = self._test_class.get_objects(self.context, _pager=pager,
subnet_id=self._subnet_id)
fields_sorted = sorted([obj['order'] for obj in self.obj_fields],
reverse=True)
self.assertEqual(fields_sorted, [obj.order for obj in objs])
def test_get_objects_sort_by_address_asc_using_pager(self):
self._create_dnsnameservers()
pager = obj_base.Pager(sorts=[('address', True)])
objs = self._test_class.get_objects(self.context, _pager=pager)
fields_sorted = sorted([obj['address'] for obj in self.obj_fields])
self.assertEqual(fields_sorted, [obj.address for obj in objs])
class RouteObjectIfaceTestCase(obj_test_base.BaseObjectIfaceTestCase):
_test_class = subnet.Route
class RouteDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
testlib_api.SqlTestCase):
_test_class = subnet.Route
def setUp(self):
super(RouteDbObjectTestCase, self).setUp()
self.update_obj_fields(
{'subnet_id': lambda: self._create_test_subnet_id()})
class SubnetServiceTypeObjectIfaceTestCase(
obj_test_base.BaseObjectIfaceTestCase):
_test_class = subnet.SubnetServiceType
class SubnetServiceTypeDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
testlib_api.SqlTestCase):
_test_class = subnet.SubnetServiceType
def setUp(self):
super(SubnetServiceTypeDbObjectTestCase, self).setUp()
self.update_obj_fields(
{'subnet_id': lambda: self._create_test_subnet_id()})
class SubnetObjectIfaceTestCase(obj_test_base.BaseObjectIfaceTestCase):
_test_class = subnet.Subnet
def setUp(self):
super(SubnetObjectIfaceTestCase, self).setUp()
self.pager_map[subnet.DNSNameServer.obj_name()] = (
obj_base.Pager(sorts=[('order', True)]))
# Base class will mock those out only when rbac_db_model is set for the
# object. Since subnets don't have their own models but only derive
# shared value from networks, we need to unconditionally mock those
# entry points out here, otherwise they will trigger database access,
# which is not allowed in 'Iface' test classes.
mock.patch.object(
rbac_db.RbacNeutronDbObjectMixin,
'is_shared_with_tenant', return_value=False).start()
mock.patch.object(
rbac_db.RbacNeutronDbObjectMixin,
'get_shared_with_tenant').start()
class SubnetDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
testlib_api.SqlTestCase):
_test_class = subnet.Subnet
CORE_PLUGIN = 'neutron.db.db_base_plugin_v2.NeutronDbPluginV2'
def setUp(self):
super(SubnetDbObjectTestCase, self).setUp()
# set up plugin because some models used here require a plugin
# (specifically, rbac models and their get_valid_actions validators)
self.setup_coreplugin(self.CORE_PLUGIN)
network_id = self._create_test_network_id()
self.update_obj_fields(
{'network_id': network_id,
'segment_id': lambda: self._create_test_segment_id(network_id)})
def test_get_dns_nameservers_in_order(self):
obj = self._make_object(self.obj_fields[0])
obj.create()
dns_nameservers = [(2, '1.2.3.4'), (1, '5.6.7.8'), (4, '7.7.7.7')]
for order, address in dns_nameservers:
dns = subnet.DNSNameServer(self.context, order=order,
address=address,
subnet_id=obj.id)
dns.create()
new = self._test_class.get_object(self.context, id=obj.id)
self.assertEqual(1, new.dns_nameservers[0].order)
self.assertEqual(2, new.dns_nameservers[1].order)
self.assertEqual(4, new.dns_nameservers[-1].order)
def _create_shared_network_rbac_entry(self, network):
attrs = {
'object_id': network['id'],
'target_tenant': '*',
'action': rbac_db_models.ACCESS_SHARED
}
obj_db_api.create_object(net_obj.NetworkRBAC, self.context, attrs)
def test_get_subnet_shared_true(self):
network = self._create_test_network()
self._create_shared_network_rbac_entry(network)
subnet_data = dict(self.obj_fields[0])
subnet_data['network_id'] = network['id']
obj = self._make_object(subnet_data)
# check if shared will be load by 'obj_load_attr' and using extra query
# by RbacNeutronDbObjectMixin get_shared_with_tenant
self.assertTrue(obj.shared)
obj.create()
# here the shared should be load by is_network_shared
self.assertTrue(obj.shared)
new = self._test_class.get_object(self.context,
**obj._get_composite_keys())
# again, the shared should be load by is_network_shared
self.assertTrue(new.shared)
def test_filter_by_shared(self):
network = self._create_test_network()
self._create_shared_network_rbac_entry(network)
subnet_data = dict(self.obj_fields[0])
subnet_data['network_id'] = network['id']
obj = self._make_object(subnet_data)
obj.create()
result = self._test_class.get_objects(self.context, shared=True)
self.assertEqual(obj, result[0])
def test_get_shared_subnet_with_another_tenant(self):
network_shared = self._create_test_network()
self._create_shared_network_rbac_entry(network_shared)
subnet_data = dict(self.obj_fields[0])
subnet_data['network_id'] = network_shared['id']
shared_subnet = self._make_object(subnet_data)
shared_subnet.create()
priv_subnet = self._make_object(self.obj_fields[1])
priv_subnet.create()
# Situation here:
# - we have one network with a subnet that are private
# - shared network with its subnet
# creating new context, user should have access to one shared network
all_subnets = self._test_class.get_objects(self.context)
self.assertEqual(2, len(all_subnets))
# access with new tenant_id, should be able to access to one subnet
new_ctx = context.Context('', uuidutils.generate_uuid())
public_subnets = self._test_class.get_objects(new_ctx)
self.assertEqual([shared_subnet], public_subnets)
# test get_object to fetch the private and then the shared subnet
fetched_private_subnet = self._test_class.get_object(new_ctx,
id=priv_subnet.id)
self.assertIsNone(fetched_private_subnet)
fetched_public_subnet = (
self._test_class.get_object(new_ctx, id=shared_subnet.id))
self.assertEqual(shared_subnet, fetched_public_subnet)
def test_get_service_types(self):
obj = self._make_object(self.obj_fields[0])
obj.create()
service_type_obj = subnet.SubnetServiceType(
self.context, subnet_id=obj.id, service_type='dhcp-agent')
service_type_obj.create()
listed_obj = subnet.Subnet.get_object(self.context, id=obj.id)
self.assertEqual([service_type_obj.service_type],
listed_obj.service_types)
# Try to load the service_types by obj_load_attr
obj1 = self._make_object(self.obj_fields[0])
self.assertEqual([service_type_obj.service_type],
obj1.service_types)
class NetworkSubnetLockTestCase(obj_test_base.BaseObjectIfaceTestCase):
_test_class = subnet.NetworkSubnetLock
class NetworkSubnetLockDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
testlib_api.SqlTestCase):
_test_class = subnet.NetworkSubnetLock
def setUp(self):
super(NetworkSubnetLockDbObjectTestCase, self).setUp()
self.update_obj_fields(
{'network_id': lambda: self._create_test_network_id()})
def test_lock_subnet_update(self):
obj = self._make_object(self.obj_fields[0])
obj.create()
subnet_id = self._create_test_subnet_id(network_id=obj.network_id)
subnet.NetworkSubnetLock.lock_subnet(self.context, obj.network_id,
subnet_id)
obj = subnet.NetworkSubnetLock.get_object(self.context,
network_id=obj.network_id)
self.assertEqual(subnet_id, obj.subnet_id)
def test_lock_subnet_create(self):
network_id = self._create_test_network_id()
subnet_id = self._create_test_subnet_id(network_id=network_id)
obj = subnet.NetworkSubnetLock.get_object(self.context,
network_id=network_id)
self.assertIsNone(obj)
subnet.NetworkSubnetLock.lock_subnet(self.context, network_id,
subnet_id)
obj = subnet.NetworkSubnetLock.get_object(self.context,
network_id=network_id)
self.assertEqual(network_id, obj.network_id)
self.assertEqual(subnet_id, obj.subnet_id)
View Git Blame Copy Permalink