Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

EC2: Add additional VPC Endpoint Services (#7509)

Browse Source
This commit is contained in:
Bert Blommers 2024-03-22 07:38:57 -01:00 committed by GitHub
parent b22683eb98
commit 33d3745263
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
32 changed files with 297 additions and 306 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
moto/acm/models.py
View File

@ -409,15 +409,6 @@ class AWSCertificateManagerBackend(BaseBackend):
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)

9
moto/applicationautoscaling/models.py
View File

@ -73,15 +73,6 @@ class ApplicationAutoscalingBackend(BaseBackend):
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"]:

9
moto/athena/models.py
View File

@ -173,15 +173,6 @@ class AthenaBackend(BaseBackend):
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,

11
moto/autoscaling/models.py
View File

@ -882,17 +882,6 @@ class AutoScalingBackend(BaseBackend):
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,

9
moto/awslambda/models.py
View File

@ -1893,15 +1893,6 @@ class LambdaBackend(BaseBackend):
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,

9
moto/cloudwatch/models.py
View File

@ -442,15 +442,6 @@ class CloudWatchBackend(BaseBackend):
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

9
moto/codecommit/models.py
View File

@ -40,15 +40,6 @@ class CodeCommitBackend(BaseBackend):
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]:

9
moto/config/models.py
View File

@ -917,15 +917,6 @@ class ConfigBackend(BaseBackend):
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(

9
moto/datasync/models.py
View File

@ -103,15 +103,6 @@ class DataSyncBackend(BaseBackend):
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:

9
moto/dms/models.py
View File

@ -18,15 +18,6 @@ class DatabaseMigrationServiceBackend(BaseBackend):
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,

9
moto/ds/models.py
View File

@ -200,15 +200,6 @@ class DirectoryServiceBackend(BaseBackend):
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.

11
moto/ec2/models/__init__.py
View File

@ -163,17 +163,6 @@ class EC2Backend(
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:

309
moto/ec2/models/vpcs.py
View File

@ -6,6 +6,7 @@ from collections import defaultdict
from operator import itemgetter
from typing import Any, Dict, List, Optional
from moto.core.base_backend import BaseBackend
from moto.core.common_models import CloudFormationModel
from ..exceptions import (
@ -37,52 +38,290 @@ from ..utils import (
from .availability_zones_and_regions import RegionsAndZonesBackend
from .core import TaggedEC2Resource
# We used to load the entirety of Moto into memory, and check every module if it's supported
# But having a fixed list is much more performant
# Maintaining it is more difficult, but the contents of this list does not change very often
# List of Moto services that exposes a non-standard EndpointService config
IMPLEMENTED_ENDPOINT_SERVICES = [
"acm",
"applicationautoscaling",
"athena",
"autoscaling",
"awslambda",
"cloudformation",
"cloudwatch",
"codecommit",
"codepipeline",
"config",
"datasync",
"dms",
"ds",
"dynamodb",
"ec2",
"ecr",
"ecs",
"elasticbeanstalk",
"elbv2",
"emr",
"events",
"firehose",
"glue",
"iot",
"kinesis",
"kms",
"logs",
"rds",
"redshift",
"redshiftdata",
"route53resolver",
"s3",
"sagemaker",
]
# List of names used by AWS that are implemented by our services
COVERED_ENDPOINT_SERVICES = IMPLEMENTED_ENDPOINT_SERVICES + [
"ecr.api",
"ecr.dkr",
"iot.data",
"kinesis-firehose",
"kinesis-streams",
"redshift-data",
"sagemaker.api",
]
# All endpoints offered by AWS
# We expose a sensible default for these services, if we don't implement a service-specific non-standard
AWS_ENDPOINT_SERVICES = [
"access-analyzer",
"account",
"acm-pca",
"airflow.api",
"airflow.env",
"airflow.ops",
"analytics-omics",
"app-integrations",
"appconfig",
"appconfigdata",
"application-autoscaling",
"appmesh",
"appmesh-envoy-management",
"apprunner",
"apprunner.requests",
"appstream.api",
"appstream.streaming",
"appsync-api",
"aps",
"aps-workspaces",
"athena",
"auditmanager",
"autoscaling",
"autoscaling-plans",
"awsconnector",
"b2bi",
"backup",
"backup-gateway",
"batch",
"bedrock",
"bedrock-agent",
"bedrock-agent-runtime",
"bedrock-runtime",
"billingconductor",
"braket",
"cases",
"cassandra",
"cassandra-fips",
"cleanrooms",
"cloudcontrolapi",
"cloudcontrolapi-fips",
"clouddirectory",
"cloudformation",
"cloudhsmv2",
"cloudtrail",
"codeartifact.api",
"codeartifact.repositories",
"codebuild",
"codebuild-fips",
"codecommit",
"codecommit-fips",
"codedeploy",
"codedeploy-commands-secure",
"codeguru-profiler",
"codeguru-reviewer",
"codepipeline",
"codestar-connections.api",
"codewhisperer",
"comprehend",
"comprehendmedical",
"config",
"connect-campaigns",
"console",
"control-storage-omics",
"data-servicediscovery",
"data-servicediscovery-fips",
"databrew",
"dataexchange",
"datasync",
"datazone",
"deviceadvisor.iot",
"devops-guru",
"dms",
"dms-fips",
"drs",
"ds",
"dynamodb",
"ebs",
"ec2",
"ec2messages",
"ecr.api",
"ecr.dkr",
"ecs",
"ecs-agent",
"ecs-telemetry",
"eks",
"eks-auth",
"elastic-inference.runtime",
"elasticache",
"elasticache-fips",
"elasticbeanstalk",
"elasticbeanstalk-health",
"elasticfilesystem",
"elasticfilesystem-fips",
"elasticloadbalancing",
"elasticmapreduce",
"email-smtp",
"emr-containers",
"emr-serverless",
"emrwal.prod",
"entityresolution",
"events",
"evidently",
"evidently-dataplane",
"execute-api",
"finspace",
"finspace-api",
"fis",
"forecast",
"forecast-fips",
"forecastquery",
"forecastquery-fips",
"frauddetector",
"fsx",
"fsx-fips",
"git-codecommit",
"git-codecommit-fips",
"glue",
"grafana",
"grafana-workspace",
"greengrass",
"groundstation",
"guardduty-data",
"guardduty-data-fips",
"healthlake",
"identitystore",
"imagebuilder",
"inspector2",
"iot.credentials",
"iot.data",
"iot.fleethub.api",
"iotfleetwise",
"iotroborunner",
"iotsitewise.api",
"iotsitewise.data",
"iottwinmaker.api",
"iottwinmaker.data",
"iotwireless.api",
"kendra",
"kinesis-firehose",
"kinesis-streams",
"kms",
"kms-fips",
"lakeformation",
"lambda",
"license-manager",
"license-manager-fips",
"license-manager-user-subscriptions",
"logs",
"lookoutequipment",
"lookoutmetrics",
"lookoutvision",
"lorawan.cups",
"lorawan.lns",
"m2",
"macie2",
"managedblockchain-query",
"managedblockchain.bitcoin.mainnet",
"managedblockchain.bitcoin.testnet",
"mediaconnect",
"medical-imaging",
"memory-db",
"memorydb-fips",
"mgn",
"migrationhub-orchestrator",
"models-v2-lex",
"monitoring",
"neptune-graph",
"networkmonitor",
"nimble",
"organizations",
"organizations-fips",
"panorama",
"payment-cryptography.controlplane",
"payment-cryptography.dataplane",
"pca-connector-ad",
"personalize",
"personalize-events",
"personalize-runtime",
"pinpoint",
"pinpoint-sms-voice-v2",
"polly",
"private-networks",
"profile",
"proton",
"qldb.session",
"rds",
"rds-data",
"redshift",
"redshift-data",
"redshift-fips",
"refactor-spaces",
"rekognition",
"rekognition-fips",
"robomaker",
"rolesanywhere",
"rum",
"rum-dataplane",
"runtime-medical-imaging",
"runtime-v2-lex",
"s3",
"s3",
"s3-outposts",
"s3express",
"sagemaker.api",
"sagemaker.featurestore-runtime",
"sagemaker.metrics",
"sagemaker.runtime",
"sagemaker.runtime-fips",
"scn",
"secretsmanager",
"securityhub",
"servicecatalog",
"servicecatalog-appregistry",
"servicediscovery",
"servicediscovery-fips",
"signin",
"simspaceweaver",
"snow-device-management",
"sns",
"sqs",
"ssm",
"ssm-contacts",
"ssm-incidents",
"ssmmessages",
"states",
"storage-omics",
"storagegateway",
"streaming-rekognition",
"streaming-rekognition-fips",
"sts",
"swf",
"swf-fips",
"sync-states",
"synthetics",
"tags-omics",
"textract",
"textract-fips",
"thinclient.api",
"timestream-influxdb",
"tnb",
"transcribe",
"transcribestreaming",
"transfer",
"transfer.server",
"translate",
"trustedadvisor",
"verifiedpermissions",
"voiceid",
"vpc-lattice",
"wisdom",
"workflows-omics",
"workspaces",
"xray",
]
MAX_NUMBER_OF_ENDPOINT_SERVICES_RESULTS = 1000
DEFAULT_VPC_ENDPOINT_SERVICES: List[Dict[str, str]] = []
DEFAULT_VPC_ENDPOINT_SERVICES: Dict[str, List[Dict[str, str]]] = {}
ENDPOINT_SERVICE_COLLECTION_LOCK = threading.Lock()
@ -772,9 +1011,10 @@ class VPCBackend:
) -> List[Dict[str, str]]:
"""Return list of default services using list of backends."""
with ENDPOINT_SERVICE_COLLECTION_LOCK:
if DEFAULT_VPC_ENDPOINT_SERVICES:
return DEFAULT_VPC_ENDPOINT_SERVICES
if region in DEFAULT_VPC_ENDPOINT_SERVICES:
return DEFAULT_VPC_ENDPOINT_SERVICES[region]
DEFAULT_VPC_ENDPOINT_SERVICES[region] = []
zones = [
zone.name
for zones in RegionsAndZonesBackend.zones.values()
@ -792,15 +1032,22 @@ class VPCBackend:
region, zones
)
if service:
DEFAULT_VPC_ENDPOINT_SERVICES.extend(service)
DEFAULT_VPC_ENDPOINT_SERVICES[region].extend(service)
if "global" in account_backend:
service = account_backend["global"].default_vpc_endpoint_service(
region, zones
)
if service:
DEFAULT_VPC_ENDPOINT_SERVICES.extend(service)
return DEFAULT_VPC_ENDPOINT_SERVICES
DEFAULT_VPC_ENDPOINT_SERVICES[region].extend(service)
# Return sensible defaults, for services that do not offer a custom implementation
for aws_service in AWS_ENDPOINT_SERVICES:
if aws_service not in COVERED_ENDPOINT_SERVICES:
service_configs = BaseBackend.default_vpc_endpoint_service_factory(region, zones, aws_service)
DEFAULT_VPC_ENDPOINT_SERVICES[region].extend(service_configs)
return DEFAULT_VPC_ENDPOINT_SERVICES[region]
@staticmethod
def _matches_service_by_tags(

9
moto/ecs/models.py
View File

@ -962,15 +962,6 @@ class EC2ContainerServiceBackend(BaseBackend):
self.services: Dict[str, Service] = {}
self.container_instances: Dict[str, Dict[str, ContainerInstance]] = {}
@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, "ecs"
)
def _get_cluster(self, name: str) -> Cluster:
# short name or full ARN of the cluster
cluster_name = name.split("/")[-1]

11
moto/elasticbeanstalk/models.py
View File

@ -84,17 +84,6 @@ class EBBackend(BaseBackend):
super().__init__(region_name, account_id)
self.applications: Dict[str, FakeApplication] = dict()
@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, "elasticbeanstalk"
) + BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "elasticbeanstalk-health"
)
def create_application(self, application_name: str) -> FakeApplication:
if application_name in self.applications:
raise InvalidParameterValueError(

9
moto/elbv2/models.py
View File

@ -722,15 +722,6 @@ class ELBv2Backend(BaseBackend):
self.load_balancers: Dict[str, FakeLoadBalancer] = OrderedDict()
self.tagging_service = 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, "elasticloadbalancing"
)
@property
def ec2_backend(self) -> Any: # type: ignore[misc]
"""

9
moto/emr/models.py
View File

@ -556,15 +556,6 @@ class ElasticMapReduceBackend(BaseBackend):
self.instance_groups: Dict[str, FakeInstanceGroup] = {}
self.security_configurations: Dict[str, FakeSecurityConfiguration] = {}
@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, "elasticmapreduce"
)
@property
def ec2_backend(self) -> Any: # type: ignore[misc]
"""

9
moto/events/models.py
View File

@ -1026,15 +1026,6 @@ class EventsBackend(BaseBackend):
self.partner_event_sources: Dict[str, PartnerEventSource] = {}
self.approved_parent_event_bus_names: List[str] = []
@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, "events"
)
def _add_default_event_bus(self) -> None:
self.event_buses["default"] = EventBus(
self.account_id, self.region_name, "default"

9
moto/glue/models.py
View File

@ -113,15 +113,6 @@ class GlueBackend(BaseBackend):
self.num_schemas = 0
self.num_schema_versions = 0
@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, "glue"
)
def create_database(
self,
database_name: str,

9
moto/kms/models.py
View File

@ -260,15 +260,6 @@ class KmsBackend(BaseBackend):
self.key_to_aliases: Dict[str, Set[str]] = defaultdict(set)
self.tagger = TaggingService(key_name="TagKey", value_name="TagValue")
@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, "kms"
)
def _generate_default_keys(self, alias_name: str) -> Optional[str]:
"""Creates default kms keys"""
if alias_name in RESERVED_ALIASES:

9
moto/logs/models.py
View File

@ -766,15 +766,6 @@ class LogsBackend(BaseBackend):
self.tagger = TaggingService()
self.export_tasks: Dict[str, ExportTask] = dict()
@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, "logs"
)
def create_log_group(
self, log_group_name: str, tags: Dict[str, str], **kwargs: Any
) -> LogGroup:

11
moto/rds/models.py
View File

@ -1736,17 +1736,6 @@ class RDSBackend(BaseBackend):
]
return self._db_cluster_options
@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, "rds"
) + BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "rds-data"
)
def create_db_instance(self, db_kwargs: Dict[str, Any]) -> Database:
database_id = db_kwargs["db_instance_identifier"]
self._validate_db_identifier(database_id)

11
moto/redshift/models.py
View File

@ -595,17 +595,6 @@ class RedshiftBackend(BaseBackend):
}
self.snapshot_copy_grants: Dict[str, SnapshotCopyGrant] = {}
@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, "redshift"
) + BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "redshift-data", policy_supported=False
)
def enable_snapshot_copy(self, **kwargs: Any) -> Cluster:
cluster_identifier = kwargs["cluster_identifier"]
cluster = self.clusters[cluster_identifier]

9
moto/redshiftdata/models.py
View File

@ -117,6 +117,15 @@ class RedshiftDataAPIServiceBackend(BaseBackend):
super().__init__(region_name, account_id)
self.statements: Dict[str, Statement] = {}
@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, "redshift-data", policy_supported=False
)
def cancel_statement(self, statement_id: str) -> None:
_validate_uuid(statement_id)

9
moto/secretsmanager/models.py
View File

@ -361,15 +361,6 @@ class SecretsManagerBackend(BaseBackend):
super().__init__(region_name, account_id)
self.secrets = SecretsStore()
@staticmethod
def default_vpc_endpoint_service(
service_region: str, zones: List[str]
) -> List[Dict[str, str]]:
"""Default VPC endpoint services."""
return BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "secretsmanager"
)
def _is_valid_identifier(self, identifier: str) -> bool:
return identifier in self.secrets

9
moto/sns/models.py
View File

@ -409,15 +409,6 @@ class SNSBackend(BaseBackend):
"+447700900907",
]
@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, "sns"
)
def get_sms_attributes(self, filter_list: Set[str]) -> Dict[str, str]:
if len(filter_list) > 0:
return {k: v for k, v in self.sms_attributes.items() if k in filter_list}

9
moto/sqs/models.py
View File

@ -679,15 +679,6 @@ class SQSBackend(BaseBackend):
super().__init__(region_name, account_id)
self.queues: Dict[str, Queue] = {}
@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, "sqs"
)
def create_queue(
self, name: str, tags: Optional[Dict[str, str]] = None, **kwargs: Any
) -> Queue:

11
moto/ssm/models.py
View File

@ -1188,17 +1188,6 @@ class SimpleSystemManagerBackend(BaseBackend):
self.windows: Dict[str, FakeMaintenanceWindow] = dict()
self.baselines: Dict[str, FakePatchBaseline] = dict()
@staticmethod
def default_vpc_endpoint_service(
service_region: str, zones: List[str]
) -> List[Dict[str, str]]:
"""Default VPC endpoint services."""
return BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "ssm"
) + BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "ssmmessages"
)
def _generate_document_information(
self, ssm_document: Document, document_format: str
) -> Dict[str, Any]:

9
moto/sts/models.py
View File

@ -81,15 +81,6 @@ class STSBackend(BaseBackend):
super().__init__(region_name, account_id)
self.assumed_roles: List[AssumedRole] = []
@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, "sts"
)
def get_session_token(self, duration: int) -> Token:
return Token(duration=duration)

11
moto/transcribe/models.py
View File

@ -480,17 +480,6 @@ class TranscribeBackend(BaseBackend):
self.medical_vocabularies: Dict[str, FakeMedicalVocabulary] = {}
self.vocabularies: Dict[str, FakeVocabulary] = {}
@staticmethod
def default_vpc_endpoint_service(
service_region: str, zones: List[str]
) -> List[Dict[str, str]]:
"""Default VPC endpoint services."""
return BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "transcribe"
) + BaseBackend.default_vpc_endpoint_service_factory(
service_region, zones, "transcribestreaming"
)
def start_transcription_job(
self,
transcription_job_name: str,

9
moto/xray/models.py
View File

@ -250,15 +250,6 @@ class XRayBackend(BaseBackend):
self._telemetry_records: List[TelemetryRecords] = []
self._segment_collection = SegmentCollection()
@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, "xray"
)
def add_telemetry_records(self, src: Any) -> None:
self._telemetry_records.append(TelemetryRecords.from_json(src))

10
tests/test_core/test_ec2_vpc_endpoint_services.py
View File

@ -51,6 +51,16 @@ def test_describe_vpc_endpoint_services_bad_args() -> None:
assert "The token 'foo' is invalid" in err["Message"]
@mock_aws
def test_describe_vpc_endpoint_services_unimplemented_service() -> None:
"""Verify exceptions are raised for bad arguments."""
ec2 = boto3.client("ec2", region_name="us-east-1")
service_name = "com.amazonaws.us-east-1.bedrock-agent-runtime"
resp = ec2.describe_vpc_endpoint_services(ServiceNames=[service_name])
assert resp["ServiceNames"] == [service_name]
@mock_aws
def test_describe_vpc_default_endpoint_services() -> None:
"""Test successfull calls as well as the next_token arg."""