KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use update_security_group method in tempest

Best Python code snippet using tempest_python

test_security_groups.py

Source:test_security_groups.py

...157                } for port in [80, 443]]158            }159        )160        with self.stubber:161            main.update_security_group(self.managed_sg, self.allowed_ranges)162            self.stubber.assert_no_pending_responses()163    def test_update_security_group_adds_partial_rules(self):164        """ Test update_security_group only adds relevant rules """165        self.stubber.add_response(166            'authorize_security_group_ingress',167            {},168            {169                "GroupId": f'sg-{self.managed_id}',170                "IpPermissions": [{171                    "FromPort": port,172                    "ToPort": port,173                    "IpProtocol": "tcp",174                    "IpRanges": [{175                        "Description": "GitHub",176                        "CidrIp": ip_range177                    } for ip_range in self.allowed_ranges[1:]]178                } for port in [80, 443]]179            }180        )181        with self.stubber:182            main.update_security_group({183                "GroupId": f'sg-{self.managed_id}',184                "VpcId": f'vpc-{self.managed_id}',185                "IpPermissionsEgress": [],186                "IpPermissions": [{187                    "FromPort": port,188                    "ToPort": port,189                    "IpRanges": [{190                        "Description": "GitHub",191                        "CidrIp": self.allowed_ranges[0]192                    }]193                } for port in [80, 443]]194            }, self.allowed_ranges)195            self.stubber.assert_no_pending_responses()196    def test_update_security_group_removes_ingress_rules(self):197        """ Test update_security_group removes any rules NOT in the allowed list """198        ranges = [item for item in self.allowed_ranges] # This MUST be a copy, not a reference199        ranges.append("10.0.0.0/8")200        self.stubber.add_response(201            'revoke_security_group_ingress',202            {},203            {204                "GroupId": f'sg-{self.managed_id}',205                "IpPermissions": [{206                    "FromPort": port,207                    "ToPort": port,208                    "IpProtocol": "tcp",209                    "IpRanges": [{210                        "Description": "GitHub",211                        "CidrIp": "10.0.0.0/8"212                    }]213                } for port in [80, 443]]214            }215        )216        with self.stubber:217            main.update_security_group({218                "GroupId": f'sg-{self.managed_id}',219                "VpcId": f'vpc-{self.managed_id}',220                "IpPermissionsEgress": [],221                "IpPermissions": [{222                    "FromPort": port,223                    "ToPort": port,224                    "IpRanges": [{225                        "Description": "GitHub",226                        "CidrIp": ip_range227                    } for ip_range in ranges]228                } for port in [80, 443]]229            }, self.allowed_ranges)230            self.stubber.assert_no_pending_responses()231    def test_update_security_group_removes_egress_rules(self):232        """ Test update_security_group removes ALL egress rules """233        self.stubber.add_response(234            'revoke_security_group_egress',235            {},236            {237                "GroupId": f'sg-{self.managed_id}',238                "IpPermissions": [{239                    "FromPort": 80,240                    "ToPort": 80,241                    "IpProtocol": "tcp",242                    "IpRanges": [{243                        "Description": "TestNet",244                        "CidrIp": "0.0.0.0/0"245                    }]246                }]247            }248        )249        with self.stubber:250            main.update_security_group({251                "GroupId": f'sg-{self.managed_id}',252                "VpcId": f'vpc-{self.managed_id}',253                "IpPermissionsEgress": [{254                    "FromPort": 80,255                    "ToPort": 80,256                    "IpProtocol": "tcp",257                    "IpRanges": [{258                        "Description": "TestNet",259                        "CidrIp": "0.0.0.0/0"260                    }]261                }],262                "IpPermissions": []263            }, [])264        self.stubber.assert_no_pending_responses()

lambda_function.py

Source:lambda_function.py

...55    global_https_updated = 056    region_http_updated = 057    region_https_updated = 058    for group in global_http_group:59        if update_security_group(client, group, new_ranges["GLOBAL"], INGRESS_PORTS['Http']):60            global_http_updated += 161            result.append('Updated ' + group['GroupId'])62    for group in global_https_group:63        if update_security_group(client, group, new_ranges["GLOBAL"], INGRESS_PORTS['Https']):64            global_https_updated += 165            result.append('Updated ' + group['GroupId'])66    for group in region_http_group:67        if update_security_group(client, group, new_ranges["REGION"], INGRESS_PORTS['Http']):68            region_http_updated += 169            result.append('Updated ' + group['GroupId'])70    for group in region_https_group:71        if update_security_group(client, group, new_ranges["REGION"], INGRESS_PORTS['Https']):72            region_https_updated += 173            result.append('Updated ' + group['GroupId'])74    result.append('Updated ' + str(global_http_updated) + ' of ' + str(len(global_http_group)) + ' CloudFront_g HttpSecurityGroups')75    result.append('Updated ' + str(global_https_updated) + ' of ' + str(len(global_https_group)) + ' CloudFront_g HttpsSecurityGroups')76    result.append('Updated ' + str(region_http_updated) + ' of ' + str(len(region_http_group)) + ' CloudFront_r HttpSecurityGroups')77    result.append('Updated ' + str(region_https_updated) + ' of ' + str(len(region_https_group)) + ' CloudFront_r HttpsSecurityGroups')78    return result79def update_security_group(client, group, new_ranges, port):80    added = 081    removed = 082    if len(group['IpPermissions']) > 0:83        for permission in group['IpPermissions']:84            if permission['FromPort'] <= port and permission['ToPort'] >= port :85                old_prefixes = list()86                to_revoke = list()87                to_add = list()88                for range in permission['IpRanges']:89                    cidr = range['CidrIp']90                    old_prefixes.append(cidr)91                    if new_ranges.count(cidr) == 0:92                        to_revoke.append(range)93                        print(group['GroupId'] + ": Revoking " + cidr + ":" + str(permission['ToPort']))...

test_security_group.py

Source:test_security_group.py

...33    def test_kwargs(self):34        ec2 = boto3.client('ec2')35        sg_id = ec2.create_security_group(Description='test', GroupName='test_kwargs')['GroupId']36        with pytest.raises(ValueError):37            update_security_group()38        with pytest.raises(ValueError):39            update_security_group(security_group_id=sg_id)40        with pytest.raises(ValueError):41            update_security_group(security_group_id=sg_id, port=123)42        with pytest.raises(ValueError):43            update_security_group(security_group_id=sg_id, port=123, protocol='tcp')44        with pytest.raises(ValueError):45            update_security_group(security_group_id=sg_id, port=123, protocol='tcp', allowed_ipv4_addresses=[])46        assert update_security_group(security_group_id=sg_id, port=123, protocol='tcp', allowed_ipv4_addresses=[], allowed_ipv6_addresses=[])47    def test_update(self):48        ec2 = boto3.client('ec2')49        sg_id = ec2.create_security_group(Description='test', GroupName='test_update')['GroupId']50        event['allowed_ipv4_addresses'] = [ipv4_addresses[0]]51        expected_result['authorized_ipv4_addresses'] = [ipv4_addresses[0]]52        assert update_security_group(security_group_id=sg_id, **event) == expected_result53        event['allowed_ipv4_addresses'] = [ipv4_addresses[0], ipv4_addresses[1]]54        expected_result['authorized_ipv4_addresses'] = [ipv4_addresses[1]]55        assert update_security_group(security_group_id=sg_id, **event) == expected_result56        event['allowed_ipv4_addresses'] = [ipv4_addresses[0], ipv4_addresses[1], ipv4_addresses[2]]57        expected_result['authorized_ipv4_addresses'] = [ipv4_addresses[2]]58        assert update_security_group(security_group_id=sg_id, **event) == expected_result59        event['allowed_ipv4_addresses'] = [ipv4_addresses[0], ipv4_addresses[1]]60        expected_result['authorized_ipv4_addresses'] = []61        expected_result['revoked_ipv4_addresses'] = [ipv4_addresses[2]]62        assert update_security_group(security_group_id=sg_id, **event) == expected_result63        event['allowed_ipv4_addresses'] = [ipv4_addresses[0]]64        expected_result['revoked_ipv4_addresses'] = [ipv4_addresses[1]]65        assert update_security_group(security_group_id=sg_id, **event) == expected_result66        event['allowed_ipv4_addresses'] = []67        expected_result['revoked_ipv4_addresses'] = [ipv4_addresses[0]]...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.