x
/
group-based-policy
Code Issues Proposed changes

Adding icmp_code and icmp_type for SG rule 

Change-Id: Ibcb64c3cd3232b81d4ab32228ba330a3a88a506f
(cherry picked from commit 1f6ba97795)
This commit is contained in:
ajay goyal 2019-08-05 18:08:29 -07:00
parent 647da5e17b
commit b3e5abe05a
2 changed files with 62 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
gbpservice/neutron/plugins/ml2plus/drivers/apic_aim/mechanism_driver.py
View File

@ -2485,6 +2485,14 @@ class ApicMechanismDriver(api_plus.MechanismDriver,
ip_protocol=(sg_rule['protocol'] if sg_rule['protocol']
else 'unspecified'),
remote_ips=remote_ips,
icmp_code=(sg_rule['port_range_min']
if (sg_rule['port_range_min'] and
sg_rule['protocol'].lower() == 'icmp')
else 'unspecified'),
icmp_type=(sg_rule['port_range_max']
if (sg_rule['port_range_max'] and
sg_rule['protocol'].lower() == 'icmp')
else 'unspecified'),
from_port=(sg_rule['port_range_min']
if sg_rule['port_range_min'] else 'unspecified'),
to_port=(sg_rule['port_range_max']

54
gbpservice/neutron/tests/unit/plugins/ml2plus/test_apic_aim.py
View File

@ -1016,6 +1016,17 @@ class TestAimMapping(ApicAimTestCase):
self.assertEqual((str(sg_rule['port_range_max']) if
sg_rule['port_range_max'] else 'unspecified'),
aim_sg_rule.to_port)
if (sg_rule['protocol'] and sg_rule['protocol'].lower() == 'icmp'):
if (sg_rule['port_range_min']):
self.assertEqual(str(sg_rule['port_range_min']),
aim_sg_rule.icmp_code)
else:
self.assertEqual(aim_sg_rule.icmp_code, 'unspecified')
if (sg_rule['port_range_max']):
self.assertEqual(str(sg_rule['port_range_max']),
aim_sg_rule.icmp_type)
else:
self.assertEqual(aim_sg_rule.icmp_type, 'unspecified')
def _check_router(self, router, expected_gw_ips, scopes=None,
unscoped_project=None, is_svi_net=False):
@ -1334,6 +1345,25 @@ class TestAimMapping(ApicAimTestCase):
sg_rule = self._make_security_group_rule(
self.fmt, rules)['security_group_rules'][0]
self._check_sg_rule(sg_id, sg_rule)
rule2 = self._build_security_group_rule(
sg_id, 'ingress', n_constants.PROTO_NAME_ICMP, '8', '100',
remote_ip_prefix='1.1.1.1/0', remote_group_id=None,
ethertype=n_constants.IPv4)
rules = {'security_group_rules': [rule2['security_group_rule']]}
sg_rule = self._make_security_group_rule(
self.fmt, rules)['security_group_rules'][0]
self._check_sg_rule(sg_id, sg_rule)
rule3 = self._build_security_group_rule(
sg_id, 'ingress', n_constants.PROTO_NAME_ICMP, None, None,
remote_ip_prefix='1.1.1.1/0', remote_group_id=None,
ethertype=n_constants.IPv4)
rules = {'security_group_rules': [rule3['security_group_rule']]}
sg_rule = self._make_security_group_rule(
self.fmt, rules)['security_group_rules'][0]
self._check_sg_rule(sg_id, sg_rule)
sg = self._show('security-groups', sg_id)['security_group']
self._check_sg(sg)
@ -8411,6 +8441,30 @@ class TestPortOnPhysicalNode(TestPortVlanNetwork):
sg_rule1['id'], 'default', default_sg_id, tenant_aname)
self.assertEqual(aim_sg_rule.remote_ips, ['10.0.1.100'])
rule2 = self._build_security_group_rule(
default_sg_id, 'ingress', n_constants.PROTO_NAME_ICMP, '2', '33',
remote_group_id=default_sg_id, ethertype=n_constants.IPv4)
rules = {'security_group_rules': [rule2['security_group_rule']]}
sg_rule2 = self._make_security_group_rule(
self.fmt, rules)['security_group_rules'][0]
aim_sg_rule = self._get_sg_rule(
sg_rule2['id'], 'default', default_sg_id, tenant_aname)
self.assertEqual(aim_sg_rule.remote_ips, ['10.0.1.100'])
self.assertEqual(aim_sg_rule.icmp_code, '2')
self.assertEqual(aim_sg_rule.icmp_type, '33')
rule3 = self._build_security_group_rule(
default_sg_id, 'ingress', n_constants.PROTO_NAME_ICMP, None, None,
remote_group_id=default_sg_id, ethertype=n_constants.IPv4)
rules = {'security_group_rules': [rule3['security_group_rule']]}
sg_rule3 = self._make_security_group_rule(
self.fmt, rules)['security_group_rules'][0]
aim_sg_rule = self._get_sg_rule(
sg_rule3['id'], 'default', default_sg_id, tenant_aname)
self.assertEqual(aim_sg_rule.remote_ips, ['10.0.1.100'])
self.assertEqual(aim_sg_rule.icmp_code, 'unspecified')
self.assertEqual(aim_sg_rule.icmp_type, 'unspecified')
# delete SG from port
data = {'port': {'security_groups': []}}
port = self._update('ports', port['id'], data)['port']