Skip to content

Commit

Permalink
EC2: Add additional VPC Endpoint Services (#7509)
Browse files Browse the repository at this point in the history
  • Loading branch information
bblommers committed Mar 22, 2024
1 parent b22683e commit 33d3745
Show file tree
Hide file tree
Showing 32 changed files with 287 additions and 296 deletions.
9 changes: 0 additions & 9 deletions moto/acm/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -409,15 +409,6 @@ def __init__(self, region_name: str, account_id: str):
self._certificates: Dict[str, CertBundle] = {}
self._idempotency_tokens: Dict[str, Any] = {}

@staticmethod
def default_vpc_endpoint_service(
service_region: str, zones: List[str]
) -> List[Dict[str, str]]:
"""Default VPC endpoint service."""
return BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "acm-pca"
)

def set_certificate_in_use_by(self, arn: str, load_balancer_name: str) -> None:
if arn not in self._certificates:
raise CertificateNotFound(arn=arn, account_id=self.account_id)
Expand Down
9 changes: 0 additions & 9 deletions moto/applicationautoscaling/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -73,15 +73,6 @@ def __init__(self, region_name: str, account_id: str) -> None:
self.policies: Dict[str, FakeApplicationAutoscalingPolicy] = {}
self.scheduled_actions: List[FakeScheduledAction] = list()

@staticmethod
def default_vpc_endpoint_service(
service_region: str, zones: List[str]
) -> List[Dict[str, str]]:
"""Default VPC endpoint service."""
return BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "application-autoscaling"
)

def describe_scalable_targets(
self, namespace: str, r_ids: Union[None, List[str]], dimension: Union[None, str]
) -> List["FakeScalableTarget"]:
Expand Down
9 changes: 0 additions & 9 deletions moto/athena/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -173,15 +173,6 @@ def __init__(self, region_name: str, account_id: str):
name="primary", description="", configuration=dict(), tags=[]
)

@staticmethod
def default_vpc_endpoint_service(
service_region: str, zones: List[str]
) -> List[Dict[str, str]]:
"""Default VPC endpoint service."""
return BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "athena"
)

def create_work_group(
self,
name: str,
Expand Down
11 changes: 0 additions & 11 deletions moto/autoscaling/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -882,17 +882,6 @@ def __init__(self, region_name: str, account_id: str):
self.elb_backend: ELBBackend = elb_backends[self.account_id][region_name]
self.elbv2_backend: ELBv2Backend = elbv2_backends[self.account_id][region_name]

@staticmethod
def default_vpc_endpoint_service(
service_region: str, zones: List[str]
) -> List[Dict[str, Any]]: # type: ignore[misc]
"""Default VPC endpoint service."""
return BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "autoscaling"
) + BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "autoscaling-plans"
)

def create_launch_configuration(
self,
name: str,
Expand Down
9 changes: 0 additions & 9 deletions moto/awslambda/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -1893,15 +1893,6 @@ def __init__(self, region_name: str, account_id: str):
self._event_source_mappings: Dict[str, EventSourceMapping] = {}
self._layers = LayerStorage()

@staticmethod
def default_vpc_endpoint_service(
service_region: str, zones: List[str]
) -> List[Dict[str, str]]:
"""Default VPC endpoint service."""
return BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "lambda"
)

def create_alias(
self,
name: str,
Expand Down
9 changes: 0 additions & 9 deletions moto/cloudwatch/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -442,15 +442,6 @@ def __init__(self, region_name: str, account_id: str):
self.paged_metric_data: Dict[str, List[MetricDatumBase]] = {}
self.tagger = TaggingService()

@staticmethod
def default_vpc_endpoint_service(
service_region: str, zones: List[str]
) -> List[Dict[str, str]]:
"""Default VPC endpoint service."""
return BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "monitoring"
)

@property
# Retrieve a list of all OOTB metrics that are provided by metrics providers
# Computed on the fly
Expand Down
9 changes: 0 additions & 9 deletions moto/codecommit/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,15 +40,6 @@ def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id)
self.repositories: Dict[str, CodeCommit] = {}

@staticmethod
def default_vpc_endpoint_service(
service_region: str, zones: List[str]
) -> List[Dict[str, str]]:
"""Default VPC endpoint service."""
return BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "codecommit"
)

def create_repository(
self, repository_name: str, repository_description: str
) -> Dict[str, str]:
Expand Down
9 changes: 0 additions & 9 deletions moto/config/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -917,15 +917,6 @@ def __init__(self, region_name: str, account_id: str):
self.config_schema: Optional[AWSServiceSpec] = None
self.retention_configuration: Optional[RetentionConfiguration] = None

@staticmethod
def default_vpc_endpoint_service(
service_region: str, zones: List[str]
) -> List[Dict[str, Any]]: # type: ignore[misc]
"""List of dicts representing default VPC endpoints for this service."""
return BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "config"
)

def _validate_resource_types(self, resource_list: List[str]) -> None:
if not self.config_schema:
self.config_schema = AWSServiceSpec(
Expand Down
9 changes: 0 additions & 9 deletions moto/datasync/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -103,15 +103,6 @@ def __init__(self, region_name: str, account_id: str):
self.tasks: Dict[str, Task] = OrderedDict()
self.task_executions: Dict[str, TaskExecution] = OrderedDict()

@staticmethod
def default_vpc_endpoint_service(
service_region: str, zones: List[str]
) -> List[Dict[str, Any]]: # type: ignore[misc]
"""Default VPC endpoint service."""
return BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "datasync"
)

def create_location(
self, location_uri: str, typ: str, metadata: Dict[str, Any]
) -> str:
Expand Down
9 changes: 0 additions & 9 deletions moto/dms/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,15 +18,6 @@ def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id)
self.replication_tasks: Dict[str, "FakeReplicationTask"] = {}

@staticmethod
def default_vpc_endpoint_service(
service_region: str, zones: List[str]
) -> List[Dict[str, Any]]: # type: ignore[misc]
"""Default VPC endpoint service."""
return BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "dms"
)

def create_replication_task(
self,
replication_task_identifier: str,
Expand Down
9 changes: 0 additions & 9 deletions moto/ds/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -200,15 +200,6 @@ def __init__(self, region_name: str, account_id: str):
self.directories: Dict[str, Directory] = {}
self.tagger = TaggingService()

@staticmethod
def default_vpc_endpoint_service(
service_region: str, zones: List[str]
) -> List[Dict[str, str]]:
"""List of dicts representing default VPC endpoints for this service."""
return BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "ds"
)

def _verify_subnets(self, region: str, vpc_settings: Dict[str, Any]) -> None:
"""Verify subnets are valid, else raise an exception.
Expand Down
11 changes: 0 additions & 11 deletions moto/ec2/models/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -163,17 +163,6 @@ def __init__(self, region_name: str, account_id: str):
self.create_subnet(vpc.id, cidr_block, availability_zone=az_name)
ip[2] += 16 # type: ignore

@staticmethod
def default_vpc_endpoint_service(
service_region: str, zones: List[str]
) -> List[Dict[str, Any]]: # type: ignore[misc]
"""Default VPC endpoint service."""
return BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "ec2"
) + BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "ec2messages"
)

# Use this to generate a proper error template response when in a response
# handler.
def raise_error(self, code: str, message: str) -> None:
Expand Down

0 comments on commit 33d3745

Please sign in to comment.