¡@

Home 

OpenStack Study: test_security_group.py

OpenStack Index

**** CubicPower OpenStack Study ****

# Copyright (c) 2013 OpenStack Foundation

# Copyright 2013, Nachi Ueno, NTT MCL, Inc.

# All Rights Reserved.

#

# 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.api.v2 import attributes

from neutron.extensions import securitygroup as ext_sg

from neutron import manager

from neutron.tests.unit import test_extension_security_group as test_sg

from neutron.tests.unit import test_security_groups_rpc as test_sg_rpc

PLUGIN_NAME = 'neutron.plugins.ml2.plugin.Ml2Plugin'

NOTIFIER = 'neutron.plugins.ml2.rpc.AgentNotifierApi'

**** CubicPower OpenStack Study ****

class Ml2SecurityGroupsTestCase(test_sg.SecurityGroupDBTestCase):

_plugin_name = PLUGIN_NAME

**** CubicPower OpenStack Study ****

    def setUp(self, plugin=None):

        test_sg_rpc.set_firewall_driver(test_sg_rpc.FIREWALL_HYBRID_DRIVER)

        notifier_p = mock.patch(NOTIFIER)

        notifier_cls = notifier_p.start()

        self.notifier = mock.Mock()

        notifier_cls.return_value = self.notifier

        self._attribute_map_bk_ = {}

        for item in attributes.RESOURCE_ATTRIBUTE_MAP:

            self._attribute_map_bk_[item] = (attributes.

                                             RESOURCE_ATTRIBUTE_MAP[item].

                                             copy())

        super(Ml2SecurityGroupsTestCase, self).setUp(PLUGIN_NAME)

**** CubicPower OpenStack Study ****

    def tearDown(self):

        super(Ml2SecurityGroupsTestCase, self).tearDown()

        attributes.RESOURCE_ATTRIBUTE_MAP = self._attribute_map_bk_

**** CubicPower OpenStack Study ****

class TestMl2SecurityGroups(Ml2SecurityGroupsTestCase, test_sg.TestSecurityGroups, test_sg_rpc.SGNotificationTestMixin):

**** CubicPower OpenStack Study ****

    def setUp(self):

        super(TestMl2SecurityGroups, self).setUp()

        plugin = manager.NeutronManager.get_plugin()

        plugin.start_rpc_listener()

**** CubicPower OpenStack Study ****

    def test_security_group_get_port_from_device(self):

        with self.network() as n:

            with self.subnet(n):

                with self.security_group() as sg:

                    security_group_id = sg['security_group']['id']

                    res = self._create_port(self.fmt, n['network']['id'])

                    port = self.deserialize(self.fmt, res)

                    fixed_ips = port['port']['fixed_ips']

                    data = {'port': {'fixed_ips': fixed_ips,

                                     'name': port['port']['name'],

                                     ext_sg.SECURITYGROUPS:

                                     [security_group_id]}}

                    req = self.new_update_request('ports', data,

                                                  port['port']['id'])

                    res = self.deserialize(self.fmt,

                                           req.get_response(self.api))

                    port_id = res['port']['id']

                    plugin = manager.NeutronManager.get_plugin()

                    port_dict = plugin.callbacks.get_port_from_device(port_id)

                    self.assertEqual(port_id, port_dict['id'])

                    self.assertEqual([security_group_id],

                                     port_dict[ext_sg.SECURITYGROUPS])

                    self.assertEqual([], port_dict['security_group_rules'])

                    self.assertEqual([fixed_ips[0]['ip_address']],

                                     port_dict['fixed_ips'])

                    self._delete('ports', port_id)

**** CubicPower OpenStack Study ****

    def test_security_group_get_port_from_device_with_no_port(self):

        plugin = manager.NeutronManager.get_plugin()

        port_dict = plugin.callbacks.get_port_from_device('bad_device_id')

        self.assertIsNone(port_dict)

**** CubicPower OpenStack Study ****

class TestMl2SecurityGroupsXML(TestMl2SecurityGroups):

fmt = 'xml'

**** CubicPower OpenStack Study ****

class TestMl2SGServerRpcCallBack( Ml2SecurityGroupsTestCase, test_sg_rpc.SGServerRpcCallBackMixinTestCase):

pass

**** CubicPower OpenStack Study ****

class TestMl2SGServerRpcCallBackXML( Ml2SecurityGroupsTestCase, test_sg_rpc.SGServerRpcCallBackMixinTestCaseXML):

pass