openstack
/
neutron
Code Issues Proposed changes

Merge "Delete sg rule which remote is the deleted sg"

This commit is contained in:
Zuul 2023-07-01 05:12:35 +00:00 committed by Gerrit Code Review
parent 33bf186467 67a0b07287
commit bce16f6f3d
2 changed files with 55 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
neutron/api/rpc/handlers/securitygroups_rpc.py
View File

@ -310,6 +310,13 @@ class SecurityGroupServerAPIShim(sg_rpc_base.SecurityGroupInfoAPIMixin):
for rule in rules:
self.rcache.record_resource_delete(context, 'SecurityGroupRule',
rule.id)
# If there's a rule which remote is the deleted sg, remove that also.
rules = self.rcache.match_resources_with_func(
'SecurityGroupRule',
lambda sg_rule: sg_rule.remote_group_id == existing.id)
for rule in rules:
self.rcache.record_resource_delete(context, 'SecurityGroupRule',
rule.id)
def _handle_sg_rule_delete(self, rtype, event, trigger, payload):
existing = payload.states[0]

49
neutron/tests/unit/api/rpc/handlers/test_securitygroups_rpc.py
View File

@ -121,13 +121,14 @@ class SecurityGroupServerAPIShimTestCase(base.BaseTestCase):
return_value=False)
def _make_security_group_ovo(self, *args, **kwargs):
attrs = {'id': uuidutils.generate_uuid(), 'revision_number': 1}
r_group = kwargs.get('remote_group_id') or attrs['id']
sg_rule = securitygroup.SecurityGroupRule(
id=uuidutils.generate_uuid(),
security_group_id=attrs['id'],
direction='ingress',
ethertype='IPv4', protocol='tcp',
port_range_min=400,
remote_group_id=attrs['id'],
remote_group_id=r_group,
revision_number=1,
remote_address_group_id=kwargs.get('remote_address_group_id',
None),
@ -263,6 +264,52 @@ class SecurityGroupServerAPIShimTestCase(base.BaseTestCase):
self.sg_agent.security_groups_member_updated.assert_called_with(
{s1.id})
def test_sg_delete_events_with_remote(self):
s1 = self._make_security_group_ovo(remote_group_id='')
s2 = self._make_security_group_ovo(remote_group_id=s1.id)
rules = self.rcache.get_resources(
'SecurityGroupRule',
filters={'security_group_id': (s1.id, s2.id)})
self.assertEqual(2, len(rules))
self.assertEqual(s1.id, rules[0].remote_group_id)
self.shim._clear_child_sg_rules(
'SecurityGroup', 'after_delete', '',
events.DBEventPayload(
context=self.ctx,
states=[s1]
)
)
rules = self.rcache.get_resources(
'SecurityGroupRule',
filters={'security_group_id': (s1.id, s2.id)})
self.assertEqual(0, len(rules))
def test_sg_delete_events_without_remote(self):
s1 = self._make_security_group_ovo()
s2 = self._make_security_group_ovo()
rules = self.rcache.get_resources(
'SecurityGroupRule',
filters={'security_group_id': (s1.id, s2.id)})
self.assertEqual(2, len(rules))
self.assertEqual(s1.id, rules[0].remote_group_id)
self.shim._clear_child_sg_rules(
'SecurityGroup', 'after_delete', '',
events.DBEventPayload(
context=self.ctx,
states=[s1]
)
)
s1_rules = self.rcache.get_resources(
'SecurityGroupRule',
filters={'security_group_id': (s1.id, )})
self.assertEqual(0, len(s1_rules))
s2_rules = self.rcache.get_resources(
'SecurityGroupRule',
filters={'security_group_id': (s2.id, )})
self.assertEqual(1, len(s2_rules))
def test_get_secgroup_ids_for_address_group(self):
ag = self._make_address_group_ovo()
sg1 = self._make_security_group_ovo(remote_address_group_id=ag.id)