"Fossies" - the Fresh Open Source Software Archive

Member "neutron-14.0.3/neutron/tests/unit/objects/test_ports.py" (22 Oct 2019, 19625 Bytes) of package /linux/misc/openstack/neutron-14.0.3.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "test_ports.py": 14.0.2_vs_14.0.3.

    1 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    2 #    not use this file except in compliance with the License. You may obtain
    3 #    a copy of the License at
    4 #
    5 #         http://www.apache.org/licenses/LICENSE-2.0
    6 #
    7 #    Unless required by applicable law or agreed to in writing, software
    8 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
    9 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   10 #    License for the specific language governing permissions and limitations
   11 #    under the License.
   12 
   13 import mock
   14 from neutron_lib import constants
   15 from oslo_utils import uuidutils
   16 import testscenarios
   17 
   18 from neutron.objects import base as obj_base
   19 from neutron.objects import network
   20 from neutron.objects import ports
   21 from neutron.objects.qos import binding
   22 from neutron.objects.qos import policy
   23 from neutron.tests import tools
   24 from neutron.tests.unit.objects import test_base as obj_test_base
   25 from neutron.tests.unit import testlib_api
   26 
   27 
   28 class SecurityGroupPortBindingIfaceObjTestCase(
   29         obj_test_base.BaseObjectIfaceTestCase):
   30     _test_class = ports.SecurityGroupPortBinding
   31 
   32 
   33 class SecurityGroupPortBindingDbObjectTestCase(
   34         obj_test_base.BaseDbObjectTestCase):
   35     _test_class = ports.SecurityGroupPortBinding
   36 
   37 
   38 class BasePortBindingDbObjectTestCase(obj_test_base._BaseObjectTestCase,
   39                                       testlib_api.SqlTestCase):
   40     def setUp(self):
   41         super(BasePortBindingDbObjectTestCase, self).setUp()
   42         self.update_obj_fields(
   43             {'port_id': lambda: self._create_test_port_id()})
   44 
   45 
   46 class PortBindingIfaceObjTestCase(obj_test_base.BaseObjectIfaceTestCase):
   47     _test_class = ports.PortBinding
   48 
   49 
   50 class PortBindingDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
   51                                   BasePortBindingDbObjectTestCase):
   52     _test_class = ports.PortBinding
   53 
   54 
   55 class DistributedPortBindingIfaceObjTestCase(
   56         obj_test_base.BaseObjectIfaceTestCase):
   57     _test_class = ports.DistributedPortBinding
   58 
   59 
   60 class DistributedPortBindingDbObjectTestCase(
   61         obj_test_base.BaseDbObjectTestCase,
   62         BasePortBindingDbObjectTestCase):
   63     _test_class = ports.DistributedPortBinding
   64 
   65 
   66 # TODO(ihrachys): this test case copies some functions from the base module.
   67 # This is because we currently cannot inherit from the base class that contains
   68 # those functions, because that same class provides test cases that we don't
   69 # want to execute. Ideally, we would need to copy paste, but that would require
   70 # some significant refactoring in the base test classes. Leaving it for a
   71 # follow up.
   72 class PortBindingVifDetailsTestCase(testscenarios.WithScenarios,
   73                                     obj_test_base._BaseObjectTestCase,
   74                                     testlib_api.SqlTestCase):
   75 
   76     scenarios = [
   77         (cls.__name__, {'_test_class': cls})
   78         for cls in (ports.PortBinding, ports.DistributedPortBinding)
   79     ]
   80 
   81     def setUp(self):
   82         super(PortBindingVifDetailsTestCase, self).setUp()
   83         self._create_test_network()
   84         getter = lambda: self._create_port(network_id=self._network['id']).id
   85         self.update_obj_fields({'port_id': getter})
   86 
   87     def _create_port(self, **port_attrs):
   88         attrs = {'project_id': uuidutils.generate_uuid(),
   89                  'admin_state_up': True,
   90                  'status': 'ACTIVE',
   91                  'device_id': 'fake_device',
   92                  'device_owner': 'fake_owner',
   93                  'mac_address': tools.get_random_EUI()}
   94         attrs.update(port_attrs)
   95         port = ports.Port(self.context, **attrs)
   96         port.create()
   97         return port
   98 
   99     def _create_test_network(self):
  100         self._network = network.Network(self.context,
  101                                         name='test-network1')
  102         self._network.create()
  103 
  104     def _make_object(self, fields):
  105         fields = obj_test_base.get_non_synthetic_fields(
  106             self._test_class, fields
  107         )
  108         return self._test_class(
  109             self.context,
  110             **obj_test_base.remove_timestamps_from_fields(
  111                 fields, self._test_class.fields))
  112 
  113     def test_vif_details(self):
  114         vif_details = {'item1': 'val1', 'item2': 'val2'}
  115         obj = self._make_object(self.obj_fields[0])
  116         obj.vif_details = vif_details
  117         obj.create()
  118 
  119         obj = self._test_class.get_object(
  120             self.context, **obj._get_composite_keys())
  121         self.assertEqual(vif_details, obj.vif_details)
  122 
  123         vif_details['item1'] = 1.23
  124         del vif_details['item2']
  125         vif_details['item3'] = True
  126 
  127         obj.vif_details = vif_details
  128         obj.update()
  129 
  130         obj = self._test_class.get_object(
  131             self.context, **obj._get_composite_keys())
  132         self.assertEqual(vif_details, obj.vif_details)
  133 
  134         obj.vif_details = None
  135         obj.update()
  136         # here the obj is reloaded from DB,
  137         # so we test if vif_details is still none
  138         self.assertIsNone(obj.vif_details)
  139 
  140         obj = self._test_class.get_object(
  141             self.context, **obj._get_composite_keys())
  142         self.assertIsNone(obj.vif_details)
  143 
  144     def test_null_vif_details_in_db(self):
  145         # the null case for vif_details in our db model is an
  146         # empty string. add that here to simulate it correctly
  147         # in the tests
  148         kwargs = self.get_random_db_fields()
  149         kwargs['vif_details'] = ''
  150         db_obj = self._test_class.db_model(**kwargs)
  151         obj_fields = self._test_class.modify_fields_from_db(db_obj)
  152         obj = self._test_class(self.context, **obj_fields)
  153         self.assertIsNone(obj.vif_details)
  154 
  155 
  156 class IPAllocationIfaceObjTestCase(obj_test_base.BaseObjectIfaceTestCase):
  157 
  158     _test_class = ports.IPAllocation
  159 
  160 
  161 class IPAllocationDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
  162                                    testlib_api.SqlTestCase):
  163 
  164     _test_class = ports.IPAllocation
  165 
  166     def setUp(self):
  167         super(IPAllocationDbObjectTestCase, self).setUp()
  168         network_id = self._create_test_network_id()
  169         port_id = self._create_test_port_id(network_id=network_id)
  170         self.update_obj_fields(
  171             {'port_id': port_id, 'network_id': network_id,
  172              'subnet_id': lambda: self._create_test_subnet_id(network_id)})
  173 
  174 
  175 class PortDNSIfaceObjTestCase(obj_test_base.BaseObjectIfaceTestCase):
  176 
  177     _test_class = ports.PortDNS
  178 
  179 
  180 class PortDNSDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
  181                               testlib_api.SqlTestCase):
  182 
  183     _test_class = ports.PortDNS
  184 
  185     def setUp(self):
  186         super(PortDNSDbObjectTestCase, self).setUp()
  187         self.update_obj_fields(
  188             {'port_id': lambda: self._create_test_port_id()})
  189 
  190 
  191 class PortBindingLevelIfaceObjTestCase(
  192         obj_test_base.BaseObjectIfaceTestCase):
  193 
  194     _test_class = ports.PortBindingLevel
  195 
  196     def setUp(self):
  197         super(PortBindingLevelIfaceObjTestCase, self).setUp()
  198         self.pager_map[self._test_class.obj_name()] = (
  199             obj_base.Pager(sorts=[('port_id', True), ('level', True)]))
  200 
  201 
  202 class PortBindingLevelDbObjectTestCase(
  203         obj_test_base.BaseDbObjectTestCase, testlib_api.SqlTestCase):
  204 
  205     _test_class = ports.PortBindingLevel
  206 
  207     def setUp(self):
  208         super(PortBindingLevelDbObjectTestCase, self).setUp()
  209         self.update_obj_fields(
  210             {'port_id': lambda: self._create_test_port_id(),
  211              'segment_id': lambda: self._create_test_segment_id()})
  212 
  213 
  214 class PortIfaceObjTestCase(obj_test_base.BaseObjectIfaceTestCase):
  215 
  216     _test_class = ports.Port
  217 
  218     def setUp(self):
  219         super(PortIfaceObjTestCase, self).setUp()
  220         self.pager_map[ports.PortBindingLevel.obj_name()] = (
  221             obj_base.Pager(sorts=[('port_id', True), ('level', True)]))
  222 
  223 
  224 class PortDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
  225                            testlib_api.SqlTestCase):
  226 
  227     _test_class = ports.Port
  228 
  229     def setUp(self):
  230         super(PortDbObjectTestCase, self).setUp()
  231         network_id = self._create_test_network_id()
  232         segment_id = self._create_test_segment_id(network_id)
  233         subnet_id = self._create_test_subnet_id(network_id)
  234         self.update_obj_fields(
  235             {'network_id': network_id,
  236              'fixed_ips': {'subnet_id': subnet_id,
  237                            'network_id': network_id},
  238              'device_owner': 'not_a_router',
  239              'binding_levels': {'segment_id': segment_id}})
  240 
  241     def test_security_group_ids(self):
  242         groups = []
  243         objs = []
  244         for i in range(2):
  245             groups.append(self._create_test_security_group_id())
  246             objs.append(self._make_object(self.obj_fields[i]))
  247             objs[i].security_group_ids = {groups[i]}
  248             objs[i].create()
  249 
  250         self.assertEqual([objs[0]],
  251                          ports.Port.get_objects(
  252                              self.context, security_group_ids=(groups[0], )))
  253         self.assertEqual([objs[1]],
  254                          ports.Port.get_objects(
  255                              self.context, security_group_ids=(groups[1], )))
  256 
  257         sg3_id = self._create_test_security_group_id()
  258         objs[0].security_group_ids = {sg3_id}
  259         objs[0].update()
  260 
  261         objs[0] = ports.Port.get_object(self.context, id=objs[0].id)
  262         self.assertEqual({sg3_id}, objs[0].security_group_ids)
  263 
  264         objs[0].security_group_ids = set()
  265         objs[0].update()
  266 
  267         objs[0] = ports.Port.get_object(self.context, id=objs[0].id)
  268         self.assertFalse(objs[0].security_group_ids)
  269 
  270     def test_security_group_ids_and_port_id(self):
  271         objs = []
  272         group = self._create_test_security_group_id()
  273         for i in range(2):
  274             objs.append(self._make_object(self.obj_fields[i]))
  275             objs[i].security_group_ids = {group}
  276             objs[i].create()
  277 
  278         for i in range(2):
  279             self.assertEqual(
  280                 [objs[i]],
  281                 ports.Port.get_objects(
  282                     self.context, id=(objs[i].id, ),
  283                     security_group_ids=(group, )))
  284 
  285     def test__attach_security_group(self):
  286         obj = self._make_object(self.obj_fields[0])
  287         obj.create()
  288 
  289         sg_id = self._create_test_security_group_id()
  290         obj._attach_security_group(sg_id)
  291 
  292         obj = ports.Port.get_object(self.context, id=obj.id)
  293         self.assertIn(sg_id, obj.security_group_ids)
  294 
  295         sg2_id = self._create_test_security_group_id()
  296         obj._attach_security_group(sg2_id)
  297 
  298         obj = ports.Port.get_object(self.context, id=obj.id)
  299         self.assertIn(sg2_id, obj.security_group_ids)
  300 
  301     @mock.patch.object(policy.QosPolicy, 'unset_default')
  302     def test_qos_policy_id(self, *mocks):
  303         policy_obj = policy.QosPolicy(self.context)
  304         policy_obj.create()
  305 
  306         obj = self._make_object(self.obj_fields[0])
  307         obj.qos_policy_id = policy_obj.id
  308         obj.create()
  309 
  310         obj = ports.Port.get_object(self.context, id=obj.id)
  311         self.assertEqual(policy_obj.id, obj.qos_policy_id)
  312 
  313         policy_obj2 = policy.QosPolicy(self.context)
  314         policy_obj2.create()
  315 
  316         obj.qos_policy_id = policy_obj2.id
  317         obj.update()
  318 
  319         obj = ports.Port.get_object(self.context, id=obj.id)
  320         self.assertEqual(policy_obj2.id, obj.qos_policy_id)
  321 
  322         obj.qos_policy_id = None
  323         obj.update()
  324 
  325         obj = ports.Port.get_object(self.context, id=obj.id)
  326         self.assertIsNone(obj.qos_policy_id)
  327 
  328     @mock.patch.object(policy.QosPolicy, 'unset_default')
  329     def test__attach_qos_policy(self, *mocks):
  330         obj = self._make_object(self.obj_fields[0])
  331         obj.create()
  332 
  333         policy_obj = policy.QosPolicy(self.context)
  334         policy_obj.create()
  335         obj._attach_qos_policy(policy_obj.id)
  336 
  337         obj = ports.Port.get_object(self.context, id=obj.id)
  338         self.assertEqual(policy_obj.id, obj.qos_policy_id)
  339         qos_binding_obj = binding.QosPolicyPortBinding.get_object(
  340             self.context, port_id=obj.id)
  341         self.assertEqual(qos_binding_obj.policy_id, obj.qos_policy_id)
  342         old_policy_id = policy_obj.id
  343 
  344         policy_obj2 = policy.QosPolicy(self.context)
  345         policy_obj2.create()
  346         obj._attach_qos_policy(policy_obj2.id)
  347 
  348         obj = ports.Port.get_object(self.context, id=obj.id)
  349         self.assertEqual(policy_obj2.id, obj.qos_policy_id)
  350         qos_binding_obj2 = binding.QosPolicyPortBinding.get_object(
  351             self.context, port_id=obj.id)
  352         self.assertEqual(qos_binding_obj2.policy_id, obj.qos_policy_id)
  353         qos_binding_obj = binding.QosPolicyPortBinding.get_objects(
  354             self.context, policy_id=old_policy_id)
  355         self.assertEqual(0, len(qos_binding_obj))
  356 
  357     @mock.patch.object(policy.QosPolicy, 'unset_default')
  358     def test_qos_network_policy_id(self, *mocks):
  359         policy_obj = policy.QosPolicy(self.context)
  360         policy_obj.create()
  361 
  362         obj = self._make_object(self.obj_fields[0])
  363         obj.create()
  364         obj = ports.Port.get_object(self.context, id=obj.id)
  365         self.assertIsNone(obj.qos_network_policy_id)
  366         self.assertIsNone(obj.qos_policy_id)
  367 
  368         network = self._create_test_network(qos_policy_id=policy_obj.id)
  369         self.update_obj_fields({'network_id': network.id})
  370         obj = self._make_object(self.obj_fields[1])
  371         obj.create()
  372         obj = ports.Port.get_object(self.context, id=obj.id)
  373         self.assertEqual(policy_obj.id, obj.qos_network_policy_id)
  374         self.assertIsNone(obj.qos_policy_id)
  375 
  376     def test_get_objects_queries_constant(self):
  377         self.skipTest(
  378             'Port object loads segment info without relationships')
  379 
  380     def test_v1_1_to_v1_0_drops_data_plane_status(self):
  381         port_new = self._create_test_port()
  382         port_v1_0 = port_new.obj_to_primitive(target_version='1.0')
  383         self.assertNotIn('data_plane_status',
  384                          port_v1_0['versioned_object.data'])
  385 
  386     def test_v1_2_to_v1_1_drops_segment_id_in_binding_levels(self):
  387         port_new = self._create_test_port()
  388         segment = network.NetworkSegment(
  389             self.context,
  390             # TODO(ihrachys) we should be able to create a segment object
  391             # without explicitly specifying id, but it's currently not working
  392             id=uuidutils.generate_uuid(),
  393             network_id=port_new.network_id,
  394             network_type='vxlan')
  395         segment.create()
  396 
  397         # TODO(ihrachys) we should be able to create / update level objects via
  398         # Port object, but it's currently not working
  399         binding = ports.PortBindingLevel(
  400             self.context, port_id=port_new.id,
  401             host='host1', level=0, segment_id=segment.id)
  402         binding.create()
  403 
  404         port_new = ports.Port.get_object(self.context, id=port_new.id)
  405         port_v1_1 = port_new.obj_to_primitive(target_version='1.1')
  406 
  407         lvl = port_v1_1['versioned_object.data']['binding_levels'][0]
  408         self.assertNotIn('segment_id', lvl['versioned_object.data'])
  409 
  410         # check that we also downgraded level object version
  411         self.assertEqual('1.0', lvl['versioned_object.version'])
  412 
  413         # finally, prove that binding primitive is now identical to direct
  414         # downgrade of the binding object
  415         binding_v1_0 = binding.obj_to_primitive(target_version='1.0')
  416         self.assertEqual(binding_v1_0, lvl)
  417 
  418     def test_v1_3_to_v1_2_unlists_distributed_bindings(self):
  419         port_new = self._create_test_port()
  420 
  421         # empty list transforms into None
  422         port_v1_2 = port_new.obj_to_primitive(target_version='1.2')
  423         port_data = port_v1_2['versioned_object.data']
  424         self.assertIsNone(port_data['distributed_binding'])
  425 
  426         # now insert a distributed binding
  427         binding = ports.DistributedPortBinding(
  428             self.context,
  429             host='host1', port_id=port_new.id, status='ACTIVE',
  430             vnic_type='vnic_type1', vif_type='vif_type1')
  431         binding.create()
  432 
  433         # refetch port object to include binding
  434         port_new = ports.Port.get_object(self.context, id=port_new.id)
  435 
  436         # new primitive should contain the binding data
  437         port_v1_2 = port_new.obj_to_primitive(target_version='1.2')
  438         port_data = port_v1_2['versioned_object.data']
  439         binding_data = (
  440             port_data['distributed_binding']['versioned_object.data'])
  441         self.assertEqual(binding.host, binding_data['host'])
  442 
  443     def test_v1_4_to_v1_3_converts_binding_to_portbinding_object(self):
  444         port_v1_4 = self._create_test_port()
  445         port_v1_3 = port_v1_4.obj_to_primitive(target_version='1.3')
  446 
  447         # Port has no bindings, so binding attribute should be None
  448         self.assertIsNone(port_v1_3['versioned_object.data']['binding'])
  449         active_binding = ports.PortBinding(self.context, port_id=port_v1_4.id,
  450                                            host='host1', vif_type='type')
  451         inactive_binding = ports.PortBinding(
  452             self.context, port_id=port_v1_4.id, host='host2', vif_type='type',
  453             status=constants.INACTIVE)
  454         active_binding.create()
  455         inactive_binding.create()
  456         port_v1_4 = ports.Port.get_object(self.context, id=port_v1_4.id)
  457         port_v1_3 = port_v1_4.obj_to_primitive(target_version='1.3')
  458         binding = port_v1_3['versioned_object.data']['binding']
  459 
  460         # Port has active binding, so the binding attribute should point to it
  461         self.assertEqual('host1', binding['versioned_object.data']['host'])
  462         active_binding.delete()
  463         port_v1_4 = ports.Port.get_object(self.context, id=port_v1_4.id)
  464         port_v1_3 = port_v1_4.obj_to_primitive(target_version='1.3')
  465 
  466         # Port has no active bindings, so binding attribute should be None
  467         self.assertIsNone(port_v1_3['versioned_object.data']['binding'])
  468 
  469         # bindings attribute in V1.4 port should have one inactive binding
  470         primitive = port_v1_4.obj_to_primitive()
  471         self.assertEqual(1,
  472                          len(primitive['versioned_object.data']['bindings']))
  473         binding = primitive['versioned_object.data']['bindings'][0]
  474         self.assertEqual(constants.INACTIVE,
  475                          binding['versioned_object.data']['status'])
  476 
  477         # Port with no binding attribute should be handled without raising
  478         # exception
  479         primitive['versioned_object.data'].pop('bindings')
  480         port_v1_4_no_binding = port_v1_4.obj_from_primitive(primitive)
  481         port_v1_4_no_binding.obj_to_primitive(target_version='1.3')
  482 
  483     def test_v1_5_to_v1_4_drops_qos_network_policy_id(self):
  484         port_new = self._create_test_port()
  485         port_v1_4 = port_new.obj_to_primitive(target_version='1.4')
  486         self.assertNotIn('qos_network_policy_id',
  487                          port_v1_4['versioned_object.data'])
  488 
  489     def test_get_ports_ids_by_security_groups_except_router(self):
  490         sg_id = self._create_test_security_group_id()
  491         filter_owner = constants.ROUTER_INTERFACE_OWNERS_SNAT
  492         obj = self._make_object(self.obj_fields[0])
  493         obj.create()
  494         obj.security_group_ids = {sg_id}
  495         obj.update()
  496         self.assertEqual(1, len(
  497             ports.Port.get_ports_ids_by_security_groups(
  498                 self.context, security_group_ids=(sg_id, ),
  499                 excluded_device_owners=filter_owner)))
  500         obj.device_owner = constants.DEVICE_OWNER_ROUTER_SNAT
  501         obj.update()
  502         self.assertEqual(0, len(
  503             ports.Port.get_ports_ids_by_security_groups(
  504                 self.context, security_group_ids=(sg_id, ),
  505                 excluded_device_owners=filter_owner)))