KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use update_integration_response method in localstack

Best Python code snippet using localstack_python

apigateway.py

Source:apigateway.py

...417        resp = self.apigateway_client.update_method(restApiId=restid, resourceId=resourceid, httpMethod=method,418                                                    operation=operation)419        super(Apigateway, self).query_information(query=resp)420        return resp421    def update_integration_response(self, restid, resourceid, method, statuscode, operation):422        """423        This function updates an integration response424        :param method: the method that is requested425        :type method: basestring426        :param restid: the id of the rest api object427        :type restid: basestring428        :param resourceid: id of a single resource object429        :type resourceid: basestring430        :param statuscode: the statuscode where the integration response is431        :type statuscode: basestring432        :param operation: a list of patchOperations433        :type operation: list434        :return: the updated integration response object435        """436        if self.dryrun:437            logger.info("Dryrun requested no changes will be done")438            return None439        resp = self.apigateway_client.update_integration_response(restApiId=restid, resourceId=resourceid,440                                                                  httpMethod=method, statusCode=statuscode,441                                                                  patchOperations=operation)442        super(Apigateway, self).query_information(query=resp)443        return resp444    def generate_resourcehash(self, restid):445        """446        This function collects and returns a hash with resource object and their ids.447        This is used to find any resources that should be deleted or added448        :param restid: the id of the rest api object449        :type restid: basestring450        :return: a dict with resource name with resource id-s451        :rtype: dict452        """453        resources = self.get_resource(restid=restid)454        ret = {}455        for resource in resources:456            ret[resource['path']] = resource['id']457        return ret458    def create_resource(self, restid, parentid, pathpart):459        """460        This function creates a resource object461        :param restid: the id of the rest api object462        :type restid: basestring463        :param parentid: the parent id of the created resource, should be rest api464        :type parentid: basestring465        :param pathpart: The pathpart where the resource be466        :type pathpart: basestring467        :return: the resource object created468        """469        if self.dryrun:470            logger.info("Dryrun requested no changes will be done")471            return None472        resp = self.apigateway_client.create_resource(restApiId=restid, parentId=parentid, pathPart=pathpart)473        super(Apigateway, self).query_information(query=resp)474        return resp475    def delete_resource(self, restid, resourceid):476        """477        This function deletes a resource object478        :param restid: the id of the rest api object479        :type restid: basestring480        :param resourceid: id of a single resource object481        :type resourceid: basestring482        :return: None483        """484        if self.dryrun:485            logger.info("Dryrun requested no changes will be done")486            return None487        resp = self.apigateway_client.delete_resource(restApiId=restid, resourceId=resourceid)488        super(Apigateway, self).query_information(query=resp)489        return resp490    def delete_integration_response(self, restid, resourceid, method, statuscode):491        """492        This function deletes an integration response493        :param method: the method that is requested494        :type method: basestring495        :param restid: the id of the rest api object496        :type restid: basestring497        :param resourceid: id of a single resource object498        :type resourceid: basestring499        :param statuscode: the statuscode to delete500        :type statuscode: basestring501        :return: None502        """503        if self.dryrun:504            logger.info("Dryrun requested no changes will be done")505            return None506        resp = self.apigateway_client.delete_integration_response(restApiId=restid, resourceId=resourceid,507                                                                  httpMethod=method, statusCode=statuscode)508        super(Apigateway, self).query_information(query=resp)509        return resp510    def method_exists(self, restid, resourceid, method):511        try:512            self.get_method(restid=restid, resourceid=resourceid, method=method)513            return True514        except:515            return False516    def method_response_exists(self, restid, resourceid, method, statuscode):517        try:518            self.get_method_response(restid=restid, resourceid=resourceid, method=method, statuscode=statuscode)519            return True520        except:521            return False522    def integration_response_exists(self, restid, resourceid, method, status):523        try:524            self.get_integration_response(restid=restid, resourceid=resourceid, method=method, status=status)525            return True526        except:527            return False528    def integration_exists(self, restid, resourceid, method):529        try:530            self.get_integration(restid=restid, resourceid=resourceid, method=method)531            return True532        except:533            return False534    def compare_method(self, restid, resourceid, method, json_data):535        """536        This function compares a json data to the current method to detect an updates that need to be done537        :param method: the method that is requested538        :type method: basestring539        :param restid: the id of the rest api object540        :type restid: basestring541        :param resourceid: id of a single resource object542        :type resourceid: basestring543        :param json_data: the json data from the model that is the representation of the current state544        :type json_data: dict545        :return: None546        """547        logger.info("Looking at restid: %s, resourceid: %s, and method: %s" % (restid, resourceid, method))548        # First we test if the top level method is created or we need to create it549        if not self.method_exists(restid=restid, resourceid=resourceid, method=method):550            logger.info("Need to create method: %s" % method)551            cur_method = self.create_method(restid=restid, resourceid=resourceid, method=method,552                                            authorizationtype=json_data['authorizationType'], further_opts=json_data)553        else:554            cur_method = self.get_method(restid=restid, resourceid=resourceid, method=method)555            logger.info("Method existed, need to compare for changes")556            for element in ['authorizationType', 'apiKeyRequired', 'requestParameters', 'requestModels']:557                if (element in json_data and element in cur_method) and json_data[element] != cur_method[element]:558                    logger.warning("Need to update %s" % element)559                    self.update_method(restid=restid, resourceid=resourceid, method=method, operation=[560                        {'op': 'replace', 'path': "/%s" % element, 'value': json_data[element]}])561                if element not in json_data:562                    logger.debug("Upload template missing key %s, skipping" % element)563                if element not in cur_method and element in json_data:564                    logger.warning("Not defined in current method need to update current method with %s" % element)565        # Check if method needs to be deleted566        if 'methodResponses' in cur_method:567            for statuscode in cur_method['methodResponses']:568                if statuscode not in json_data['methodResponses']:569                    logger.warning("This method response needs to be deleted %s" % statuscode)570                    self.delete_method_response(restid=restid, resourceid=resourceid, method=method,571                                                statuscode=statuscode)572        # iterate over status codes and check we need to create or update573        for statuscode in json_data['methodResponses']:574            if not self.method_response_exists(restid=restid, resourceid=resourceid, method=method,575                                               statuscode=statuscode):576                logger.debug("Creating method response %s" % statuscode)577                self.create_method_response(restid=restid, resourceid=resourceid, method=method, statuscode=statuscode,578                                            further_ops=json_data['methodResponses'][statuscode])579            else:580                cur_response = self.get_method_response(restid=restid, resourceid=resourceid, method=method,581                                                        statuscode=statuscode)582                logger.debug("Need to compare the responses")583                dictdiffer = DictDiffer(cur_response, json_data['methodResponses'][statuscode])584                for remove_statuscode in dictdiffer.added():585                    logger.info("Need to remove statuscode: %s" % remove_statuscode)586                    self.delete_method_response(restid=restid, resourceid=resourceid, method=method,587                                                statuscode=remove_statuscode)588                for add_statuscode in dictdiffer.removed():589                    logger.info("Need to add statuscode: %s" % add_statuscode)590                    self.create_method_response(restid=restid, resourceid=resourceid, method=method,591                                                statuscode=add_statuscode,592                                                further_ops=json_data['methodResponses'][add_statuscode])593                for changed_statuscode in dictdiffer.changed():594                    logger.info("Need to update statuscode: %s" % changed_statuscode)595                    cur_method_statuscode = cur_method['methodResponses'][changed_statuscode]596                    json_data_statuscode = json_data['methodmethod']['methodResponses'][changed_statuscode]597                    for element in ['responseParameters', 'responseTemplates']:598                        if element not in json_data_statuscode:599                            continue600                        change_dictdiffer = DictDiffer(601                            cur_method_statuscode[element],602                            json_data_statuscode[element])603                        for add_int_statuscode in change_dictdiffer.removed():604                            logger.info("method response is missing, adding: %s" % add_int_statuscode)605                            self.update_method_response(restid=restid, resourceid=resourceid, method=method,606                                                        statuscode=changed_statuscode, operation=[607                                    {'op': 'add', 'path': "/%s/%s" % (element, add_int_statuscode), 'value':608                                        json_data_statuscode[element][add_int_statuscode]}])609                        for remove_int_statuscode in change_dictdiffer.added():610                            logger.info("Method response is present, deleting: %s" % remove_int_statuscode)611                            self.update_method_response(restid=restid, resourceid=resourceid, method=method,612                                                        statuscode=changed_statuscode, operation=[613                                    {'op': 'remove', 'path': "/%s/%s" % (element, remove_int_statuscode)}])614                        for change_int_statuscode in change_dictdiffer.changed():615                            logger.info("There is a change in value, need to update: %s" % change_int_statuscode)616                            self.update_method_response(restid=restid, resourceid=resourceid, method=method,617                                                        statuscode=changed_statuscode, operation=[618                                    {'op': 'replace', 'path': "/%s/%s" % (element, change_int_statuscode), 'value':619                                        json_data_statuscode[element][change_int_statuscode]}])620        # method integration621        if self.integration_exists(restid=restid, resourceid=resourceid, method=method):622            cur_method_integration = self.get_integration(restid=restid, resourceid=resourceid, method=method)623            dictdiffer_integration_response = DictDiffer(cur_method_integration['integrationResponses'],624                                                         json_data['methodIntegration']['integrationResponses'])625            for remove_response in dictdiffer_integration_response.added():626                logger.info("Need to remove integration response: %s" % remove_response)627                self.delete_integration_response(restid=restid, resourceid=resourceid, method=method,628                                                 statuscode=remove_response)629            for add_response in dictdiffer_integration_response.removed():630                logger.info("Need to add integration response: %s" % add_response)631                self.create_integration_response(restid=restid, resourceid=resourceid, method=method,632                                                 statuscode=add_response,633                                                 further_opts=json_data['methodIntegration']['integrationResponses'][634                                                     add_response])635            for changed_response in dictdiffer_integration_response.changed():636                logger.info("Need to change response value: %s" % changed_response)637                cur_method_integration_response = cur_method_integration['integrationResponses'][changed_response]638                json_data_integration_response = json_data['methodIntegration']['integrationResponses'][639                    changed_response]640                if 'selectionPattern' in cur_method_integration_response:641                    if (cur_method_integration_response['selectionPattern'] !=642                            json_data_integration_response['selectionPattern']):643                        logger.debug("selectionPattern needs to be updated")644                        self.update_integration_response(restid=restid, resourceid=resourceid, method=method,645                                                         statuscode=changed_response, operation=[646                                {'op': 'replace', 'path': '/selectionPattern',647                                 'value': json_data_integration_response['selectionPattern']}])648                for element in ['responseParameters', 'responseTemplates']:649                    if element not in json_data_integration_response:650                        continue651                    change_dictdiffer = DictDiffer(652                        cur_method_integration_response[element],653                        json_data_integration_response[element])654                    for add_int_response in change_dictdiffer.removed():655                        logger.debug("Need to add the integration response: %s" % add_int_response)656                        self.update_integration_response(restid=restid, resourceid=resourceid, method=method,657                                                         statuscode=changed_response, operation=[658                                {'op': 'add', 'path': "/%s/%s" % (element, add_int_response), 'value':659                                    json_data_integration_response[element][add_int_response]}])660                    for remove_int_response in change_dictdiffer.added():661                        logger.debug("Need to remove the integration response: %s" % remove_int_response)662                        self.update_integration_response(restid=restid, resourceid=resourceid, method=method,663                                                         statuscode=changed_response, operation=[664                                {'op': 'remove', 'path': "/%s/%s" % (element, remove_int_response)}])665                    for change_int_response in change_dictdiffer.changed():666                        logger.debug("Need to update the integration response: %s" % change_int_response)667                        self.update_integration_response(restid=restid, resourceid=resourceid, method=method,668                                                         statuscode=changed_response, operation=[669                                {'op': 'replace', 'path': "/%s/%s" % (element, change_int_response), 'value':670                                    json_data_integration_response[element][change_int_response]}])671        else:672            logger.debug("Need to create method integration")673            cur_method_integration = self.create_integration(restid=restid, resourceid=resourceid, method=method,674                                                             integration_type=json_data['methodIntegration'][675                                                                 'type'],676                                                             further_opts=json_data['methodIntegration'])677            for integration_response in json_data['methodIntegration']['integrationResponses']:678                logger.debug("Need to create the method integrations for the new method")679                self.create_integration_response(restid=restid, resourceid=resourceid, method=method,680                                                 statuscode=integration_response,681                                                 further_opts=...

test_update_integration_response.py

Source:test_update_integration_response.py

1# coding: utf-82"""3    Opsgenie REST API4    Opsgenie OpenAPI Specification  # noqa: E5015    OpenAPI spec version: 2.0.06    7    Generated by: https://github.com/swagger-api/swagger-codegen.git8"""9from __future__ import absolute_import10import unittest11import swagger_client12from swagger_client.models.update_integration_response import UpdateIntegrationResponse  # noqa: E50113from swagger_client.rest import ApiException14class TestUpdateIntegrationResponse(unittest.TestCase):15    """UpdateIntegrationResponse unit test stubs"""16    def setUp(self):17        pass18    def tearDown(self):19        pass20    def testUpdateIntegrationResponse(self):21        """Test UpdateIntegrationResponse"""22        # FIXME: construct object with mandatory attributes with example values23        # model = swagger_client.models.update_integration_response.UpdateIntegrationResponse()  # noqa: E50124        pass25if __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.