¡@

Home 

OpenStack Study: test_ryu_security_group.py

OpenStack Index

**** CubicPower OpenStack Study ****

# vim: tabstop=4 shiftwidth=4 softtabstop=4

#

# Copyright 2012, 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 contextlib

import mock

from neutron.api.v2 import attributes

from neutron.extensions import securitygroup as ext_sg

from neutron import manager

from neutron.tests.unit.ryu import fake_ryu

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.ryu.'

'ryu_neutron_plugin.RyuNeutronPluginV2')

NOTIFIER = ('neutron.plugins.ryu.'

'ryu_neutron_plugin.AgentNotifierApi')

**** CubicPower OpenStack Study ****

class RyuSecurityGroupsTestCase(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)

        self.fake_ryu = fake_ryu.patch_fake_ryu_client().start()

        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(RyuSecurityGroupsTestCase, self).setUp(PLUGIN_NAME)

**** CubicPower OpenStack Study ****

    def tearDown(self):

        super(RyuSecurityGroupsTestCase, self).tearDown()

        attributes.RESOURCE_ATTRIBUTE_MAP = self._attribute_map_bk_

**** CubicPower OpenStack Study ****

class TestRyuSecurityGroups(RyuSecurityGroupsTestCase, test_sg.TestSecurityGroups, test_sg_rpc.SGNotificationTestMixin):

**** CubicPower OpenStack Study ****

    def test_security_group_get_port_from_device(self):

        with contextlib.nested(self.network(),

                               self.security_group()) as (n, sg):

            with self.subnet(n):

                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 TestRyuSecurityGroupsXML(TestRyuSecurityGroups):

fmt = 'xml'