diff --git a/moto/ec2/models/vpcs.py b/moto/ec2/models/vpcs.py index 3457a396509d..09cdbca96b78 100644 --- a/moto/ec2/models/vpcs.py +++ b/moto/ec2/models/vpcs.py @@ -32,7 +32,6 @@ create_dns_entries, generic_filter, random_ipv6_cidr, - random_private_ip, random_vpc_cidr_association_id, random_vpc_ep_id, random_vpc_id, @@ -992,7 +991,15 @@ def create_vpc_endpoint( network_interface_ids = [] for subnet_id in subnet_ids or []: self.get_subnet(subnet_id) # type: ignore[attr-defined] - eni = self.create_network_interface(subnet_id, random_private_ip()) # type: ignore[attr-defined] + eni = self.create_network_interface( # type: ignore[attr-defined] + subnet_id, + private_ip_address=None, + group_ids=security_group_ids, + description=f"VPC Endpoint Interface {vpc_endpoint_id}", + ) + eni.interface_type = "vpc_endpoint" + eni.requester_managed = True + eni.status = "in-use" network_interface_ids.append(eni.id) dns_entries = create_dns_entries(service_name, vpc_endpoint_id) diff --git a/moto/ec2/utils.py b/moto/ec2/utils.py index 362a116b889a..0f2423b978ae 100644 --- a/moto/ec2/utils.py +++ b/moto/ec2/utils.py @@ -286,7 +286,7 @@ def generate_dns_from_ip(ip: Any, dns_type: str = "internal") -> str: def random_mac_address() -> str: - return f"02:00:00:{random.randint(0, 255)}02x:{random.randint(0, 255)}02x:{random.randint(0, 255)}02x" + return f"02:00:00:{random.randint(0, 255):02x}:{random.randint(0, 255):02x}:{random.randint(0, 255):02x}" def randor_ipv4_cidr() -> str: diff --git a/tests/test_ec2/test_elastic_network_interfaces.py b/tests/test_ec2/test_elastic_network_interfaces.py index 9d64adc236dc..8b74b468ced7 100644 --- a/tests/test_ec2/test_elastic_network_interfaces.py +++ b/tests/test_ec2/test_elastic_network_interfaces.py @@ -38,6 +38,7 @@ def test_elastic_network_interfaces(): assert eni["RequesterManaged"] is False assert eni["PrivateIpAddresses"][0]["PrivateIpAddress"].startswith("10.") is True assert "Association" not in eni + assert len(eni["MacAddress"]) == 17 with pytest.raises(ClientError) as ex: ec2client.delete_network_interface(NetworkInterfaceId=eni_id, DryRun=True) diff --git a/tests/test_ec2/test_vpcs.py b/tests/test_ec2/test_vpcs.py index 12eba28aff52..b995467c2cc3 100644 --- a/tests/test_ec2/test_vpcs.py +++ b/tests/test_ec2/test_vpcs.py @@ -1,5 +1,7 @@ +import ipaddress import json import random +from time import sleep from unittest import SkipTest from uuid import uuid4 @@ -1036,6 +1038,61 @@ def test_create_vpc_endpoint__policy(): assert vpc_end_point["PolicyDocument"] == "my policy document" +@ec2_aws_verified(create_vpc=True, create_subnet=True, create_sg=True) +@pytest.mark.aws_verified +def test_create_interface_vpc_endpoint_network_interfaces( + ec2_client=None, vpc_id=None, subnet_id=None, sg_id=None +): + vpc_end_point_id = ec2_client.create_vpc_endpoint( + VpcId=vpc_id, + ServiceName="com.amazonaws.us-east-1.execute-api", + VpcEndpointType="Interface", + PrivateDnsEnabled=False, + SubnetIds=[subnet_id], + SecurityGroupIds=[sg_id], + )["VpcEndpoint"]["VpcEndpointId"] + + try: + vpc_end_point = wait_for_vpc_endpoint_available(ec2_client, vpc_end_point_id) + + eni = ec2_client.describe_network_interfaces( + NetworkInterfaceIds=vpc_end_point["NetworkInterfaceIds"] + )["NetworkInterfaces"][0] + assert ipaddress.ip_address(eni["PrivateIpAddress"]) in ipaddress.ip_network( + "10.0.0.0/24" + ) + assert [group["GroupId"] for group in eni["Groups"]] == [sg_id] + assert eni["InterfaceType"] == "vpc_endpoint" + assert eni["RequesterManaged"] is True + assert eni["Status"] == "in-use" + assert eni["Description"] == f"VPC Endpoint Interface {vpc_end_point_id}" + finally: + ec2_client.delete_vpc_endpoints(VpcEndpointIds=[vpc_end_point_id]) + # the subnet cannot be deleted while the endpoint's network interface exists + wait_for_subnet_network_interfaces_deleted(ec2_client, subnet_id) + + +def wait_for_vpc_endpoint_available(ec2_client, vpc_end_point_id): + for _ in range(60): + vpc_end_point = ec2_client.describe_vpc_endpoints( + VpcEndpointIds=[vpc_end_point_id] + )["VpcEndpoints"][0] + if vpc_end_point["State"].lower() == "available": + return vpc_end_point + sleep(5) + raise AssertionError(f"VPC endpoint {vpc_end_point_id} never became available") + + +def wait_for_subnet_network_interfaces_deleted(ec2_client, subnet_id): + for _ in range(60): + interfaces = ec2_client.describe_network_interfaces( + Filters=[{"Name": "subnet-id", "Values": [subnet_id]}] + )["NetworkInterfaces"] + if not interfaces: + return + sleep(5) + + @mock_aws def test_describe_vpc_gateway_end_points(): ec2 = boto3.client("ec2", region_name="us-west-1")