¡@

Home 

OpenStack Study: test_packet_filter.py

OpenStack Index

**** 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)