¡@

OpenStack Study: test_mlnx_security_group.py

OpenStack Index

**** CubicPower OpenStack Study ****

# 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

import webob.exc

from neutron.api.v2 import attributes

from neutron.extensions import securitygroup as ext_sg

from neutron.plugins.mlnx.db import mlnx_db_v2 as mlnx_db

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

'mlnx_plugin.MellanoxEswitchPlugin')

NOTIFIER = ('neutron.plugins.mlnx.'

'agent_notify_api.AgentNotifierApi')

**** CubicPower OpenStack Study ****

class MlnxSecurityGroupsTestCase(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_IPTABLES_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(MlnxSecurityGroupsTestCase, self).setUp(PLUGIN_NAME)

**** CubicPower OpenStack Study ****

def tearDown(self):

attributes.RESOURCE_ATTRIBUTE_MAP = self._attribute_map_bk_

super(MlnxSecurityGroupsTestCase, self).tearDown()

**** CubicPower OpenStack Study ****

class TestMlnxSecurityGroups(MlnxSecurityGroupsTestCase, test_sg.TestSecurityGroups, test_sg_rpc.SGNotificationTestMixin):

pass

**** CubicPower OpenStack Study ****

class TestMlnxSecurityGroupsXML(TestMlnxSecurityGroups):

fmt = 'xml'

**** CubicPower OpenStack Study ****

class TestMlnxSecurityGroupsDB(MlnxSecurityGroupsTestCase):

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

if res.status_int >= 400:

raise webob.exc.HTTPClientError(code=res.status_int)

res = self.deserialize(self.fmt,

req.get_response(self.api))

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

device_id = port_id[:8]

port_dict = mlnx_db.get_port_from_device(device_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['port']['id'])

**** CubicPower OpenStack Study ****

def test_security_group_get_port_from_device_with_no_port(self):

port_dict = mlnx_db.get_port_from_device('bad_device_id')

self.assertIsNone(port_dict)

**** CubicPower OpenStack Study ****

class TestMlnxSecurityGroupsDBXML(TestMlnxSecurityGroupsDB):

fmt = 'xml'