# Copyright (c) 2017 Intel Corporation. # # 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 from neutron.objects import base as base_obj from neutron.objects import classification as n_class_obj from neutron_classifier.objects import classification as class_obj from neutron_classifier.services.classification import plugin from neutron_classifier.tests import base from neutron_lib import context from oslo_utils import uuidutils class TestPlugin(base.BaseClassificationTestCase): def setUp(self): super(TestPlugin, self).setUp() self.setup_coreplugin(load_plugins=False) mock.patch('neutron.objects.db.api.create_object').start() mock.patch('neutron.objects.db.api.update_object').start() mock.patch('neutron.objects.db.api.delete_object').start() mock.patch('neutron.objects.db.api.get_object').start() mock.patch('neutron_classifier.services.classification.advertiser' '.ClassificationAdvertiser').start() self.cl_plugin = plugin.ClassificationPlugin() self.ctxt = context.Context('fake_user', 'fake_tenant') mock.patch.object(self.ctxt.session, 'refresh').start() mock.patch.object(self.ctxt.session, 'expunge').start() mock.patch('neutron_classifier.objects.classification').start() self._generate_test_classifications() def _generate_test_classifications(self): self.test_classification = { 'classification': {'id': uuidutils.generate_uuid(), 'name': 'test_ethernet_classification', 'description': 'Test Ethernet Classification', 'project_id': uuidutils.generate_uuid(), 'shared': True, 'negated': True, 'c_type': 'ethernet', 'definition': {'src_addr': '00:07:E9:63:CE:53', 'dst_addr': '00:07:E9:42:AC:28', 'ethertype': 8100}}} self.test_classification_broken_headers = { 'id': self.test_classification['classification']['id'], 'name': 'test_ethernet_classification', 'description': 'Test Ethernet Classification', 'project_id': self.test_classification[ 'classification']['project_id'], 'shared': True, 'negated': True, 'c_type': 'ethernet', 'src_addr': '00:07:E9:63:CE:53', 'dst_addr': '00:07:E9:42:AC:28', 'ethertype': 8100} self.test_classification_2 = { 'classification': {'id': uuidutils.generate_uuid(), 'name': 'test_second_ethernet_cl', 'description': 'Test Second ' 'Ethernet Classification', 'project_id': uuidutils.generate_uuid(), 'shared': False, 'negated': False, 'c_type': 'ethernet', 'definition': {'src_addr': '00:54:TY:89:G6:67', 'dst_adrr': '00:54:TY:65:T7:44', 'ethertype': 8100}}} self.test_classification_2_broken_headers = { 'id': self.test_classification_2['classification']['id'], 'name': 'test_second_ethernet_cl', 'description': 'Test Second Ethernet Classification', 'project_id': self.test_classification_2[ 'classification']['project_id'], 'shared': False, 'negated': False, 'c_type': 'ethernet', 'src_addr': '00:54:TY:89:G6:67', 'dst_addr': '00:54:TY:65:T7:44', 'ethertype': 8100} def test_classification_break_out_headers(self): cl = self.cl_plugin.break_out_headers( self.test_classification) self.test_classification_broken_headers.pop('id', None) self.assertEqual(self.test_classification_broken_headers, cl) def test_merge_header(self): cl = self.cl_plugin.merge_header( self.test_classification_broken_headers) self.assertEqual(self.test_classification['classification'], cl) @mock.patch.object(class_obj.EthernetClassification, 'create') @mock.patch.object(class_obj.EthernetClassification, 'id', return_value=uuidutils.generate_uuid()) def test_create_classification(self, mock_ethernet_id, mock_ethernet_create): mock_manager = mock.Mock() mock_manager.attach_mock(mock_ethernet_create, 'create') mock_manager.attach_mock(mock_ethernet_id, 'id') mock_manager.reset_mock() mock_manager.start() self.test_classification['classification'].pop('id', None) val = self.cl_plugin.create_classification( self.ctxt, self.test_classification) expected_val = self.test_classification['classification'] expected_val['id'] = class_obj.EthernetClassification.id self.assertEqual(expected_val, val) mock_manager.create.assert_called_once() @mock.patch.object(plugin.ClassificationPlugin, 'merge_header') @mock.patch.object(n_class_obj.ClassificationBase, 'get_object') @mock.patch.object(class_obj.EthernetClassification, 'update_object') @mock.patch.object(class_obj.EthernetClassification, 'id', return_value=uuidutils.generate_uuid()) def test_update_classification(self, mock_id, mock_ethernet_update, mock_class_get, mock_merge): mock_manager = mock.Mock() mock_manager.attach_mock(mock_id, 'id') mock_manager.attach_mock(mock_ethernet_update, 'update') mock_manager.attach_mock(mock_class_get, 'get_classification') mock_manager.attach_mock(mock_merge, 'merge_header') mock_manager.reset_mock() mock_manager.start() c_obj = class_obj.EthernetClassification( self.ctxt, **self.test_classification_broken_headers) ethernet_classification_update = {'classification': { 'name': 'test_ethernet_classification Version 2', 'description': 'Test Ethernet Classification Version 2'}} mock_manager.get_classification().c_type = 'ethernet' self.cl_plugin.update_classification( self.ctxt, c_obj.id, ethernet_classification_update) classification_update_mock_call = mock.call.update( self.ctxt, {'description': 'Test Ethernet Classification Version 2', 'name': 'test_ethernet_classification Version 2'}, id=c_obj.id) self.assertIn(classification_update_mock_call, mock_manager.mock_calls) self.assertEqual(mock_manager.get_classification.call_count, 2) @mock.patch.object(n_class_obj.ClassificationBase, 'get_object') @mock.patch.object(class_obj.EthernetClassification, 'get_object') def test_delete_classification(self, mock_ethernet_get, mock_base_get): mock_manager = mock.Mock() mock_manager.attach_mock(mock_base_get, 'get_object') mock_manager.attach_mock(mock_ethernet_get, 'get_object') eth_class_obj = class_obj.EthernetClassification( self.ctxt, **self.test_classification_broken_headers) eth_class_obj.delete = mock.Mock() base_class_obj = n_class_obj.ClassificationBase( self.ctxt, **self.test_classification_broken_headers) mock_base_get.return_value = base_class_obj mock_ethernet_get.return_value = eth_class_obj mock_manager.reset_mock() self.cl_plugin.delete_classification( self.ctxt, base_class_obj.id) get_obj_mock_call = mock.call.get_object( self.ctxt, id=self.test_classification_broken_headers['id']) self.assertIn(get_obj_mock_call, mock_manager.mock_calls) self.assertEqual([get_obj_mock_call, get_obj_mock_call, get_obj_mock_call, get_obj_mock_call], mock_manager.mock_calls) self.assertTrue(eth_class_obj.delete.assert_called_once) @mock.patch.object(n_class_obj.ClassificationBase, 'get_object') @mock.patch.object(class_obj.EthernetClassification, 'get_object') def test_get_classification(self, mock_ethernet_get, mock_base_get): mock_manager = mock.Mock() mock_manager.attach_mock(mock_base_get, 'get_object') mock_manager.attach_mock(mock_ethernet_get, 'get_object') eth_classification = self.test_classification[ 'classification'] definition = eth_classification.pop('definition') base_class_obj = n_class_obj.ClassificationBase( self.ctxt, **eth_classification) eth_class_obj = class_obj.EthernetClassification( self.ctxt, **self.test_classification_broken_headers) mock_base_get.return_value = base_class_obj mock_ethernet_get.return_value = eth_class_obj eth_classification['definition'] = definition mock_manager.reset_mock() value = self.cl_plugin.get_classification( self.ctxt, eth_classification['id']) get_obj_mock_call = mock.call.get_object( self.ctxt, id=eth_classification['id']) self.assertIn(get_obj_mock_call, mock_manager.mock_calls) self.assertEqual([get_obj_mock_call, get_obj_mock_call], mock_manager.mock_calls) self.assertTrue(eth_classification, value) @mock.patch.object(n_class_obj.ClassificationBase, 'get_objects') @mock.patch.object(class_obj.EthernetClassification, 'get_objects') @mock.patch.object(base_obj, 'Pager') def test_get_classifications(self, mock_pager, mock_ethernet_get, mock_base_get): mock_manager = mock.Mock() mock_manager.attach_mock(mock_base_get, 'get_objects') mock_manager.attach_mock(mock_ethernet_get, 'get_objects') eth_cl_1 = self.test_classification['classification'] eth_cl_2 = self.test_classification_2['classification'] definition = eth_cl_1.pop('definition') definition_2 = eth_cl_2.pop('definition') base_class_obj_1 = n_class_obj.ClassificationBase( self.ctxt, **eth_cl_1) base_class_obj_2 = n_class_obj.ClassificationBase( self.ctxt, **eth_cl_2) eth_class_obj_1 = class_obj.EthernetClassification( self.ctxt, **self.test_classification_broken_headers) eth_class_obj_2 = class_obj.EthernetClassification( self.ctxt, **self.test_classification_2_broken_headers) base_list = [base_class_obj_1, base_class_obj_2] eth_list = [eth_class_obj_1, eth_class_obj_2] mock_base_get.return_value = base_list mock_ethernet_get.return_value = eth_list mock_pager.return_value = None eth_cl_1['definition'] = definition eth_cl_2['defintion'] = definition_2 result_list = [eth_cl_1, eth_cl_2] mock_manager.reset_mock() value = self.cl_plugin.get_classifications( self.ctxt, filters={'c_type': ['ethernet']}) get_objs_mock_call = mock.call.get_objects( self.ctxt, _pager=None) self.assertIn(get_objs_mock_call, mock_manager.mock_calls) self.assertEqual([get_objs_mock_call], mock_manager.mock_calls) self.assertTrue(result_list, value)