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

How to use update_integration method in localstack

Best Python code snippet using localstack_python

cidr_block_vending_machine_stack.py

Source:cidr_block_vending_machine_stack.py

1from aws_cdk import (2    core,3    aws_iam,4    aws_lambda,5    aws_dynamodb,6    aws_apigateway7)8class CidrBlockVendingMachineStack(core.Stack):9    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:10        super().__init__(scope, id, **kwargs)11        organization_id = self.node.try_get_context("organizationId")12        # create dynamo table13        allocation_table = aws_dynamodb.Table(14            self, "CidrBlockTable",15            partition_key=aws_dynamodb.Attribute(16                name="vpcCidrBlock",17                type=aws_dynamodb.AttributeType.STRING18            )19        )20        # create producer lambda function21        create_lambda = aws_lambda.Function(self, "create_lambda_function",22                                              runtime=aws_lambda.Runtime.PYTHON_3_6,23                                              handler="create.lambda_handler",24                                              code=aws_lambda.Code.asset("./src/"))25        create_lambda.add_environment("TABLE_NAME", allocation_table.table_name)26        create_lambda.add_environment("MASTER_CIDR_BLOCK", "10.0.0.0/12")27        create_lambda.add_environment("VPC_NETMASK", "24")28        create_lambda.add_environment("SUBNET_NETMASK", "26")29        # grant permission to lambda to write to demo table30        allocation_table.grant_write_data(create_lambda)31        allocation_table.grant_read_data(create_lambda)32        # API gateway ... Allow own Organizations33        api_policy = aws_iam.PolicyDocument()34        api_policy.add_statements(35            aws_iam.PolicyStatement(36                effect=aws_iam.Effect.ALLOW,37                principals=[aws_iam.AnyPrincipal()],38                actions=["execute-api:Invoke"],39                resources=[core.Fn.join('', ['execute-api:/', '*'])]))40        api_policy.add_statements(41            aws_iam.PolicyStatement(42                effect=aws_iam.Effect.DENY,43                actions=["execute-api:Invoke"],44                conditions={45                    "StringNotEquals": {46                        "aws:PrincipalOrgID": [47                            organization_id48                        ]49                    }50                },51                principals=[aws_iam.AnyPrincipal()],52                resources=[core.Fn.join('', ['execute-api:/', '*'])]))53        base_api = aws_apigateway.RestApi(self, 'ApiGateway', rest_api_name='cidr_vending_machine', policy=api_policy)54        vpc_api = base_api.root.add_resource('vpc')55        rest_api_role = aws_iam.Role(self,'RestAPIRole',56            assumed_by=aws_iam.ServicePrincipal('apigateway.amazonaws.com'),57        )58        allocation_table.grant_read_write_data(rest_api_role)59        60        patch_request_string = """61            {{62            "TableName": "{}",63                "Key": {{64                        "vpcCidrBlock": {{65                            "S": "$input.params('cidr_block')"66                        }}67                    }},68                    "UpdateExpression": "set vpcId = :v",69                    "ConditionExpression": "accountId = :v2",70                    "ExpressionAttributeValues" : {{71                        ":v": {{"S": "$input.params('vpc_id')"}},72                        ":v2": {{"S": "$context.identity.accountId"}}73                    }},74                    "ReturnValues": "ALL_NEW"75            }}"""76        delete_request_string = """77            {{78            "TableName": "{}",79                "Key": {{80                        "vpcCidrBlock": {{81                            "S": "$input.params('cidr_block')"82                        }}83                    }},84                "ConditionExpression": "accountId = :v2",85                "ExpressionAttributeValues" : {{86                    ":v2": {{"S": "$context.identity.accountId"}}87                }}88            }}"""89        network_integration = aws_apigateway.LambdaIntegration(create_lambda)90        update_integration = aws_apigateway.AwsIntegration(91            service='dynamodb',92            action='UpdateItem',93            integration_http_method='POST',94            options=aws_apigateway.IntegrationOptions(95                request_templates={96                    "application/json": patch_request_string.format(allocation_table.table_name)97                },98                integration_responses=[aws_apigateway.IntegrationResponse(99                    status_code="200"100                )],101                credentials_role=rest_api_role102            )103        )104        delete_integration = aws_apigateway.AwsIntegration(105            service='dynamodb',106            action='DeleteItem',107            integration_http_method='POST',108            options=aws_apigateway.IntegrationOptions(109                request_templates={110                    "application/json": delete_request_string.format(allocation_table.table_name)111                },112                integration_responses=[aws_apigateway.IntegrationResponse(113                    status_code="200"114                )],115                credentials_role=rest_api_role116            )117        )118        vpc_api.add_method('POST', network_integration, authorization_type=aws_apigateway.AuthorizationType.IAM)119        vpc_api.add_method('DELETE', delete_integration, authorization_type=aws_apigateway.AuthorizationType.IAM, 120            method_responses=[aws_apigateway.MethodResponse(status_code="200")])121        vpc_api.add_method('PATCH', update_integration,authorization_type=aws_apigateway.AuthorizationType.IAM,...

test_integrations.py

Source:test_integrations.py

...65        integration_params["user"] = "original_user"66        integrations[origin_name] = {"params": integration_params}67        self.assertTrue(isinstance(integrations[origin_name].get_info(), dict))68        self.assertTrue(len(integrations[origin_name]) > 5)69    def update_integration(self, _type, integrations, to_update={'user': 'updated_user'}):70        origin_name = f"{_type}_{self.integration_suffix}"71        integration = integrations[origin_name]72        self.assertTrue(integration is not None)73        update_params = self.integration_creds[_type]74        update_params["type"] = _type75        # update_params["user"] = "updated_user"76        if to_update is not None:77            update_params.update(to_update)78        integration.update({"params": update_params})79        self.assertTrue(isinstance(integration.get_info(), dict))80        if to_update is not None:81            for k in to_update:82                self.assertTrue(integration.get_info()[k] == to_update[k])83    def test_1_list_info(self):84        self.list_info(self.integrations)85    def test_2_add_clickhouse(self):86        self.add_integration("clickhouse", self.integrations)87    def test_3_update_clickhouse(self):88        self.update_integration("clickhouse", self.integrations)89    def test_4_add_mysql(self):90        self.add_integration("mysql", self.integrations)91    def test_5_update_mysql(self):92        self.update_integration("mysql", self.integrations)93    def test_6_add_mongo(self):94        self.add_integration("mongodb", self.integrations)95    def test_7_update_mongo(self):96        self.update_integration("mongodb", self.integrations)97    def test_8_add_mariadb(self):98        self.add_integration("mariadb", self.integrations)99    def test_9_update_mariadb(self):100        self.update_integration("mariadb", self.integrations)101    def test_10_add_postgres(self):102        self.add_integration("postgres", self.integrations)103    def test_11_update_postgres(self):104        self.update_integration("postgres", self.integrations)105    def test_12_add_snowflake(self):106        self.add_integration("snowflake", self.integrations)107    def test_13_update_snowflake(self):108        self.update_integration("snowflake", self.integrations, to_update={'test': True})109if __name__ == "__main__":110    if len(sys.argv) > 1 and sys.argv[-1] == "--no_backend_instance":111        # need to remove if from arg list112        # mustn't provide it into unittest.main113        sys.argv.pop()114        TestDatasources.start_backend = False...

main.py

Source:main.py

...4api_ids = {5}6methods_to_update = ['GET', 'POST', 'PUT', 'DELETE', 'UPDATE']7value_to_update = '120000'8def update_integration():9    for api, id in api_ids.items():10        print(f'Starting for {api}')11        resources = client.get_resources(12            restApiId=id,13            limit=500,14        )15        for resource in resources['items']:16            if 'resourceMethods' not in resource.keys():17                continue18            for method in resource['resourceMethods'].keys():19                if method in methods_to_update:20                    try:21                        response = client.update_integration(22                            restApiId=id,23                            resourceId=resource['id'],24                            httpMethod=method,25                            patchOperations=[26                                {27                                    'op': 'replace',28                                    'path': '/timeoutInMillis',29                                    'value': value_to_update30                                },31                            ]32                        )33                        print(f"Completed for id: {resource['id']}")34                    except Exception as e:35                        print(e)36                        print(f"Could not complete for id: {resource['id']}")37                        continue38                else:39                    continue40    41if __name__ == '__main__':...

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.