8000 Replicator: support generating specific VPC ids by simonrw · Pull Request #11772 · localstack/localstack · GitHub
[go: up one dir, main page]
Skip to content

Replicator: support generating specific VPC ids #11772

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Nov 13, 2024
Merged

Replicator: support generating specific VPC ids #11772

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
fa79dab
Make create_vpc function more user friendly
simonrw Nov 1, 2024
b690534
Support customisation of the VPC id
simonrw Nov 1, 2024
732e393
Simplify vpc id generation based on review feedback
simonrw Nov 13, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 37 additions & 5 deletions localstack-core/localstack/services/ec2/patches.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,18 +2,49 @@
from typing import Optional

from moto.ec2 import models as ec2_models
from moto.utilities.id_generator import Tags

from localstack.constants import TAG_KEY_CUSTOM_ID
from localstack.services.ec2.exceptions import (
InvalidSecurityGroupDuplicateCustomIdError,
InvalidSubnetDuplicateCustomIdError,
InvalidVpcDuplicateCustomIdError,
)
from localstack.utils.id_generator import (
ExistingIds,
ResourceIdentifier,
localstack_id,
)
from localstack.utils.patch import patch

LOG = logging.getLogger(__name__)


@localstack_id
def generate_vpc_id(
resource_identifier: ResourceIdentifier,
existing_ids: ExistingIds = None,
tags: Tags = None,
) -> str:
# We return an empty string here to differentiate between when a custom ID was used, or when it was randomly generated by `moto`.
return ""


class VpcIdentifier(ResourceIdentifier):
service = "ec2"
resource = "vpc"

def __init__(self, account_id: str, region: str, cidr_block: str):
super().__init__(account_id, region, name=cidr_block)

def generate(self, existing_ids: ExistingIds = None, tags: Tags = None) -> str:
return generate_vpc_id(
resource_identifier=self,
existing_ids=existing_ids,
tags=tags,
)


def apply_patches():
@patch(ec2_models.subnets.SubnetBackend.create_subnet)
def ec2_create_subnet(
Expand Down Expand Up @@ -82,22 +113,23 @@ def ec2_create_security_group(
def ec2_create_vpc(
fn: ec2_models.vpcs.VPCBackend.create_vpc,
self: ec2_models.vpcs.VPCBackend,
cidr_block: str,
*args,
tags: Optional[list[dict[str, str]]] = None,
is_default: bool = False,
**kwargs,
):
# Extract custom ID from tags if it exists
tags: list[dict[str, str]] = tags or []
custom_ids = [tag["Value"] for tag in tags if tag["Key"] == TAG_KEY_CUSTOM_ID]
custom_id = custom_ids[0] if len(custom_ids) > 0 else None
resource_identifier = VpcIdentifier(self.account_id, self.region_name, cidr_block)
custom_id = resource_identifier.generate(tags=tags)

# Check if custom id is unique
if custom_id and custom_id in self.vpcs:
raise InvalidVpcDuplicateCustomIdError(custom_id)

# Generate VPC with moto library
result: ec2_models.vpcs.VPC = fn(self, *args, tags=tags, is_default=is_default, **kwargs)
result: ec2_models.vpcs.VPC = fn(
self, cidr_block, *args, tags=tags, is_default=is_default, **kwargs
)
vpc_id = result.id

if custom_id:
Expand Down
17 changes: 16 additions & 1 deletion tests/aws/services/ec2/test_ec2.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
)

from localstack.constants import TAG_KEY_CUSTOM_ID
from localstack.services.ec2.patches import VpcIdentifier
from localstack.testing.pytest import markers
from localstack.utils.strings import short_uid
from localstack.utils.sync import retry
Expand Down Expand Up @@ -48,8 +49,9 @@ def create_vpc(aws_client):

def _create_vpc(
cidr_block: str,
tag_specifications: list[dict],
tag_specifications: list[dict] | None = None,
):
tag_specifications = tag_specifications or []
vpc = aws_client.ec2.create_vpc(CidrBlock=cidr_block, TagSpecifications=tag_specifications)
vpcs.append(vpc["Vpc"]["VpcId"])
return vpc
Expand Down Expand Up @@ -816,3 +818,16 @@ def test_pickle_ec2_backend(pickle_backends, aws_client):
_ = aws_client.ec2.describe_account_attributes()
pickle_backends(ec2_backends)
assert pickle_backends(ec2_backends)


@markers.aws.only_localstack
def test_create_specific_vpc_id(account_id, region_name, create_vpc, set_resource_custom_id):
cidr_block = "10.0.0.0/16"
custom_id = "my-custom-id"
set_resource_custom_id(
VpcIdentifier(account_id=account_id, region=region_name, cidr_block=cidr_block),
f"vpc-{custom_id}",
)

vpc = create_vpc(cidr_block=cidr_block)
assert vpc["Vpc"]["VpcId"] == f"vpc-{custom_id}"
Loading
0