¡@

Home 

OpenStack Study: test_midonet_plugin.py

OpenStack Index

**** CubicPower OpenStack Study ****

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

# Copyright (C) 2012 Midokura Japan K.K.

# Copyright (C) 2013 Midokura PTE LTD

# 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.

#

# @author: Rossella Sblendido, Midokura Europe SARL

# @author: Ryu Ishimoto, Midokura Japan KK

# @author: Tomoe Sugihara, Midokura Japan KK

import mock

import os

import sys

import neutron.common.test_lib as test_lib

from neutron.extensions import portbindings

from neutron.tests.unit import _test_extension_portbindings as test_bindings

import neutron.tests.unit.midonet.mock_lib as mock_lib

import neutron.tests.unit.test_db_plugin as test_plugin

import neutron.tests.unit.test_extension_security_group as sg

import neutron.tests.unit.test_l3_plugin as test_l3_plugin

MIDOKURA_PKG_PATH = "neutron.plugins.midonet.plugin"

MIDONET_PLUGIN_NAME = ('%s.MidonetPluginV2' % MIDOKURA_PKG_PATH)

# Need to mock the midonetclient module since the plugin will try to load it.

sys.modules["midonetclient"] = mock.Mock()

**** CubicPower OpenStack Study ****

class MidonetPluginV2TestCase(test_plugin.NeutronDbPluginV2TestCase):

**** CubicPower OpenStack Study ****

    def setUp(self,

              plugin=MIDONET_PLUGIN_NAME,

              ext_mgr=None,

              service_plugins=None):

        self.mock_api = mock.patch(

            'neutron.plugins.midonet.midonet_lib.MidoClient')

        etc_path = os.path.join(os.path.dirname(__file__), 'etc')

        test_lib.test_config['config_files'] = [os.path.join(

            etc_path, 'midonet.ini.test')]

        self.instance = self.mock_api.start()

        mock_cfg = mock_lib.MidonetLibMockConfig(self.instance.return_value)

        mock_cfg.setup()

        super(MidonetPluginV2TestCase, self).setUp(plugin=plugin,

                                                   ext_mgr=ext_mgr)

**** CubicPower OpenStack Study ****

    def tearDown(self):

        super(MidonetPluginV2TestCase, self).tearDown()

        self.mock_api.stop()

**** CubicPower OpenStack Study ****

class TestMidonetNetworksV2(test_plugin.TestNetworksV2, MidonetPluginV2TestCase):

pass

**** CubicPower OpenStack Study ****

class TestMidonetL3NatTestCase(test_l3_plugin.L3NatDBIntTestCase, MidonetPluginV2TestCase):

**** CubicPower OpenStack Study ****

    def setUp(self,

              plugin=MIDONET_PLUGIN_NAME,

              ext_mgr=None,

              service_plugins=None):

        super(TestMidonetL3NatTestCase, self).setUp(plugin=plugin,

                                                    ext_mgr=None,

                                                    service_plugins=None)

**** CubicPower OpenStack Study ****

    def test_floatingip_with_invalid_create_port(self):

        self._test_floatingip_with_invalid_create_port(MIDONET_PLUGIN_NAME)

**** CubicPower OpenStack Study ****

    def test_floatingip_assoc_no_port(self):

        with self.subnet(cidr='200.0.0.0/24') as public_sub:

            self._set_net_external(public_sub['subnet']['network_id'])

            res = super(TestMidonetL3NatTestCase, self)._create_floatingip(

                self.fmt, public_sub['subnet']['network_id'])

            # Cleanup

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

            self._delete('floatingips', floatingip['floatingip']['id'])

        self.assertFalse(self.instance.return_value.add_static_nat.called)

**** CubicPower OpenStack Study ****

    def test_floatingip_assoc_with_port(self):

        with self.subnet(cidr='200.0.0.0/24') as public_sub:

            self._set_net_external(public_sub['subnet']['network_id'])

            with self.port() as private_port:

                with self.router() as r:

                    # We need to hook up the private subnet to the external

                    # network in order to associate the fip.

                    sid = private_port['port']['fixed_ips'][0]['subnet_id']

                    private_sub = {'subnet': {'id': sid}}

                    self._add_external_gateway_to_router(

                        r['router']['id'],

                        public_sub['subnet']['network_id'])

                    # Check that get_link_port was called - if not, Source NAT

                    # will not be set up correctly on the MidoNet side

                    self.assertTrue(

                        self.instance.return_value.get_link_port.called)

                    self._router_interface_action('add', r['router']['id'],

                                                  private_sub['subnet']['id'],

                                                  None)

                    # Create the fip.

                    res = super(TestMidonetL3NatTestCase,

                                self)._create_floatingip(

                                    self.fmt,

                                    public_sub['subnet']['network_id'],

                                    port_id=private_port['port']['id'])

                    # Cleanup the resources used for the test

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

                    self._delete('floatingips', floatingip['floatingip']['id'])

                    self._remove_external_gateway_from_router(

                        r['router']['id'],

                        public_sub['subnet']['network_id'])

                    self._router_interface_action('remove',

                                                  r['router']['id'],

                                                  private_sub['subnet']['id'],

                                                  None)

        self.assertTrue(self.instance.return_value.add_static_nat.called)

**** CubicPower OpenStack Study ****

class TestMidonetSecurityGroupsTestCase(sg.SecurityGroupDBTestCase):

_plugin_name = ('%s.MidonetPluginV2' % MIDOKURA_PKG_PATH)

**** CubicPower OpenStack Study ****

    def setUp(self):

        self.mock_api = mock.patch(

            'neutron.plugins.midonet.midonet_lib.MidoClient')

        etc_path = os.path.join(os.path.dirname(__file__), 'etc')

        test_lib.test_config['config_files'] = [os.path.join(

            etc_path, 'midonet.ini.test')]

        self.instance = self.mock_api.start()

        mock_cfg = mock_lib.MidonetLibMockConfig(self.instance.return_value)

        mock_cfg.setup()

        super(TestMidonetSecurityGroupsTestCase, self).setUp(self._plugin_name)

**** CubicPower OpenStack Study ****

class TestMidonetSecurityGroup(sg.TestSecurityGroups, TestMidonetSecurityGroupsTestCase):

pass

**** CubicPower OpenStack Study ****

class TestMidonetSubnetsV2(test_plugin.TestSubnetsV2, MidonetPluginV2TestCase):

# IPv6 is not supported by MidoNet yet. Ignore tests that attempt to

# create IPv6 subnet.

**** CubicPower OpenStack Study ****

    def test_create_subnet_inconsistent_ipv6_cidrv4(self):

        pass

**** CubicPower OpenStack Study ****

    def test_create_subnet_inconsistent_ipv6_dns_v4(self):

        pass

**** CubicPower OpenStack Study ****

    def test_create_subnet_with_v6_allocation_pool(self):

        pass

**** CubicPower OpenStack Study ****

    def test_update_subnet_inconsistent_ipv6_gatewayv4(self):

        pass

**** CubicPower OpenStack Study ****

    def test_update_subnet_inconsistent_ipv6_hostroute_dst_v4(self):

        pass

**** CubicPower OpenStack Study ****

    def test_update_subnet_inconsistent_ipv6_hostroute_np_v4(self):

        pass

**** CubicPower OpenStack Study ****

    def test_create_subnet_inconsistent_ipv6_gatewayv4(self):

        pass

**** CubicPower OpenStack Study ****

    def test_create_subnet_dhcp_disabled(self):

        super(TestMidonetSubnetsV2, self)._test_create_subnet(

            enable_dhcp=False)

        self.assertFalse(self.instance.return_value.create_dhcp.called)

**** CubicPower OpenStack Study ****

class TestMidonetPortsV2(test_plugin.TestPortsV2, MidonetPluginV2TestCase):

# IPv6 is not supported by MidoNet yet. Ignore tests that attempt to

# create IPv6 subnet.

**** CubicPower OpenStack Study ****

    def test_requested_subnet_id_v4_and_v6(self):

        pass

**** CubicPower OpenStack Study ****

    def test_vif_port_binding(self):

        with self.port(name='myname') as port:

            self.assertEqual('midonet', port['port']['binding:vif_type'])

            self.assertTrue(port['port']['admin_state_up'])

**** CubicPower OpenStack Study ****

class TestMidonetPluginPortBinding(test_bindings.PortBindingsTestCase, MidonetPluginV2TestCase):

VIF_TYPE = portbindings.VIF_TYPE_MIDONET

HAS_PORT_FILTER = True

**** CubicPower OpenStack Study ****

    def setUp(self):

        super(TestMidonetPluginPortBinding, self).setUp()