KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use endpoint method in localstack
Best Python code snippet using localstack_python
test_discovery.py
Source:test_discovery.py 
...73 with http_stubber as stubber:74 self.add_describe_endpoints_response(stubber, discovered_endpoint)75 client.describe_table(TableName='sometable')76 self.assert_endpoint_discovery_used(stubber, discovered_endpoint)77 def test_endpoint_discovery_with_invalid_endpoint(self):78 response = {79 'Error': {80 'Code': 'InvalidEndpointException',81 'Message': 'Test Error',82 }83 }84 response_body = json.dumps(response).encode()85 config = Config(endpoint_discovery_enabled=True)86 client, http_stubber = self.create_client(config=config)87 with http_stubber as stubber:88 stubber.add_response(status=421, body=response_body)89 with self.assertRaises(ClientError):90 client.describe_table(TableName='sometable')91 def test_endpoint_discovery_disabled(self):92 config = Config(endpoint_discovery_enabled=False)93 client, http_stubber = self.create_client(config=config)94 with http_stubber as stubber:95 stubber.add_response(status=200, body=b'{}')96 client.describe_table(TableName='sometable')97 self.assertEqual(len(stubber.requests), 1)98 def test_endpoint_discovery_no_config_default(self):99 client, http_stubber = self.create_client()100 with http_stubber as stubber:101 stubber.add_response(status=200, body=b'{}')102 client.describe_table(TableName='sometable')103 self.assertEqual(len(stubber.requests), 1)104 def test_endpoint_discovery_default_required_endpoint(self):105 discovered_endpoint = "https://discovered.domain"106 client, http_stubber = self.create_client(107 service_name="test-discovery-endpoint"108 )109 with http_stubber as stubber:110 self.add_describe_endpoints_response(stubber, discovered_endpoint)111 client.test_discovery_required(Foo="bar")112 self.assert_endpoint_discovery_used(stubber, discovered_endpoint)113 def test_endpoint_discovery_required_with_discovery_enabled(self):114 discovered_endpoint = "https://discovered.domain"115 config = Config(endpoint_discovery_enabled=True)116 client, http_stubber = self.create_client(117 service_name="test-discovery-endpoint", config=config118 )119 with http_stubber as stubber:120 self.add_describe_endpoints_response(stubber, discovered_endpoint)121 client.test_discovery_required(Foo="bar")122 self.assert_endpoint_discovery_used(stubber, discovered_endpoint)123 def test_endpoint_discovery_required_with_discovery_disabled(self):124 discovered_endpoint = "https://discovered.domain"125 config = Config(endpoint_discovery_enabled=False)126 client, http_stubber = self.create_client(127 service_name="test-discovery-endpoint", config=config128 )129 self.add_describe_endpoints_response(http_stubber, discovered_endpoint)130 with self.assertRaises(EndpointDiscoveryRequired):131 client.test_discovery_required(Foo="bar")132 def test_endpoint_discovery_required_with_custom_endpoint(self):133 endpoint = "https://custom.domain/"134 client, http_stubber = self.create_client(135 service_name="test-discovery-endpoint", endpoint_url=endpoint136 )137 with http_stubber as stubber:138 stubber.add_response(status=200, body=b'{}')139 client.test_discovery_required(Foo="bar")140 self.assert_discovery_skipped(141 stubber, b"test-discovery-endpoint.TestDiscoveryRequired"142 )143 self.assert_endpoint_used(stubber.requests[0].url, endpoint)144 def test_endpoint_discovery_disabled_with_custom_endpoint(self):145 endpoint = "https://custom.domain/"146 config = Config(endpoint_discovery_enabled=False)147 client, http_stubber = self.create_client(148 service_name="test-discovery-endpoint",149 config=config,150 endpoint_url=endpoint,151 )152 with http_stubber as stubber:153 stubber.add_response(status=200, body=b'{}')154 client.test_discovery_required(Foo="bar")155 self.assert_discovery_skipped(156 stubber, b"test-discovery-endpoint.TestDiscoveryRequired"157 )158 self.assert_endpoint_used(stubber.requests[0].url, endpoint)159 def test_endpoint_discovery_enabled_with_custom_endpoint(self):160 endpoint = "https://custom.domain/"161 config = Config(endpoint_discovery_enabled=True)162 client, http_stubber = self.create_client(163 service_name="test-discovery-endpoint",164 config=config,165 endpoint_url=endpoint,166 )167 with http_stubber as stubber:168 stubber.add_response(status=200, body=b'{}')169 client.test_discovery_required(Foo="bar")170 self.assert_discovery_skipped(171 stubber, b"test-discovery-endpoint.TestDiscoveryRequired"172 )173 self.assert_endpoint_used(stubber.requests[0].url, endpoint)174 def test_endpoint_discovery_optional_with_custom_endpoint(self):175 endpoint = "https://custom.domain/"176 client, http_stubber = self.create_client(177 service_name="test-discovery-endpoint", endpoint_url=endpoint178 )179 with http_stubber as stubber:180 stubber.add_response(status=200, body=b'{}')181 client.test_discovery_optional(Foo="bar")182 self.assert_discovery_skipped(183 stubber, b"test-discovery-endpoint.TestDiscoveryOptional"184 )185 self.assert_endpoint_used(stubber.requests[0].url, endpoint)186 def test_endpoint_discovery_optional_disabled_with_custom_endpoint(self):187 endpoint = "https://custom.domain/"188 config = Config(endpoint_discovery_enabled=False)189 client, http_stubber = self.create_client(190 service_name="test-discovery-endpoint",191 config=config,192 endpoint_url=endpoint,193 )194 with http_stubber as stubber:195 stubber.add_response(status=200, body=b'{}')196 client.test_discovery_optional(Foo="bar")197 self.assert_discovery_skipped(198 stubber,199 b"test-discovery-endpoint.TestDiscoveryOptional",200 )201 self.assert_endpoint_used(stubber.requests[0].url, endpoint)202 def test_endpoint_discovery_default_optional_endpoint(self):203 client, http_stubber = self.create_client(204 service_name="test-discovery-endpoint"205 )206 with http_stubber as stubber:207 stubber.add_response(status=200, body=b'{}')208 client.test_discovery_optional(Foo="bar")209 self.assertEqual(len(stubber.requests), 1)210 def test_endpoint_discovery_enabled_optional_endpoint(self):211 discovered_endpoint = 'https://discovered.domain'212 config = Config(endpoint_discovery_enabled=True)213 client, http_stubber = self.create_client(214 service_name="test-discovery-endpoint", config=config215 )216 with http_stubber as stubber:217 self.add_describe_endpoints_response(stubber, discovered_endpoint)218 client.test_discovery_optional(Foo="bar")219 self.assert_endpoint_discovery_used(stubber, discovered_endpoint)220 def test_endpoint_discovery_manual_auto_on_required_endpoint(self):221 discovered_endpoint = 'https://discovered.domain'222 config = Config(endpoint_discovery_enabled=" aUto \n")223 client, http_stubber = self.create_client(224 service_name="test-discovery-endpoint", config=config225 )226 with http_stubber as stubber:227 self.add_describe_endpoints_response(stubber, discovered_endpoint)228 client.test_discovery_required(Foo="bar")229 self.assert_endpoint_discovery_used(stubber, discovered_endpoint)230 def test_endpoint_discovery_enabled_with_random_string(self):231 config = Config(endpoint_discovery_enabled="bad value")232 with self.assertRaises(InvalidEndpointDiscoveryConfigurationError):233 client, http_stubber = self.create_client(234 service_name="test-discovery-endpoint", config=config...
test_event_alias.py
Source:test_event_alias.py
1import pytest2from botocore.session import Session3# The list of services which were available when we switched over from using4# endpoint prefix in event to using service id. These should all accept5# either.6SERVICES = {7 "acm": {"endpoint_prefix": "acm", "service_id": "acm"},8 "acm-pca": {"endpoint_prefix": "acm-pca", "service_id": "acm-pca"},9 "alexaforbusiness": {10 "endpoint_prefix": "a4b",11 "service_id": "alexa-for-business",12 },13 "apigateway": {14 "endpoint_prefix": "apigateway",15 "service_id": "api-gateway",16 },17 "application-autoscaling": {"service_id": "application-auto-scaling"},18 "appstream": {"endpoint_prefix": "appstream2", "service_id": "appstream"},19 "appsync": {"endpoint_prefix": "appsync", "service_id": "appsync"},20 "athena": {"endpoint_prefix": "athena", "service_id": "athena"},21 "autoscaling": {22 "endpoint_prefix": "autoscaling",23 "service_id": "auto-scaling",24 },25 "autoscaling-plans": {"service_id": "auto-scaling-plans"},26 "batch": {"endpoint_prefix": "batch", "service_id": "batch"},27 "budgets": {"endpoint_prefix": "budgets", "service_id": "budgets"},28 "ce": {"endpoint_prefix": "ce", "service_id": "cost-explorer"},29 "cloud9": {"endpoint_prefix": "cloud9", "service_id": "cloud9"},30 "clouddirectory": {31 "endpoint_prefix": "clouddirectory",32 "service_id": "clouddirectory",33 },34 "cloudformation": {35 "endpoint_prefix": "cloudformation",36 "service_id": "cloudformation",37 },38 "cloudfront": {39 "endpoint_prefix": "cloudfront",40 "service_id": "cloudfront",41 },42 "cloudhsm": {"endpoint_prefix": "cloudhsm", "service_id": "cloudhsm"},43 "cloudhsmv2": {44 "endpoint_prefix": "cloudhsmv2",45 "service_id": "cloudhsm-v2",46 },47 "cloudsearch": {48 "endpoint_prefix": "cloudsearch",49 "service_id": "cloudsearch",50 },51 "cloudsearchdomain": {52 "endpoint_prefix": "cloudsearchdomain",53 "service_id": "cloudsearch-domain",54 },55 "cloudtrail": {56 "endpoint_prefix": "cloudtrail",57 "service_id": "cloudtrail",58 },59 "cloudwatch": {60 "endpoint_prefix": "monitoring",61 "service_id": "cloudwatch",62 },63 "codebuild": {"endpoint_prefix": "codebuild", "service_id": "codebuild"},64 "codecommit": {65 "endpoint_prefix": "codecommit",66 "service_id": "codecommit",67 },68 "codedeploy": {69 "endpoint_prefix": "codedeploy",70 "service_id": "codedeploy",71 },72 "codepipeline": {73 "endpoint_prefix": "codepipeline",74 "service_id": "codepipeline",75 },76 "codestar": {"endpoint_prefix": "codestar", "service_id": "codestar"},77 "cognito-identity": {78 "endpoint_prefix": "cognito-identity",79 "service_id": "cognito-identity",80 },81 "cognito-idp": {82 "endpoint_prefix": "cognito-idp",83 "service_id": "cognito-identity-provider",84 },85 "cognito-sync": {86 "endpoint_prefix": "cognito-sync",87 "service_id": "cognito-sync",88 },89 "comprehend": {90 "endpoint_prefix": "comprehend",91 "service_id": "comprehend",92 },93 "config": {"endpoint_prefix": "config", "service_id": "config-service"},94 "connect": {"endpoint_prefix": "connect", "service_id": "connect"},95 "cur": {96 "endpoint_prefix": "cur",97 "service_id": "cost-and-usage-report-service",98 },99 "datapipeline": {100 "endpoint_prefix": "datapipeline",101 "service_id": "data-pipeline",102 },103 "dax": {"endpoint_prefix": "dax", "service_id": "dax"},104 "devicefarm": {105 "endpoint_prefix": "devicefarm",106 "service_id": "device-farm",107 },108 "directconnect": {109 "endpoint_prefix": "directconnect",110 "service_id": "direct-connect",111 },112 "discovery": {113 "endpoint_prefix": "discovery",114 "service_id": "application-discovery-service",115 },116 "dlm": {"endpoint_prefix": "dlm", "service_id": "dlm"},117 "dms": {118 "endpoint_prefix": "dms",119 "service_id": "database-migration-service",120 },121 "ds": {"endpoint_prefix": "ds", "service_id": "directory-service"},122 "dynamodb": {"endpoint_prefix": "dynamodb", "service_id": "dynamodb"},123 "dynamodbstreams": {124 "endpoint_prefix": "streams.dynamodb",125 "service_id": "dynamodb-streams",126 },127 "ec2": {"endpoint_prefix": "ec2", "service_id": "ec2"},128 "ecr": {"endpoint_prefix": "ecr", "service_id": "ecr"},129 "ecs": {"endpoint_prefix": "ecs", "service_id": "ecs"},130 "efs": {"endpoint_prefix": "elasticfilesystem", "service_id": "efs"},131 "eks": {"endpoint_prefix": "eks", "service_id": "eks"},132 "elasticache": {133 "endpoint_prefix": "elasticache",134 "service_id": "elasticache",135 },136 "elasticbeanstalk": {137 "endpoint_prefix": "elasticbeanstalk",138 "service_id": "elastic-beanstalk",139 },140 "elastictranscoder": {141 "endpoint_prefix": "elastictranscoder",142 "service_id": "elastic-transcoder",143 },144 "elb": {145 "endpoint_prefix": "elasticloadbalancing",146 "service_id": "elastic-load-balancing",147 },148 "elbv2": {"service_id": "elastic-load-balancing-v2"},149 "emr": {"endpoint_prefix": "elasticmapreduce", "service_id": "emr"},150 "es": {"endpoint_prefix": "es", "service_id": "elasticsearch-service"},151 "events": {"endpoint_prefix": "events", "service_id": "cloudwatch-events"},152 "firehose": {"endpoint_prefix": "firehose", "service_id": "firehose"},153 "fms": {"endpoint_prefix": "fms", "service_id": "fms"},154 "gamelift": {"endpoint_prefix": "gamelift", "service_id": "gamelift"},155 "glacier": {"endpoint_prefix": "glacier", "service_id": "glacier"},156 "glue": {"endpoint_prefix": "glue", "service_id": "glue"},157 "greengrass": {158 "endpoint_prefix": "greengrass",159 "service_id": "greengrass",160 },161 "guardduty": {"endpoint_prefix": "guardduty", "service_id": "guardduty"},162 "health": {"endpoint_prefix": "health", "service_id": "health"},163 "iam": {"endpoint_prefix": "iam", "service_id": "iam"},164 "importexport": {165 "endpoint_prefix": "importexport",166 "service_id": "importexport",167 },168 "inspector": {"endpoint_prefix": "inspector", "service_id": "inspector"},169 "iot": {"endpoint_prefix": "iot", "service_id": "iot"},170 "iot-data": {171 "endpoint_prefix": "data.iot",172 "service_id": "iot-data-plane",173 },174 "iot-jobs-data": {175 "endpoint_prefix": "data.jobs.iot",176 "service_id": "iot-jobs-data-plane",177 },178 "iot1click-devices": {179 "endpoint_prefix": "devices.iot1click",180 "service_id": "iot-1click-devices-service",181 },182 "iot1click-projects": {183 "endpoint_prefix": "projects.iot1click",184 "service_id": "iot-1click-projects",185 },186 "iotanalytics": {187 "endpoint_prefix": "iotanalytics",188 "service_id": "iotanalytics",189 },190 "kinesis": {"endpoint_prefix": "kinesis", "service_id": "kinesis"},191 "kinesis-video-archived-media": {192 "service_id": "kinesis-video-archived-media"193 },194 "kinesis-video-media": {"service_id": "kinesis-video-media"},195 "kinesisanalytics": {196 "endpoint_prefix": "kinesisanalytics",197 "service_id": "kinesis-analytics",198 },199 "kinesisvideo": {200 "endpoint_prefix": "kinesisvideo",201 "service_id": "kinesis-video",202 },203 "kms": {"endpoint_prefix": "kms", "service_id": "kms"},204 "lambda": {"endpoint_prefix": "lambda", "service_id": "lambda"},205 "lex-models": {206 "endpoint_prefix": "models.lex",207 "service_id": "lex-model-building-service",208 },209 "lex-runtime": {210 "endpoint_prefix": "runtime.lex",211 "service_id": "lex-runtime-service",212 },213 "lightsail": {"endpoint_prefix": "lightsail", "service_id": "lightsail"},214 "logs": {"endpoint_prefix": "logs", "service_id": "cloudwatch-logs"},215 "machinelearning": {216 "endpoint_prefix": "machinelearning",217 "service_id": "machine-learning",218 },219 "macie": {"endpoint_prefix": "macie", "service_id": "macie"},220 "marketplace-entitlement": {221 "endpoint_prefix": "entitlement.marketplace",222 "service_id": "marketplace-entitlement-service",223 },224 "marketplacecommerceanalytics": {225 "endpoint_prefix": "marketplacecommerceanalytics",226 "service_id": "marketplace-commerce-analytics",227 },228 "mediaconvert": {229 "endpoint_prefix": "mediaconvert",230 "service_id": "mediaconvert",231 },232 "medialive": {"endpoint_prefix": "medialive", "service_id": "medialive"},233 "mediapackage": {234 "endpoint_prefix": "mediapackage",235 "service_id": "mediapackage",236 },237 "mediastore": {238 "endpoint_prefix": "mediastore",239 "service_id": "mediastore",240 },241 "mediastore-data": {242 "endpoint_prefix": "data.mediastore",243 "service_id": "mediastore-data",244 },245 "mediatailor": {246 "endpoint_prefix": "api.mediatailor",247 "service_id": "mediatailor",248 },249 "meteringmarketplace": {250 "endpoint_prefix": "metering.marketplace",251 "service_id": "marketplace-metering",252 },253 "mgh": {"endpoint_prefix": "mgh", "service_id": "migration-hub"},254 "mobile": {"endpoint_prefix": "mobile", "service_id": "mobile"},255 "mq": {"endpoint_prefix": "mq", "service_id": "mq"},256 "mturk": {"endpoint_prefix": "mturk-requester", "service_id": "mturk"},257 "neptune": {"service_id": "neptune"},258 "opsworks": {"endpoint_prefix": "opsworks", "service_id": "opsworks"},259 "opsworkscm": {260 "endpoint_prefix": "opsworks-cm",261 "service_id": "opsworkscm",262 },263 "organizations": {264 "endpoint_prefix": "organizations",265 "service_id": "organizations",266 },267 "pi": {"endpoint_prefix": "pi", "service_id": "pi"},268 "pinpoint": {"endpoint_prefix": "pinpoint", "service_id": "pinpoint"},269 "polly": {"endpoint_prefix": "polly", "service_id": "polly"},270 "pricing": {"endpoint_prefix": "api.pricing", "service_id": "pricing"},271 "rds": {"endpoint_prefix": "rds", "service_id": "rds"},272 "redshift": {"endpoint_prefix": "redshift", "service_id": "redshift"},273 "rekognition": {274 "endpoint_prefix": "rekognition",275 "service_id": "rekognition",276 },277 "resource-groups": {278 "endpoint_prefix": "resource-groups",279 "service_id": "resource-groups",280 },281 "resourcegroupstaggingapi": {282 "endpoint_prefix": "tagging",283 "service_id": "resource-groups-tagging-api",284 },285 "route53": {"endpoint_prefix": "route53", "service_id": "route-53"},286 "route53domains": {287 "endpoint_prefix": "route53domains",288 "service_id": "route-53-domains",289 },290 "s3": {"endpoint_prefix": "s3", "service_id": "s3"},291 "sagemaker": {292 "endpoint_prefix": "api.sagemaker",293 "service_id": "sagemaker",294 },295 "sagemaker-runtime": {296 "endpoint_prefix": "runtime.sagemaker",297 "service_id": "sagemaker-runtime",298 },299 "sdb": {"endpoint_prefix": "sdb", "service_id": "simpledb"},300 "secretsmanager": {301 "endpoint_prefix": "secretsmanager",302 "service_id": "secrets-manager",303 },304 "serverlessrepo": {305 "endpoint_prefix": "serverlessrepo",306 "service_id": "serverlessapplicationrepository",307 },308 "servicecatalog": {309 "endpoint_prefix": "servicecatalog",310 "service_id": "service-catalog",311 },312 "servicediscovery": {313 "endpoint_prefix": "servicediscovery",314 "service_id": "servicediscovery",315 },316 "ses": {"endpoint_prefix": "email", "service_id": "ses"},317 "shield": {"endpoint_prefix": "shield", "service_id": "shield"},318 "sms": {"endpoint_prefix": "sms", "service_id": "sms"},319 "snowball": {"endpoint_prefix": "snowball", "service_id": "snowball"},320 "sns": {"endpoint_prefix": "sns", "service_id": "sns"},321 "sqs": {"endpoint_prefix": "sqs", "service_id": "sqs"},322 "ssm": {"endpoint_prefix": "ssm", "service_id": "ssm"},323 "stepfunctions": {"endpoint_prefix": "states", "service_id": "sfn"},324 "storagegateway": {325 "endpoint_prefix": "storagegateway",326 "service_id": "storage-gateway",327 },328 "sts": {"endpoint_prefix": "sts", "service_id": "sts"},329 "support": {"endpoint_prefix": "support", "service_id": "support"},330 "swf": {"endpoint_prefix": "swf", "service_id": "swf"},331 "transcribe": {332 "endpoint_prefix": "transcribe",333 "service_id": "transcribe",334 },335 "translate": {"endpoint_prefix": "translate", "service_id": "translate"},336 "waf": {"endpoint_prefix": "waf", "service_id": "waf"},337 "waf-regional": {338 "endpoint_prefix": "waf-regional",339 "service_id": "waf-regional",340 },341 "workdocs": {"endpoint_prefix": "workdocs", "service_id": "workdocs"},342 "workmail": {"endpoint_prefix": "workmail", "service_id": "workmail"},343 "workspaces": {344 "endpoint_prefix": "workspaces",345 "service_id": "workspaces",346 },347 "xray": {"endpoint_prefix": "xray", "service_id": "xray"},348}349def _event_aliases():350 for client_name in SERVICES.keys():351 service_id = SERVICES[client_name]['service_id']352 yield client_name, service_id353def _event_aliases_with_endpoint_prefix():354 for client_name in SERVICES.keys():355 endpoint_prefix = SERVICES[client_name].get('endpoint_prefix')356 if endpoint_prefix is not None:357 yield client_name, endpoint_prefix358@pytest.mark.parametrize(359 "client_name, endpoint_prefix", _event_aliases_with_endpoint_prefix()360)361def test_event_alias_by_endpoint_prefix(client_name, endpoint_prefix):362 _assert_handler_called(client_name, endpoint_prefix)363@pytest.mark.parametrize("client_name, service_id", _event_aliases())364def test_event_alias_by_service_id(client_name, service_id):365 _assert_handler_called(client_name, service_id)366@pytest.mark.parametrize("client_name, service_id", _event_aliases())367def test_event_alias_by_client_name(client_name, service_id):368 _assert_handler_called(client_name, client_name)369def _assert_handler_called(client_name, event_part):370 hook_calls = []371 def _hook(**kwargs):372 hook_calls.append(kwargs['event_name'])373 session = _get_session()374 session.register('creating-client-class.%s' % event_part, _hook)375 session.create_client(client_name)376 assert len(hook_calls) == 1377def _get_session():378 session = Session()379 session.set_credentials('foo', 'bar')380 session.set_config_variable('region', 'us-west-2')381 session.config_filename = 'no-exist-foo'...
test_eventbridge.py
Source:test_eventbridge.py
...35 "EventBusName": "my-bus",36 },37 ]38 }39 def _assert_multi_region_endpoint(self, request, endpoint_id, suffix=None):40 if suffix is None:41 suffix = "amazonaws.com"42 assert (43 request.url == f"https://{endpoint_id}.endpoint.events.{suffix}/"44 )45 def _assert_sigv4a_headers(self, request):46 assert request.headers["x-amz-region-set"] == b"*"47 assert request.headers["authorization"].startswith(48 b"AWS4-ECDSA-P256-SHA256 Credential="49 )50 def _assert_params_in_body(self, request, params):51 assert len(params) > 052 body = json.loads(request.body)53 for key, value in params:54 assert body[key] == value55 def test_put_event_default_endpoint(self):56 client, stubber = self.create_stubbed_eventbridge_client(57 with_default_responses=True,58 )59 with stubber:60 client.put_events(**self._default_put_events_args())61 assert (62 stubber.requests[0].url63 == "https://events.us-east-1.amazonaws.com/"64 )65 assert b"EndpointId" not in stubber.requests[0].body66 def test_put_event_default_endpoint_explicit_configs(self):67 client, stubber = self.create_stubbed_eventbridge_client(68 with_default_responses=True,69 config=Config(70 use_dualstack_endpoint=False,71 use_fips_endpoint=False,72 ),73 )74 with stubber:75 client.put_events(**self._default_put_events_args())76 assert (77 stubber.requests[0].url78 == "https://events.us-east-1.amazonaws.com/"79 )80 assert b"EndpointId" not in stubber.requests[0].body81 @requires_crt()82 def test_put_event_endpoint_id(self):83 client, stubber = self.create_stubbed_eventbridge_client(84 with_default_responses=True,85 )86 default_args = self._default_put_events_args()87 endpoint_id = "abc123.456def"88 with stubber:89 client.put_events(EndpointId=endpoint_id, **default_args)90 self._assert_params_in_body(91 stubber.requests[0],92 [93 ("EndpointId", endpoint_id),94 ],95 )96 self._assert_multi_region_endpoint(stubber.requests[0], endpoint_id)97 self._assert_sigv4a_headers(stubber.requests[0])98 @requires_crt()99 def test_put_event_endpoint_id_explicit_config(self):100 client, stubber = self.create_stubbed_eventbridge_client(101 with_default_responses=True,102 config=Config(103 use_dualstack_endpoint=False,104 use_fips_endpoint=False,105 ),106 )107 default_args = self._default_put_events_args()108 endpoint_id = "abc123.456def"109 with stubber:110 client.put_events(EndpointId=endpoint_id, **default_args)111 self._assert_params_in_body(112 stubber.requests[0],113 [114 ("EndpointId", endpoint_id),115 ],116 )117 self._assert_multi_region_endpoint(stubber.requests[0], endpoint_id)118 self._assert_sigv4a_headers(stubber.requests[0])119 @requires_crt()120 def test_put_event_bad_endpoint_id(self):121 client, stubber = self.create_stubbed_eventbridge_client(122 with_default_responses=True,123 )124 default_args = self._default_put_events_args()125 endpoint_id = "badactor.com?foo=bar"126 with pytest.raises(InvalidEndpointConfigurationError) as e:127 client.put_events(EndpointId=endpoint_id, **default_args)128 assert "EndpointId is not a valid hostname component" in str(e.value)129 @requires_crt()130 def test_put_event_bad_endpoint_id_explicit_config(self):131 client, stubber = self.create_stubbed_eventbridge_client(132 with_default_responses=True,133 config=Config(134 use_dualstack_endpoint=False,135 use_fips_endpoint=False,136 ),137 )138 default_args = self._default_put_events_args()139 endpoint_id = "badactor.com?foo=bar"140 with pytest.raises(InvalidEndpointConfigurationError) as e:141 client.put_events(EndpointId=endpoint_id, **default_args)142 assert "EndpointId is not a valid hostname component" in str(e.value)143 @requires_crt()144 def test_put_event_empty_endpoint_id(self):145 client, stubber = self.create_stubbed_eventbridge_client(146 with_default_responses=True,147 )148 default_args = self._default_put_events_args()149 endpoint_id = ""150 with pytest.raises(InvalidEndpointConfigurationError) as e:151 client.put_events(EndpointId=endpoint_id, **default_args)152 assert "EndpointId must not be a zero length string" in str(e.value)153 @requires_crt()154 def test_put_event_empty_endpoint_id_explicit_config(self):155 client, stubber = self.create_stubbed_eventbridge_client(156 with_default_responses=True,157 config=Config(158 use_dualstack_endpoint=False,159 use_fips_endpoint=False,160 ),161 )162 default_args = self._default_put_events_args()163 endpoint_id = ""164 with pytest.raises(InvalidEndpointConfigurationError) as e:165 client.put_events(EndpointId=endpoint_id, **default_args)166 assert "EndpointId must not be a zero length string" in str(e.value)167 def test_put_event_default_dualstack_endpoint(self):168 config = Config(use_dualstack_endpoint=True, use_fips_endpoint=False)169 client, stubber = self.create_stubbed_eventbridge_client(170 with_default_responses=True, config=config171 )172 default_args = self._default_put_events_args()173 with stubber:174 client.put_events(**default_args)175 assert stubber.requests[0].url == "https://events.us-east-1.api.aws/"176 @requires_crt()177 def test_put_events_endpoint_id_dualstack(self):178 config = Config(use_dualstack_endpoint=True, use_fips_endpoint=False)179 client, stubber = self.create_stubbed_eventbridge_client(180 with_default_responses=True, config=config181 )182 default_args = self._default_put_events_args()183 endpoint_id = "abc123.456def"184 with stubber:185 client.put_events(EndpointId=endpoint_id, **default_args)186 self._assert_params_in_body(187 stubber.requests[0],188 [189 ("EndpointId", endpoint_id),190 ],191 )192 self._assert_multi_region_endpoint(193 stubber.requests[0], endpoint_id, suffix="api.aws"194 )195 self._assert_sigv4a_headers(stubber.requests[0])196 def test_put_events_default_fips_endpoint(self):197 config = Config(use_dualstack_endpoint=False, use_fips_endpoint=True)198 client, stubber = self.create_stubbed_eventbridge_client(199 with_default_responses=True, config=config200 )201 default_args = self._default_put_events_args()202 with stubber:203 client.put_events(**default_args)204 assert (205 stubber.requests[0].url206 == "https://events-fips.us-east-1.amazonaws.com/"207 )208 @requires_crt()209 def test_put_events_endpoint_id_fips(self):210 config = Config(use_dualstack_endpoint=False, use_fips_endpoint=True)211 client, stubber = self.create_stubbed_eventbridge_client(212 with_default_responses=True, config=config213 )214 default_args = self._default_put_events_args()215 endpoint_id = "abc123.456def"216 with pytest.raises(InvalidEndpointConfigurationError) as e:217 client.put_events(EndpointId=endpoint_id, **default_args)218 assert (219 "FIPS is not supported with EventBridge multi-region endpoints"220 in str(e.value)221 )222 def test_put_events_default_dualstack_fips_endpoint(self):223 config = Config(use_dualstack_endpoint=True, use_fips_endpoint=True)224 client, stubber = self.create_stubbed_eventbridge_client(225 with_default_responses=True, config=config226 )227 default_args = self._default_put_events_args()228 with stubber:229 client.put_events(**default_args)230 assert (231 stubber.requests[0].url == "https://events-fips.us-east-1.api.aws/"232 )233 @requires_crt()234 def test_put_events_endpoint_id_dualstack_fips(self):235 config = Config(use_dualstack_endpoint=True, use_fips_endpoint=True)236 client, stubber = self.create_stubbed_eventbridge_client(237 with_default_responses=True, config=config238 )239 default_args = self._default_put_events_args()240 endpoint_id = "abc123.456def"241 with pytest.raises(InvalidEndpointConfigurationError) as e:242 client.put_events(EndpointId=endpoint_id, **default_args)243 assert (244 "FIPS is not supported with EventBridge multi-region endpoints"245 in str(e.value)246 )247 def test_put_events_default_gov_endpoint(self):248 client, stubber = self.create_stubbed_eventbridge_client(249 with_default_responses=True,250 region="us-iso-east-1",251 )252 default_args = self._default_put_events_args()253 with stubber:254 client.put_events(**default_args)255 assert (256 stubber.requests[0].url257 == "https://events.us-iso-east-1.c2s.ic.gov/"258 )259 @requires_crt()260 def test_put_events_endpoint_id_gov(self):261 client, stubber = self.create_stubbed_eventbridge_client(262 with_default_responses=True,263 region="us-iso-east-1",264 )265 default_args = self._default_put_events_args()266 endpoint_id = "abc123.456def"267 with stubber:268 client.put_events(EndpointId=endpoint_id, **default_args)269 self._assert_params_in_body(270 stubber.requests[0],271 [272 ("EndpointId", endpoint_id),273 ],274 )275 self._assert_multi_region_endpoint(276 stubber.requests[0], endpoint_id, suffix="c2s.ic.gov"277 )278 self._assert_sigv4a_headers(stubber.requests[0])279 def test_put_events_default_custom_endpoint(self):280 client, stubber = self.create_stubbed_eventbridge_client(281 with_default_responses=True, endpoint_url="https://example.org"282 )283 default_args = self._default_put_events_args()284 with stubber:285 client.put_events(**default_args)286 assert stubber.requests[0].url == "https://example.org/"287 @requires_crt()288 def test_put_events_endpoint_id_custom(self):289 client, stubber = self.create_stubbed_eventbridge_client(290 with_default_responses=True, endpoint_url="https://example.org"291 )292 default_args = self._default_put_events_args()293 endpoint_id = "abc123.456def"...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
