8000 Add EC2 support for GetSecurityGroupsForVpc API operation (#12602) · localstack/localstack@bf228f3 · GitHub
[go: up one dir, main page]
Skip to content

Commit bf228f3

Browse files
iamramtiniamramtin
committed
Add EC2 support for GetSecurityGroupsForVpc API operation (#12602)
Signed-off-by: Ramtin Mesgari <26694963+iamramtin@users.noreply.github.com>
1 parent dcea04f commit bf228f3

File tree

2 files changed

+118
-0
lines changed

2 files changed

+118
-0
lines changed

‎localstack-core/localstack/services/ec2/provider.py

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,8 @@
5050
DnsOptionsSpecification,
5151
DnsRecordIpType,
5252
Ec2Api,
53+
GetSecurityGroupsForVpcRequest,
54+
GetSecurityGroupsForVpcResult,
5355
InstanceType,
5456
IpAddressType,
5557
LaunchTemplate,
@@ -70,6 +72,7 @@
7072
RevokeSecurityGroupEgressRequest,
7173
RevokeSecurityGroupEgressResult,
7274
RIProductDescription,
75+
SecurityGroupForVpc,
10000
7376
String,
7477
SubnetConfigurationsList,
7578
Tenancy,
@@ -539,6 +542,33 @@ def create_flow_logs(
539542

540543
return response
541544

545+
@handler("GetSecurityGroupsForVpc", expand=False)
546+
def get_security_groups_for_vpc(
547+
self,
548+
context: RequestContext,
549+
get_security_groups_for_vpc_request: GetSecurityGroupsForVpcRequest,
550+
) -> GetSecurityGroupsForVpcResult:
551+
vpc_id = get_security_groups_for_vpc_request.get("VpcId")
552+
if vpc_id:
553+
next_token = get_security_groups_for_vpc_request.get("NextToken", None)
554+
backend = get_ec2_backend(context.account_id, context.region)
555+
filters = {"vpc-id": [vpc_id]}
556+
filtered_sgs = backend.describe_security_groups(filters=filters)
557+
558+
sgs = [
559+
SecurityGroupForVpc(
560+
Description=sg.description,
561+
GroupId=sg.id,
562+
GroupName=sg.name,
563+
OwnerId=context.account_id,
564+
PrimaryVpcId=sg.vpc_id,
565+
Tags=[{"Key": k, "Value": v} for k, v in (sg.get_tags() or {}).items()],
566+
)
567+
for sg in filtered_sgs
568+
]
569+
return GetSecurityGroupsForVpcResult(SecurityGroupForVpcs=sgs, NextToken=next_token)
570+
return call_moto(context)
571+
542572

543573
@patch(SubnetBackend.modify_subnet_attribute)
544574
def modify_subnet_attribute(fn, self, subnet_id, attr_name, attr_value):

‎tests/aws/services/ec2/test_ec2.py

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -482,6 +482,94 @@ def test_create_vpc_with_custom_id(self, aws_client, create_vpc):
482482
assert e.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
483483
assert e.value.response["Error"]["Code"] == "InvalidVpc.DuplicateCustomId"
484484

485+
@markers.aws.only_localstack
486+
def test_get_security_groups_for_vpc(self, cleanups, aws_client, create_vpc):
487+
# Create a VPC
488+
vpc: dict = create_vpc(
489+
cidr_block="10.0.0.0/16",
490+
tag_specifications=[
491+
{
492+
"ResourceType": "vpc",
493+
"Tags": [
494+
{"Key": "Name", "Value": "test-vpc"},
495+
],
496+
}
497+
],
498+
)
499+
vpc_id: str = vpc["Vpc"]["VpcId"]
500+
501+
# Create security groups in the VPC
502+
sg1_response = aws_client.ec2.create_security_group(
503+
GroupName="sg-1",
504+
Description="Test Security Group 1",
505+
VpcId=vpc_id
506+
)
507+
sg1_id = sg1_response["GroupId"]
508+
509+
sg2_response = aws_client.ec2.create_security_group(
510+
GroupName="sg-2",
511+
Description="Test Security Group 2",
512+
VpcId=vpc_id
513+
)
514+
sg2_id = sg2_response["GroupId"]
515+
516+
# Create a security group in a different VPC (default VPC)
517+
default_vpc_response = aws_client.ec2.describe_vpcs(
518+
Filters=[{"Name": "isDefault", "Values": ["true"]}]
519+
)
520+
default_vpc_id = default_vpc_response["Vpcs"][0]["VpcId"]
521+
522+
sg3_response = aws_client.ec2.create_security_group(
523+
GroupName="sg-3",
524+
Description="Test Security Group 3",
525+
VpcId=default_vpc_id
526+
)
527+
sg3_id = sg3_response["GroupId"]
528+
529+
# Call GetSecurityGroupsForVpc
530+
response = aws_client.ec2.get_security_groups_for_vpc(VpcId=vpc_id)
531+
532+
# Verify response
533+
assert "SecurityGroupForVpcs" in response
534+
sg_ids = [sg["GroupId"] for sg in response["SecurityGroupForVpcs"]]
535+
536+
# Should only include the security groups we created in the VPC
537+
assert sg1_id in sg_ids
538+
assert sg2_id in sg_ids
539+
assert sg3_id not in sg_ids
540+
541+
# Should include the default security group for the VPC
542+
default_sg = next(
543+
(sg for sg in response["SecurityGroupForVpcs"] if sg["GroupName"] == "default"),
544+
None
545+
)
546+
assert default_sg is not None
547+
548+
cleanups.append(lambda: aws_client.ec2.delete_security_group(GroupId=sg1_id))
549+
cleanups.append(lambda: aws_client.ec2.delete_security_group(GroupId=sg2_id))
550+
cleanups.append(lambda: aws_client.ec2.delete_security_group(GroupId=sg3_id))
551+
cleanups.append(lambda: aws_client.ec2.delete_vpc(VpcId=vpc_id))
552+
553+
@markers.aws.only_localstack
554+
def test_get_security_groups_for_vpc_invalid_inputs(self, aws_client):
555+
# Test with non-existent VPC ID
556+
response = aws_client.ec2.get_security_groups_for_vpc(VpcId="vpc-nonexistent")
557+
assert "SecurityGroupForVpcs" in response
558+
assert len(response["SecurityGroupForVpcs"]) == 0
559+
560+
# Test with invalid VPC ID format
561+
response = aws_client.ec2.get_security_groups_for_vpc(VpcId="123-invalid-vpc-format")
562+
assert "SecurityGroupForVpcs" in response
563+
assert len(response["SecurityGroupForVpcs"]) == 0
564+
565+
# Test with missing VPC ID parameter
566+
with pytest.raises(Exception):
567+
aws_client.ec2.get_security_groups_for_vpc()
568+
569+
# Test with empty string for VPC ID
570+
with pytest.raises(Exception):
571+
aws_client.ec2.get_security_groups_for_vpc(VpcId="")
572+
485573
@markers.aws.only_localstack
486574
def test_create_subnet_with_tags(self, cleanups, aws_client, create_vpc):
487575
# Create a VPC.

0 commit comments

Comments
 (0)
0