openstack/neutron
Code Issues Proposed changes

Merge "Trigger provider security group update for RA"

Browse Source
This commit is contained in:
Jenkins 2014-07-05 01:15:52 +00:00 committed by Gerrit Code Review
commit 6e877945c9
3 changed files with 106 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
neutron/db/securitygroups_rpc_base.py
View File

@ -118,6 +118,12 @@ class SecurityGroupServerRpcMixin(sg_db.SecurityGroupDbMixin):
"""
if port['device_owner'] == q_const.DEVICE_OWNER_DHCP:
self.notifier.security_groups_provider_updated(context)
# For IPv6, provider rule need to be updated in case router
# interface is created or updated after VM port is created.
elif port['device_owner'] == q_const.DEVICE_OWNER_ROUTER_INTF:
if any(netaddr.IPAddress(fixed_ip['ip_address']).version == 6
for fixed_ip in port['fixed_ips']):
self.notifier.security_groups_provider_updated(context)
else:
self.notifier.security_groups_member_updated(
context, port.get(ext_sg.SECURITYGROUPS))
@ -246,8 +252,6 @@ class SecurityGroupServerRpcCallbackMixin(object):
gateway_ip = subnet['gateway_ip']
if subnet['ip_version'] != 6 or not gateway_ip:
continue
# TODO(xuhanp): Figure out how to call the following code
# each time router is created or updated.
if not netaddr.IPAddress(gateway_ip).is_link_local():
if subnet['ipv6_ra_mode']:
gateway_ip = self._get_lla_gateway_ip_for_subnet(context,

12
neutron/tests/unit/oneconvergence/test_security_group.py
View File

@ -87,6 +87,12 @@ class TestOneConvergenceSGServerRpcCallBack(
def test_security_group_rule_for_device_ipv6_multi_router_interfaces(self):
self.skipTest("NVSD Plugin does not support IPV6.")
def test_notify_security_group_ipv6_gateway_port_added(self):
self.skipTest("NVSD Plugin does not support IPV6.")
def test_notify_security_group_ipv6_normal_port_added(self):
self.skipTest("NVSD Plugin does not support IPV6.")
class TestOneConvergenceSGServerRpcCallBackXML(
OneConvergenceSecurityGroupsTestCase,
@ -112,6 +118,12 @@ class TestOneConvergenceSGServerRpcCallBackXML(
def test_security_group_rule_for_device_ipv6_multi_router_interfaces(self):
self.skipTest("NVSD Plugin does not support IPV6.")
def test_notify_security_group_ipv6_gateway_port_added(self):
self.skipTest("NVSD Plugin does not support IPV6.")
def test_notify_security_group_ipv6_normal_port_added(self):
self.skipTest("NVSD Plugin does not support IPV6.")
class TestOneConvergenceSecurityGroups(OneConvergenceSecurityGroupsTestCase,
test_sg.TestSecurityGroups,

88
neutron/tests/unit/test_security_groups_rpc.py
View File

@ -64,6 +64,94 @@ class SGServerRpcCallBackMixinTestCase(test_sg.SecurityGroupDBTestCase):
super(SGServerRpcCallBackMixinTestCase, self).setUp(plugin)
self.rpc = FakeSGCallback()
def _test_security_group_port(self, device_owner, gw_ip,
cidr, ip_version, ip_address):
with self.network() as net:
with self.subnet(net,
gateway_ip=gw_ip,
cidr=cidr,
ip_version=ip_version) as subnet:
with mock.patch.object(
self.notifier,
'security_groups_provider_updated') as mock_notifier:
kwargs = {
'fixed_ips': [{'subnet_id': subnet['subnet']['id'],
'ip_address': ip_address}]}
if device_owner:
kwargs['device_owner'] = device_owner
res = self._create_port(
self.fmt, net['network']['id'], **kwargs)
res = self.deserialize(self.fmt, res)
port_id = res['port']['id']
if device_owner == const.DEVICE_OWNER_ROUTER_INTF:
data = {'port': {'fixed_ips': []}}
req = self.new_update_request('ports', data, port_id)
res = self.deserialize(self.fmt,
req.get_response(self.api))
self._delete('ports', port_id)
return mock_notifier
def test_notify_security_group_ipv6_gateway_port_added(self):
if getattr(self, "notifier", None) is None:
self.skipTest("Notifier mock is not set so security group "
"RPC calls can't be tested")
mock_notifier = self._test_security_group_port(
const.DEVICE_OWNER_ROUTER_INTF,
'2001:0db8::1',
'2001:0db8::/64',
6,
'2001:0db8::1')
self.assertTrue(mock_notifier.called)
def test_notify_security_group_ipv6_normal_port_added(self):
if getattr(self, "notifier", None) is None:
self.skipTest("Notifier mock is not set so security group "
"RPC calls can't be tested")
mock_notifier = self._test_security_group_port(
None,
'2001:0db8::1',
'2001:0db8::/64',
6,
'2001:0db8::3')
self.assertFalse(mock_notifier.called)
def test_notify_security_group_ipv4_dhcp_port_added(self):
if getattr(self, "notifier", None) is None:
self.skipTest("Notifier mock is not set so security group "
"RPC calls can't be tested")
mock_notifier = self._test_security_group_port(
const.DEVICE_OWNER_DHCP,
'192.168.1.1',
'192.168.1.0/24',
4,
'192.168.1.2')
self.assertTrue(mock_notifier.called)
def test_notify_security_group_ipv4_gateway_port_added(self):
if getattr(self, "notifier", None) is None:
self.skipTest("Notifier mock is not set so security group "
"RPC calls can't be tested")
mock_notifier = self._test_security_group_port(
const.DEVICE_OWNER_ROUTER_INTF,
'192.168.1.1',
'192.168.1.0/24',
4,
'192.168.1.1')
self.assertFalse(mock_notifier.called)
def test_notify_security_group_ipv4_normal_port_added(self):
if getattr(self, "notifier", None) is None:
self.skipTest("Notifier mock is not set so security group "
"RPC calls can't be tested")
mock_notifier = self._test_security_group_port(
None,
'192.168.1.1',
'192.168.1.0/24',
4,
'192.168.1.3')
self.assertFalse(mock_notifier.called)
def test_security_group_rules_for_devices_ipv4_ingress(self):
fake_prefix = FAKE_PREFIX[const.IPv4]
with self.network() as n: