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.