Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use create_rest_api method in localstack

Best Python code snippet using localstack_python

test_apigateway.py

Source:test_apigateway.py

...12@freeze_time("2015-01-01")13@mock_apigateway14def test_create_and_get_rest_api():15    client = boto3.client("apigateway", region_name="us-west-2")16    response = client.create_rest_api(name="my_api", description="this is my api")17    api_id = response["id"]18    response = client.get_rest_api(restApiId=api_id)19    response.pop("ResponseMetadata")20    response.pop("createdDate")21    response.should.equal(22        {23            "id": api_id,24            "name": "my_api",25            "description": "this is my api",26            "version": "V1",27            "binaryMediaTypes": [],28            "apiKeySource": "HEADER",29            "endpointConfiguration": {"types": ["EDGE"]},30            "tags": {},31            "disableExecuteApiEndpoint": False,32        }33    )34@mock_apigateway35def test_upate_rest_api():36    client = boto3.client("apigateway", region_name="us-west-2")37    response = client.create_rest_api(name="my_api", description="this is my api")38    api_id = response["id"]39    patchOperations = [40        {"op": "replace", "path": "/name", "value": "new-name"},41        {"op": "replace", "path": "/description", "value": "new-description"},42        {"op": "replace", "path": "/apiKeySource", "value": "AUTHORIZER"},43        {"op": "replace", "path": "/binaryMediaTypes", "value": "image/jpeg"},44        {"op": "replace", "path": "/disableExecuteApiEndpoint", "value": "True"},45    ]46    response = client.update_rest_api(restApiId=api_id, patchOperations=patchOperations)47    response.pop("ResponseMetadata")48    response.pop("createdDate")49    response.pop("binaryMediaTypes")50    response.should.equal(51        {52            "id": api_id,53            "name": "new-name",54            "version": "V1",55            "description": "new-description",56            "apiKeySource": "AUTHORIZER",57            "endpointConfiguration": {"types": ["EDGE"]},58            "tags": {},59            "disableExecuteApiEndpoint": True,60        }61    )62    # should fail with wrong apikeysoruce63    patchOperations = [64        {"op": "replace", "path": "/apiKeySource", "value": "Wrong-value-AUTHORIZER"}65    ]66    with pytest.raises(ClientError) as ex:67        response = client.update_rest_api(68            restApiId=api_id, patchOperations=patchOperations69        )70    ex.value.response["Error"]["Message"].should.equal(71        "1 validation error detected: Value 'Wrong-value-AUTHORIZER' at 'createRestApiInput.apiKeySource' failed to satisfy constraint: Member must satisfy enum value set: [AUTHORIZER, HEADER]"72    )73    ex.value.response["Error"]["Code"].should.equal("ValidationException")74@mock_apigateway75def test_upate_rest_api_invalid_api_id():76    client = boto3.client("apigateway", region_name="us-west-2")77    patchOperations = [78        {"op": "replace", "path": "/apiKeySource", "value": "AUTHORIZER"}79    ]80    with pytest.raises(ClientError) as ex:81        client.update_rest_api(restApiId="api_id", patchOperations=patchOperations)82    ex.value.response["Error"]["Code"].should.equal("NotFoundException")83@mock_apigateway84def test_list_and_delete_apis():85    client = boto3.client("apigateway", region_name="us-west-2")86    response = client.create_rest_api(name="my_api", description="this is my api")87    api_id = response["id"]88    client.create_rest_api(name="my_api2", description="this is my api2")89    response = client.get_rest_apis()90    len(response["items"]).should.equal(2)91    client.delete_rest_api(restApiId=api_id)92    response = client.get_rest_apis()93    len(response["items"]).should.equal(1)94@mock_apigateway95def test_create_rest_api_with_tags():96    client = boto3.client("apigateway", region_name="us-west-2")97    response = client.create_rest_api(98        name="my_api", description="this is my api", tags={"MY_TAG1": "MY_VALUE1"}99    )100    api_id = response["id"]101    response = client.get_rest_api(restApiId=api_id)102    assert "tags" in response103    response["tags"].should.equal({"MY_TAG1": "MY_VALUE1"})104@mock_apigateway105def test_create_rest_api_with_policy():106    client = boto3.client("apigateway", region_name="us-west-2")107    policy = '{"Version": "2012-10-17","Statement": []}'108    response = client.create_rest_api(109        name="my_api", description="this is my api", policy=policy110    )111    api_id = response["id"]112    response = client.get_rest_api(restApiId=api_id)113    assert "policy" in response114    response["policy"].should.equal(policy)115@mock_apigateway116def test_create_rest_api_invalid_apikeysource():117    client = boto3.client("apigateway", region_name="us-west-2")118    with pytest.raises(ClientError) as ex:119        client.create_rest_api(120            name="my_api",121            description="this is my api",122            apiKeySource="not a valid api key source",123        )124    ex.value.response["Error"]["Code"].should.equal("ValidationException")125@mock_apigateway126def test_create_rest_api_valid_apikeysources():127    client = boto3.client("apigateway", region_name="us-west-2")128    # 1. test creating rest api with HEADER apiKeySource129    response = client.create_rest_api(130        name="my_api", description="this is my api", apiKeySource="HEADER",131    )132    api_id = response["id"]133    response = client.get_rest_api(restApiId=api_id)134    response["apiKeySource"].should.equal("HEADER")135    # 2. test creating rest api with AUTHORIZER apiKeySource136    response = client.create_rest_api(137        name="my_api2", description="this is my api", apiKeySource="AUTHORIZER",138    )139    api_id = response["id"]140    response = client.get_rest_api(restApiId=api_id)141    response["apiKeySource"].should.equal("AUTHORIZER")142@mock_apigateway143def test_create_rest_api_invalid_endpointconfiguration():144    client = boto3.client("apigateway", region_name="us-west-2")145    with pytest.raises(ClientError) as ex:146        client.create_rest_api(147            name="my_api",148            description="this is my api",149            endpointConfiguration={"types": ["INVALID"]},150        )151    ex.value.response["Error"]["Code"].should.equal("ValidationException")152@mock_apigateway153def test_create_rest_api_valid_endpointconfigurations():154    client = boto3.client("apigateway", region_name="us-west-2")155    # 1. test creating rest api with PRIVATE endpointConfiguration156    response = client.create_rest_api(157        name="my_api",158        description="this is my api",159        endpointConfiguration={"types": ["PRIVATE"]},160    )161    api_id = response["id"]162    response = client.get_rest_api(restApiId=api_id)163    response["endpointConfiguration"].should.equal(164        {"types": ["PRIVATE"],}165    )166    # 2. test creating rest api with REGIONAL endpointConfiguration167    response = client.create_rest_api(168        name="my_api2",169        description="this is my api",170        endpointConfiguration={"types": ["REGIONAL"]},171    )172    api_id = response["id"]173    response = client.get_rest_api(restApiId=api_id)174    response["endpointConfiguration"].should.equal(175        {"types": ["REGIONAL"],}176    )177    # 3. test creating rest api with EDGE endpointConfiguration178    response = client.create_rest_api(179        name="my_api3",180        description="this is my api",181        endpointConfiguration={"types": ["EDGE"]},182    )183    api_id = response["id"]184    response = client.get_rest_api(restApiId=api_id)185    response["endpointConfiguration"].should.equal(186        {"types": ["EDGE"],}187    )188@mock_apigateway189def test_create_resource__validate_name():190    client = boto3.client("apigateway", region_name="us-west-2")191    response = client.create_rest_api(name="my_api", description="this is my api")192    api_id = response["id"]193    resources = client.get_resources(restApiId=api_id)194    root_id = [resource for resource in resources["items"] if resource["path"] == "/"][195        0196    ]["id"]197    invalid_names = ["/users", "users/", "users/{user_id}", "us{er", "us+er"]198    valid_names = ["users", "{user_id}", "{proxy+}", "user_09", "good-dog"]199    # All invalid names should throw an exception200    for name in invalid_names:201        with pytest.raises(ClientError) as ex:202            client.create_resource(restApiId=api_id, parentId=root_id, pathPart=name)203        ex.value.response["Error"]["Code"].should.equal("BadRequestException")204        ex.value.response["Error"]["Message"].should.equal(205            "Resource's path part only allow a-zA-Z0-9._- and curly braces at the beginning and the end and an optional plus sign before the closing brace."206        )207    # All valid names  should go through208    for name in valid_names:209        client.create_resource(restApiId=api_id, parentId=root_id, pathPart=name)210@mock_apigateway211def test_create_resource():212    client = boto3.client("apigateway", region_name="us-west-2")213    response = client.create_rest_api(name="my_api", description="this is my api")214    api_id = response["id"]215    resources = client.get_resources(restApiId=api_id)216    root_id = [resource for resource in resources["items"] if resource["path"] == "/"][217        0218    ]["id"]219    root_resource = client.get_resource(restApiId=api_id, resourceId=root_id)220    # this is hard to match against, so remove it221    root_resource["ResponseMetadata"].pop("HTTPHeaders", None)222    root_resource["ResponseMetadata"].pop("RetryAttempts", None)223    root_resource.should.equal(224        {"path": "/", "id": root_id, "ResponseMetadata": {"HTTPStatusCode": 200},}225    )226    client.create_resource(restApiId=api_id, parentId=root_id, pathPart="users")227    resources = client.get_resources(restApiId=api_id)["items"]228    len(resources).should.equal(2)229    non_root_resource = [resource for resource in resources if resource["path"] != "/"][230        0231    ]232    client.delete_resource(restApiId=api_id, resourceId=non_root_resource["id"])233    len(client.get_resources(restApiId=api_id)["items"]).should.equal(1)234@mock_apigateway235def test_child_resource():236    client = boto3.client("apigateway", region_name="us-west-2")237    response = client.create_rest_api(name="my_api", description="this is my api")238    api_id = response["id"]239    resources = client.get_resources(restApiId=api_id)240    root_id = [resource for resource in resources["items"] if resource["path"] == "/"][241        0242    ]["id"]243    response = client.create_resource(244        restApiId=api_id, parentId=root_id, pathPart="users"245    )246    users_id = response["id"]247    response = client.create_resource(248        restApiId=api_id, parentId=users_id, pathPart="tags"249    )250    tags_id = response["id"]251    child_resource = client.get_resource(restApiId=api_id, resourceId=tags_id)252    # this is hard to match against, so remove it253    child_resource["ResponseMetadata"].pop("HTTPHeaders", None)254    child_resource["ResponseMetadata"].pop("RetryAttempts", None)255    child_resource.should.equal(256        {257            "path": "/users/tags",258            "pathPart": "tags",259            "parentId": users_id,260            "id": tags_id,261            "ResponseMetadata": {"HTTPStatusCode": 200},262        }263    )264@mock_apigateway265def test_create_method():266    client = boto3.client("apigateway", region_name="us-west-2")267    response = client.create_rest_api(name="my_api", description="this is my api")268    api_id = response["id"]269    resources = client.get_resources(restApiId=api_id)270    root_id = [resource for resource in resources["items"] if resource["path"] == "/"][271        0272    ]["id"]273    client.put_method(274        restApiId=api_id, resourceId=root_id, httpMethod="GET", authorizationType="none"275    )276    response = client.get_method(restApiId=api_id, resourceId=root_id, httpMethod="GET")277    # this is hard to match against, so remove it278    response["ResponseMetadata"].pop("HTTPHeaders", None)279    response["ResponseMetadata"].pop("RetryAttempts", None)280    response.should.equal(281        {282            "httpMethod": "GET",283            "authorizationType": "none",284            "apiKeyRequired": False,285            "ResponseMetadata": {"HTTPStatusCode": 200},286        }287    )288@mock_apigateway289def test_create_method_apikeyrequired():290    client = boto3.client("apigateway", region_name="us-west-2")291    response = client.create_rest_api(name="my_api", description="this is my api")292    api_id = response["id"]293    resources = client.get_resources(restApiId=api_id)294    root_id = [resource for resource in resources["items"] if resource["path"] == "/"][295        0296    ]["id"]297    client.put_method(298        restApiId=api_id,299        resourceId=root_id,300        httpMethod="GET",301        authorizationType="none",302        apiKeyRequired=True,303    )304    response = client.get_method(restApiId=api_id, resourceId=root_id, httpMethod="GET")305    # this is hard to match against, so remove it306    response["ResponseMetadata"].pop("HTTPHeaders", None)307    response["ResponseMetadata"].pop("RetryAttempts", None)308    response.should.equal(309        {310            "httpMethod": "GET",311            "authorizationType": "none",312            "apiKeyRequired": True,313            "ResponseMetadata": {"HTTPStatusCode": 200},314        }315    )316@mock_apigateway317def test_create_method_response():318    client = boto3.client("apigateway", region_name="us-west-2")319    response = client.create_rest_api(name="my_api", description="this is my api")320    api_id = response["id"]321    resources = client.get_resources(restApiId=api_id)322    root_id = [resource for resource in resources["items"] if resource["path"] == "/"][323        0324    ]["id"]325    client.put_method(326        restApiId=api_id, resourceId=root_id, httpMethod="GET", authorizationType="none"327    )328    response = client.get_method(restApiId=api_id, resourceId=root_id, httpMethod="GET")329    response = client.put_method_response(330        restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"331    )332    # this is hard to match against, so remove it333    response["ResponseMetadata"].pop("HTTPHeaders", None)334    response["ResponseMetadata"].pop("RetryAttempts", None)335    response.should.equal(336        {"ResponseMetadata": {"HTTPStatusCode": 200}, "statusCode": "200"}337    )338    response = client.get_method_response(339        restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"340    )341    # this is hard to match against, so remove it342    response["ResponseMetadata"].pop("HTTPHeaders", None)343    response["ResponseMetadata"].pop("RetryAttempts", None)344    response.should.equal(345        {"ResponseMetadata": {"HTTPStatusCode": 200}, "statusCode": "200"}346    )347    response = client.delete_method_response(348        restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"349    )350    # this is hard to match against, so remove it351    response["ResponseMetadata"].pop("HTTPHeaders", None)352    response["ResponseMetadata"].pop("RetryAttempts", None)353    response.should.equal({"ResponseMetadata": {"HTTPStatusCode": 200}})354@mock_apigateway355def test_integrations():356    client = boto3.client("apigateway", region_name="us-west-2")357    response = client.create_rest_api(name="my_api", description="this is my api")358    api_id = response["id"]359    resources = client.get_resources(restApiId=api_id)360    root_id = [resource for resource in resources["items"] if resource["path"] == "/"][361        0362    ]["id"]363    client.put_method(364        restApiId=api_id, resourceId=root_id, httpMethod="GET", authorizationType="none"365    )366    client.put_method_response(367        restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"368    )369    response = client.put_integration(370        restApiId=api_id,371        resourceId=root_id,372        httpMethod="GET",373        type="HTTP",374        uri="http://httpbin.org/robots.txt",375        integrationHttpMethod="POST",376    )377    # this is hard to match against, so remove it378    response["ResponseMetadata"].pop("HTTPHeaders", None)379    response["ResponseMetadata"].pop("RetryAttempts", None)380    response.should.equal(381        {382            "ResponseMetadata": {"HTTPStatusCode": 200},383            "httpMethod": "GET",384            "integrationResponses": {385                "200": {386                    "responseTemplates": {"application/json": None},387                    "statusCode": 200,388                }389            },390            "type": "HTTP",391            "uri": "http://httpbin.org/robots.txt",392        }393    )394    response = client.get_integration(395        restApiId=api_id, resourceId=root_id, httpMethod="GET"396    )397    # this is hard to match against, so remove it398    response["ResponseMetadata"].pop("HTTPHeaders", None)399    response["ResponseMetadata"].pop("RetryAttempts", None)400    response.should.equal(401        {402            "ResponseMetadata": {"HTTPStatusCode": 200},403            "httpMethod": "GET",404            "integrationResponses": {405                "200": {406                    "responseTemplates": {"application/json": None},407                    "statusCode": 200,408                }409            },410            "type": "HTTP",411            "uri": "http://httpbin.org/robots.txt",412        }413    )414    response = client.get_resource(restApiId=api_id, resourceId=root_id)415    # this is hard to match against, so remove it416    response["ResponseMetadata"].pop("HTTPHeaders", None)417    response["ResponseMetadata"].pop("RetryAttempts", None)418    response["resourceMethods"]["GET"]["methodIntegration"].should.equal(419        {420            "httpMethod": "GET",421            "integrationResponses": {422                "200": {423                    "responseTemplates": {"application/json": None},424                    "statusCode": 200,425                }426            },427            "type": "HTTP",428            "uri": "http://httpbin.org/robots.txt",429        }430    )431    client.delete_integration(restApiId=api_id, resourceId=root_id, httpMethod="GET")432    response = client.get_resource(restApiId=api_id, resourceId=root_id)433    response["resourceMethods"]["GET"].shouldnt.contain("methodIntegration")434    # Create a new integration with a requestTemplates config435    client.put_method(436        restApiId=api_id,437        resourceId=root_id,438        httpMethod="POST",439        authorizationType="none",440    )441    templates = {442        # example based on443        # http://docs.aws.amazon.com/apigateway/latest/developerguide/api-as-kinesis-proxy-export-swagger-with-extensions.html444        "application/json": '{\n    "StreamName": "$input.params(\'stream-name\')",\n    "Records": []\n}'445    }446    test_uri = "http://example.com/foobar.txt"447    response = client.put_integration(448        restApiId=api_id,449        resourceId=root_id,450        httpMethod="POST",451        type="HTTP",452        uri=test_uri,453        requestTemplates=templates,454        integrationHttpMethod="POST",455    )456    # this is hard to match against, so remove it457    response["ResponseMetadata"].pop("HTTPHeaders", None)458    response["ResponseMetadata"].pop("RetryAttempts", None)459    response["ResponseMetadata"].should.equal({"HTTPStatusCode": 200})460    response = client.get_integration(461        restApiId=api_id, resourceId=root_id, httpMethod="POST"462    )463    response["uri"].should.equal(test_uri)464    response["requestTemplates"].should.equal(templates)465@mock_apigateway466def test_integration_response():467    client = boto3.client("apigateway", region_name="us-west-2")468    response = client.create_rest_api(name="my_api", description="this is my api")469    api_id = response["id"]470    resources = client.get_resources(restApiId=api_id)471    root_id = [resource for resource in resources["items"] if resource["path"] == "/"][472        0473    ]["id"]474    client.put_method(475        restApiId=api_id, resourceId=root_id, httpMethod="GET", authorizationType="none"476    )477    client.put_method_response(478        restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"479    )480    client.put_integration(481        restApiId=api_id,482        resourceId=root_id,483        httpMethod="GET",484        type="HTTP",485        uri="http://httpbin.org/robots.txt",486        integrationHttpMethod="POST",487    )488    response = client.put_integration_response(489        restApiId=api_id,490        resourceId=root_id,491        httpMethod="GET",492        statusCode="200",493        selectionPattern="foobar",494        responseTemplates={},495    )496    # this is hard to match against, so remove it497    response["ResponseMetadata"].pop("HTTPHeaders", None)498    response["ResponseMetadata"].pop("RetryAttempts", None)499    response.should.equal(500        {501            "statusCode": "200",502            "selectionPattern": "foobar",503            "ResponseMetadata": {"HTTPStatusCode": 200},504            "responseTemplates": {"application/json": None},505        }506    )507    response = client.get_integration_response(508        restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"509    )510    # this is hard to match against, so remove it511    response["ResponseMetadata"].pop("HTTPHeaders", None)512    response["ResponseMetadata"].pop("RetryAttempts", None)513    response.should.equal(514        {515            "statusCode": "200",516            "selectionPattern": "foobar",517            "ResponseMetadata": {"HTTPStatusCode": 200},518            "responseTemplates": {"application/json": None},519        }520    )521    response = client.get_method(restApiId=api_id, resourceId=root_id, httpMethod="GET")522    # this is hard to match against, so remove it523    response["ResponseMetadata"].pop("HTTPHeaders", None)524    response["ResponseMetadata"].pop("RetryAttempts", None)525    response["methodIntegration"]["integrationResponses"].should.equal(526        {527            "200": {528                "responseTemplates": {"application/json": None},529                "selectionPattern": "foobar",530                "statusCode": "200",531            }532        }533    )534    response = client.delete_integration_response(535        restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"536    )537    response = client.get_method(restApiId=api_id, resourceId=root_id, httpMethod="GET")538    response["methodIntegration"]["integrationResponses"].should.equal({})539    # adding a new method and perfomring put intergration with contentHandling as CONVERT_TO_BINARY540    client.put_method(541        restApiId=api_id, resourceId=root_id, httpMethod="PUT", authorizationType="none"542    )543    client.put_method_response(544        restApiId=api_id, resourceId=root_id, httpMethod="PUT", statusCode="200"545    )546    client.put_integration(547        restApiId=api_id,548        resourceId=root_id,549        httpMethod="PUT",550        type="HTTP",551        uri="http://httpbin.org/robots.txt",552        integrationHttpMethod="POST",553    )554    response = client.put_integration_response(555        restApiId=api_id,556        resourceId=root_id,557        httpMethod="PUT",558        statusCode="200",559        selectionPattern="foobar",560        responseTemplates={},561        contentHandling="CONVERT_TO_BINARY",562    )563    # this is hard to match against, so remove it564    response["ResponseMetadata"].pop("HTTPHeaders", None)565    response["ResponseMetadata"].pop("RetryAttempts", None)566    response.should.equal(567        {568            "statusCode": "200",569            "selectionPattern": "foobar",570            "ResponseMetadata": {"HTTPStatusCode": 200},571            "responseTemplates": {"application/json": None},572            "contentHandling": "CONVERT_TO_BINARY",573        }574    )575    response = client.get_integration_response(576        restApiId=api_id, resourceId=root_id, httpMethod="PUT", statusCode="200"577    )578    # this is hard to match against, so remove it579    response["ResponseMetadata"].pop("HTTPHeaders", None)580    response["ResponseMetadata"].pop("RetryAttempts", None)581    response.should.equal(582        {583            "statusCode": "200",584            "selectionPattern": "foobar",585            "ResponseMetadata": {"HTTPStatusCode": 200},586            "responseTemplates": {"application/json": None},587            "contentHandling": "CONVERT_TO_BINARY",588        }589    )590@mock_apigateway591@mock_cognitoidp592def test_update_authorizer_configuration():593    client = boto3.client("apigateway", region_name="us-west-2")594    authorizer_name = "my_authorizer"595    response = client.create_rest_api(name="my_api", description="this is my api")596    api_id = response["id"]597    cognito_client = boto3.client("cognito-idp", region_name="us-west-2")598    user_pool_arn = cognito_client.create_user_pool(PoolName="my_cognito_pool")[599        "UserPool"600    ]["Arn"]601    response = client.create_authorizer(602        restApiId=api_id,603        name=authorizer_name,604        type="COGNITO_USER_POOLS",605        providerARNs=[user_pool_arn],606        identitySource="method.request.header.Authorization",607    )608    authorizer_id = response["id"]609    response = client.get_authorizer(restApiId=api_id, authorizerId=authorizer_id)610    # createdDate is hard to match against, remove it611    response.pop("createdDate", None)612    # this is hard to match against, so remove it613    response["ResponseMetadata"].pop("HTTPHeaders", None)614    response["ResponseMetadata"].pop("RetryAttempts", None)615    response.should.equal(616        {617            "id": authorizer_id,618            "name": authorizer_name,619            "type": "COGNITO_USER_POOLS",620            "providerARNs": [user_pool_arn],621            "identitySource": "method.request.header.Authorization",622            "authorizerResultTtlInSeconds": 300,623            "ResponseMetadata": {"HTTPStatusCode": 200},624        }625    )626    client.update_authorizer(627        restApiId=api_id,628        authorizerId=authorizer_id,629        patchOperations=[{"op": "replace", "path": "/type", "value": "TOKEN"}],630    )631    authorizer = client.get_authorizer(restApiId=api_id, authorizerId=authorizer_id)632    authorizer.should.have.key("type").which.should.equal("TOKEN")633    client.update_authorizer(634        restApiId=api_id,635        authorizerId=authorizer_id,636        patchOperations=[{"op": "replace", "path": "/type", "value": "REQUEST"}],637    )638    authorizer = client.get_authorizer(restApiId=api_id, authorizerId=authorizer_id)639    authorizer.should.have.key("type").which.should.equal("REQUEST")640    # TODO: implement mult-update tests641    try:642        client.update_authorizer(643            restApiId=api_id,644            authorizerId=authorizer_id,645            patchOperations=[646                {"op": "add", "path": "/notasetting", "value": "eu-west-1"}647            ],648        )649        assert False.should.be.ok  # Fail, should not be here650    except Exception:651        assert True.should.be.ok652@mock_apigateway653def test_non_existent_authorizer():654    client = boto3.client("apigateway", region_name="us-west-2")655    response = client.create_rest_api(name="my_api", description="this is my api")656    api_id = response["id"]657    client.get_authorizer.when.called_with(658        restApiId=api_id, authorizerId="xxx"659    ).should.throw(ClientError)660@mock_apigateway661@mock_cognitoidp662def test_create_authorizer():663    client = boto3.client("apigateway", region_name="us-west-2")664    authorizer_name = "my_authorizer"665    response = client.create_rest_api(name="my_api", description="this is my api")666    api_id = response["id"]667    cognito_client = boto3.client("cognito-idp", region_name="us-west-2")668    user_pool_arn = cognito_client.create_user_pool(PoolName="my_cognito_pool")[669        "UserPool"670    ]["Arn"]671    response = client.create_authorizer(672        restApiId=api_id,673        name=authorizer_name,674        type="COGNITO_USER_POOLS",675        providerARNs=[user_pool_arn],676        identitySource="method.request.header.Authorization",677    )678    authorizer_id = response["id"]679    response = client.get_authorizer(restApiId=api_id, authorizerId=authorizer_id)680    # createdDate is hard to match against, remove it681    response.pop("createdDate", None)682    # this is hard to match against, so remove it683    response["ResponseMetadata"].pop("HTTPHeaders", None)684    response["ResponseMetadata"].pop("RetryAttempts", None)685    response.should.equal(686        {687            "id": authorizer_id,688            "name": authorizer_name,689            "type": "COGNITO_USER_POOLS",690            "providerARNs": [user_pool_arn],691            "identitySource": "method.request.header.Authorization",692            "authorizerResultTtlInSeconds": 300,693            "ResponseMetadata": {"HTTPStatusCode": 200},694        }695    )696    authorizer_name2 = "my_authorizer2"697    response = client.create_authorizer(698        restApiId=api_id,699        name=authorizer_name2,700        type="COGNITO_USER_POOLS",701        providerARNs=[user_pool_arn],702        identitySource="method.request.header.Authorization",703    )704    authorizer_id2 = response["id"]705    response = client.get_authorizers(restApiId=api_id)706    # this is hard to match against, so remove it707    response["ResponseMetadata"].pop("HTTPHeaders", None)708    response["ResponseMetadata"].pop("RetryAttempts", None)709    response["items"][0]["id"].should.match(710        r"{0}|{1}".format(authorizer_id2, authorizer_id)711    )712    response["items"][1]["id"].should.match(713        r"{0}|{1}".format(authorizer_id2, authorizer_id)714    )715    new_authorizer_name_with_vars = "authorizer_with_vars"716    response = client.create_authorizer(717        restApiId=api_id,718        name=new_authorizer_name_with_vars,719        type="COGNITO_USER_POOLS",720        providerARNs=[user_pool_arn],721        identitySource="method.request.header.Authorization",722    )723    authorizer_id3 = response["id"]724    # this is hard to match against, so remove it725    response["ResponseMetadata"].pop("HTTPHeaders", None)726    response["ResponseMetadata"].pop("RetryAttempts", None)727    response.should.equal(728        {729            "name": new_authorizer_name_with_vars,730            "id": authorizer_id3,731            "type": "COGNITO_USER_POOLS",732            "providerARNs": [user_pool_arn],733            "identitySource": "method.request.header.Authorization",734            "authorizerResultTtlInSeconds": 300,735            "ResponseMetadata": {"HTTPStatusCode": 200},736        }737    )738    stage = client.get_authorizer(restApiId=api_id, authorizerId=authorizer_id3)739    stage["name"].should.equal(new_authorizer_name_with_vars)740    stage["id"].should.equal(authorizer_id3)741    stage["type"].should.equal("COGNITO_USER_POOLS")742    stage["providerARNs"].should.equal([user_pool_arn])743    stage["identitySource"].should.equal("method.request.header.Authorization")744    stage["authorizerResultTtlInSeconds"].should.equal(300)745@mock_apigateway746@mock_cognitoidp747def test_delete_authorizer():748    client = boto3.client("apigateway", region_name="us-west-2")749    authorizer_name = "my_authorizer"750    response = client.create_rest_api(name="my_api", description="this is my api")751    api_id = response["id"]752    cognito_client = boto3.client("cognito-idp", region_name="us-west-2")753    user_pool_arn = cognito_client.create_user_pool(PoolName="my_cognito_pool")[754        "UserPool"755    ]["Arn"]756    response = client.create_authorizer(757        restApiId=api_id,758        name=authorizer_name,759        type="COGNITO_USER_POOLS",760        providerARNs=[user_pool_arn],761        identitySource="method.request.header.Authorization",762    )763    authorizer_id = response["id"]764    response = client.get_authorizer(restApiId=api_id, authorizerId=authorizer_id)765    # createdDate is hard to match against, remove it766    response.pop("createdDate", None)767    # this is hard to match against, so remove it768    response["ResponseMetadata"].pop("HTTPHeaders", None)769    response["ResponseMetadata"].pop("RetryAttempts", None)770    response.should.equal(771        {772            "id": authorizer_id,773            "name": authorizer_name,774            "type": "COGNITO_USER_POOLS",775            "providerARNs": [user_pool_arn],776            "identitySource": "method.request.header.Authorization",777            "authorizerResultTtlInSeconds": 300,778            "ResponseMetadata": {"HTTPStatusCode": 200},779        }780    )781    authorizer_name2 = "my_authorizer2"782    response = client.create_authorizer(783        restApiId=api_id,784        name=authorizer_name2,785        type="COGNITO_USER_POOLS",786        providerARNs=[user_pool_arn],787        identitySource="method.request.header.Authorization",788    )789    authorizer_id2 = response["id"]790    authorizers = client.get_authorizers(restApiId=api_id)["items"]791    sorted([authorizer["name"] for authorizer in authorizers]).should.equal(792        sorted([authorizer_name2, authorizer_name])793    )794    # delete stage795    response = client.delete_authorizer(restApiId=api_id, authorizerId=authorizer_id2)796    response["ResponseMetadata"]["HTTPStatusCode"].should.equal(202)797    # verify other stage still exists798    authorizers = client.get_authorizers(restApiId=api_id)["items"]799    sorted([authorizer["name"] for authorizer in authorizers]).should.equal(800        sorted([authorizer_name])801    )802@mock_apigateway803def test_update_stage_configuration():804    client = boto3.client("apigateway", region_name="us-west-2")805    stage_name = "staging"806    response = client.create_rest_api(name="my_api", description="this is my api")807    api_id = response["id"]808    create_method_integration(client, api_id)809    response = client.create_deployment(810        restApiId=api_id, stageName=stage_name, description="1.0.1"811    )812    deployment_id = response["id"]813    response = client.get_deployment(restApiId=api_id, deploymentId=deployment_id)814    # createdDate is hard to match against, remove it815    response.pop("createdDate", None)816    # this is hard to match against, so remove it817    response["ResponseMetadata"].pop("HTTPHeaders", None)818    response["ResponseMetadata"].pop("RetryAttempts", None)819    response.should.equal(820        {821            "id": deployment_id,822            "ResponseMetadata": {"HTTPStatusCode": 200},823            "description": "1.0.1",824        }825    )826    response = client.create_deployment(827        restApiId=api_id, stageName=stage_name, description="1.0.2"828    )829    deployment_id2 = response["id"]830    stage = client.get_stage(restApiId=api_id, stageName=stage_name)831    stage["stageName"].should.equal(stage_name)832    stage["deploymentId"].should.equal(deployment_id2)833    stage.shouldnt.have.key("cacheClusterSize")834    client.update_stage(835        restApiId=api_id,836        stageName=stage_name,837        patchOperations=[838            {"op": "replace", "path": "/cacheClusterEnabled", "value": "True"}839        ],840    )841    stage = client.get_stage(restApiId=api_id, stageName=stage_name)842    stage.should.have.key("cacheClusterSize").which.should.equal("0.5")843    client.update_stage(844        restApiId=api_id,845        stageName=stage_name,846        patchOperations=[847            {"op": "replace", "path": "/cacheClusterSize", "value": "1.6"}848        ],849    )850    stage = client.get_stage(restApiId=api_id, stageName=stage_name)851    stage.should.have.key("cacheClusterSize").which.should.equal("1.6")852    client.update_stage(853        restApiId=api_id,854        stageName=stage_name,855        patchOperations=[856            {"op": "replace", "path": "/deploymentId", "value": deployment_id},857            {"op": "replace", "path": "/variables/environment", "value": "dev"},858            {"op": "replace", "path": "/variables/region", "value": "eu-west-1"},859            {"op": "replace", "path": "/*/*/caching/dataEncrypted", "value": "True"},860            {"op": "replace", "path": "/cacheClusterEnabled", "value": "True"},861            {862                "op": "replace",863                "path": "/description",864                "value": "stage description update",865            },866            {"op": "replace", "path": "/cacheClusterSize", "value": "1.6"},867        ],868    )869    client.update_stage(870        restApiId=api_id,871        stageName=stage_name,872        patchOperations=[873            {"op": "remove", "path": "/variables/region", "value": "eu-west-1"}874        ],875    )876    stage = client.get_stage(restApiId=api_id, stageName=stage_name)877    stage["description"].should.match("stage description update")878    stage["cacheClusterSize"].should.equal("1.6")879    stage["variables"]["environment"].should.match("dev")880    stage["variables"].should_not.have.key("region")881    stage["cacheClusterEnabled"].should.be.true882    stage["deploymentId"].should.match(deployment_id)883    stage["methodSettings"].should.have.key("*/*")884    stage["methodSettings"]["*/*"].should.have.key(885        "cacheDataEncrypted"886    ).which.should.be.true887    try:888        client.update_stage(889            restApiId=api_id,890            stageName=stage_name,891            patchOperations=[892                {"op": "add", "path": "/notasetting", "value": "eu-west-1"}893            ],894        )895        assert False.should.be.ok  # Fail, should not be here896    except Exception:897        assert True.should.be.ok898@mock_apigateway899def test_non_existent_stage():900    client = boto3.client("apigateway", region_name="us-west-2")901    response = client.create_rest_api(name="my_api", description="this is my api")902    api_id = response["id"]903    client.get_stage.when.called_with(restApiId=api_id, stageName="xxx").should.throw(904        ClientError905    )906@mock_apigateway907def test_create_stage():908    client = boto3.client("apigateway", region_name="us-west-2")909    stage_name = "staging"910    response = client.create_rest_api(name="my_api", description="this is my api")911    api_id = response["id"]912    create_method_integration(client, api_id)913    response = client.create_deployment(restApiId=api_id, stageName=stage_name)914    deployment_id = response["id"]915    response = client.get_deployment(restApiId=api_id, deploymentId=deployment_id)916    # createdDate is hard to match against, remove it917    response.pop("createdDate", None)918    # this is hard to match against, so remove it919    response["ResponseMetadata"].pop("HTTPHeaders", None)920    response["ResponseMetadata"].pop("RetryAttempts", None)921    response.should.equal(922        {923            "id": deployment_id,924            "ResponseMetadata": {"HTTPStatusCode": 200},925            "description": "",926        }927    )928    response = client.create_deployment(restApiId=api_id, stageName=stage_name)929    deployment_id2 = response["id"]930    response = client.get_deployments(restApiId=api_id)931    # this is hard to match against, so remove it932    response["ResponseMetadata"].pop("HTTPHeaders", None)933    response["ResponseMetadata"].pop("RetryAttempts", None)934    response["items"][0].pop("createdDate")935    response["items"][1].pop("createdDate")936    response["items"][0]["id"].should.match(937        r"{0}|{1}".format(deployment_id2, deployment_id)938    )939    response["items"][1]["id"].should.match(940        r"{0}|{1}".format(deployment_id2, deployment_id)941    )942    new_stage_name = "current"943    response = client.create_stage(944        restApiId=api_id, stageName=new_stage_name, deploymentId=deployment_id2945    )946    # this is hard to match against, so remove it947    response["ResponseMetadata"].pop("HTTPHeaders", None)948    response["ResponseMetadata"].pop("RetryAttempts", None)949    response.should.equal(950        {951            "stageName": new_stage_name,952            "deploymentId": deployment_id2,953            "methodSettings": {},954            "variables": {},955            "ResponseMetadata": {"HTTPStatusCode": 200},956            "description": "",957            "cacheClusterEnabled": False,958        }959    )960    stage = client.get_stage(restApiId=api_id, stageName=new_stage_name)961    stage["stageName"].should.equal(new_stage_name)962    stage["deploymentId"].should.equal(deployment_id2)963    new_stage_name_with_vars = "stage_with_vars"964    response = client.create_stage(965        restApiId=api_id,966        stageName=new_stage_name_with_vars,967        deploymentId=deployment_id2,968        variables={"env": "dev"},969    )970    # this is hard to match against, so remove it971    response["ResponseMetadata"].pop("HTTPHeaders", None)972    response["ResponseMetadata"].pop("RetryAttempts", None)973    response.should.equal(974        {975            "stageName": new_stage_name_with_vars,976            "deploymentId": deployment_id2,977            "methodSettings": {},978            "variables": {"env": "dev"},979            "ResponseMetadata": {"HTTPStatusCode": 200},980            "description": "",981            "cacheClusterEnabled": False,982        }983    )984    stage = client.get_stage(restApiId=api_id, stageName=new_stage_name_with_vars)985    stage["stageName"].should.equal(new_stage_name_with_vars)986    stage["deploymentId"].should.equal(deployment_id2)987    stage["variables"].should.have.key("env").which.should.match("dev")988    new_stage_name = "stage_with_vars_and_cache_settings"989    response = client.create_stage(990        restApiId=api_id,991        stageName=new_stage_name,992        deploymentId=deployment_id2,993        variables={"env": "dev"},994        cacheClusterEnabled=True,995        description="hello moto",996    )997    # this is hard to match against, so remove it998    response["ResponseMetadata"].pop("HTTPHeaders", None)999    response["ResponseMetadata"].pop("RetryAttempts", None)1000    response.should.equal(1001        {1002            "stageName": new_stage_name,1003            "deploymentId": deployment_id2,1004            "methodSettings": {},1005            "variables": {"env": "dev"},1006            "ResponseMetadata": {"HTTPStatusCode": 200},1007            "description": "hello moto",1008            "cacheClusterEnabled": True,1009            "cacheClusterSize": "0.5",1010        }1011    )1012    stage = client.get_stage(restApiId=api_id, stageName=new_stage_name)1013    stage["cacheClusterSize"].should.equal("0.5")1014    new_stage_name = "stage_with_vars_and_cache_settings_and_size"1015    response = client.create_stage(1016        restApiId=api_id,1017        stageName=new_stage_name,1018        deploymentId=deployment_id2,1019        variables={"env": "dev"},1020        cacheClusterEnabled=True,1021        cacheClusterSize="1.6",1022        description="hello moto",1023    )1024    # this is hard to match against, so remove it1025    response["ResponseMetadata"].pop("HTTPHeaders", None)1026    response["ResponseMetadata"].pop("RetryAttempts", None)1027    response.should.equal(1028        {1029            "stageName": new_stage_name,1030            "deploymentId": deployment_id2,1031            "methodSettings": {},1032            "variables": {"env": "dev"},1033            "ResponseMetadata": {"HTTPStatusCode": 200},1034            "description": "hello moto",1035            "cacheClusterEnabled": True,1036            "cacheClusterSize": "1.6",1037        }1038    )1039    stage = client.get_stage(restApiId=api_id, stageName=new_stage_name)1040    stage["stageName"].should.equal(new_stage_name)1041    stage["deploymentId"].should.equal(deployment_id2)1042    stage["variables"].should.have.key("env").which.should.match("dev")1043    stage["cacheClusterSize"].should.equal("1.6")1044@mock_apigateway1045def test_create_deployment_requires_REST_methods():1046    client = boto3.client("apigateway", region_name="us-west-2")1047    stage_name = "staging"1048    response = client.create_rest_api(name="my_api", description="this is my api")1049    api_id = response["id"]1050    with pytest.raises(ClientError) as ex:1051        client.create_deployment(restApiId=api_id, stageName=stage_name)["id"]1052    ex.value.response["Error"]["Code"].should.equal("BadRequestException")1053    ex.value.response["Error"]["Message"].should.equal(1054        "The REST API doesn't contain any methods"1055    )1056@mock_apigateway1057def test_create_deployment_requires_REST_method_integrations():1058    client = boto3.client("apigateway", region_name="us-west-2")1059    stage_name = "staging"1060    response = client.create_rest_api(name="my_api", description="this is my api")1061    api_id = response["id"]1062    resources = client.get_resources(restApiId=api_id)1063    root_id = [resource for resource in resources["items"] if resource["path"] == "/"][1064        01065    ]["id"]1066    client.put_method(1067        restApiId=api_id, resourceId=root_id, httpMethod="GET", authorizationType="NONE"1068    )1069    with pytest.raises(ClientError) as ex:1070        client.create_deployment(restApiId=api_id, stageName=stage_name)["id"]1071    ex.value.response["Error"]["Code"].should.equal("BadRequestException")1072    ex.value.response["Error"]["Message"].should.equal(1073        "No integration defined for method"1074    )1075@mock_apigateway1076def test_create_simple_deployment_with_get_method():1077    client = boto3.client("apigateway", region_name="us-west-2")1078    stage_name = "staging"1079    response = client.create_rest_api(name="my_api", description="this is my api")1080    api_id = response["id"]1081    create_method_integration(client, api_id)1082    deployment = client.create_deployment(restApiId=api_id, stageName=stage_name)1083    assert "id" in deployment1084@mock_apigateway1085def test_create_simple_deployment_with_post_method():1086    client = boto3.client("apigateway", region_name="us-west-2")1087    stage_name = "staging"1088    response = client.create_rest_api(name="my_api", description="this is my api")1089    api_id = response["id"]1090    create_method_integration(client, api_id, httpMethod="POST")1091    deployment = client.create_deployment(restApiId=api_id, stageName=stage_name)1092    assert "id" in deployment1093@mock_apigateway1094# https://github.com/aws/aws-sdk-js/issues/25881095def test_put_integration_response_requires_responseTemplate():1096    client = boto3.client("apigateway", region_name="us-west-2")1097    response = client.create_rest_api(name="my_api", description="this is my api")1098    api_id = response["id"]1099    resources = client.get_resources(restApiId=api_id)1100    root_id = [resource for resource in resources["items"] if resource["path"] == "/"][1101        01102    ]["id"]1103    client.put_method(1104        restApiId=api_id, resourceId=root_id, httpMethod="GET", authorizationType="NONE"1105    )1106    client.put_method_response(1107        restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"1108    )1109    client.put_integration(1110        restApiId=api_id,1111        resourceId=root_id,1112        httpMethod="GET",1113        type="HTTP",1114        uri="http://httpbin.org/robots.txt",1115        integrationHttpMethod="POST",1116    )1117    with pytest.raises(ClientError) as ex:1118        client.put_integration_response(1119            restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"1120        )1121    ex.value.response["Error"]["Code"].should.equal("BadRequestException")1122    ex.value.response["Error"]["Message"].should.equal("Invalid request input")1123    # Works fine if responseTemplate is defined1124    client.put_integration_response(1125        restApiId=api_id,1126        resourceId=root_id,1127        httpMethod="GET",1128        statusCode="200",1129        responseTemplates={},1130    )1131@mock_apigateway1132def test_put_integration_response_with_response_template():1133    client = boto3.client("apigateway", region_name="us-west-2")1134    response = client.create_rest_api(name="my_api", description="this is my api")1135    api_id = response["id"]1136    resources = client.get_resources(restApiId=api_id)1137    root_id = [resource for resource in resources["items"] if resource["path"] == "/"][1138        01139    ]["id"]1140    client.put_method(1141        restApiId=api_id, resourceId=root_id, httpMethod="GET", authorizationType="NONE"1142    )1143    client.put_method_response(1144        restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"1145    )1146    client.put_integration(1147        restApiId=api_id,1148        resourceId=root_id,1149        httpMethod="GET",1150        type="HTTP",1151        uri="http://httpbin.org/robots.txt",1152        integrationHttpMethod="POST",1153    )1154    with pytest.raises(ClientError) as ex:1155        client.put_integration_response(1156            restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"1157        )1158    ex.value.response["Error"]["Code"].should.equal("BadRequestException")1159    ex.value.response["Error"]["Message"].should.equal("Invalid request input")1160    client.put_integration_response(1161        restApiId=api_id,1162        resourceId=root_id,1163        httpMethod="GET",1164        statusCode="200",1165        selectionPattern="foobar",1166        responseTemplates={"application/json": json.dumps({"data": "test"})},1167    )1168    response = client.get_integration_response(1169        restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"1170    )1171    # this is hard to match against, so remove it1172    response["ResponseMetadata"].pop("HTTPHeaders", None)1173    response["ResponseMetadata"].pop("RetryAttempts", None)1174    response.should.equal(1175        {1176            "statusCode": "200",1177            "selectionPattern": "foobar",1178            "ResponseMetadata": {"HTTPStatusCode": 200},1179            "responseTemplates": {"application/json": json.dumps({"data": "test"})},1180        }1181    )1182@mock_apigateway1183def test_put_integration_validation():1184    client = boto3.client("apigateway", region_name="us-west-2")1185    response = client.create_rest_api(name="my_api", description="this is my api")1186    api_id = response["id"]1187    resources = client.get_resources(restApiId=api_id)1188    root_id = [resource for resource in resources["items"] if resource["path"] == "/"][1189        01190    ]["id"]1191    client.put_method(1192        restApiId=api_id, resourceId=root_id, httpMethod="GET", authorizationType="NONE"1193    )1194    client.put_method_response(1195        restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"1196    )1197    http_types = ["HTTP", "HTTP_PROXY"]1198    aws_types = ["AWS", "AWS_PROXY"]1199    types_requiring_integration_method = http_types + aws_types1200    types_not_requiring_integration_method = ["MOCK"]1201    for type in types_requiring_integration_method:1202        # Ensure that integrations of these types fail if no integrationHttpMethod is provided1203        with pytest.raises(ClientError) as ex:1204            client.put_integration(1205                restApiId=api_id,1206                resourceId=root_id,1207                httpMethod="GET",1208                type=type,1209                uri="http://httpbin.org/robots.txt",1210            )1211        ex.value.response["Error"]["Code"].should.equal("BadRequestException")1212        ex.value.response["Error"]["Message"].should.equal(1213            "Enumeration value for HttpMethod must be non-empty"1214        )1215    for type in types_not_requiring_integration_method:1216        # Ensure that integrations of these types do not need the integrationHttpMethod1217        client.put_integration(1218            restApiId=api_id,1219            resourceId=root_id,1220            httpMethod="GET",1221            type=type,1222            uri="http://httpbin.org/robots.txt",1223        )1224    for type in http_types:1225        # Ensure that it works fine when providing the integrationHttpMethod-argument1226        client.put_integration(1227            restApiId=api_id,1228            resourceId=root_id,1229            httpMethod="GET",1230            type=type,1231            uri="http://httpbin.org/robots.txt",1232            integrationHttpMethod="POST",1233        )1234    for type in ["AWS"]:1235        # Ensure that it works fine when providing the integrationHttpMethod + credentials1236        client.put_integration(1237            restApiId=api_id,1238            resourceId=root_id,1239            credentials="arn:aws:iam::{}:role/service-role/testfunction-role-oe783psq".format(1240                ACCOUNT_ID1241            ),1242            httpMethod="GET",1243            type=type,1244            uri="arn:aws:apigateway:us-west-2:s3:path/b/k",1245            integrationHttpMethod="POST",1246        )1247    for type in aws_types:1248        # Ensure that credentials are not required when URI points to a Lambda stream1249        client.put_integration(1250            restApiId=api_id,1251            resourceId=root_id,1252            httpMethod="GET",1253            type=type,1254            uri="arn:aws:apigateway:eu-west-1:lambda:path/2015-03-31/functions/arn:aws:lambda:eu-west-1:012345678901:function:MyLambda/invocations",1255            integrationHttpMethod="POST",1256        )1257    for type in ["AWS_PROXY"]:1258        # Ensure that aws_proxy does not support S31259        with pytest.raises(ClientError) as ex:1260            client.put_integration(1261                restApiId=api_id,1262                resourceId=root_id,1263                credentials="arn:aws:iam::{}:role/service-role/testfunction-role-oe783psq".format(1264                    ACCOUNT_ID1265                ),1266                httpMethod="GET",1267                type=type,1268                uri="arn:aws:apigateway:us-west-2:s3:path/b/k",1269                integrationHttpMethod="POST",1270            )1271        ex.value.response["Error"]["Code"].should.equal("BadRequestException")1272        ex.value.response["Error"]["Message"].should.equal(1273            "Integrations of type 'AWS_PROXY' currently only supports Lambda function and Firehose stream invocations."1274        )1275    for type in aws_types:1276        # Ensure that the Role ARN is for the current account1277        with pytest.raises(ClientError) as ex:1278            client.put_integration(1279                restApiId=api_id,1280                resourceId=root_id,1281                credentials="arn:aws:iam::000000000000:role/service-role/testrole",1282                httpMethod="GET",1283                type=type,1284                uri="arn:aws:apigateway:us-west-2:s3:path/b/k",1285                integrationHttpMethod="POST",1286            )1287        ex.value.response["Error"]["Code"].should.equal("AccessDeniedException")1288        ex.value.response["Error"]["Message"].should.equal(1289            "Cross-account pass role is not allowed."1290        )1291    for type in ["AWS"]:1292        # Ensure that the Role ARN is specified for aws integrations1293        with pytest.raises(ClientError) as ex:1294            client.put_integration(1295                restApiId=api_id,1296                resourceId=root_id,1297                httpMethod="GET",1298                type=type,1299                uri="arn:aws:apigateway:us-west-2:s3:path/b/k",1300                integrationHttpMethod="POST",1301            )1302        ex.value.response["Error"]["Code"].should.equal("BadRequestException")1303        ex.value.response["Error"]["Message"].should.equal(1304            "Role ARN must be specified for AWS integrations"1305        )1306    for type in http_types:1307        # Ensure that the URI is valid HTTP1308        with pytest.raises(ClientError) as ex:1309            client.put_integration(1310                restApiId=api_id,1311                resourceId=root_id,1312                httpMethod="GET",1313                type=type,1314                uri="non-valid-http",1315                integrationHttpMethod="POST",1316            )1317        ex.value.response["Error"]["Code"].should.equal("BadRequestException")1318        ex.value.response["Error"]["Message"].should.equal(1319            "Invalid HTTP endpoint specified for URI"1320        )1321    for type in aws_types:1322        # Ensure that the URI is an ARN1323        with pytest.raises(ClientError) as ex:1324            client.put_integration(1325                restApiId=api_id,1326                resourceId=root_id,1327                httpMethod="GET",1328                type=type,1329                uri="non-valid-arn",1330                integrationHttpMethod="POST",1331            )1332        ex.value.response["Error"]["Code"].should.equal("BadRequestException")1333        ex.value.response["Error"]["Message"].should.equal(1334            "Invalid ARN specified in the request"1335        )1336    for type in aws_types:1337        # Ensure that the URI is a valid ARN1338        with pytest.raises(ClientError) as ex:1339            client.put_integration(1340                restApiId=api_id,1341                resourceId=root_id,1342                httpMethod="GET",1343                type=type,1344                uri="arn:aws:iam::0000000000:role/service-role/asdf",1345                integrationHttpMethod="POST",1346            )1347        ex.value.response["Error"]["Code"].should.equal("BadRequestException")1348        ex.value.response["Error"]["Message"].should.equal(1349            "AWS ARN for integration must contain path or action"1350        )1351@mock_apigateway1352def test_delete_stage():1353    client = boto3.client("apigateway", region_name="us-west-2")1354    stage_name = "staging"1355    response = client.create_rest_api(name="my_api", description="this is my api")1356    api_id = response["id"]1357    create_method_integration(client, api_id)1358    deployment_id1 = client.create_deployment(restApiId=api_id, stageName=stage_name)[1359        "id"1360    ]1361    deployment_id2 = client.create_deployment(restApiId=api_id, stageName=stage_name)[1362        "id"1363    ]1364    new_stage_name = "current"1365    client.create_stage(1366        restApiId=api_id, stageName=new_stage_name, deploymentId=deployment_id11367    )1368    new_stage_name_with_vars = "stage_with_vars"1369    client.create_stage(1370        restApiId=api_id,1371        stageName=new_stage_name_with_vars,1372        deploymentId=deployment_id2,1373        variables={"env": "dev"},1374    )1375    stages = client.get_stages(restApiId=api_id)["item"]1376    sorted([stage["stageName"] for stage in stages]).should.equal(1377        sorted([new_stage_name, new_stage_name_with_vars, stage_name])1378    )1379    # delete stage1380    response = client.delete_stage(restApiId=api_id, stageName=new_stage_name_with_vars)1381    response["ResponseMetadata"]["HTTPStatusCode"].should.equal(202)1382    # verify other stage still exists1383    stages = client.get_stages(restApiId=api_id)["item"]1384    sorted([stage["stageName"] for stage in stages]).should.equal(1385        sorted([new_stage_name, stage_name])1386    )1387@mock_apigateway1388def test_deployment():1389    client = boto3.client("apigateway", region_name="us-west-2")1390    stage_name = "staging"1391    response = client.create_rest_api(name="my_api", description="this is my api")1392    api_id = response["id"]1393    create_method_integration(client, api_id)1394    response = client.create_deployment(restApiId=api_id, stageName=stage_name)1395    deployment_id = response["id"]1396    response = client.get_deployment(restApiId=api_id, deploymentId=deployment_id)1397    # createdDate is hard to match against, remove it1398    response.pop("createdDate", None)1399    # this is hard to match against, so remove it1400    response["ResponseMetadata"].pop("HTTPHeaders", None)1401    response["ResponseMetadata"].pop("RetryAttempts", None)1402    response.should.equal(1403        {1404            "id": deployment_id,1405            "ResponseMetadata": {"HTTPStatusCode": 200},1406            "description": "",1407        }1408    )1409    response = client.get_deployments(restApiId=api_id)1410    response["items"][0].pop("createdDate")1411    response["items"].should.equal([{"id": deployment_id, "description": ""}])1412    client.delete_deployment(restApiId=api_id, deploymentId=deployment_id)1413    response = client.get_deployments(restApiId=api_id)1414    len(response["items"]).should.equal(0)1415    # test deployment stages1416    stage = client.get_stage(restApiId=api_id, stageName=stage_name)1417    stage["stageName"].should.equal(stage_name)1418    stage["deploymentId"].should.equal(deployment_id)1419    client.update_stage(1420        restApiId=api_id,1421        stageName=stage_name,1422        patchOperations=[1423            {"op": "replace", "path": "/description", "value": "_new_description_"}1424        ],1425    )1426    stage = client.get_stage(restApiId=api_id, stageName=stage_name)1427    stage["stageName"].should.equal(stage_name)1428    stage["deploymentId"].should.equal(deployment_id)1429    stage["description"].should.equal("_new_description_")1430@mock_apigateway1431def test_create_domain_names():1432    client = boto3.client("apigateway", region_name="us-west-2")1433    domain_name = "testDomain"1434    test_certificate_name = "test.certificate"1435    test_certificate_private_key = "testPrivateKey"1436    # success case with valid params1437    response = client.create_domain_name(1438        domainName=domain_name,1439        certificateName=test_certificate_name,1440        certificatePrivateKey=test_certificate_private_key,1441    )1442    response["domainName"].should.equal(domain_name)1443    response["certificateName"].should.equal(test_certificate_name)1444    # without domain name it should throw BadRequestException1445    with pytest.raises(ClientError) as ex:1446        client.create_domain_name(domainName="")1447    ex.value.response["Error"]["Message"].should.equal("No Domain Name specified")1448    ex.value.response["Error"]["Code"].should.equal("BadRequestException")1449@mock_apigateway1450def test_get_domain_names():1451    client = boto3.client("apigateway", region_name="us-west-2")1452    # without any domain names already present1453    result = client.get_domain_names()1454    result["items"].should.equal([])1455    domain_name = "testDomain"1456    test_certificate_name = "test.certificate"1457    response = client.create_domain_name(1458        domainName=domain_name, certificateName=test_certificate_name1459    )1460    response["domainName"].should.equal(domain_name)1461    response["certificateName"].should.equal(test_certificate_name)1462    response["domainNameStatus"].should.equal("AVAILABLE")1463    # after adding a new domain name1464    result = client.get_domain_names()1465    result["items"][0]["domainName"].should.equal(domain_name)1466    result["items"][0]["certificateName"].should.equal(test_certificate_name)1467    result["items"][0]["domainNameStatus"].should.equal("AVAILABLE")1468@mock_apigateway1469def test_get_domain_name():1470    client = boto3.client("apigateway", region_name="us-west-2")1471    domain_name = "testDomain"1472    # quering an invalid domain name which is not present1473    with pytest.raises(ClientError) as ex:1474        client.get_domain_name(domainName=domain_name)1475    ex.value.response["Error"]["Message"].should.equal("Invalid Domain Name specified")1476    ex.value.response["Error"]["Code"].should.equal("NotFoundException")1477    # adding a domain name1478    client.create_domain_name(domainName=domain_name)1479    # retrieving the data of added domain name.1480    result = client.get_domain_name(domainName=domain_name)1481    result["domainName"].should.equal(domain_name)1482    result["domainNameStatus"].should.equal("AVAILABLE")1483@mock_apigateway1484def test_create_model():1485    client = boto3.client("apigateway", region_name="us-west-2")1486    response = client.create_rest_api(name="my_api", description="this is my api")1487    rest_api_id = response["id"]1488    dummy_rest_api_id = "a12b3c4d"1489    model_name = "testModel"1490    description = "test model"1491    content_type = "application/json"1492    # success case with valid params1493    response = client.create_model(1494        restApiId=rest_api_id,1495        name=model_name,1496        description=description,1497        contentType=content_type,1498    )1499    response["name"].should.equal(model_name)1500    response["description"].should.equal(description)1501    # with an invalid rest_api_id it should throw NotFoundException1502    with pytest.raises(ClientError) as ex:1503        client.create_model(1504            restApiId=dummy_rest_api_id,1505            name=model_name,1506            description=description,1507            contentType=content_type,1508        )1509    ex.value.response["Error"]["Message"].should.equal("Invalid Rest API Id specified")1510    ex.value.response["Error"]["Code"].should.equal("NotFoundException")1511    with pytest.raises(ClientError) as ex:1512        client.create_model(1513            restApiId=rest_api_id,1514            name="",1515            description=description,1516            contentType=content_type,1517        )1518    ex.value.response["Error"]["Message"].should.equal("No Model Name specified")1519    ex.value.response["Error"]["Code"].should.equal("BadRequestException")1520@mock_apigateway1521def test_get_api_models():1522    client = boto3.client("apigateway", region_name="us-west-2")1523    response = client.create_rest_api(name="my_api", description="this is my api")1524    rest_api_id = response["id"]1525    model_name = "testModel"1526    description = "test model"1527    content_type = "application/json"1528    # when no models are present1529    result = client.get_models(restApiId=rest_api_id)1530    result["items"].should.equal([])1531    # add a model1532    client.create_model(1533        restApiId=rest_api_id,1534        name=model_name,1535        description=description,1536        contentType=content_type,1537    )1538    # get models after adding1539    result = client.get_models(restApiId=rest_api_id)1540    result["items"][0]["name"] = model_name1541    result["items"][0]["description"] = description1542@mock_apigateway1543def test_get_model_by_name():1544    client = boto3.client("apigateway", region_name="us-west-2")1545    response = client.create_rest_api(name="my_api", description="this is my api")1546    rest_api_id = response["id"]1547    dummy_rest_api_id = "a12b3c4d"1548    model_name = "testModel"1549    description = "test model"1550    content_type = "application/json"1551    # add a model1552    client.create_model(1553        restApiId=rest_api_id,1554        name=model_name,1555        description=description,1556        contentType=content_type,1557    )1558    # get models after adding1559    result = client.get_model(restApiId=rest_api_id, modelName=model_name)1560    result["name"] = model_name1561    result["description"] = description1562    with pytest.raises(ClientError) as ex:1563        client.get_model(restApiId=dummy_rest_api_id, modelName=model_name)1564    ex.value.response["Error"]["Message"].should.equal("Invalid Rest API Id specified")1565    ex.value.response["Error"]["Code"].should.equal("NotFoundException")1566@mock_apigateway1567def test_get_model_with_invalid_name():1568    client = boto3.client("apigateway", region_name="us-west-2")1569    response = client.create_rest_api(name="my_api", description="this is my api")1570    rest_api_id = response["id"]1571    # test with an invalid model name1572    with pytest.raises(ClientError) as ex:1573        client.get_model(restApiId=rest_api_id, modelName="fake")1574    ex.value.response["Error"]["Message"].should.equal("Invalid Model Name specified")1575    ex.value.response["Error"]["Code"].should.equal("NotFoundException")1576@mock_apigateway1577def test_http_proxying_integration():1578    responses_mock.add(1579        responses_mock.GET, "http://httpbin.org/robots.txt", body="a fake response"1580    )1581    region_name = "us-west-2"1582    client = boto3.client("apigateway", region_name=region_name)1583    response = client.create_rest_api(name="my_api", description="this is my api")1584    api_id = response["id"]1585    resources = client.get_resources(restApiId=api_id)1586    root_id = [resource for resource in resources["items"] if resource["path"] == "/"][1587        01588    ]["id"]1589    client.put_method(1590        restApiId=api_id, resourceId=root_id, httpMethod="GET", authorizationType="none"1591    )1592    client.put_method_response(1593        restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"1594    )1595    response = client.put_integration(1596        restApiId=api_id,1597        resourceId=root_id,...

api_gateway.py

Source:api_gateway.py

...7        self.platform = platform8    def verify(self):9        client = self.platform.get_client('apigateway')10        api_name = self.platform.get_test_name('api')11        api_id = client.create_rest_api(name=api_name,12                                        description='Api for Revizor Tests',13                                        endpointConfiguration={14                                            'types': ['EDGE', ]15                                        })['id']16        assert any([api for api in client.get_rest_apis()['items'] if api['name'] == api_name])17        assert client.get_rest_api(restApiId=api_id)['name'] == api_name18        client.delete_rest_api(restApiId=api_id)19        with world.assert_raises(boto_exceptions.ClientError, 'NotFound'):20            client.get_rest_api(restApiId=api_id)21    def verify_denied(self, error_text):22        client = self.platform.get_client('apigateway')23        with world.assert_raises(boto_exceptions.ClientError, error_text):24            client.get_rest_apis()25    def verify_policy(self, prefix=False, pattern=False):26        client = self.platform.get_client('apigateway')27        if prefix:28            api_name = self.platform.get_test_name('api_')29            with world.assert_raises(boto_exceptions.ClientError,30                                     "Action 'CreateRestApi' violates policy 'csg.resource.name.prefix'"):31                client.create_rest_api(name=api_name,32                                       description='Api for Revizor Tests',33                                       endpointConfiguration={34                                           'types': ['EDGE', ]35                                       })36        if pattern:37            api_name = 'tmp_%s' % self.platform.get_test_name('api')38            with world.assert_raises(boto_exceptions.ClientError,39                                     "Action 'CreateRestApi' violates policy 'csg.resource.name.validation_pattern'"):40                client.create_rest_api(name=api_name,41                                       description='Api for Revizor Tests',42                                       endpointConfiguration={43                                           'types': ['EDGE', ]44                                       })45        api_name = 'tmp_%s' % self.platform.get_test_name('api_')46        api_id = client.create_rest_api(name=api_name,47                                        description='Api for Revizor Tests',48                                        endpointConfiguration={49                                            'types': ['EDGE', ]50                                        })['id']51        assert any([api for api in client.get_rest_apis()['items'] if api['name'] == api_name])...

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.