¡@
OpenStack Study: test_packet_filter.py**** CubicPower OpenStack Study ****
# Copyright (c) 2013 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 contextlib import mock import webob.exc from neutron.api.v2 import attributes from neutron import context from neutron.plugins.nec.common import exceptions as nexc from neutron.plugins.nec.extensions import packetfilter as ext_pf from neutron.tests.unit.nec import test_nec_plugin from neutron.tests.unit import test_db_plugin as test_plugin NEC_PLUGIN_PF_INI = """ [DEFAULT] api_extensions_path = neutron/plugins/nec/extensions [OFC] driver = neutron.tests.unit.nec.stub_ofc_driver.StubOFCDriver enable_packet_filter = True """
**** CubicPower OpenStack Study **** class PacketfilterExtensionManager(ext_pf.Packetfilter): @classmethod
**** CubicPower OpenStack Study **** def get_resources(cls): # Add the resources to the global attribute map # This is done here as the setup process won't # initialize the main API router which extends # the global attribute map attributes.RESOURCE_ATTRIBUTE_MAP.update( {'packet_filters': ext_pf.PACKET_FILTER_ATTR_MAP}) return super(PacketfilterExtensionManager, cls).get_resources()
**** CubicPower OpenStack Study **** class TestNecPluginPacketFilterBase(test_nec_plugin.NecPluginV2TestCase): _nec_ini = NEC_PLUGIN_PF_INI
**** CubicPower OpenStack Study **** def setUp(self): ext_mgr = PacketfilterExtensionManager() super(TestNecPluginPacketFilterBase, self).setUp(ext_mgr=ext_mgr)
**** CubicPower OpenStack Study **** def _create_packet_filter(self, fmt, net_id, expected_res_status=None, arg_list=None, **kwargs): data = {'packet_filter': {'network_id': net_id, 'tenant_id': self._tenant_id, 'priority': '1', 'action': 'ALLOW'}} for arg in (('name', 'admin_state_up', 'action', 'priority', 'in_port', 'src_mac', 'dst_mac', 'eth_type', 'src_cidr', 'dst_cidr', 'protocol', 'src_port', 'dst_port') + (arg_list or ())): # Arg must be present if arg in kwargs: data['packet_filter'][arg] = kwargs[arg] pf_req = self.new_create_request('packet_filters', data, fmt) if (kwargs.get('set_context') and 'tenant_id' in kwargs): # create a specific auth context for this request pf_req.environ['neutron.context'] = context.Context( '', kwargs['tenant_id']) pf_res = pf_req.get_response(self.ext_api) if expected_res_status: self.assertEqual(pf_res.status_int, expected_res_status) return pf_res
**** CubicPower OpenStack Study **** def _make_packet_filter(self, fmt, net_id, expected_res_status=None, **kwargs): res = self._create_packet_filter(fmt, net_id, expected_res_status, **kwargs) # Things can go wrong - raise HTTP exc with res code only # so it can be caught by unit tests if res.status_int >= 400: raise webob.exc.HTTPClientError(code=res.status_int) return self.deserialize(fmt, res) @contextlib.contextmanager
**** CubicPower OpenStack Study **** def packet_filter_on_network(self, network=None, fmt=None, do_delete=True, **kwargs): with test_plugin.optional_ctx(network, self.network) as network_to_use: net_id = network_to_use['network']['id'] pf = self._make_packet_filter(fmt or self.fmt, net_id, **kwargs) yield pf if do_delete: self._delete('packet_filters', pf['packet_filter']['id']) @contextlib.contextmanager
**** CubicPower OpenStack Study **** def packet_filter_on_port(self, port=None, fmt=None, do_delete=True, set_portinfo=True, **kwargs): with test_plugin.optional_ctx(port, self.port) as port_to_use: net_id = port_to_use['port']['network_id'] port_id = port_to_use['port']['id'] if set_portinfo: portinfo = {'id': port_id, 'port_no': kwargs.get('port_no', 123)} kw = {'added': [portinfo]} if 'datapath_id' in kwargs: kw['datapath_id'] = kwargs['datapath_id'] self.rpcapi_update_ports(**kw) kwargs['in_port'] = port_id pf = self._make_packet_filter(fmt or self.fmt, net_id, **kwargs) self.assertEqual(port_id, pf['packet_filter']['in_port']) yield pf if do_delete: self._delete('packet_filters', pf['packet_filter']['id'])
**** CubicPower OpenStack Study **** class TestNecPluginPacketFilter(TestNecPluginPacketFilterBase):
**** CubicPower OpenStack Study **** def setUp(self): super(TestNecPluginPacketFilter, self).setUp() # Remove attributes explicitly from mock object to check # a case where there are no update_filter and validate_*. del self.ofc.driver.update_filter del self.ofc.driver.validate_filter_create del self.ofc.driver.validate_filter_update
**** CubicPower OpenStack Study **** def test_list_packet_filters(self): self._list('packet_filters')
**** CubicPower OpenStack Study **** def test_create_pf_on_network_no_ofc_creation(self): with self.packet_filter_on_network(admin_state_up=False) as pf: self.assertEqual(pf['packet_filter']['status'], 'DOWN') self.assertFalse(self.ofc.create_ofc_packet_filter.called) self.assertFalse(self.ofc.delete_ofc_packet_filter.called)
**** CubicPower OpenStack Study **** def test_create_pf_on_port_no_ofc_creation(self): with self.packet_filter_on_port(admin_state_up=False, set_portinfo=False) as pf: self.assertEqual(pf['packet_filter']['status'], 'DOWN') self.assertFalse(self.ofc.create_ofc_packet_filter.called) self.assertFalse(self.ofc.delete_ofc_packet_filter.called)
**** CubicPower OpenStack Study **** def test_create_pf_on_network_with_ofc_creation(self): with self.packet_filter_on_network() as pf: pf_id = pf['packet_filter']['id'] self.assertEqual(pf['packet_filter']['status'], 'ACTIVE') ctx = mock.ANY pf_dict = mock.ANY expected = [ mock.call.exists_ofc_packet_filter(ctx, pf_id), mock.call.create_ofc_packet_filter(ctx, pf_id, pf_dict), mock.call.exists_ofc_packet_filter(ctx, pf_id), mock.call.delete_ofc_packet_filter(ctx, pf_id), ] self.ofc.assert_has_calls(expected) self.assertEqual(self.ofc.create_ofc_packet_filter.call_count, 1) self.assertEqual(self.ofc.delete_ofc_packet_filter.call_count, 1)
**** CubicPower OpenStack Study **** def test_create_pf_on_port_with_ofc_creation(self): with self.packet_filter_on_port() as pf: pf_id = pf['packet_filter']['id'] self.assertEqual(pf['packet_filter']['status'], 'ACTIVE') ctx = mock.ANY pf_dict = mock.ANY expected = [ mock.call.exists_ofc_packet_filter(ctx, pf_id), mock.call.create_ofc_packet_filter(ctx, pf_id, pf_dict), mock.call.exists_ofc_packet_filter(ctx, pf_id), mock.call.delete_ofc_packet_filter(ctx, pf_id), ] self.ofc.assert_has_calls(expected) self.assertEqual(self.ofc.create_ofc_packet_filter.call_count, 1) self.assertEqual(self.ofc.delete_ofc_packet_filter.call_count, 1)
**** CubicPower OpenStack Study **** def _test_create_pf_with_protocol(self, protocol, expected_eth_type): with self.packet_filter_on_network(protocol=protocol) as pf: pf_data = pf['packet_filter'] self.assertEqual(protocol, pf_data['protocol']) self.assertEqual(expected_eth_type, pf_data['eth_type'])
**** CubicPower OpenStack Study **** def test_create_pf_with_protocol_tcp(self): self._test_create_pf_with_protocol('TCP', 0x800)
**** CubicPower OpenStack Study **** def test_create_pf_with_protocol_udp(self): self._test_create_pf_with_protocol('UDP', 0x800)
**** CubicPower OpenStack Study **** def test_create_pf_with_protocol_icmp(self): self._test_create_pf_with_protocol('ICMP', 0x800)
**** CubicPower OpenStack Study **** def test_create_pf_with_protocol_arp(self): self._test_create_pf_with_protocol('ARP', 0x806)
**** CubicPower OpenStack Study **** def test_create_pf_with_inconsistent_protocol_and_eth_type(self): with self.packet_filter_on_network(protocol='TCP') as pf: pf_data = pf['packet_filter'] pf_id = pf_data['id'] self.assertEqual('TCP', pf_data['protocol']) self.assertEqual(0x800, pf_data['eth_type']) data = {'packet_filter': {'eth_type': 0x806}} self._update('packet_filters', pf_id, data, expected_code=409)
**** CubicPower OpenStack Study **** def test_create_pf_with_invalid_priority(self): with self.network() as net: net_id = net['network']['id'] kwargs = {'priority': 'high'} self._create_packet_filter(self.fmt, net_id, webob.exc.HTTPBadRequest.code, **kwargs) self.assertFalse(self.ofc.create_ofc_packet_filter.called)
**** CubicPower OpenStack Study **** def test_create_pf_with_ofc_creation_failure(self): self.ofc.set_raise_exc('create_ofc_packet_filter', nexc.OFCException(reason='hoge')) with self.packet_filter_on_network() as pf: pf_id = pf['packet_filter']['id'] pf_ref = self._show('packet_filters', pf_id) self.assertEqual(pf_ref['packet_filter']['status'], 'ERROR') self.ofc.set_raise_exc('create_ofc_packet_filter', None) # Retry activate packet_filter (even if there is no change). data = {'packet_filter': {}} self._update('packet_filters', pf_id, data) pf_ref = self._show('packet_filters', pf_id) self.assertEqual(pf_ref['packet_filter']['status'], 'ACTIVE') ctx = mock.ANY pf_dict = mock.ANY expected = [ mock.call.exists_ofc_packet_filter(ctx, pf_id), mock.call.create_ofc_packet_filter(ctx, pf_id, pf_dict), mock.call.exists_ofc_packet_filter(ctx, pf_id), mock.call.exists_ofc_packet_filter(ctx, pf_id), mock.call.create_ofc_packet_filter(ctx, pf_id, pf_dict), ] self.ofc.assert_has_calls(expected) self.assertEqual(self.ofc.create_ofc_packet_filter.call_count, 2)
**** CubicPower OpenStack Study **** def test_show_pf_on_network(self): kwargs = { 'name': 'test-pf-net', 'admin_state_up': False, 'action': 'DENY', 'priority': '102', 'src_mac': '00:11:22:33:44:55', 'dst_mac': '66:77:88:99:aa:bb', 'eth_type': '2048', 'src_cidr': '192.168.1.0/24', 'dst_cidr': '192.168.2.0/24', 'protocol': 'TCP', 'src_port': '35001', 'dst_port': '22' } with self.packet_filter_on_network(**kwargs) as pf: pf_id = pf['packet_filter']['id'] pf_ref = self._show('packet_filters', pf_id) # convert string to int. kwargs.update({'priority': 102, 'eth_type': 2048, 'src_port': 35001, 'dst_port': 22, 'in_port': None}) self.assertEqual(pf_id, pf_ref['packet_filter']['id']) for key in kwargs: self.assertEqual(kwargs[key], pf_ref['packet_filter'][key])
**** CubicPower OpenStack Study **** def test_show_pf_on_network_with_wildcards(self): kwargs = { 'name': 'test-pf-net', 'admin_state_up': False, 'action': 'DENY', 'priority': '102', } with self.packet_filter_on_network(**kwargs) as pf: pf_id = pf['packet_filter']['id'] pf_ref = self._show('packet_filters', pf_id) # convert string to int. kwargs.update({'priority': 102, 'in_port': None, 'src_mac': None, 'dst_mac': None, 'eth_type': None, 'src_cidr': None, 'dst_cidr': None, 'protocol': None, 'src_port': None, 'dst_port': None}) self.assertEqual(pf_id, pf_ref['packet_filter']['id']) for key in kwargs: self.assertEqual(kwargs[key], pf_ref['packet_filter'][key])
**** CubicPower OpenStack Study **** def test_show_pf_on_port(self): kwargs = { 'name': 'test-pf-port', 'admin_state_up': False, 'action': 'DENY', 'priority': '0o147', 'src_mac': '00:11:22:33:44:55', 'dst_mac': '66:77:88:99:aa:bb', 'eth_type': 2048, 'src_cidr': '192.168.1.0/24', 'dst_cidr': '192.168.2.0/24', 'protocol': 'TCP', 'dst_port': '0x50' } with self.packet_filter_on_port(**kwargs) as pf: pf_id = pf['packet_filter']['id'] pf_ref = self._show('packet_filters', pf_id) # convert string to int. kwargs.update({'priority': 103, 'eth_type': 2048, 'dst_port': 80, # wildcard field is None in a response. 'src_port': None}) self.assertEqual(pf_id, pf_ref['packet_filter']['id']) self.assertTrue(pf_ref['packet_filter']['in_port']) for key in kwargs: self.assertEqual(kwargs[key], pf_ref['packet_filter'][key])
**** CubicPower OpenStack Study **** def test_show_pf_not_found(self): pf_id = '00000000-ffff-ffff-ffff-000000000000' self._show('packet_filters', pf_id, expected_code=webob.exc.HTTPNotFound.code)
**** CubicPower OpenStack Study **** def test_update_pf_on_network(self): ctx = mock.ANY pf_dict = mock.ANY with self.packet_filter_on_network(admin_state_up=False) as pf: pf_id = pf['packet_filter']['id'] self.assertFalse(self.ofc.create_ofc_packet_filter.called) data = {'packet_filter': {'admin_state_up': True}} self._update('packet_filters', pf_id, data) self.ofc.create_ofc_packet_filter.assert_called_once_with( ctx, pf_id, pf_dict) self.assertFalse(self.ofc.delete_ofc_packet_filter.called) data = {'packet_filter': {'admin_state_up': False}} self._update('packet_filters', pf_id, data) self.ofc.delete_ofc_packet_filter.assert_called_once_with( ctx, pf_id)
**** CubicPower OpenStack Study **** def test_update_pf_on_port(self): ctx = mock.ANY pf_dict = mock.ANY with self.packet_filter_on_port(admin_state_up=False) as pf: pf_id = pf['packet_filter']['id'] self.assertFalse(self.ofc.create_ofc_packet_filter.called) data = {'packet_filter': {'admin_state_up': True}} self._update('packet_filters', pf_id, data) self.ofc.create_ofc_packet_filter.assert_called_once_with( ctx, pf_id, pf_dict) self.assertFalse(self.ofc.delete_ofc_packet_filter.called) data = {'packet_filter': {'admin_state_up': False}} self._update('packet_filters', pf_id, data) self.ofc.delete_ofc_packet_filter.assert_called_once_with( ctx, pf_id)
**** CubicPower OpenStack Study **** def test_delete_pf_with_error_status(self): self.ofc.set_raise_exc('create_ofc_packet_filter', nexc.OFCException(reason='fake')) with self.packet_filter_on_network() as pf: pf_id = pf['packet_filter']['id'] pf_ref = self._show('packet_filters', pf_id) self.assertEqual(pf_ref['packet_filter']['status'], 'ERROR') ctx = mock.ANY pf_dict = mock.ANY expected = [ mock.call.exists_ofc_packet_filter(ctx, pf_id), mock.call.create_ofc_packet_filter(ctx, pf_id, pf_dict), mock.call.exists_ofc_packet_filter(ctx, pf_id), ] self.ofc.assert_has_calls(expected) self.assertEqual(1, self.ofc.create_ofc_packet_filter.call_count) self.assertEqual(0, self.ofc.delete_ofc_packet_filter.call_count)
**** CubicPower OpenStack Study **** def test_activate_pf_on_port_triggered_by_update_port(self): ctx = mock.ANY pf_dict = mock.ANY with self.packet_filter_on_port(set_portinfo=False) as pf: pf_id = pf['packet_filter']['id'] in_port_id = pf['packet_filter']['in_port'] self.assertFalse(self.ofc.create_ofc_packet_filter.called) portinfo = {'id': in_port_id, 'port_no': 123} kw = {'added': [portinfo]} self.rpcapi_update_ports(**kw) self.ofc.create_ofc_packet_filter.assert_called_once_with( ctx, pf_id, pf_dict) self.assertFalse(self.ofc.delete_ofc_packet_filter.called) kw = {'removed': [in_port_id]} self.rpcapi_update_ports(**kw) self.ofc.delete_ofc_packet_filter.assert_called_once_with( ctx, pf_id) # Ensure pf was created before in_port has activated. ctx = mock.ANY pf_dict = mock.ANY port_dict = mock.ANY expected = [ mock.call.exists_ofc_packet_filter(ctx, pf_id), mock.call.create_ofc_packet_filter(ctx, pf_id, pf_dict), mock.call.exists_ofc_port(ctx, in_port_id), mock.call.create_ofc_port(ctx, in_port_id, port_dict), mock.call.exists_ofc_port(ctx, in_port_id), mock.call.delete_ofc_port(ctx, in_port_id, port_dict), mock.call.exists_ofc_packet_filter(ctx, pf_id), mock.call.delete_ofc_packet_filter(ctx, pf_id), ] self.ofc.assert_has_calls(expected) self.assertEqual(self.ofc.create_ofc_packet_filter.call_count, 1) self.assertEqual(self.ofc.delete_ofc_packet_filter.call_count, 1)
**** CubicPower OpenStack Study **** def test_activate_pf_while_exists_on_ofc(self): ctx = mock.ANY with self.packet_filter_on_network() as pf: pf_id = pf['packet_filter']['id'] self.ofc.set_raise_exc('delete_ofc_packet_filter', nexc.OFCException(reason='hoge')) # This update request will make plugin reactivate pf. data = {'packet_filter': {'priority': 1000}} self._update('packet_filters', pf_id, data, expected_code=webob.exc.HTTPInternalServerError.code) self.ofc.set_raise_exc('delete_ofc_packet_filter', None) ctx = mock.ANY pf_dict = mock.ANY expected = [ mock.call.exists_ofc_packet_filter(ctx, pf_id), mock.call.create_ofc_packet_filter(ctx, pf_id, pf_dict), mock.call.exists_ofc_packet_filter(ctx, pf_id), mock.call.delete_ofc_packet_filter(ctx, pf_id), mock.call.exists_ofc_packet_filter(ctx, pf_id), mock.call.delete_ofc_packet_filter(ctx, pf_id), ] self.ofc.assert_has_calls(expected) self.assertEqual(self.ofc.delete_ofc_packet_filter.call_count, 2)
**** CubicPower OpenStack Study **** def test_deactivate_pf_with_ofc_deletion_failure(self): ctx = mock.ANY with self.packet_filter_on_network() as pf: pf_id = pf['packet_filter']['id'] self.ofc.set_raise_exc('delete_ofc_packet_filter', nexc.OFCException(reason='hoge')) data = {'packet_filter': {'admin_state_up': False}} self._update('packet_filters', pf_id, data, expected_code=webob.exc.HTTPInternalServerError.code) pf_ref = self._show('packet_filters', pf_id) self.assertEqual(pf_ref['packet_filter']['status'], 'ERROR') self.ofc.set_raise_exc('delete_ofc_packet_filter', None) data = {'packet_filter': {'priority': 1000}} self._update('packet_filters', pf_id, data) pf_ref = self._show('packet_filters', pf_id) self.assertEqual(pf_ref['packet_filter']['status'], 'DOWN') ctx = mock.ANY pf_dict = mock.ANY expected = [ mock.call.exists_ofc_packet_filter(ctx, pf_id), mock.call.create_ofc_packet_filter(ctx, pf_id, pf_dict), mock.call.exists_ofc_packet_filter(ctx, pf_id), mock.call.delete_ofc_packet_filter(ctx, pf_id), mock.call.exists_ofc_packet_filter(ctx, pf_id), mock.call.delete_ofc_packet_filter(ctx, pf_id), mock.call.exists_ofc_packet_filter(ctx, pf_id), ] self.ofc.assert_has_calls(expected) self.assertEqual(self.ofc.delete_ofc_packet_filter.call_count, 2)
**** CubicPower OpenStack Study **** def test_delete_pf_with_ofc_deletion_failure(self): self.ofc.set_raise_exc('delete_ofc_packet_filter', nexc.OFCException(reason='hoge')) with self.packet_filter_on_network() as pf: pf_id = pf['packet_filter']['id'] self._delete('packet_filters', pf_id, expected_code=webob.exc.HTTPInternalServerError.code) pf_ref = self._show('packet_filters', pf_id) self.assertEqual(pf_ref['packet_filter']['status'], 'ERROR') self.ofc.set_raise_exc('delete_ofc_packet_filter', None) # Then, self._delete('packet_filters', pf_id) will success. ctx = mock.ANY pf_dict = mock.ANY expected = [ mock.call.exists_ofc_packet_filter(ctx, pf_id), mock.call.create_ofc_packet_filter(ctx, pf_id, pf_dict), mock.call.exists_ofc_packet_filter(ctx, pf_id), mock.call.delete_ofc_packet_filter(ctx, pf_id), mock.call.exists_ofc_packet_filter(ctx, pf_id), mock.call.delete_ofc_packet_filter(ctx, pf_id), ] self.ofc.assert_has_calls(expected) self.assertEqual(self.ofc.delete_ofc_packet_filter.call_count, 2)
**** CubicPower OpenStack Study **** def test_auto_delete_pf_in_network_deletion(self): with self.packet_filter_on_network(admin_state_up=False, do_delete=False) as pf: pf_id = pf['packet_filter']['id'] self._show('packet_filters', pf_id, expected_code=webob.exc.HTTPNotFound.code)
**** CubicPower OpenStack Study **** def test_auto_delete_pf_in_port_deletion(self): with self.port(no_delete=True) as port: network = self._show('networks', port['port']['network_id']) with self.packet_filter_on_network(network=network) as pfn: with self.packet_filter_on_port(port=port, do_delete=False, set_portinfo=False) as pf: pf_id = pf['packet_filter']['id'] in_port_id = pf['packet_filter']['in_port'] self._delete('ports', in_port_id) # Check the packet filter on the port is deleted. self._show('packet_filters', pf_id, expected_code=webob.exc.HTTPNotFound.code) # Check the packet filter on the network is not deleted. self._show('packet_filters', pfn['packet_filter']['id'])
**** CubicPower OpenStack Study **** def test_no_pf_activation_while_port_operations(self): with self.packet_filter_on_port() as pf: in_port_id = pf['packet_filter']['in_port'] self.assertEqual(self.ofc.create_ofc_packet_filter.call_count, 1) self.assertEqual(self.ofc.delete_ofc_packet_filter.call_count, 0) data = {'port': {'admin_state_up': False}} self._update('ports', in_port_id, data) self.assertEqual(self.ofc.create_ofc_packet_filter.call_count, 1) self.assertEqual(self.ofc.delete_ofc_packet_filter.call_count, 0) data = {'port': {'admin_state_up': True}} self._update('ports', in_port_id, data) self.assertEqual(self.ofc.create_ofc_packet_filter.call_count, 1) self.assertEqual(self.ofc.delete_ofc_packet_filter.call_count, 0)
**** CubicPower OpenStack Study **** class TestNecPluginPacketFilterWithValidate(TestNecPluginPacketFilterBase):
**** CubicPower OpenStack Study **** def setUp(self): super(TestNecPluginPacketFilterWithValidate, self).setUp() # Remove attributes explicitly from mock object to check # a case where there are no update_filter. del self.ofc.driver.update_filter self.validate_create = self.ofc.driver.validate_filter_create self.validate_update = self.ofc.driver.validate_filter_update
**** CubicPower OpenStack Study **** def test_create_pf_on_network(self): with self.packet_filter_on_network() as pf: pf_id = pf['packet_filter']['id'] self.assertEqual(pf['packet_filter']['status'], 'ACTIVE') ctx = mock.ANY pf_dict = mock.ANY expected = [ mock.call.driver.validate_filter_create(ctx, pf_dict), mock.call.exists_ofc_packet_filter(ctx, pf_id), mock.call.create_ofc_packet_filter(ctx, pf_id, pf_dict), mock.call.exists_ofc_packet_filter(ctx, pf_id), mock.call.delete_ofc_packet_filter(ctx, pf_id), ] self.ofc.assert_has_calls(expected) self.assertEqual(self.ofc.create_ofc_packet_filter.call_count, 1) self.assertEqual(self.ofc.delete_ofc_packet_filter.call_count, 1)
**** CubicPower OpenStack Study **** def test_update_pf_on_network(self): ctx = mock.ANY pf_dict = mock.ANY with self.packet_filter_on_network(admin_state_up=False) as pf: pf_id = pf['packet_filter']['id'] self.assertFalse(self.ofc.create_ofc_packet_filter.called) data = {'packet_filter': {'admin_state_up': True}} self._update('packet_filters', pf_id, data) self.ofc.create_ofc_packet_filter.assert_called_once_with( ctx, pf_id, pf_dict) self.ofc.driver.validate_filter_update.assert_called_once_with( ctx, data['packet_filter']) self.assertFalse(self.ofc.delete_ofc_packet_filter.called) data = {'packet_filter': {'admin_state_up': False}} self._update('packet_filters', pf_id, data) self.ofc.delete_ofc_packet_filter.assert_called_once_with( ctx, pf_id) self.assertEqual( 2, self.ofc.driver.validate_filter_update.call_count)
**** CubicPower OpenStack Study **** def test_create_pf_on_network_with_validation_error(self): self.validate_create.side_effect = ext_pf.PacketFilterInvalidPriority( min=1, max=65535) with self.network() as net: net_id = net['network']['id'] e = self.assertRaises(webob.exc.HTTPClientError, self._make_packet_filter, self.fmt, net_id, expected_res_status=400) self.assertEqual(400, e.status_int)
**** CubicPower OpenStack Study **** def test_update_pf_on_network_with_validation_error(self): self.validate_update.side_effect = ( ext_pf.PacketFilterUpdateNotSupported(field='priority')) with self.packet_filter_on_network() as pf: pf_id = pf['packet_filter']['id'] pf_ref = self._show('packet_filters', pf_id) self.assertEqual(pf_ref['packet_filter']['status'], 'ACTIVE') data = {'packet_filter': {'priority': 1000}} self._update('packet_filters', pf_id, data, expected_code=400)
**** CubicPower OpenStack Study **** class TestNecPluginPacketFilterWithFilterUpdate(TestNecPluginPacketFilterBase):
**** CubicPower OpenStack Study **** def setUp(self): super(TestNecPluginPacketFilterWithFilterUpdate, self).setUp() # Remove attributes explicitly from mock object to check # a case where there are no update_filter and validate_*. del self.ofc.driver.validate_filter_create del self.ofc.driver.validate_filter_update
**** CubicPower OpenStack Study **** def test_update_pf_toggle_admin_state(self): ctx = mock.ANY pf_dict = mock.ANY with self.packet_filter_on_network(admin_state_up=False) as pf: pf_id = pf['packet_filter']['id'] self.assertFalse(self.ofc.create_ofc_packet_filter.called) data = {'packet_filter': {'admin_state_up': True}} self._update('packet_filters', pf_id, data) self.ofc.create_ofc_packet_filter.assert_called_once_with( ctx, pf_id, pf_dict) self.assertFalse(self.ofc.delete_ofc_packet_filter.called) data = {'packet_filter': {'admin_state_up': False}} self._update('packet_filters', pf_id, data) self.ofc.delete_ofc_packet_filter.assert_called_once_with( ctx, pf_id)
**** CubicPower OpenStack Study **** def test_update_pf_change_field(self): ctx = mock.ANY with self.packet_filter_on_network(admin_state_up=True) as pf: pf_id = pf['packet_filter']['id'] self.assertTrue(self.ofc.create_ofc_packet_filter.called) data = {'packet_filter': {'src_mac': '12:34:56:78:9a:bc'}} self._update('packet_filters', pf_id, data) self.ofc.update_ofc_packet_filter.assert_called_once_with( ctx, pf_id, data['packet_filter']) self.assertEqual(1, self.ofc.update_ofc_packet_filter.call_count) self.assertFalse(self.ofc.delete_ofc_packet_filter.called) data = {'packet_filter': {'admin_state_up': False}} self._update('packet_filters', pf_id, data) self.ofc.delete_ofc_packet_filter.assert_called_once_with( ctx, pf_id) data = {'packet_filter': {'src_mac': '11:22:33:44:55:66'}} self._update('packet_filters', pf_id, data) self.assertEqual(1, self.ofc.update_ofc_packet_filter.call_count) data = {'packet_filter': {'admin_state_up': True}} self._update('packet_filters', pf_id, data) data = {'packet_filter': {'src_mac': '66:55:44:33:22:11'}} self._update('packet_filters', pf_id, data) self.assertEqual(2, self.ofc.update_ofc_packet_filter.call_count)
|