Merge "Adding unittests for security-group implementation"

This commit is contained in:
Jenkins 2015-11-17 18:58:23 +00:00 committed by Gerrit Code Review
commit 402c3bb90a
4 changed files with 98 additions and 11 deletions

View File

@ -231,16 +231,15 @@ def _init_default_section(name, description, nsgroup_id):
else:
section = firewall.create_empty_section(
name, description, [nsgroup_id], [])
# TODO(roeyc): Add aditional rules to allow IPV6 NDP.
block_rule = firewall.get_firewall_rule_dict(
'Block All', action=firewall.DROP)
# TODO(roeyc): Add additional rules to allow IPV6 NDP.
dhcp_client = firewall.get_nsservice(firewall.L4_PORT_SET_NSSERVICE,
l4_protocol=firewall.UDP,
source_ports=[67],
destination_ports=[68])
dhcp_client_rule_in = firewall.get_firewall_rule_dict(
'DHCP-Client-IN', direction=firewall.IN, service=dhcp_client)
'DHCP Reply', direction=firewall.IN, service=dhcp_client)
dhcp_server = (
firewall.get_nsservice(firewall.L4_PORT_SET_NSSERVICE,
@ -248,7 +247,7 @@ def _init_default_section(name, description, nsgroup_id):
source_ports=[68],
destination_ports=[67]))
dhcp_client_rule_out = firewall.get_firewall_rule_dict(
'DHCP-Client-OUT', direction=firewall.OUT, service=dhcp_server)
'DHCP Request', direction=firewall.OUT, service=dhcp_server)
firewall.add_rules_in_section([dhcp_client_rule_out,
dhcp_client_rule_in,

View File

@ -836,8 +836,6 @@ class NsxV3Plugin(addr_pair_db.AllowedAddressPairsMixin,
context, id, {'port': original_port}, updated_port,
original_port)
#TODO(roeyc): add port to nsgroups
return updated_port
def _extract_external_gw(self, context, router, is_extract=True):

View File

@ -0,0 +1,95 @@
# Copyright (c) 2015 VMware, 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.tests.unit.extensions import test_securitygroup as ext_sg
from vmware_nsx.plugins.nsx_v3 import plugin as nsx_plugin
from vmware_nsx.tests.unit.nsx_v3 import test_plugin as test_nsxv3
class TestSecurityGroups(test_nsxv3.NsxV3PluginTestCaseMixin,
ext_sg.TestSecurityGroups):
@mock.patch.object(nsx_plugin.security.firewall, 'remove_nsgroup_member')
@mock.patch.object(nsx_plugin.security.firewall, 'add_nsgroup_member')
@mock.patch.object(nsx_plugin.security.firewall, 'create_nsgroup')
def test_create_port_with_multiple_security_groups(self,
create_nsgroup_mock,
add_member_mock,
remove_member_mock):
NSG_IDS = ['11111111-1111-1111-1111-111111111111',
'22222222-2222-2222-2222-222222222222',
'33333333-3333-3333-3333-333333333333']
count = [-1]
def _create_nsgroup_mock(x, y, z):
count[0] += 1
return {'id': NSG_IDS[count[0]]}
create_nsgroup_mock.side_effect = _create_nsgroup_mock
super(TestSecurityGroups,
self).test_create_port_with_multiple_security_groups()
# The first nsgroup is associated with the default secgroup, which is
# not added to this port.
calls = [mock.call(NSG_IDS[1], mock.ANY, mock.ANY),
mock.call(NSG_IDS[2], mock.ANY, mock.ANY)]
add_member_mock.assert_has_calls(calls, any_order=True)
@mock.patch.object(nsx_plugin.security.firewall, 'remove_nsgroup_member')
@mock.patch.object(nsx_plugin.security.firewall, 'add_nsgroup_member')
@mock.patch.object(nsx_plugin.security.firewall, 'create_nsgroup')
def test_update_port_with_multiple_security_groups(self,
create_nsgroup_mock,
add_member_mock,
remove_member_mock):
NSG_IDS = ['11111111-1111-1111-1111-111111111111',
'22222222-2222-2222-2222-222222222222',
'33333333-3333-3333-3333-333333333333']
count = [-1]
def _create_nsgroup_mock(x, y, z):
count[0] += 1
return {'id': NSG_IDS[count[0]]}
create_nsgroup_mock.side_effect = _create_nsgroup_mock
super(TestSecurityGroups,
self).test_update_port_with_multiple_security_groups()
calls = [mock.call(NSG_IDS[0], mock.ANY, mock.ANY),
mock.call(NSG_IDS[1], mock.ANY, mock.ANY),
mock.call(NSG_IDS[2], mock.ANY, mock.ANY)]
add_member_mock.assert_has_calls(calls, any_order=True)
remove_member_mock.assert_called_with(NSG_IDS[0], mock.ANY)
@mock.patch.object(nsx_plugin.security.firewall, 'remove_nsgroup_member')
@mock.patch.object(nsx_plugin.security.firewall, 'add_nsgroup_member')
@mock.patch.object(nsx_plugin.security.firewall, 'create_nsgroup')
def test_update_port_remove_security_group_empty_list(self,
create_nsgroup_mock,
add_member_mock,
remove_member_mock):
NSG_ID = '11111111-1111-1111-1111-111111111111'
create_nsgroup_mock.side_effect = lambda x, y, z: {'id': NSG_ID}
super(TestSecurityGroups,
self).test_update_port_remove_security_group_empty_list()
add_member_mock.assert_called_with(NSG_ID, mock.ANY, mock.ANY)
remove_member_mock.assert_called_with(NSG_ID, mock.ANY)

View File

@ -31,7 +31,6 @@ from neutron.tests.unit.extensions import test_extraroute as test_ext_route
from neutron.tests.unit.extensions import test_l3 as test_l3_plugin
from neutron.tests.unit.extensions \
import test_l3_ext_gw_mode as test_ext_gw_mode
from neutron.tests.unit.extensions import test_securitygroup as ext_sg
from neutron import version
from oslo_config import cfg
@ -133,10 +132,6 @@ class TestPortsV2(test_plugin.TestPortsV2, NsxV3PluginTestCaseMixin):
data['port']['fixed_ips'])
class TestSecurityGroups(NsxV3PluginTestCaseMixin, ext_sg.TestSecurityGroups):
pass
class DHCPOptsTestCase(test_dhcpopts.TestExtraDhcpOpt,
NsxV3PluginTestCaseMixin):