How to use path_based_url method in localstack

Best Python code snippet using localstack_python

test_apigateway.py

Source:test_apigateway.py Github

copy

Full Screen

...175 {"data": '{"foo": "bar2"}'},176 {"data": '{"foo": "bar3"}'},177 ]178 }179 url = path_based_url(180 api_id=result["id"],181 stage_name=self.TEST_STAGE_NAME,182 path=self.API_PATH_DATA_INBOUND,183 )184 # list Kinesis streams via API Gateway185 result = requests.get(url)186 result = json.loads(to_str(result.content))187 assert "StreamNames" in result188 # post test data to Kinesis via API Gateway189 result = requests.post(url, data=json.dumps(test_data))190 result = json.loads(to_str(result.content))191 assert 0 == result["FailedRecordCount"]192 assert len(test_data["records"]) == len(result["Records"])193 # clean up194 kinesis = aws_stack.create_external_boto_client("kinesis")195 kinesis.delete_stream(StreamName=self.TEST_STREAM_KINESIS_API_GW)196 def test_api_gateway_sqs_integration_with_event_source(self):197 # create target SQS stream198 queue_name = f"queue-{short_uid()}"199 queue_url = aws_stack.create_sqs_queue(queue_name)["QueueUrl"]200 # create API Gateway and connect it to the target queue201 result = connect_api_gateway_to_sqs(202 "test_gateway4",203 stage_name=self.TEST_STAGE_NAME,204 queue_arn=queue_name,205 path=self.API_PATH_DATA_INBOUND,206 )207 # create event source for sqs lambda processor208 self.create_lambda_function(self.TEST_LAMBDA_SQS_HANDLER_NAME)209 event_source_data = {210 "FunctionName": self.TEST_LAMBDA_SQS_HANDLER_NAME,211 "EventSourceArn": aws_stack.sqs_queue_arn(queue_name),212 "Enabled": True,213 }214 add_event_source(event_source_data)215 # generate test data216 test_data = {"spam": "eggs & beans"}217 url = path_based_url(218 api_id=result["id"],219 stage_name=self.TEST_STAGE_NAME,220 path=self.API_PATH_DATA_INBOUND,221 )222 result = requests.post(url, data=json.dumps(test_data))223 assert 200 == result.status_code224 parsed_json = xmltodict.parse(result.content)225 result = parsed_json["SendMessageResponse"]["SendMessageResult"]226 body_md5 = result["MD5OfMessageBody"]227 assert "b639f52308afd65866c86f274c59033f" == body_md5228 # clean up229 sqs_client = aws_stack.create_external_boto_client("sqs")230 sqs_client.delete_queue(QueueUrl=queue_url)231 lambda_client = aws_stack.create_external_boto_client("lambda")232 lambda_client.delete_function(FunctionName=self.TEST_LAMBDA_SQS_HANDLER_NAME)233 def test_api_gateway_sqs_integration(self):234 # create target SQS stream235 queue_name = f"queue-{short_uid()}"236 aws_stack.create_sqs_queue(queue_name)237 # create API Gateway and connect it to the target queue238 result = connect_api_gateway_to_sqs(239 "test_gateway4",240 stage_name=self.TEST_STAGE_NAME,241 queue_arn=queue_name,242 path=self.API_PATH_DATA_INBOUND,243 )244 # generate test data245 test_data = {"spam": "eggs"}246 url = path_based_url(247 api_id=result["id"],248 stage_name=self.TEST_STAGE_NAME,249 path=self.API_PATH_DATA_INBOUND,250 )251 result = requests.post(url, data=json.dumps(test_data))252 assert 200 == result.status_code253 messages = aws_stack.sqs_receive_message(queue_name)["Messages"]254 assert 1 == len(messages)255 assert test_data == json.loads(base64.b64decode(messages[0]["Body"]))256 def test_api_gateway_lambda_integration(self, apigateway_client, create_lambda_function):257 """258 API gateway to lambda integration test returns a response with the same body as the lambda function input.259 """260 fn_name = f"test-{short_uid()}"261 create_lambda_function(262 func_name=fn_name,263 handler_file=TEST_LAMBDA_PYTHON_ECHO,264 runtime=LAMBDA_RUNTIME_PYTHON39,265 )266 lambda_arn = aws_stack.lambda_function_arn(fn_name)267 api_id, _, root = create_rest_api(apigateway_client, name="aws lambda api")268 resource_id, _ = create_rest_resource(269 apigateway_client, restApiId=api_id, parentId=root, pathPart="test"270 )271 # create method and integration272 create_rest_resource_method(273 apigateway_client,274 restApiId=api_id,275 resourceId=resource_id,276 httpMethod="GET",277 authorizationType="NONE",278 )279 create_rest_api_integration(280 apigateway_client,281 restApiId=api_id,282 resourceId=resource_id,283 httpMethod="GET",284 integrationHttpMethod="GET",285 type="AWS",286 uri=f"arn:aws:apigateway:{aws_stack.get_region()}:lambda:path//2015-03-31/functions/{lambda_arn}/invocations",287 )288 url = api_invoke_url(api_id=api_id, stage="local", path="/test")289 response = requests.get(url)290 body = response.json()291 # authorizer contains an object that does not contain the authorizer type ('lambda', 'sns')292 assert body.get("requestContext").get("authorizer") == {"context": {}, "identity": {}}293 @pytest.mark.parametrize("int_type", ["custom", "proxy"])294 def test_api_gateway_http_integrations(self, int_type, monkeypatch):295 monkeypatch.setattr(config, "DISABLE_CUSTOM_CORS_APIGATEWAY", False)296 test_port = get_free_tcp_port()297 backend_url = "http://localhost:%s%s" % (test_port, self.API_PATH_HTTP_BACKEND)298 # start test HTTP backend299 proxy = self.start_http_backend(test_port)300 # create API Gateway and connect it to the HTTP_PROXY/HTTP backend301 result = self.connect_api_gateway_to_http(302 int_type, "test_gateway2", backend_url, path=self.API_PATH_HTTP_BACKEND303 )304 url = path_based_url(305 api_id=result["id"],306 stage_name=self.TEST_STAGE_NAME,307 path=self.API_PATH_HTTP_BACKEND,308 )309 # make sure CORS headers are present310 origin = "localhost"311 result = requests.options(url, headers={"origin": origin})312 assert result.status_code == 200313 assert re.match(result.headers["Access-Control-Allow-Origin"].replace("*", ".*"), origin)314 assert "POST" in result.headers["Access-Control-Allow-Methods"]315 assert "PATCH" in result.headers["Access-Control-Allow-Methods"]316 custom_result = json.dumps({"foo": "bar"})317 # make test GET request to gateway318 result = requests.get(url)319 assert 200 == result.status_code320 expected = custom_result if int_type == "custom" else "{}"321 assert expected == json.loads(to_str(result.content))["data"]322 # make test POST request to gateway323 data = json.dumps({"data": 123})324 result = requests.post(url, data=data)325 assert 200 == result.status_code326 expected = custom_result if int_type == "custom" else data327 assert expected == json.loads(to_str(result.content))["data"]328 # make test POST request with non-JSON content type329 data = "test=123"330 ctype = "application/x-www-form-urlencoded"331 result = requests.post(url, data=data, headers={"content-type": ctype})332 assert 200 == result.status_code333 content = json.loads(to_str(result.content))334 headers = CaseInsensitiveDict(content["headers"])335 expected = custom_result if int_type == "custom" else data336 assert expected == content["data"]337 assert ctype == headers["content-type"]338 # clean up339 proxy.stop()340 def test_api_gateway_lambda_proxy_integration(self):341 self._test_api_gateway_lambda_proxy_integration(342 self.TEST_LAMBDA_PROXY_BACKEND, self.API_PATH_LAMBDA_PROXY_BACKEND343 )344 def test_api_gateway_lambda_proxy_integration_with_path_param(self):345 self._test_api_gateway_lambda_proxy_integration(346 self.TEST_LAMBDA_PROXY_BACKEND_WITH_PATH_PARAM,347 self.API_PATH_LAMBDA_PROXY_BACKEND_WITH_PATH_PARAM,348 )349 def test_api_gateway_lambda_proxy_integration_with_is_base_64_encoded(self):350 # Test the case where `isBase64Encoded` is enabled.351 content = b"hello, please base64 encode me"352 def _mutate_data(data) -> None:353 data["return_is_base_64_encoded"] = True354 data["return_raw_body"] = base64.b64encode(content).decode("utf8")355 test_result = self._test_api_gateway_lambda_proxy_integration_no_asserts(356 self.TEST_LAMBDA_PROXY_BACKEND_WITH_IS_BASE64,357 self.API_PATH_LAMBDA_PROXY_BACKEND_WITH_IS_BASE64,358 data_mutator_fn=_mutate_data,359 )360 # Ensure that `invoke_rest_api_integration_backend` correctly decodes the base64 content361 assert test_result.result.status_code == 203362 assert test_result.result.content == content363 def _test_api_gateway_lambda_proxy_integration_no_asserts(364 self,365 fn_name: str,366 path: str,367 data_mutator_fn: Optional[Callable] = None,368 ) -> ApiGatewayLambdaProxyIntegrationTestResult:369 """370 Perform the setup needed to do a POST against a Lambda Proxy Integration;371 then execute the POST.372 :param data_mutator_fn: a Callable[[Dict], None] that lets us mutate the373 data dictionary before sending it off to the lambda.374 """375 self.create_lambda_function(fn_name)376 # create API Gateway and connect it to the Lambda proxy backend377 lambda_uri = aws_stack.lambda_function_arn(fn_name)378 invocation_uri = "arn:aws:apigateway:%s:lambda:path/2015-03-31/functions/%s/invocations"379 target_uri = invocation_uri % (aws_stack.get_region(), lambda_uri)380 result = testutil.connect_api_gateway_to_http_with_lambda_proxy(381 "test_gateway2", target_uri, path=path, stage_name=self.TEST_STAGE_NAME382 )383 api_id = result["id"]384 path_map = get_rest_api_paths(api_id)385 _, resource = get_resource_for_path(path, path_map)386 # make test request to gateway and check response387 path_with_replace = path.replace("{test_param1}", "foo1")388 path_with_params = path_with_replace + "?foo=foo&bar=bar&bar=baz"389 url = path_based_url(api_id=api_id, stage_name=self.TEST_STAGE_NAME, path=path_with_params)390 # These values get read in `lambda_integration.py`391 data = {"return_status_code": 203, "return_headers": {"foo": "bar123"}}392 if data_mutator_fn:393 assert callable(data_mutator_fn)394 data_mutator_fn(data)395 result = requests.post(396 url,397 data=json.dumps(data),398 headers={"User-Agent": "python-requests/testing"},399 )400 return ApiGatewayLambdaProxyIntegrationTestResult(401 data=data,402 resource=resource,403 result=result,404 url=url,405 path_with_replace=path_with_replace,406 )407 def _test_api_gateway_lambda_proxy_integration(408 self,409 fn_name: str,410 path: str,411 ) -> None:412 test_result = self._test_api_gateway_lambda_proxy_integration_no_asserts(fn_name, path)413 data, resource, result, url, path_with_replace = test_result414 assert result.status_code == 203415 assert result.headers.get("foo") == "bar123"416 assert "set-cookie" in result.headers417 try:418 parsed_body = json.loads(to_str(result.content))419 except json.decoder.JSONDecodeError as e:420 raise Exception(421 "Couldn't json-decode content: {}".format(to_str(result.content))422 ) from e423 assert parsed_body.get("return_status_code") == 203424 assert parsed_body.get("return_headers") == {"foo": "bar123"}425 assert parsed_body.get("queryStringParameters") == {"foo": "foo", "bar": ["bar", "baz"]}426 request_context = parsed_body.get("requestContext")427 source_ip = request_context["identity"].pop("sourceIp")428 assert re.match(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$", source_ip)429 expected_path = "/" + self.TEST_STAGE_NAME + "/lambda/foo1"430 assert expected_path == request_context["path"]431 assert request_context.get("stageVariables") is None432 assert get_aws_account_id() == request_context["accountId"]433 assert resource.get("id") == request_context["resourceId"]434 assert self.TEST_STAGE_NAME == request_context["stage"]435 assert "python-requests/testing" == request_context["identity"]["userAgent"]436 assert "POST" == request_context["httpMethod"]437 assert "HTTP/1.1" == request_context["protocol"]438 assert "requestTimeEpoch" in request_context439 assert "requestTime" in request_context440 assert "requestId" in request_context441 # assert that header keys are lowercase (as in AWS)442 headers = parsed_body.get("headers") or {}443 header_names = list(headers.keys())444 assert "Host" in header_names445 assert "Content-Length" in header_names446 assert "User-Agent" in header_names447 result = requests.delete(url, data=json.dumps(data))448 assert 204 == result.status_code449 # send message with non-ASCII chars450 body_msg = "🙀 - 参よ"451 result = requests.post(url, data=json.dumps({"return_raw_body": body_msg}))452 assert body_msg == to_str(result.content)453 # send message with binary data454 binary_msg = b"\xff \xaa \x11"455 result = requests.post(url, data=binary_msg)456 result_content = json.loads(to_str(result.content))457 assert "/yCqIBE=" == result_content["body"]458 assert ["isBase64Encoded"]459 def test_api_gateway_lambda_proxy_integration_any_method(self):460 self._test_api_gateway_lambda_proxy_integration_any_method(461 self.TEST_LAMBDA_PROXY_BACKEND_ANY_METHOD,462 self.API_PATH_LAMBDA_PROXY_BACKEND_ANY_METHOD,463 )464 def test_api_gateway_lambda_proxy_integration_any_method_with_path_param(self):465 self._test_api_gateway_lambda_proxy_integration_any_method(466 self.TEST_LAMBDA_PROXY_BACKEND_ANY_METHOD_WITH_PATH_PARAM,467 self.API_PATH_LAMBDA_PROXY_BACKEND_ANY_METHOD_WITH_PATH_PARAM,468 )469 def test_api_gateway_lambda_asynchronous_invocation(self, apigateway_client):470 api_gateway_name = f"api_gateway_{short_uid()}"471 rest_api_id = apigateway_client.create_rest_api(name=api_gateway_name)["id"]472 fn_name = f"test-{short_uid()}"473 testutil.create_lambda_function(474 handler_file=TEST_LAMBDA_NODEJS, func_name=fn_name, runtime=LAMBDA_RUNTIME_NODEJS12X475 )476 lambda_arn = aws_stack.lambda_function_arn(fn_name)477 spec_file = load_file(TEST_IMPORT_REST_API_ASYNC_LAMBDA)478 spec_file = spec_file.replace("${lambda_invocation_arn}", lambda_arn)479 apigateway_client.put_rest_api(restApiId=rest_api_id, body=spec_file, mode="overwrite")480 url = path_based_url(api_id=rest_api_id, stage_name="latest", path="/wait/3")481 result = requests.get(url)482 assert result.status_code == 200483 assert result.content == b""484 def test_api_gateway_mock_integration(self, apigateway_client):485 rest_api_name = f"apigw-{short_uid()}"486 rest_api_id = apigateway_client.create_rest_api(name=rest_api_name)["id"]487 spec_file = load_file(TEST_IMPORT_MOCK_INTEGRATION)488 apigateway_client.put_rest_api(restApiId=rest_api_id, body=spec_file, mode="overwrite")489 url = path_based_url(api_id=rest_api_id, stage_name="latest", path="/echo/foobar")490 response = requests.get(url)491 assert response._content == b'{"echo": "foobar", "response": "mocked"}'492 def test_api_gateway_authorizer_crud(self):493 apig = aws_stack.create_external_boto_client("apigateway")494 authorizer = apig.create_authorizer(495 restApiId=self.TEST_API_GATEWAY_ID, **self.TEST_API_GATEWAY_AUTHORIZER496 )497 authorizer_id = authorizer.get("id")498 create_result = apig.get_authorizer(499 restApiId=self.TEST_API_GATEWAY_ID, authorizerId=authorizer_id500 )501 # ignore boto3 stuff502 del create_result["ResponseMetadata"]503 create_expected = clone(self.TEST_API_GATEWAY_AUTHORIZER)504 create_expected["id"] = authorizer_id505 assert create_expected == create_result506 apig.update_authorizer(507 restApiId=self.TEST_API_GATEWAY_ID,508 authorizerId=authorizer_id,509 patchOperations=self.TEST_API_GATEWAY_AUTHORIZER_OPS,510 )511 update_result = apig.get_authorizer(512 restApiId=self.TEST_API_GATEWAY_ID, authorizerId=authorizer_id513 )514 # ignore boto3 stuff515 del update_result["ResponseMetadata"]516 update_expected = apply_patch(create_expected, self.TEST_API_GATEWAY_AUTHORIZER_OPS)517 assert update_expected == update_result518 apig.delete_authorizer(restApiId=self.TEST_API_GATEWAY_ID, authorizerId=authorizer_id)519 with pytest.raises(Exception):520 apig.get_authorizer(self.TEST_API_GATEWAY_ID, authorizer_id)521 def test_apigateway_with_lambda_integration(self):522 apigw_client = aws_stack.create_external_boto_client("apigateway")523 # create Lambda function524 lambda_name = f"apigw-lambda-{short_uid()}"525 self.create_lambda_function(lambda_name)526 lambda_uri = aws_stack.lambda_function_arn(lambda_name)527 target_uri = aws_stack.apigateway_invocations_arn(lambda_uri)528 # create REST API529 api = apigw_client.create_rest_api(name="test-api", description="")530 api_id = api["id"]531 root_res_id = apigw_client.get_resources(restApiId=api_id)["items"][0]["id"]532 api_resource = apigw_client.create_resource(533 restApiId=api_id, parentId=root_res_id, pathPart="test"534 )535 apigw_client.put_method(536 restApiId=api_id,537 resourceId=api_resource["id"],538 httpMethod="GET",539 authorizationType="NONE",540 )541 rs = apigw_client.put_integration(542 restApiId=api_id,543 resourceId=api_resource["id"],544 httpMethod="GET",545 integrationHttpMethod="POST",546 type="AWS",547 uri=target_uri,548 timeoutInMillis=3000,549 contentHandling="CONVERT_TO_BINARY",550 requestTemplates={"application/json": '{"param1": "$input.params(\'param1\')"}'},551 )552 integration_keys = [553 "httpMethod",554 "type",555 "cacheKeyParameters",556 "uri",557 "cacheNamespace",558 "timeoutInMillis",559 "contentHandling",560 "requestParameters",561 ]562 assert 200 == rs["ResponseMetadata"]["HTTPStatusCode"]563 for key in integration_keys:564 assert key in rs565 assert "responseTemplates" not in rs566 apigw_client.create_deployment(restApiId=api_id, stageName=self.TEST_STAGE_NAME)567 rs = apigw_client.get_integration(568 restApiId=api_id, resourceId=api_resource["id"], httpMethod="GET"569 )570 assert 200 == rs["ResponseMetadata"]["HTTPStatusCode"]571 assert "AWS" == rs["type"]572 assert "POST" == rs["httpMethod"]573 assert target_uri == rs["uri"]574 # invoke the gateway endpoint575 url = path_based_url(api_id=api_id, stage_name=self.TEST_STAGE_NAME, path="/test")576 response = requests.get(f"{url}?param1=foobar")577 assert response.status_code < 400578 content = response.json()579 assert '{"param1": "foobar"}' == content.get("body").get("body")580 # additional checks from https://github.com/localstack/localstack/issues/5041581 # pass Signature param582 response = requests.get(f"{url}?param1=foobar&Signature=1")583 assert response.status_code == 200584 content = response.json()585 assert '{"param1": "foobar"}' == content.get("body").get("body")586 assert {"Signature": "1", "param1": "foobar"} == content.get("body").get(587 "queryStringParameters"588 )589 # delete integration590 rs = apigw_client.delete_integration(591 restApiId=api_id,592 resourceId=api_resource["id"],593 httpMethod="GET",594 )595 assert 204 == rs["ResponseMetadata"]["HTTPStatusCode"]596 with pytest.raises(ClientError) as ctx:597 # This call should not be successful as the integration is deleted598 apigw_client.get_integration(599 restApiId=api_id, resourceId=api_resource["id"], httpMethod="GET"600 )601 assert ctx.value.response["Error"]["Code"] == "NotFoundException"602 # clean up603 lambda_client = aws_stack.create_external_boto_client("lambda")604 lambda_client.delete_function(FunctionName=lambda_name)605 apigw_client.delete_rest_api(restApiId=api_id)606 def test_api_gateway_handle_domain_name(self):607 domain_name = "%s.example.com" % short_uid()608 apigw_client = aws_stack.create_external_boto_client("apigateway")609 rs = apigw_client.create_domain_name(domainName=domain_name)610 assert 200 == rs["ResponseMetadata"]["HTTPStatusCode"]611 rs = apigw_client.get_domain_name(domainName=domain_name)612 assert 200 == rs["ResponseMetadata"]["HTTPStatusCode"]613 assert domain_name == rs["domainName"]614 # clean up615 apigw_client.delete_domain_name(domainName=domain_name)616 def _test_api_gateway_lambda_proxy_integration_any_method(self, fn_name, path):617 self.create_lambda_function(fn_name)618 # create API Gateway and connect it to the Lambda proxy backend619 lambda_uri = aws_stack.lambda_function_arn(fn_name)620 target_uri = aws_stack.apigateway_invocations_arn(lambda_uri)621 result = testutil.connect_api_gateway_to_http_with_lambda_proxy(622 "test_gateway3",623 target_uri,624 methods=["ANY"],625 path=path,626 stage_name=self.TEST_STAGE_NAME,627 )628 # make test request to gateway and check response629 path = path.replace("{test_param1}", "foo1")630 url = path_based_url(api_id=result["id"], stage_name=self.TEST_STAGE_NAME, path=path)631 data = {}632 for method in ("GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"):633 body = json.dumps(data) if method in ("POST", "PUT", "PATCH") else None634 result = getattr(requests, method.lower())(url, data=body)635 if method != "DELETE":636 assert 200 == result.status_code637 parsed_body = json.loads(to_str(result.content))638 assert method == parsed_body.get("httpMethod")639 else:640 assert 204 == result.status_code641 def test_apigateway_with_custom_authorization_method(self):642 apigw_client = aws_stack.create_external_boto_client("apigateway")643 # create Lambda function644 lambda_name = "apigw-lambda-%s" % short_uid()645 self.create_lambda_function(lambda_name)646 lambda_uri = aws_stack.lambda_function_arn(lambda_name)647 # create REST API648 api = apigw_client.create_rest_api(name="test-api", description="")649 api_id = api["id"]650 root_res_id = apigw_client.get_resources(restApiId=api_id)["items"][0]["id"]651 # create authorizer at root resource652 authorizer = apigw_client.create_authorizer(653 restApiId=api_id,654 name="lambda_authorizer",655 type="TOKEN",656 authorizerUri="arn:aws:apigateway:us-east-1:lambda:path/ \657 2015-03-31/functions/{}/invocations".format(658 lambda_uri659 ),660 identitySource="method.request.header.Auth",661 )662 # create method with custom authorizer663 is_api_key_required = True664 method_response = apigw_client.put_method(665 restApiId=api_id,666 resourceId=root_res_id,667 httpMethod="GET",668 authorizationType="CUSTOM",669 authorizerId=authorizer["id"],670 apiKeyRequired=is_api_key_required,671 )672 assert authorizer["id"] == method_response["authorizerId"]673 # clean up674 lambda_client = aws_stack.create_external_boto_client("lambda")675 lambda_client.delete_function(FunctionName=lambda_name)676 apigw_client.delete_rest_api(restApiId=api_id)677 def test_create_model(self):678 client = aws_stack.create_external_boto_client("apigateway")679 response = client.create_rest_api(name="my_api", description="this is my api")680 rest_api_id = response["id"]681 dummy_rest_api_id = "_non_existing_"682 model_name = "testModel"683 description = "test model"684 content_type = "application/json"685 # success case with valid params686 response = client.create_model(687 restApiId=rest_api_id,688 name=model_name,689 description=description,690 contentType=content_type,691 )692 assert model_name == response["name"]693 assert description == response["description"]694 with pytest.raises(Exception) as ctx:695 client.create_model(696 restApiId=dummy_rest_api_id,697 name=model_name,698 description=description,699 contentType=content_type,700 )701 assert "NotFoundException" == ctx.value.response["Error"]["Code"]702 assert "Invalid Rest API Id specified" == ctx.value.response["Error"]["Message"]703 with pytest.raises(Exception) as ctx:704 client.create_model(705 restApiId=dummy_rest_api_id,706 name="",707 description=description,708 contentType=content_type,709 )710 assert "BadRequestException" == ctx.value.response["Error"]["Code"]711 assert "No Model Name specified" == ctx.value.response["Error"]["Message"]712 # clean up713 client.delete_rest_api(restApiId=rest_api_id)714 def test_get_api_models(self):715 client = aws_stack.create_external_boto_client("apigateway")716 response = client.create_rest_api(name="my_api", description="this is my api")717 rest_api_id = response["id"]718 model_name = "testModel"719 description = "test model"720 content_type = "application/json"721 # when no models are present722 result = client.get_models(restApiId=rest_api_id)723 assert [] == result["items"]724 # add a model725 client.create_model(726 restApiId=rest_api_id,727 name=model_name,728 description=description,729 contentType=content_type,730 )731 # get models after adding732 result = client.get_models(restApiId=rest_api_id)733 assert model_name == result["items"][0]["name"]734 assert description == result["items"][0]["description"]735 # clean up736 client.delete_rest_api(restApiId=rest_api_id)737 def test_request_validator(self):738 client = aws_stack.create_external_boto_client("apigateway")739 response = client.create_rest_api(name="my_api", description="this is my api")740 rest_api_id = response["id"]741 # CREATE742 name = "validator123"743 result = client.create_request_validator(restApiId=rest_api_id, name=name)744 assert 201 == result["ResponseMetadata"]["HTTPStatusCode"]745 validator_id = result["id"]746 # LIST747 result = client.get_request_validators(restApiId=rest_api_id)748 assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]749 assert [{"id": validator_id, "name": name}] == result["items"]750 # GET751 result = client.get_request_validator(752 restApiId=rest_api_id, requestValidatorId=validator_id753 )754 assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]755 assert select_attributes(result, ["id", "name"]) == {"id": validator_id, "name": name}756 # UPDATE757 result = client.update_request_validator(758 restApiId=rest_api_id, requestValidatorId=validator_id, patchOperations=[]759 )760 # DELETE761 client.delete_request_validator(restApiId=rest_api_id, requestValidatorId=validator_id)762 with pytest.raises(Exception):763 client.get_request_validator(restApiId=rest_api_id, requestValidatorId=validator_id)764 with pytest.raises(Exception):765 client.delete_request_validator(restApiId=rest_api_id, requestValidatorId=validator_id)766 # clean up767 client.delete_rest_api(restApiId=rest_api_id)768 def test_base_path_mapping(self):769 client = aws_stack.create_external_boto_client("apigateway")770 response = client.create_rest_api(name="my_api", description="this is my api")771 rest_api_id = response["id"]772 # CREATE773 domain_name = "domain1.example.com"774 client.create_domain_name(domainName=domain_name)775 root_res_id = client.get_resources(restApiId=rest_api_id)["items"][0]["id"]776 res_id = client.create_resource(777 restApiId=rest_api_id, parentId=root_res_id, pathPart="path"778 )["id"]779 client.put_method(780 restApiId=rest_api_id, resourceId=res_id, httpMethod="GET", authorizationType="NONE"781 )782 client.put_integration(783 restApiId=rest_api_id, resourceId=res_id, httpMethod="GET", type="MOCK"784 )785 depl_id = client.create_deployment(restApiId=rest_api_id)["id"]786 client.create_stage(restApiId=rest_api_id, deploymentId=depl_id, stageName="dev")787 base_path = "foo"788 result = client.create_base_path_mapping(789 domainName=domain_name,790 basePath=base_path,791 restApiId=rest_api_id,792 stage="dev",793 )794 assert result["ResponseMetadata"]["HTTPStatusCode"] in [200, 201]795 # LIST796 result = client.get_base_path_mappings(domainName=domain_name)797 assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]798 expected = {"basePath": base_path, "restApiId": rest_api_id, "stage": "dev"}799 assert [expected] == result["items"]800 # GET801 result = client.get_base_path_mapping(domainName=domain_name, basePath=base_path)802 assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]803 assert expected == select_attributes(result, ["basePath", "restApiId", "stage"])804 # UPDATE805 result = client.update_base_path_mapping(806 domainName=domain_name, basePath=base_path, patchOperations=[]807 )808 assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]809 # DELETE810 client.delete_base_path_mapping(domainName=domain_name, basePath=base_path)811 with pytest.raises(Exception):812 client.get_base_path_mapping(domainName=domain_name, basePath=base_path)813 with pytest.raises(Exception):814 client.delete_base_path_mapping(domainName=domain_name, basePath=base_path)815 def test_base_path_mapping_root(self):816 client = aws_stack.create_external_boto_client("apigateway")817 response = client.create_rest_api(name="my_api2", description="this is my api")818 rest_api_id = response["id"]819 # CREATE820 domain_name = "domain2.example.com"821 client.create_domain_name(domainName=domain_name)822 root_res_id = client.get_resources(restApiId=rest_api_id)["items"][0]["id"]823 res_id = client.create_resource(824 restApiId=rest_api_id, parentId=root_res_id, pathPart="path"825 )["id"]826 client.put_method(827 restApiId=rest_api_id, resourceId=res_id, httpMethod="GET", authorizationType="NONE"828 )829 client.put_integration(830 restApiId=rest_api_id, resourceId=res_id, httpMethod="GET", type="MOCK"831 )832 depl_id = client.create_deployment(restApiId=rest_api_id)["id"]833 client.create_stage(restApiId=rest_api_id, deploymentId=depl_id, stageName="dev")834 result = client.create_base_path_mapping(835 domainName=domain_name,836 basePath="",837 restApiId=rest_api_id,838 stage="dev",839 )840 assert result["ResponseMetadata"]["HTTPStatusCode"] in [200, 201]841 base_path = "(none)"842 # LIST843 result = client.get_base_path_mappings(domainName=domain_name)844 assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]845 expected = {"basePath": "(none)", "restApiId": rest_api_id, "stage": "dev"}846 assert [expected] == result["items"]847 # GET848 result = client.get_base_path_mapping(domainName=domain_name, basePath=base_path)849 assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]850 assert expected == select_attributes(result, ["basePath", "restApiId", "stage"])851 # UPDATE852 result = client.update_base_path_mapping(853 domainName=domain_name, basePath=base_path, patchOperations=[]854 )855 assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]856 # DELETE857 client.delete_base_path_mapping(domainName=domain_name, basePath=base_path)858 with pytest.raises(Exception):859 client.get_base_path_mapping(domainName=domain_name, basePath=base_path)860 with pytest.raises(Exception):861 client.delete_base_path_mapping(domainName=domain_name, basePath=base_path)862 def test_api_account(self):863 client = aws_stack.create_external_boto_client("apigateway")864 response = client.create_rest_api(name="my_api", description="test 123")865 rest_api_id = response["id"]866 result = client.get_account()867 assert "UsagePlans" in result["features"]868 result = client.update_account(869 patchOperations=[{"op": "add", "path": "/features/-", "value": "foobar"}]870 )871 assert "foobar" in result["features"]872 # clean up873 client.delete_rest_api(restApiId=rest_api_id)874 def test_get_model_by_name(self):875 client = aws_stack.create_external_boto_client("apigateway")876 response = client.create_rest_api(name="my_api", description="this is my api")877 rest_api_id = response["id"]878 dummy_rest_api_id = "_non_existing_"879 model_name = "testModel"880 description = "test model"881 content_type = "application/json"882 # add a model883 client.create_model(884 restApiId=rest_api_id,885 name=model_name,886 description=description,887 contentType=content_type,888 )889 # get models after adding890 result = client.get_model(restApiId=rest_api_id, modelName=model_name)891 assert model_name == result["name"]892 assert description == result["description"]893 try:894 client.get_model(restApiId=dummy_rest_api_id, modelName=model_name)895 self.fail("This call should not be successful as the model is not created.")896 except ClientError as e:897 assert "NotFoundException" == e.response["Error"]["Code"]898 assert "Invalid Rest API Id specified" == e.response["Error"]["Message"]899 def test_get_model_with_invalid_name(self):900 client = aws_stack.create_external_boto_client("apigateway")901 response = client.create_rest_api(name="my_api", description="this is my api")902 rest_api_id = response["id"]903 # test with an invalid model name904 try:905 client.get_model(restApiId=rest_api_id, modelName="fake")906 self.fail("This call should not be successful as the model is not created.")907 except ClientError as e:908 assert "NotFoundException" == e.response["Error"]["Code"]909 # clean up910 client.delete_rest_api(restApiId=rest_api_id)911 def test_put_integration_dynamodb_proxy_validation_without_response_template(self):912 api_id = self.create_api_gateway_and_deploy({})913 url = path_based_url(api_id=api_id, stage_name="staging", path="/")914 response = requests.put(915 url,916 json.dumps({"id": "id1", "data": "foobar123"}),917 )918 assert 404 == response.status_code919 def test_put_integration_dynamodb_proxy_validation_with_response_template(self):920 response_templates = {921 "application/json": json.dumps(922 {923 "TableName": "MusicCollection",924 "Item": {"id": "$.Id", "data": "$.data"},925 }926 )927 }928 api_id = self.create_api_gateway_and_deploy(response_templates)929 url = path_based_url(api_id=api_id, stage_name="staging", path="/")930 response = requests.put(931 url,932 json.dumps({"id": "id1", "data": "foobar123"}),933 )934 assert 200 == response.status_code935 dynamo_client = aws_stack.connect_to_resource("dynamodb")936 table = dynamo_client.Table("MusicCollection")937 result = table.get_item(Key={"id": "id1"})938 assert "foobar123" == result["Item"]["data"]939 def test_api_key_required_for_methods(self):940 response_templates = {941 "application/json": json.dumps(942 {943 "TableName": "MusicCollection",944 "Item": {"id": "$.Id", "data": "$.data"},945 }946 )947 }948 api_id = self.create_api_gateway_and_deploy(response_templates, True)949 url = path_based_url(api_id=api_id, stage_name="staging", path="/")950 payload = {951 "name": "TEST-PLAN-2",952 "description": "Description",953 "quota": {"limit": 10, "period": "DAY", "offset": 0},954 "throttle": {"rateLimit": 2, "burstLimit": 1},955 "apiStages": [{"apiId": api_id, "stage": "staging"}],956 "tags": {"tag_key": "tag_value"},957 }958 client = aws_stack.create_external_boto_client("apigateway")959 usage_plan_id = client.create_usage_plan(**payload)["id"]960 key_name = "testApiKey"961 key_type = "API_KEY"962 api_key = client.create_api_key(name=key_name)963 payload = {964 "usagePlanId": usage_plan_id,965 "keyId": api_key["id"],966 "keyType": key_type,967 }968 client.create_usage_plan_key(**payload)969 response = requests.put(970 url,971 json.dumps({"id": "id1", "data": "foobar123"}),972 )973 # when the api key is not passed as part of the header974 assert 403 == response.status_code975 response = requests.put(976 url,977 json.dumps({"id": "id1", "data": "foobar123"}),978 headers={"X-API-Key": api_key["value"]},979 )980 # when the api key is passed as part of the header981 assert 200 == response.status_code982 def test_multiple_api_keys_validate(self):983 response_templates = {984 "application/json": json.dumps(985 {986 "TableName": "MusicCollection",987 "Item": {"id": "$.Id", "data": "$.data"},988 }989 )990 }991 api_id = self.create_api_gateway_and_deploy(response_templates, True)992 url = path_based_url(api_id=api_id, stage_name="staging", path="/")993 client = aws_stack.create_external_boto_client("apigateway")994 # Create multiple usage plans995 usage_plan_ids = []996 for i in range(2):997 payload = {998 "name": "APIKEYTEST-PLAN-{}".format(i),999 "description": "Description",1000 "quota": {"limit": 10, "period": "DAY", "offset": 0},1001 "throttle": {"rateLimit": 2, "burstLimit": 1},1002 "apiStages": [{"apiId": api_id, "stage": "staging"}],1003 "tags": {"tag_key": "tag_value"},1004 }1005 usage_plan_ids.append(client.create_usage_plan(**payload)["id"])1006 api_keys = []1007 key_type = "API_KEY"1008 # Create multiple API Keys in each usage plan1009 for usage_plan_id in usage_plan_ids:1010 for i in range(2):1011 api_key = client.create_api_key(name="testMultipleApiKeys{}".format(i))1012 payload = {1013 "usagePlanId": usage_plan_id,1014 "keyId": api_key["id"],1015 "keyType": key_type,1016 }1017 client.create_usage_plan_key(**payload)1018 api_keys.append(api_key["value"])1019 response = requests.put(1020 url,1021 json.dumps({"id": "id1", "data": "foobar123"}),1022 )1023 # when the api key is not passed as part of the header1024 assert 403 == response.status_code1025 # check that all API keys work1026 for key in api_keys:1027 response = requests.put(1028 url,1029 json.dumps({"id": "id1", "data": "foobar123"}),1030 headers={"X-API-Key": key},1031 )1032 # when the api key is passed as part of the header1033 assert 200 == response.status_code1034 @pytest.mark.parametrize("base_path_type", ["ignore", "prepend", "split"])1035 def test_import_rest_api(self, base_path_type):1036 rest_api_name = f"restapi-{short_uid()}"1037 client = aws_stack.create_external_boto_client("apigateway")1038 rest_api_id = client.create_rest_api(name=rest_api_name)["id"]1039 spec_file = load_file(TEST_SWAGGER_FILE_JSON)1040 api_params = {"basepath": base_path_type}1041 rs = client.put_rest_api(1042 restApiId=rest_api_id, body=spec_file, mode="overwrite", parameters=api_params1043 )1044 assert 200 == rs["ResponseMetadata"]["HTTPStatusCode"]1045 resources = client.get_resources(restApiId=rest_api_id)1046 for rv in resources.get("items"):1047 for method in rv.get("resourceMethods", {}).values():1048 assert method.get("authorizationType") == "request"1049 assert method.get("authorizerId") is not None1050 spec_file = load_file(TEST_SWAGGER_FILE_YAML)1051 rs = client.put_rest_api(1052 restApiId=rest_api_id, body=spec_file, mode="overwrite", parameters=api_params1053 )1054 assert 200 == rs["ResponseMetadata"]["HTTPStatusCode"]1055 rs = client.get_resources(restApiId=rest_api_id)1056 expected_resources = 2 if base_path_type == "ignore" else 31057 assert len(rs["items"]) == expected_resources1058 abs_path = "/test" if base_path_type == "ignore" else "/base/test"1059 resource = [res for res in rs["items"] if res["path"] == abs_path][0]1060 assert "GET" in resource["resourceMethods"]1061 assert "requestParameters" in resource["resourceMethods"]["GET"]1062 assert {"integration.request.header.X-Amz-Invocation-Type": "'Event'"} == resource[1063 "resourceMethods"1064 ]["GET"]["requestParameters"]1065 url = path_based_url(api_id=rest_api_id, stage_name="dev", path=abs_path)1066 response = requests.get(url)1067 assert 200 == response.status_code1068 # clean up1069 client.delete_rest_api(restApiId=rest_api_id)1070 spec_file = load_file(TEST_IMPORT_REST_API_FILE)1071 rs = client.import_rest_api(body=spec_file, parameters=api_params)1072 assert 201 == rs["ResponseMetadata"]["HTTPStatusCode"]1073 rest_api_id = rs["id"]1074 rs = client.get_resources(restApiId=rest_api_id)1075 resources = rs["items"]1076 assert 3 == len(resources)1077 paths = [res["path"] for res in resources]1078 assert "/" in paths1079 assert "/pets" in paths1080 assert "/pets/{petId}" in paths1081 # clean up1082 client.delete_rest_api(restApiId=rest_api_id)1083 def test_step_function_integrations(self):1084 client = aws_stack.create_external_boto_client("apigateway")1085 sfn_client = aws_stack.create_external_boto_client("stepfunctions")1086 lambda_client = aws_stack.create_external_boto_client("lambda")1087 state_machine_name = "test"1088 state_machine_def = {1089 "Comment": "Hello World example",1090 "StartAt": "step1",1091 "States": {1092 "step1": {"Type": "Task", "Resource": "__tbd__", "End": True},1093 },1094 }1095 # create state machine1096 fn_name = "test-stepfunctions-apigw"1097 testutil.create_lambda_function(1098 handler_file=TEST_LAMBDA_PYTHON_ECHO,1099 func_name=fn_name,1100 runtime=LAMBDA_RUNTIME_PYTHON36,1101 )1102 role_arn = aws_stack.role_arn("sfn_role")1103 # create state machine definition1104 definition = clone(state_machine_def)1105 lambda_arn_1 = aws_stack.lambda_function_arn(fn_name)1106 definition["States"]["step1"]["Resource"] = lambda_arn_11107 definition = json.dumps(definition)1108 # create state machine1109 result = sfn_client.create_state_machine(1110 name=state_machine_name, definition=definition, roleArn=role_arn1111 )1112 sm_arn = result["stateMachineArn"]1113 # create REST API and method1114 rest_api = client.create_rest_api(name="test", description="test")1115 resources = client.get_resources(restApiId=rest_api["id"])1116 root_resource_id = resources["items"][0]["id"]1117 client.put_method(1118 restApiId=rest_api["id"],1119 resourceId=root_resource_id,1120 httpMethod="POST",1121 authorizationType="NONE",1122 )1123 def _prepare_method_integration(1124 integr_kwargs=None, resp_templates=None, action="StartExecution", overwrite=False1125 ):1126 if integr_kwargs is None:1127 integr_kwargs = {}1128 if resp_templates is None:1129 resp_templates = {}1130 if overwrite:1131 client.delete_integration(1132 restApiId=rest_api["id"],1133 resourceId=resources["items"][0]["id"],1134 httpMethod="POST",1135 )1136 uri = f"arn:aws:apigateway:{aws_stack.get_region()}:states:action/{action}"1137 client.put_integration(1138 restApiId=rest_api["id"],1139 resourceId=root_resource_id,1140 httpMethod="POST",1141 integrationHttpMethod="POST",1142 type="AWS",1143 uri=uri,1144 **integr_kwargs,1145 )1146 if resp_templates:1147 client.put_integration_response(1148 restApiId=rest_api["id"],1149 resourceId=root_resource_id,1150 selectionPattern="",1151 responseTemplates=resp_templates,1152 httpMethod="POST",1153 statusCode="200",1154 )1155 # STEP 1: test integration with request template1156 _prepare_method_integration(1157 integr_kwargs={1158 "requestTemplates": {1159 "application/json": """1160 #set($data = $util.escapeJavaScript($input.json('$')))1161 {"input": $data, "stateMachineArn": "%s"}1162 """1163 % sm_arn1164 }1165 }1166 )1167 # invoke stepfunction via API GW, assert results1168 client.create_deployment(restApiId=rest_api["id"], stageName="dev")1169 url = path_based_url(api_id=rest_api["id"], stage_name="dev", path="/")1170 test_data = {"test": "test-value"}1171 resp = requests.post(url, data=json.dumps(test_data))1172 assert 200 == resp.status_code1173 assert "executionArn" in resp.content.decode()1174 assert "startDate" in resp.content.decode()1175 # STEP 2: test integration without request template1176 _prepare_method_integration(overwrite=True)1177 test_data_1 = {1178 "input": json.dumps(test_data),1179 "name": "MyExecution",1180 "stateMachineArn": sm_arn,1181 }1182 # invoke stepfunction via API GW, assert results1183 resp = requests.post(url, data=json.dumps(test_data_1))1184 assert 200 == resp.status_code1185 assert "executionArn" in resp.content.decode()1186 assert "startDate" in resp.content.decode()1187 # STEP 3: test integration with synchronous execution1188 _prepare_method_integration(overwrite=True, action="StartSyncExecution")1189 # invoke stepfunction via API GW, assert results1190 test_data_1["name"] += "1"1191 resp = requests.post(url, data=json.dumps(test_data_1))1192 assert 200 == resp.status_code1193 content = json.loads(to_str(resp.content.decode()))1194 assert "SUCCEEDED" == content.get("status")1195 assert test_data == json.loads(content.get("output"))1196 # STEP 4: test integration with synchronous execution and response templates1197 resp_templates = {APPLICATION_JSON: "$input.path('$.output')"}1198 _prepare_method_integration(1199 resp_templates=resp_templates, overwrite=True, action="StartSyncExecution"1200 )1201 # invoke stepfunction via API GW, assert results1202 test_data_1["name"] += "2"1203 resp = requests.post(url, data=json.dumps(test_data_1))1204 assert 200 == resp.status_code1205 assert test_data == json.loads(to_str(resp.content.decode()))1206 _prepare_method_integration(overwrite=True, action="DeleteStateMachine")1207 # Remove state machine with API GW1208 resp = requests.post(url, data=json.dumps({"stateMachineArn": sm_arn}))1209 assert 200 == resp.status_code1210 # Clean up1211 lambda_client.delete_function(FunctionName=fn_name)1212 client.delete_rest_api(restApiId=rest_api["id"])1213 def test_api_gateway_http_integration_with_path_request_parameter(self):1214 client = aws_stack.create_external_boto_client("apigateway")1215 test_port = get_free_tcp_port()1216 backend_url = "http://localhost:%s/person/{id}" % test_port1217 # start test HTTP backend1218 proxy = self.start_http_backend(test_port)1219 # create rest api1220 api_rest = client.create_rest_api(name="test")1221 api_id = api_rest["id"]1222 parent_response = client.get_resources(restApiId=api_id)1223 parent_id = parent_response["items"][0]["id"]1224 resource_1 = client.create_resource(restApiId=api_id, parentId=parent_id, pathPart="person")1225 resource_1_id = resource_1["id"]1226 resource_2 = client.create_resource(1227 restApiId=api_id, parentId=resource_1_id, pathPart="{id}"1228 )1229 resource_2_id = resource_2["id"]1230 client.put_method(1231 restApiId=api_id,1232 resourceId=resource_2_id,1233 httpMethod="GET",1234 authorizationType="NONE",1235 apiKeyRequired=False,1236 requestParameters={"method.request.path.id": True},1237 )1238 client.put_integration(1239 restApiId=api_id,1240 resourceId=resource_2_id,1241 httpMethod="GET",1242 integrationHttpMethod="GET",1243 type="HTTP",1244 uri=backend_url,1245 timeoutInMillis=3000,1246 contentHandling="CONVERT_TO_BINARY",1247 requestParameters={"integration.request.path.id": "method.request.path.id"},1248 )1249 client.create_deployment(restApiId=api_id, stageName="test")1250 def _test_invoke(url):1251 result = requests.get(url)1252 content = json.loads(to_str(result.content))1253 assert 200 == result.status_code1254 assert re.search(1255 "http://.*localhost.*/person/123",1256 content["headers"].get(HEADER_LOCALSTACK_REQUEST_URL),1257 )1258 for use_hostname in [True, False]:1259 for use_ssl in [True, False] if use_hostname else [False]:1260 url = self._get_invoke_endpoint(1261 api_id,1262 stage="test",1263 path="/person/123",1264 use_hostname=use_hostname,1265 use_ssl=use_ssl,1266 )1267 _test_invoke(url)1268 # clean up1269 client.delete_rest_api(restApiId=api_id)1270 proxy.stop()1271 def _get_invoke_endpoint(1272 self, api_id, stage="test", path="/", use_hostname=False, use_ssl=False1273 ):1274 path = path or "/"1275 path = path if path.startswith(path) else f"/{path}"1276 proto = "https" if use_ssl else "http"1277 if use_hostname:1278 return f"{proto}://{api_id}.execute-api.{LOCALHOST_HOSTNAME}:{config.EDGE_PORT}/{stage}{path}"1279 return (1280 f"{proto}://localhost:{config.EDGE_PORT}/restapis/{api_id}/{stage}/_user_request_{path}"1281 )1282 def test_api_gateway_s3_get_integration(self):1283 apigw_client = aws_stack.create_external_boto_client("apigateway")1284 s3_client = aws_stack.create_external_boto_client("s3")1285 bucket_name = f"test-bucket-{short_uid()}"1286 apigateway_name = f"test-api-{short_uid()}"1287 object_name = "test.json"1288 object_content = '{ "success": "true" }'1289 object_content_type = "application/json"1290 api = apigw_client.create_rest_api(name=apigateway_name)1291 api_id = api["id"]1292 try:1293 aws_stack.get_or_create_bucket(bucket_name)1294 s3_client.put_object(1295 Bucket=bucket_name,1296 Key=object_name,1297 Body=object_content,1298 ContentType=object_content_type,1299 )1300 self.connect_api_gateway_to_s3(bucket_name, object_name, api_id, "GET")1301 apigw_client.create_deployment(restApiId=api_id, stageName="test")1302 url = path_based_url(api_id, "test", f"/{object_name}")1303 result = requests.get(url)1304 assert 200 == result.status_code1305 assert object_content == result.text1306 assert object_content_type == result.headers["content-type"]1307 finally:1308 # clean up1309 apigw_client.delete_rest_api(restApiId=api_id)1310 s3_client.delete_object(Bucket=bucket_name, Key=object_name)1311 s3_client.delete_bucket(Bucket=bucket_name)1312 def test_api_mock_integration_response_params(self):1313 resps = [1314 {1315 "statusCode": "204",1316 "httpMethod": "OPTIONS",1317 "responseParameters": {1318 "method.response.header.Access-Control-Allow-Methods": "'POST,OPTIONS'",1319 "method.response.header.Vary": "'Origin'",1320 },1321 }1322 ]1323 api_id = self.create_api_gateway_and_deploy(1324 integration_type="MOCK", integration_responses=resps1325 )1326 url = path_based_url(api_id=api_id, stage_name=self.TEST_STAGE_NAME, path="/")1327 result = requests.options(url)1328 assert result.status_code < 4001329 assert "Origin" == result.headers.get("vary")1330 assert "POST,OPTIONS" == result.headers.get("Access-Control-Allow-Methods")1331 def test_api_gateway_update_resource_path_part(self):1332 apigw_client = aws_stack.connect_to_service("apigateway")1333 api = apigw_client.create_rest_api(name="test-api", description="")1334 api_id = api["id"]1335 root_res_id = apigw_client.get_resources(restApiId=api_id)["items"][0]["id"]1336 api_resource = apigw_client.create_resource(1337 restApiId=api_id, parentId=root_res_id, pathPart="test"1338 )1339 response = apigw_client.update_resource(1340 restApiId=api_id,1341 resourceId=api_resource.get("id"),1342 patchOperations=[1343 {"op": "replace", "path": "/pathPart", "value": "demo1"},1344 ],1345 )1346 assert response.get("pathPart") == "demo1"1347 response = apigw_client.get_resource(restApiId=api_id, resourceId=api_resource.get("id"))1348 assert response.get("pathPart") == "demo1"1349 # clean up1350 apigw_client.delete_rest_api(restApiId=api_id)1351 def test_response_headers_invocation_with_apigw(self, create_lambda_function, lambda_client):1352 lambda_name = f"test_lambda_{short_uid()}"1353 lambda_resource = "/api/v1/{proxy+}"1354 lambda_path = "/api/v1/hello/world"1355 create_lambda_function(1356 func_name=lambda_name,1357 zip_file=testutil.create_zip_file(1358 TEST_LAMBDA_NODEJS_APIGW_INTEGRATION, get_content=True1359 ),1360 runtime=LAMBDA_RUNTIME_NODEJS14X,1361 handler="apigw_integration.handler",1362 )1363 lambda_uri = aws_stack.lambda_function_arn(lambda_name)1364 target_uri = f"arn:aws:apigateway:{aws_stack.get_region()}:lambda:path/2015-03-31/functions/{lambda_uri}/invocations"1365 result = testutil.connect_api_gateway_to_http_with_lambda_proxy(1366 "test_gateway",1367 target_uri,1368 path=lambda_resource,1369 stage_name=TEST_STAGE_NAME,1370 )1371 api_id = result["id"]1372 url = path_based_url(api_id=api_id, stage_name=TEST_STAGE_NAME, path=lambda_path)1373 result = requests.get(url)1374 assert result.status_code == 3001375 assert result.headers["Content-Type"] == "application/xml"1376 body = xmltodict.parse(result.content)1377 assert body.get("message") == "completed"1378 # =====================================================================1379 # Helper methods1380 # =====================================================================1381 def connect_api_gateway_to_s3(self, bucket_name, file_name, api_id, method):1382 """Connects the root resource of an api gateway to the given object of an s3 bucket."""1383 apigw_client = aws_stack.create_external_boto_client("apigateway")1384 s3_uri = "arn:aws:apigateway:{}:s3:path/{}/{{proxy}}".format(1385 aws_stack.get_region(), bucket_name1386 )1387 test_role = "test-s3-role"1388 role_arn = aws_stack.role_arn(role_name=test_role)1389 resources = apigw_client.get_resources(restApiId=api_id)1390 # using the root resource '/' directly for this test1391 root_resource_id = resources["items"][0]["id"]1392 proxy_resource = apigw_client.create_resource(1393 restApiId=api_id, parentId=root_resource_id, pathPart="{proxy+}"1394 )1395 apigw_client.put_method(1396 restApiId=api_id,1397 resourceId=proxy_resource["id"],1398 httpMethod=method,1399 authorizationType="NONE",1400 apiKeyRequired=False,1401 requestParameters={},1402 )1403 apigw_client.put_integration(1404 restApiId=api_id,1405 resourceId=proxy_resource["id"],1406 httpMethod=method,1407 type="AWS",1408 integrationHttpMethod=method,1409 uri=s3_uri,1410 credentials=role_arn,1411 requestParameters={"integration.request.path.proxy": "method.request.path.proxy"},1412 )1413 def connect_api_gateway_to_kinesis(self, gateway_name, kinesis_stream):1414 template = self.APIGATEWAY_DATA_INBOUND_TEMPLATE % kinesis_stream1415 resource_path = self.API_PATH_DATA_INBOUND.replace("/", "")1416 resources = {1417 resource_path: [1418 {1419 "httpMethod": "POST",1420 "authorizationType": "NONE",1421 "requestModels": {"application/json": "Empty"},1422 "integrations": [1423 {1424 "type": "AWS",1425 "uri": "arn:aws:apigateway:%s:kinesis:action/PutRecords"1426 % aws_stack.get_region(),1427 "requestTemplates": {"application/json": template},1428 }1429 ],1430 },1431 {1432 "httpMethod": "GET",1433 "authorizationType": "NONE",1434 "requestModels": {"application/json": "Empty"},1435 "integrations": [1436 {1437 "type": "AWS",1438 "uri": "arn:aws:apigateway:%s:kinesis:action/ListStreams"1439 % aws_stack.get_region(),1440 "requestTemplates": {"application/json": "{}"},1441 }1442 ],1443 },1444 ]1445 }1446 return aws_stack.create_api_gateway(1447 name=gateway_name, resources=resources, stage_name=self.TEST_STAGE_NAME1448 )1449 def connect_api_gateway_to_http(1450 self, int_type, gateway_name, target_url, methods=None, path=None1451 ):1452 if methods is None:1453 methods = []1454 if not methods:1455 methods = ["GET", "POST"]1456 if not path:1457 path = "/"1458 resources = {}1459 resource_path = path.replace("/", "")1460 req_templates = (1461 {"application/json": json.dumps({"foo": "bar"})} if int_type == "custom" else {}1462 )1463 resources[resource_path] = [1464 {1465 "httpMethod": method,1466 "integrations": [1467 {1468 "type": "HTTP" if int_type == "custom" else "HTTP_PROXY",1469 "uri": target_url,1470 "requestTemplates": req_templates,1471 "responseTemplates": {},1472 }1473 ],1474 }1475 for method in methods1476 ]1477 return aws_stack.create_api_gateway(1478 name=gateway_name, resources=resources, stage_name=self.TEST_STAGE_NAME1479 )1480 @staticmethod1481 def create_lambda_function(fn_name):1482 testutil.create_lambda_function(1483 handler_file=TEST_LAMBDA_PYTHON, libs=TEST_LAMBDA_LIBS, func_name=fn_name1484 )1485 def test_apigw_test_invoke_method_api(self):1486 client = aws_stack.create_external_boto_client("apigateway")1487 lambda_client = aws_stack.create_external_boto_client("lambda")1488 # create test Lambda1489 fn_name = f"test-{short_uid()}"1490 testutil.create_lambda_function(1491 handler_file=TEST_LAMBDA_NODEJS, func_name=fn_name, runtime=LAMBDA_RUNTIME_NODEJS12X1492 )1493 lambda_arn_1 = aws_stack.lambda_function_arn(fn_name)1494 # create REST API and test resource1495 rest_api = client.create_rest_api(name="test", description="test")1496 root_resource = client.get_resources(restApiId=rest_api["id"])1497 resource = client.create_resource(1498 restApiId=rest_api["id"], parentId=root_resource["items"][0]["id"], pathPart="foo"1499 )1500 # create method and integration1501 client.put_method(1502 restApiId=rest_api["id"],1503 resourceId=resource["id"],1504 httpMethod="GET",1505 authorizationType="NONE",1506 )1507 client.put_integration(1508 restApiId=rest_api["id"],1509 resourceId=resource["id"],1510 httpMethod="GET",1511 integrationHttpMethod="GET",1512 type="AWS",1513 uri="arn:aws:apigateway:{}:lambda:path//2015-03-31/functions/{}/invocations".format(1514 aws_stack.get_region(), lambda_arn_11515 ),1516 )1517 # run test_invoke_method API #11518 response = client.test_invoke_method(1519 restApiId=rest_api["id"],1520 resourceId=resource["id"],1521 httpMethod="GET",1522 pathWithQueryString="/foo",1523 )1524 assert 200 == response["ResponseMetadata"]["HTTPStatusCode"]1525 assert 200 == response.get("status")1526 assert "response from" in response.get("body").get("body")1527 # run test_invoke_method API #21528 response = client.test_invoke_method(1529 restApiId=rest_api["id"],1530 resourceId=resource["id"],1531 httpMethod="GET",1532 pathWithQueryString="/foo",1533 body='{"test": "val123"}',1534 headers={"content-type": "application/json"},1535 )1536 assert 200 == response["ResponseMetadata"]["HTTPStatusCode"]1537 assert 200 == response.get("status")1538 assert "response from" in response.get("body").get("body")1539 assert "val123" in response.get("body").get("body")1540 # Clean up1541 lambda_client.delete_function(FunctionName=fn_name)1542 client.delete_rest_api(restApiId=rest_api["id"])1543 @staticmethod1544 def start_http_backend(test_port):1545 # test listener for target HTTP backend1546 class TestListener(ProxyListener):1547 def forward_request(self, **kwargs):1548 response = Response()1549 response.status_code = 2001550 result = {1551 "data": kwargs.get("data") or "{}",1552 "headers": dict(kwargs.get("headers")),1553 }1554 response._content = json.dumps(json_safe(result))1555 return response1556 proxy = start_proxy(test_port, update_listener=TestListener())1557 return proxy1558 @staticmethod1559 def create_api_gateway_and_deploy(1560 response_templates=None,1561 is_api_key_required=False,1562 integration_type=None,1563 integration_responses=None,1564 ):1565 response_templates = response_templates or {}1566 integration_type = integration_type or "AWS_PROXY"1567 apigw_client = aws_stack.create_external_boto_client("apigateway")1568 response = apigw_client.create_rest_api(name="my_api", description="this is my api")1569 api_id = response["id"]1570 resources = apigw_client.get_resources(restApiId=api_id)1571 root_resources = [resource for resource in resources["items"] if resource["path"] == "/"]1572 root_id = root_resources[0]["id"]1573 kwargs = {}1574 if integration_type == "AWS_PROXY":1575 aws_stack.create_dynamodb_table("MusicCollection", partition_key="id")1576 kwargs[1577 "uri"1578 ] = "arn:aws:apigateway:us-east-1:dynamodb:action/PutItem&Table=MusicCollection"1579 if not integration_responses:1580 integration_responses = [{"httpMethod": "PUT", "statusCode": "200"}]1581 for resp_details in integration_responses:1582 apigw_client.put_method(1583 restApiId=api_id,1584 resourceId=root_id,1585 httpMethod=resp_details["httpMethod"],1586 authorizationType="NONE",1587 apiKeyRequired=is_api_key_required,1588 )1589 apigw_client.put_method_response(1590 restApiId=api_id,1591 resourceId=root_id,1592 httpMethod=resp_details["httpMethod"],1593 statusCode="200",1594 )1595 apigw_client.put_integration(1596 restApiId=api_id,1597 resourceId=root_id,1598 httpMethod=resp_details["httpMethod"],1599 integrationHttpMethod=resp_details["httpMethod"],1600 type=integration_type,1601 **kwargs,1602 )1603 apigw_client.put_integration_response(1604 restApiId=api_id,1605 resourceId=root_id,1606 selectionPattern="",1607 responseTemplates=response_templates,1608 **resp_details,1609 )1610 apigw_client.create_deployment(restApiId=api_id, stageName="staging")1611 return api_id1612def test_import_swagger_api(apigateway_client):1613 apigateway_client.get_rest_apis()1614 api_spec = load_test_resource("openapi.swagger.json")1615 api_spec_dict = json.loads(api_spec)1616 backend = apigateway_backends["eu-west-1"]1617 api_model = backend.create_rest_api(name="", description="")1618 imported_api = import_api_from_openapi_spec(api_model, api_spec_dict, {})1619 # test_cfn_handle_serverless_api_resource fails if we support title1620 # assert imported_api.name == api_spec_dict.get("info").get("title")1621 assert imported_api.description == api_spec_dict.get("info").get("description")1622 # assert that are no multiple authorizers1623 assert len(imported_api.authorizers) == 11624 paths = {v.path_part for k, v in imported_api.resources.items()}1625 assert paths == {"/", "pets", "{petId}"}1626 resource_methods = {v.path_part: v.resource_methods for k, v in imported_api.resources.items()}1627 methods = {kk[0] for k, v in resource_methods.items() for kk in v.items()}1628 assert methods == {"POST", "OPTIONS", "GET"}1629 assert resource_methods.get("/").get("GET")["methodResponses"] == {1630 "200": {1631 "statusCode": "200",1632 "responseModels": None,1633 "responseParameters": {"method.response.header.Content-Type": "'text/html'"},1634 }1635 }1636 assert resource_methods.get("pets").get("GET")["methodResponses"] == {1637 "200": {1638 "responseModels": {1639 "application/json": {1640 "items": {1641 "properties": {1642 "id": {"type": "integer"},1643 "price": {"type": "number"},1644 "type": {"type": "string"},1645 },1646 "type": "object",1647 },1648 "type": "array",1649 }1650 },1651 "responseParameters": {"method.response.header.Access-Control-Allow-Origin": "'*'"},1652 "statusCode": "200",1653 }1654 }1655@pytest.mark.skipif(not use_docker(), reason="Rust lambdas cannot be executed in local executor")1656def test_apigateway_rust_lambda(1657 apigateway_client, create_lambda_function, create_iam_role_with_policy1658):1659 function_name = f"test-rust-function-{short_uid()}"1660 api_gateway_name = f"api_gateway_{short_uid()}"1661 role_name = f"test_apigateway_role_{short_uid()}"1662 policy_name = f"test_apigateway_policy_{short_uid()}"1663 stage_name = "test"1664 first_name = f"test_name_{short_uid()}"1665 lambda_create_response = create_lambda_function(1666 func_name=function_name,1667 zip_file=load_file(TEST_LAMBDA_HTTP_RUST, mode="rb"),1668 handler="bootstrap.is.the.handler",1669 runtime="provided.al2",1670 )1671 role_arn = create_iam_role_with_policy(1672 RoleName=role_name,1673 PolicyName=policy_name,1674 RoleDefinition=APIGATEWAY_ASSUME_ROLE_POLICY,1675 PolicyDefinition=APIGATEWAY_LAMBDA_POLICY,1676 )1677 lambda_arn = lambda_create_response["CreateFunctionResponse"]["FunctionArn"]1678 rest_api_id = apigateway_client.create_rest_api(name=api_gateway_name)["id"]1679 try:1680 root_resource_id = apigateway_client.get_resources(restApiId=rest_api_id)["items"][0]["id"]1681 apigateway_client.put_method(1682 restApiId=rest_api_id,1683 resourceId=root_resource_id,1684 httpMethod="GET",1685 authorizationType="NONE",1686 )1687 apigateway_client.put_method_response(1688 restApiId=rest_api_id, resourceId=root_resource_id, httpMethod="GET", statusCode="200"1689 )1690 lambda_target_uri = aws_stack.apigateway_invocations_arn(1691 lambda_uri=lambda_arn, region_name=apigateway_client.meta.region_name1692 )1693 apigateway_client.put_integration(1694 restApiId=rest_api_id,1695 resourceId=root_resource_id,1696 httpMethod="GET",1697 type="AWS_PROXY",1698 integrationHttpMethod="POST",1699 uri=lambda_target_uri,1700 credentials=role_arn,1701 )1702 apigateway_client.create_deployment(restApiId=rest_api_id, stageName=stage_name)1703 url = path_based_url(1704 api_id=rest_api_id, stage_name=stage_name, path=f"/?first_name={first_name}"1705 )1706 result = requests.get(url)1707 assert result.text == f"Hello, {first_name}!"1708 finally:1709 apigateway_client.delete_rest_api(restApiId=rest_api_id)1710def test_apigw_call_api_with_aws_endpoint_url():1711 headers = aws_stack.mock_aws_request_headers("apigateway")1712 headers["Host"] = "apigateway.us-east-2.amazonaws.com:4566"1713 url = f"{get_edge_url()}/apikeys?includeValues=true&name=test%40example.org"1714 response = requests.get(url, headers=headers)1715 assert response.ok1716 content = json.loads(to_str(response.content))1717 assert isinstance(content.get("item"), list)...

Full Screen

Full Screen

test_api_gateway.py

Source:test_api_gateway.py Github

copy

Full Screen

...136 {"data": '{"foo": "bar2"}'},137 {"data": '{"foo": "bar3"}'},138 ]139 }140 url = path_based_url(141 api_id=result["id"],142 stage_name=self.TEST_STAGE_NAME,143 path=self.API_PATH_DATA_INBOUND,144 )145 # list Kinesis streams via API Gateway146 result = requests.get(url)147 result = json.loads(to_str(result.content))148 self.assertIn("StreamNames", result)149 # post test data to Kinesis via API Gateway150 result = requests.post(url, data=json.dumps(test_data))151 result = json.loads(to_str(result.content))152 self.assertEqual(0, result["FailedRecordCount"])153 self.assertEqual(len(test_data["records"]), len(result["Records"]))154 # clean up155 kinesis = aws_stack.create_external_boto_client("kinesis")156 kinesis.delete_stream(StreamName=self.TEST_STREAM_KINESIS_API_GW)157 def test_api_gateway_sqs_integration_with_event_source(self):158 # create target SQS stream159 queue_name = "queue-%s" % short_uid()160 queue_url = aws_stack.create_sqs_queue(queue_name)["QueueUrl"]161 # create API Gateway and connect it to the target queue162 result = connect_api_gateway_to_sqs(163 "test_gateway4",164 stage_name=self.TEST_STAGE_NAME,165 queue_arn=queue_name,166 path=self.API_PATH_DATA_INBOUND,167 )168 # create event source for sqs lambda processor169 self.create_lambda_function(self.TEST_LAMBDA_SQS_HANDLER_NAME)170 event_source_data = {171 "FunctionName": self.TEST_LAMBDA_SQS_HANDLER_NAME,172 "EventSourceArn": aws_stack.sqs_queue_arn(queue_name),173 "Enabled": True,174 }175 add_event_source(event_source_data)176 # generate test data177 test_data = {"spam": "eggs & beans"}178 url = path_based_url(179 api_id=result["id"],180 stage_name=self.TEST_STAGE_NAME,181 path=self.API_PATH_DATA_INBOUND,182 )183 result = requests.post(url, data=json.dumps(test_data))184 self.assertEqual(200, result.status_code)185 parsed_json = xmltodict.parse(result.content)186 result = parsed_json["SendMessageResponse"]["SendMessageResult"]187 body_md5 = result["MD5OfMessageBody"]188 self.assertEqual("b639f52308afd65866c86f274c59033f", body_md5)189 # clean up190 sqs_client = aws_stack.create_external_boto_client("sqs")191 sqs_client.delete_queue(QueueUrl=queue_url)192 lambda_client = aws_stack.create_external_boto_client("lambda")193 lambda_client.delete_function(FunctionName=self.TEST_LAMBDA_SQS_HANDLER_NAME)194 def test_api_gateway_sqs_integration(self):195 # create target SQS stream196 queue_name = "queue-%s" % short_uid()197 aws_stack.create_sqs_queue(queue_name)198 # create API Gateway and connect it to the target queue199 result = connect_api_gateway_to_sqs(200 "test_gateway4",201 stage_name=self.TEST_STAGE_NAME,202 queue_arn=queue_name,203 path=self.API_PATH_DATA_INBOUND,204 )205 # generate test data206 test_data = {"spam": "eggs"}207 url = path_based_url(208 api_id=result["id"],209 stage_name=self.TEST_STAGE_NAME,210 path=self.API_PATH_DATA_INBOUND,211 )212 result = requests.post(url, data=json.dumps(test_data))213 self.assertEqual(200, result.status_code)214 messages = aws_stack.sqs_receive_message(queue_name)["Messages"]215 self.assertEqual(1, len(messages))216 self.assertEqual(test_data, json.loads(base64.b64decode(messages[0]["Body"])))217 def test_api_gateway_http_integrations(self):218 self.run_api_gateway_http_integration("custom")219 self.run_api_gateway_http_integration("proxy")220 @patch.object(config, "DISABLE_CUSTOM_CORS_APIGATEWAY", False)221 def run_api_gateway_http_integration(self, int_type):222 test_port = get_free_tcp_port()223 backend_url = "http://localhost:%s%s" % (test_port, self.API_PATH_HTTP_BACKEND)224 # start test HTTP backend225 proxy = self.start_http_backend(test_port)226 # create API Gateway and connect it to the HTTP_PROXY/HTTP backend227 result = self.connect_api_gateway_to_http(228 int_type, "test_gateway2", backend_url, path=self.API_PATH_HTTP_BACKEND229 )230 url = path_based_url(231 api_id=result["id"],232 stage_name=self.TEST_STAGE_NAME,233 path=self.API_PATH_HTTP_BACKEND,234 )235 # make sure CORS headers are present236 origin = "localhost"237 result = requests.options(url, headers={"origin": origin})238 self.assertEqual(result.status_code, 200)239 self.assertTrue(240 re.match(result.headers["Access-Control-Allow-Origin"].replace("*", ".*"), origin)241 )242 self.assertIn("POST", result.headers["Access-Control-Allow-Methods"])243 self.assertIn("PATCH", result.headers["Access-Control-Allow-Methods"])244 custom_result = json.dumps({"foo": "bar"})245 # make test GET request to gateway246 result = requests.get(url)247 self.assertEqual(200, result.status_code)248 expected = custom_result if int_type == "custom" else "{}"249 self.assertEqual(expected, json.loads(to_str(result.content))["data"])250 # make test POST request to gateway251 data = json.dumps({"data": 123})252 result = requests.post(url, data=data)253 self.assertEqual(200, result.status_code)254 expected = custom_result if int_type == "custom" else data255 self.assertEqual(expected, json.loads(to_str(result.content))["data"])256 # make test POST request with non-JSON content type257 data = "test=123"258 ctype = "application/x-www-form-urlencoded"259 result = requests.post(url, data=data, headers={"content-type": ctype})260 self.assertEqual(200, result.status_code)261 content = json.loads(to_str(result.content))262 headers = CaseInsensitiveDict(content["headers"])263 expected = custom_result if int_type == "custom" else data264 self.assertEqual(expected, content["data"])265 self.assertEqual(ctype, headers["content-type"])266 # clean up267 proxy.stop()268 def test_api_gateway_lambda_proxy_integration(self):269 self._test_api_gateway_lambda_proxy_integration(270 self.TEST_LAMBDA_PROXY_BACKEND, self.API_PATH_LAMBDA_PROXY_BACKEND271 )272 def test_api_gateway_lambda_proxy_integration_with_path_param(self):273 self._test_api_gateway_lambda_proxy_integration(274 self.TEST_LAMBDA_PROXY_BACKEND_WITH_PATH_PARAM,275 self.API_PATH_LAMBDA_PROXY_BACKEND_WITH_PATH_PARAM,276 )277 def test_api_gateway_lambda_proxy_integration_with_is_base_64_encoded(self):278 # Test the case where `isBase64Encoded` is enabled.279 content = b"hello, please base64 encode me"280 def _mutate_data(data) -> None:281 data["return_is_base_64_encoded"] = True282 data["return_raw_body"] = base64.b64encode(content).decode("utf8")283 test_result = self._test_api_gateway_lambda_proxy_integration_no_asserts(284 self.TEST_LAMBDA_PROXY_BACKEND_WITH_IS_BASE64,285 self.API_PATH_LAMBDA_PROXY_BACKEND_WITH_IS_BASE64,286 data_mutator_fn=_mutate_data,287 )288 # Ensure that `invoke_rest_api_integration_backend` correctly decodes the base64 content289 self.assertEqual(test_result.result.status_code, 203)290 self.assertEqual(test_result.result.content, content)291 def _test_api_gateway_lambda_proxy_integration_no_asserts(292 self,293 fn_name: str,294 path: str,295 data_mutator_fn: Optional[Callable] = None,296 ) -> ApiGatewayLambdaProxyIntegrationTestResult:297 """298 Perform the setup needed to do a POST against a Lambda Proxy Integration;299 then execute the POST.300 :param data_mutator_fn: a Callable[[Dict], None] that lets us mutate the301 data dictionary before sending it off to the lambda.302 """303 self.create_lambda_function(fn_name)304 # create API Gateway and connect it to the Lambda proxy backend305 lambda_uri = aws_stack.lambda_function_arn(fn_name)306 invocation_uri = "arn:aws:apigateway:%s:lambda:path/2015-03-31/functions/%s/invocations"307 target_uri = invocation_uri % (aws_stack.get_region(), lambda_uri)308 result = testutil.connect_api_gateway_to_http_with_lambda_proxy(309 "test_gateway2", target_uri, path=path, stage_name=self.TEST_STAGE_NAME310 )311 api_id = result["id"]312 path_map = get_rest_api_paths(api_id)313 _, resource = get_resource_for_path(path, path_map)314 # make test request to gateway and check response315 path_with_replace = path.replace("{test_param1}", "foo1")316 path_with_params = path_with_replace + "?foo=foo&bar=bar&bar=baz"317 url = path_based_url(api_id=api_id, stage_name=self.TEST_STAGE_NAME, path=path_with_params)318 # These values get read in `lambda_integration.py`319 data = {"return_status_code": 203, "return_headers": {"foo": "bar123"}}320 if data_mutator_fn:321 assert callable(data_mutator_fn)322 data_mutator_fn(data)323 result = requests.post(324 url,325 data=json.dumps(data),326 headers={"User-Agent": "python-requests/testing"},327 )328 return ApiGatewayLambdaProxyIntegrationTestResult(329 data=data,330 resource=resource,331 result=result,332 url=url,333 path_with_replace=path_with_replace,334 )335 def _test_api_gateway_lambda_proxy_integration(336 self,337 fn_name: str,338 path: str,339 ) -> None:340 test_result = self._test_api_gateway_lambda_proxy_integration_no_asserts(fn_name, path)341 data, resource, result, url, path_with_replace = test_result342 self.assertEqual(result.status_code, 203)343 self.assertEqual(result.headers.get("foo"), "bar123")344 self.assertIn("set-cookie", result.headers)345 try:346 parsed_body = json.loads(to_str(result.content))347 except json.decoder.JSONDecodeError as e:348 raise Exception(349 "Couldn't json-decode content: {}".format(to_str(result.content))350 ) from e351 self.assertEqual(parsed_body.get("return_status_code"), 203)352 self.assertDictEqual(parsed_body.get("return_headers"), {"foo": "bar123"})353 self.assertDictEqual(354 parsed_body.get("queryStringParameters"),355 {"foo": "foo", "bar": ["bar", "baz"]},356 )357 request_context = parsed_body.get("requestContext")358 source_ip = request_context["identity"].pop("sourceIp")359 self.assertTrue(re.match(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$", source_ip))360 expected_path = "/" + self.TEST_STAGE_NAME + "/lambda/foo1"361 self.assertEqual(expected_path, request_context["path"])362 self.assertIsNone(request_context.get("stageVariables"))363 self.assertEqual(TEST_AWS_ACCOUNT_ID, request_context["accountId"])364 self.assertEqual(resource.get("id"), request_context["resourceId"])365 self.assertEqual(self.TEST_STAGE_NAME, request_context["stage"])366 self.assertEqual("python-requests/testing", request_context["identity"]["userAgent"])367 self.assertEqual("POST", request_context["httpMethod"])368 self.assertEqual("HTTP/1.1", request_context["protocol"])369 self.assertIn("requestTimeEpoch", request_context)370 self.assertIn("requestTime", request_context)371 self.assertIn("requestId", request_context)372 # assert that header keys are lowercase (as in AWS)373 headers = parsed_body.get("headers") or {}374 header_names = list(headers.keys())375 self.assertIn("Host", header_names)376 self.assertIn("Content-Length", header_names)377 self.assertIn("User-Agent", header_names)378 result = requests.delete(url, data=json.dumps(data))379 self.assertEqual(204, result.status_code)380 # send message with non-ASCII chars381 body_msg = "🙀 - 参よ"382 result = requests.post(url, data=json.dumps({"return_raw_body": body_msg}))383 self.assertEqual(body_msg, to_str(result.content))384 # send message with binary data385 binary_msg = b"\xff \xaa \x11"386 result = requests.post(url, data=binary_msg)387 result_content = json.loads(to_str(result.content))388 self.assertEqual("/yCqIBE=", result_content["body"])389 self.assertEqual(True, result_content["isBase64Encoded"])390 def test_api_gateway_lambda_proxy_integration_any_method(self):391 self._test_api_gateway_lambda_proxy_integration_any_method(392 self.TEST_LAMBDA_PROXY_BACKEND_ANY_METHOD,393 self.API_PATH_LAMBDA_PROXY_BACKEND_ANY_METHOD,394 )395 def test_api_gateway_lambda_proxy_integration_any_method_with_path_param(self):396 self._test_api_gateway_lambda_proxy_integration_any_method(397 self.TEST_LAMBDA_PROXY_BACKEND_ANY_METHOD_WITH_PATH_PARAM,398 self.API_PATH_LAMBDA_PROXY_BACKEND_ANY_METHOD_WITH_PATH_PARAM,399 )400 def test_api_gateway_authorizer_crud(self):401 apig = aws_stack.create_external_boto_client("apigateway")402 authorizer = apig.create_authorizer(403 restApiId=self.TEST_API_GATEWAY_ID, **self.TEST_API_GATEWAY_AUTHORIZER404 )405 authorizer_id = authorizer.get("id")406 create_result = apig.get_authorizer(407 restApiId=self.TEST_API_GATEWAY_ID, authorizerId=authorizer_id408 )409 # ignore boto3 stuff410 del create_result["ResponseMetadata"]411 create_expected = clone(self.TEST_API_GATEWAY_AUTHORIZER)412 create_expected["id"] = authorizer_id413 self.assertDictEqual(create_expected, create_result)414 apig.update_authorizer(415 restApiId=self.TEST_API_GATEWAY_ID,416 authorizerId=authorizer_id,417 patchOperations=self.TEST_API_GATEWAY_AUTHORIZER_OPS,418 )419 update_result = apig.get_authorizer(420 restApiId=self.TEST_API_GATEWAY_ID, authorizerId=authorizer_id421 )422 # ignore boto3 stuff423 del update_result["ResponseMetadata"]424 update_expected = apply_patch(create_expected, self.TEST_API_GATEWAY_AUTHORIZER_OPS)425 self.assertDictEqual(update_expected, update_result)426 apig.delete_authorizer(restApiId=self.TEST_API_GATEWAY_ID, authorizerId=authorizer_id)427 self.assertRaises(Exception, apig.get_authorizer, self.TEST_API_GATEWAY_ID, authorizer_id)428 def test_apigateway_with_lambda_integration(self):429 apigw_client = aws_stack.create_external_boto_client("apigateway")430 # create Lambda function431 lambda_name = "apigw-lambda-%s" % short_uid()432 self.create_lambda_function(lambda_name)433 lambda_uri = aws_stack.lambda_function_arn(lambda_name)434 target_uri = aws_stack.apigateway_invocations_arn(lambda_uri)435 # create REST API436 api = apigw_client.create_rest_api(name="test-api", description="")437 api_id = api["id"]438 root_res_id = apigw_client.get_resources(restApiId=api_id)["items"][0]["id"]439 api_resource = apigw_client.create_resource(440 restApiId=api_id, parentId=root_res_id, pathPart="test"441 )442 apigw_client.put_method(443 restApiId=api_id,444 resourceId=api_resource["id"],445 httpMethod="GET",446 authorizationType="NONE",447 )448 rs = apigw_client.put_integration(449 restApiId=api_id,450 resourceId=api_resource["id"],451 httpMethod="GET",452 integrationHttpMethod="POST",453 type="AWS",454 uri=target_uri,455 timeoutInMillis=3000,456 contentHandling="CONVERT_TO_BINARY",457 requestTemplates={"application/json": '{"param1": "$input.params(\'param1\')"}'},458 )459 integration_keys = [460 "httpMethod",461 "type",462 "passthroughBehavior",463 "cacheKeyParameters",464 "uri",465 "cacheNamespace",466 "timeoutInMillis",467 "contentHandling",468 "requestParameters",469 ]470 self.assertEqual(200, rs["ResponseMetadata"]["HTTPStatusCode"])471 for key in integration_keys:472 self.assertIn(key, rs)473 self.assertNotIn("responseTemplates", rs)474 apigw_client.create_deployment(restApiId=api_id, stageName=self.TEST_STAGE_NAME)475 rs = apigw_client.get_integration(476 restApiId=api_id, resourceId=api_resource["id"], httpMethod="GET"477 )478 self.assertEqual(200, rs["ResponseMetadata"]["HTTPStatusCode"])479 self.assertEqual("AWS", rs["type"])480 self.assertEqual("POST", rs["httpMethod"])481 self.assertEqual(target_uri, rs["uri"])482 # invoke the gateway endpoint483 url = path_based_url(api_id=api_id, stage_name=self.TEST_STAGE_NAME, path="/test")484 response = requests.get("%s?param1=foobar" % url)485 self.assertLess(response.status_code, 400)486 content = response.json()487 self.assertEqual("GET", content.get("httpMethod"))488 self.assertEqual(api_resource["id"], content.get("requestContext", {}).get("resourceId"))489 self.assertEqual(self.TEST_STAGE_NAME, content.get("requestContext", {}).get("stage"))490 self.assertEqual('{"param1": "foobar"}', content.get("body"))491 # additional checks from https://github.com/localstack/localstack/issues/5041492 # pass Signature param493 response = requests.get("%s?param1=foobar&Signature=1" % url)494 self.assertEqual(response.status_code, 200)495 content = response.json()496 self.assertEqual("GET", content.get("httpMethod"))497 self.assertEqual(api_resource["id"], content.get("requestContext", {}).get("resourceId"))498 self.assertEqual(self.TEST_STAGE_NAME, content.get("requestContext", {}).get("stage"))499 self.assertEqual('{"param1": "foobar"}', content.get("body"))500 # pass TestSignature param as well501 response = requests.get("%s?param1=foobar&TestSignature=1" % url)502 self.assertEqual(response.status_code, 200)503 content = response.json()504 self.assertEqual("GET", content.get("httpMethod"))505 self.assertEqual(api_resource["id"], content.get("requestContext", {}).get("resourceId"))506 self.assertEqual(self.TEST_STAGE_NAME, content.get("requestContext", {}).get("stage"))507 self.assertEqual('{"param1": "foobar"}', content.get("body"))508 # delete integration509 rs = apigw_client.delete_integration(510 restApiId=api_id,511 resourceId=api_resource["id"],512 httpMethod="GET",513 )514 self.assertEqual(200, rs["ResponseMetadata"]["HTTPStatusCode"])515 with self.assertRaises(ClientError) as ctx:516 # This call should not be successful as the integration is deleted517 apigw_client.get_integration(518 restApiId=api_id, resourceId=api_resource["id"], httpMethod="GET"519 )520 self.assertEqual(ctx.exception.response["Error"]["Code"], "NotFoundException")521 # clean up522 lambda_client = aws_stack.create_external_boto_client("lambda")523 lambda_client.delete_function(FunctionName=lambda_name)524 apigw_client.delete_rest_api(restApiId=api_id)525 def test_api_gateway_handle_domain_name(self):526 domain_name = "%s.example.com" % short_uid()527 apigw_client = aws_stack.create_external_boto_client("apigateway")528 rs = apigw_client.create_domain_name(domainName=domain_name)529 self.assertEqual(200, rs["ResponseMetadata"]["HTTPStatusCode"])530 rs = apigw_client.get_domain_name(domainName=domain_name)531 self.assertEqual(200, rs["ResponseMetadata"]["HTTPStatusCode"])532 self.assertEqual(domain_name, rs["domainName"])533 # clean up534 apigw_client.delete_domain_name(domainName=domain_name)535 def _test_api_gateway_lambda_proxy_integration_any_method(self, fn_name, path):536 self.create_lambda_function(fn_name)537 # create API Gateway and connect it to the Lambda proxy backend538 lambda_uri = aws_stack.lambda_function_arn(fn_name)539 target_uri = aws_stack.apigateway_invocations_arn(lambda_uri)540 result = testutil.connect_api_gateway_to_http_with_lambda_proxy(541 "test_gateway3",542 target_uri,543 methods=["ANY"],544 path=path,545 stage_name=self.TEST_STAGE_NAME,546 )547 # make test request to gateway and check response548 path = path.replace("{test_param1}", "foo1")549 url = path_based_url(api_id=result["id"], stage_name=self.TEST_STAGE_NAME, path=path)550 data = {}551 for method in ("GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"):552 body = json.dumps(data) if method in ("POST", "PUT", "PATCH") else None553 result = getattr(requests, method.lower())(url, data=body)554 if method != "DELETE":555 self.assertEqual(200, result.status_code)556 parsed_body = json.loads(to_str(result.content))557 self.assertEqual(method, parsed_body.get("httpMethod"))558 else:559 self.assertEqual(204, result.status_code)560 def test_apigateway_with_custom_authorization_method(self):561 apigw_client = aws_stack.create_external_boto_client("apigateway")562 # create Lambda function563 lambda_name = "apigw-lambda-%s" % short_uid()564 self.create_lambda_function(lambda_name)565 lambda_uri = aws_stack.lambda_function_arn(lambda_name)566 # create REST API567 api = apigw_client.create_rest_api(name="test-api", description="")568 api_id = api["id"]569 root_res_id = apigw_client.get_resources(restApiId=api_id)["items"][0]["id"]570 # create authorizer at root resource571 authorizer = apigw_client.create_authorizer(572 restApiId=api_id,573 name="lambda_authorizer",574 type="TOKEN",575 authorizerUri="arn:aws:apigateway:us-east-1:lambda:path/ \576 2015-03-31/functions/{}/invocations".format(577 lambda_uri578 ),579 identitySource="method.request.header.Auth",580 )581 # create method with custom authorizer582 is_api_key_required = True583 method_response = apigw_client.put_method(584 restApiId=api_id,585 resourceId=root_res_id,586 httpMethod="GET",587 authorizationType="CUSTOM",588 authorizerId=authorizer["id"],589 apiKeyRequired=is_api_key_required,590 )591 self.assertEqual(authorizer["id"], method_response["authorizerId"])592 # clean up593 lambda_client = aws_stack.create_external_boto_client("lambda")594 lambda_client.delete_function(FunctionName=lambda_name)595 apigw_client.delete_rest_api(restApiId=api_id)596 def test_create_model(self):597 client = aws_stack.create_external_boto_client("apigateway")598 response = client.create_rest_api(name="my_api", description="this is my api")599 rest_api_id = response["id"]600 dummy_rest_api_id = "_non_existing_"601 model_name = "testModel"602 description = "test model"603 content_type = "application/json"604 # success case with valid params605 response = client.create_model(606 restApiId=rest_api_id,607 name=model_name,608 description=description,609 contentType=content_type,610 )611 self.assertEqual(model_name, response["name"])612 self.assertEqual(description, response["description"])613 with self.assertRaises(Exception) as ctx:614 client.create_model(615 restApiId=dummy_rest_api_id,616 name=model_name,617 description=description,618 contentType=content_type,619 )620 self.assertEqual("NotFoundException", ctx.exception.response["Error"]["Code"])621 self.assertEqual(622 "Invalid Rest API Id specified", ctx.exception.response["Error"]["Message"]623 )624 with self.assertRaises(Exception) as ctx:625 client.create_model(626 restApiId=dummy_rest_api_id,627 name="",628 description=description,629 contentType=content_type,630 )631 self.assertEqual("BadRequestException", ctx.exception.response["Error"]["Code"])632 self.assertEqual("No Model Name specified", ctx.exception.response["Error"]["Message"])633 # clean up634 client.delete_rest_api(restApiId=rest_api_id)635 def test_get_api_models(self):636 client = aws_stack.create_external_boto_client("apigateway")637 response = client.create_rest_api(name="my_api", description="this is my api")638 rest_api_id = response["id"]639 model_name = "testModel"640 description = "test model"641 content_type = "application/json"642 # when no models are present643 result = client.get_models(restApiId=rest_api_id)644 self.assertEqual([], result["items"])645 # add a model646 client.create_model(647 restApiId=rest_api_id,648 name=model_name,649 description=description,650 contentType=content_type,651 )652 # get models after adding653 result = client.get_models(restApiId=rest_api_id)654 self.assertEqual(model_name, result["items"][0]["name"])655 self.assertEqual(description, result["items"][0]["description"])656 # clean up657 client.delete_rest_api(restApiId=rest_api_id)658 def test_request_validator(self):659 client = aws_stack.create_external_boto_client("apigateway")660 response = client.create_rest_api(name="my_api", description="this is my api")661 rest_api_id = response["id"]662 # CREATE663 name = "validator123"664 result = client.create_request_validator(restApiId=rest_api_id, name=name)665 self.assertEqual(200, result["ResponseMetadata"]["HTTPStatusCode"])666 validator_id = result["id"]667 # LIST668 result = client.get_request_validators(restApiId=rest_api_id)669 self.assertEqual(200, result["ResponseMetadata"]["HTTPStatusCode"])670 self.assertEqual([{"id": validator_id, "name": name}], result["items"])671 # GET672 result = client.get_request_validator(673 restApiId=rest_api_id, requestValidatorId=validator_id674 )675 self.assertEqual(200, result["ResponseMetadata"]["HTTPStatusCode"])676 self.assertEqual(677 select_attributes(result, ["id", "name"]),678 {"id": validator_id, "name": name},679 )680 # UPDATE681 result = client.update_request_validator(682 restApiId=rest_api_id, requestValidatorId=validator_id, patchOperations=[]683 )684 # DELETE685 client.delete_request_validator(restApiId=rest_api_id, requestValidatorId=validator_id)686 with self.assertRaises(Exception):687 client.get_request_validator(restApiId=rest_api_id, requestValidatorId=validator_id)688 with self.assertRaises(Exception):689 client.delete_request_validator(restApiId=rest_api_id, requestValidatorId=validator_id)690 # clean up691 client.delete_rest_api(restApiId=rest_api_id)692 def test_base_path_mapping(self):693 client = aws_stack.create_external_boto_client("apigateway")694 response = client.create_rest_api(name="my_api", description="this is my api")695 rest_api_id = response["id"]696 # CREATE697 domain_name = "domain1.example.com"698 client.create_domain_name(domainName=domain_name)699 root_res_id = client.get_resources(restApiId=rest_api_id)["items"][0]["id"]700 res_id = client.create_resource(701 restApiId=rest_api_id, parentId=root_res_id, pathPart="path"702 )["id"]703 client.put_method(704 restApiId=rest_api_id, resourceId=res_id, httpMethod="GET", authorizationType="NONE"705 )706 client.put_integration(707 restApiId=rest_api_id, resourceId=res_id, httpMethod="GET", type="MOCK"708 )709 depl_id = client.create_deployment(restApiId=rest_api_id)["id"]710 client.create_stage(restApiId=rest_api_id, deploymentId=depl_id, stageName="dev")711 base_path = "foo"712 result = client.create_base_path_mapping(713 domainName=domain_name,714 basePath=base_path,715 restApiId=rest_api_id,716 stage="dev",717 )718 self.assertIn(result["ResponseMetadata"]["HTTPStatusCode"], [200, 201])719 # LIST720 result = client.get_base_path_mappings(domainName=domain_name)721 self.assertEqual(200, result["ResponseMetadata"]["HTTPStatusCode"])722 expected = {"basePath": base_path, "restApiId": rest_api_id, "stage": "dev"}723 self.assertEqual([expected], result["items"])724 # GET725 result = client.get_base_path_mapping(domainName=domain_name, basePath=base_path)726 self.assertEqual(200, result["ResponseMetadata"]["HTTPStatusCode"])727 self.assertEqual(expected, select_attributes(result, ["basePath", "restApiId", "stage"]))728 # UPDATE729 result = client.update_base_path_mapping(730 domainName=domain_name, basePath=base_path, patchOperations=[]731 )732 self.assertEqual(200, result["ResponseMetadata"]["HTTPStatusCode"])733 # DELETE734 client.delete_base_path_mapping(domainName=domain_name, basePath=base_path)735 with self.assertRaises(Exception):736 client.get_base_path_mapping(domainName=domain_name, basePath=base_path)737 with self.assertRaises(Exception):738 client.delete_base_path_mapping(domainName=domain_name, basePath=base_path)739 def test_base_path_mapping_root(self):740 client = aws_stack.create_external_boto_client("apigateway")741 response = client.create_rest_api(name="my_api2", description="this is my api")742 rest_api_id = response["id"]743 # CREATE744 domain_name = "domain2.example.com"745 client.create_domain_name(domainName=domain_name)746 root_res_id = client.get_resources(restApiId=rest_api_id)["items"][0]["id"]747 res_id = client.create_resource(748 restApiId=rest_api_id, parentId=root_res_id, pathPart="path"749 )["id"]750 client.put_method(751 restApiId=rest_api_id, resourceId=res_id, httpMethod="GET", authorizationType="NONE"752 )753 client.put_integration(754 restApiId=rest_api_id, resourceId=res_id, httpMethod="GET", type="MOCK"755 )756 depl_id = client.create_deployment(restApiId=rest_api_id)["id"]757 client.create_stage(restApiId=rest_api_id, deploymentId=depl_id, stageName="dev")758 result = client.create_base_path_mapping(759 domainName=domain_name,760 basePath="",761 restApiId=rest_api_id,762 stage="dev",763 )764 self.assertIn(result["ResponseMetadata"]["HTTPStatusCode"], [200, 201])765 base_path = "(none)"766 # LIST767 result = client.get_base_path_mappings(domainName=domain_name)768 self.assertEqual(200, result["ResponseMetadata"]["HTTPStatusCode"])769 expected = {"basePath": "(none)", "restApiId": rest_api_id, "stage": "dev"}770 self.assertEqual([expected], result["items"])771 # GET772 result = client.get_base_path_mapping(domainName=domain_name, basePath=base_path)773 self.assertEqual(200, result["ResponseMetadata"]["HTTPStatusCode"])774 self.assertEqual(expected, select_attributes(result, ["basePath", "restApiId", "stage"]))775 # UPDATE776 result = client.update_base_path_mapping(777 domainName=domain_name, basePath=base_path, patchOperations=[]778 )779 self.assertEqual(200, result["ResponseMetadata"]["HTTPStatusCode"])780 # DELETE781 client.delete_base_path_mapping(domainName=domain_name, basePath=base_path)782 with self.assertRaises(Exception):783 client.get_base_path_mapping(domainName=domain_name, basePath=base_path)784 with self.assertRaises(Exception):785 client.delete_base_path_mapping(domainName=domain_name, basePath=base_path)786 def test_api_account(self):787 client = aws_stack.create_external_boto_client("apigateway")788 response = client.create_rest_api(name="my_api", description="test 123")789 rest_api_id = response["id"]790 result = client.get_account()791 self.assertIn("UsagePlans", result["features"])792 result = client.update_account(793 patchOperations=[{"op": "add", "path": "/features/-", "value": "foobar"}]794 )795 self.assertIn("foobar", result["features"])796 # clean up797 client.delete_rest_api(restApiId=rest_api_id)798 def test_get_model_by_name(self):799 client = aws_stack.create_external_boto_client("apigateway")800 response = client.create_rest_api(name="my_api", description="this is my api")801 rest_api_id = response["id"]802 dummy_rest_api_id = "_non_existing_"803 model_name = "testModel"804 description = "test model"805 content_type = "application/json"806 # add a model807 client.create_model(808 restApiId=rest_api_id,809 name=model_name,810 description=description,811 contentType=content_type,812 )813 # get models after adding814 result = client.get_model(restApiId=rest_api_id, modelName=model_name)815 self.assertEqual(model_name, result["name"])816 self.assertEqual(description, result["description"])817 try:818 client.get_model(restApiId=dummy_rest_api_id, modelName=model_name)819 self.fail("This call should not be successful as the model is not created.")820 except ClientError as e:821 self.assertEqual("NotFoundException", e.response["Error"]["Code"])822 self.assertEqual("Invalid Rest API Id specified", e.response["Error"]["Message"])823 def test_get_model_with_invalid_name(self):824 client = aws_stack.create_external_boto_client("apigateway")825 response = client.create_rest_api(name="my_api", description="this is my api")826 rest_api_id = response["id"]827 # test with an invalid model name828 try:829 client.get_model(restApiId=rest_api_id, modelName="fake")830 self.fail("This call should not be successful as the model is not created.")831 except ClientError as e:832 self.assertEqual("NotFoundException", e.response["Error"]["Code"])833 # clean up834 client.delete_rest_api(restApiId=rest_api_id)835 def test_put_integration_dynamodb_proxy_validation_without_response_template(self):836 api_id = self.create_api_gateway_and_deploy({})837 url = path_based_url(api_id=api_id, stage_name="staging", path="/")838 response = requests.put(839 url,840 json.dumps({"id": "id1", "data": "foobar123"}),841 )842 self.assertEqual(404, response.status_code)843 def test_put_integration_dynamodb_proxy_validation_with_response_template(self):844 response_templates = {845 "application/json": json.dumps(846 {847 "TableName": "MusicCollection",848 "Item": {"id": "$.Id", "data": "$.data"},849 }850 )851 }852 api_id = self.create_api_gateway_and_deploy(response_templates)853 url = path_based_url(api_id=api_id, stage_name="staging", path="/")854 response = requests.put(855 url,856 json.dumps({"id": "id1", "data": "foobar123"}),857 )858 self.assertEqual(200, response.status_code)859 dynamo_client = aws_stack.connect_to_resource("dynamodb")860 table = dynamo_client.Table("MusicCollection")861 result = table.get_item(Key={"id": "id1"})862 self.assertEqual("foobar123", result["Item"]["data"])863 def test_api_key_required_for_methods(self):864 response_templates = {865 "application/json": json.dumps(866 {867 "TableName": "MusicCollection",868 "Item": {"id": "$.Id", "data": "$.data"},869 }870 )871 }872 api_id = self.create_api_gateway_and_deploy(response_templates, True)873 url = path_based_url(api_id=api_id, stage_name="staging", path="/")874 payload = {875 "name": "TEST-PLAN-2",876 "description": "Description",877 "quota": {"limit": 10, "period": "DAY", "offset": 0},878 "throttle": {"rateLimit": 2, "burstLimit": 1},879 "apiStages": [{"apiId": api_id, "stage": "staging"}],880 "tags": {"tag_key": "tag_value"},881 }882 client = aws_stack.create_external_boto_client("apigateway")883 usage_plan_id = client.create_usage_plan(**payload)["id"]884 key_name = "testApiKey"885 key_type = "API_KEY"886 api_key = client.create_api_key(name=key_name)887 payload = {888 "usagePlanId": usage_plan_id,889 "keyId": api_key["id"],890 "keyType": key_type,891 }892 client.create_usage_plan_key(**payload)893 response = requests.put(894 url,895 json.dumps({"id": "id1", "data": "foobar123"}),896 )897 # when the api key is not passed as part of the header898 self.assertEqual(403, response.status_code)899 response = requests.put(900 url,901 json.dumps({"id": "id1", "data": "foobar123"}),902 headers={"X-API-Key": api_key["value"]},903 )904 # when the api key is passed as part of the header905 self.assertEqual(200, response.status_code)906 def test_multiple_api_keys_validate(self):907 response_templates = {908 "application/json": json.dumps(909 {910 "TableName": "MusicCollection",911 "Item": {"id": "$.Id", "data": "$.data"},912 }913 )914 }915 api_id = self.create_api_gateway_and_deploy(response_templates, True)916 url = path_based_url(api_id=api_id, stage_name="staging", path="/")917 client = aws_stack.create_external_boto_client("apigateway")918 # Create multiple usage plans919 usage_plan_ids = []920 for i in range(2):921 payload = {922 "name": "APIKEYTEST-PLAN-{}".format(i),923 "description": "Description",924 "quota": {"limit": 10, "period": "DAY", "offset": 0},925 "throttle": {"rateLimit": 2, "burstLimit": 1},926 "apiStages": [{"apiId": api_id, "stage": "staging"}],927 "tags": {"tag_key": "tag_value"},928 }929 usage_plan_ids.append(client.create_usage_plan(**payload)["id"])930 api_keys = []931 key_type = "API_KEY"932 # Create multiple API Keys in each usage plan933 for usage_plan_id in usage_plan_ids:934 for i in range(2):935 api_key = client.create_api_key(name="testMultipleApiKeys{}".format(i))936 payload = {937 "usagePlanId": usage_plan_id,938 "keyId": api_key["id"],939 "keyType": key_type,940 }941 client.create_usage_plan_key(**payload)942 api_keys.append(api_key["value"])943 response = requests.put(944 url,945 json.dumps({"id": "id1", "data": "foobar123"}),946 )947 # when the api key is not passed as part of the header948 self.assertEqual(403, response.status_code)949 # check that all API keys work950 for key in api_keys:951 response = requests.put(952 url,953 json.dumps({"id": "id1", "data": "foobar123"}),954 headers={"X-API-Key": key},955 )956 # when the api key is passed as part of the header957 self.assertEqual(200, response.status_code)958 def test_import_rest_api(self):959 rest_api_name = "restapi-%s" % short_uid()960 client = aws_stack.create_external_boto_client("apigateway")961 rest_api_id = client.create_rest_api(name=rest_api_name)["id"]962 spec_file = load_file(TEST_SWAGGER_FILE)963 rs = client.put_rest_api(restApiId=rest_api_id, body=spec_file, mode="overwrite")964 self.assertEqual(200, rs["ResponseMetadata"]["HTTPStatusCode"])965 rs = client.get_resources(restApiId=rest_api_id)966 self.assertEqual(967 2, len(rs["items"])968 ) # should contain 2 resources (including the root resource)969 resource = [res for res in rs["items"] if res["path"] == "/test"][0]970 self.assertIn("GET", resource["resourceMethods"])971 # clean up972 client.delete_rest_api(restApiId=rest_api_id)973 spec_file = load_file(TEST_IMPORT_REST_API_FILE)974 rs = client.import_rest_api(body=spec_file)975 self.assertEqual(200, rs["ResponseMetadata"]["HTTPStatusCode"])976 rest_api_id = rs["id"]977 rs = client.get_resources(restApiId=rest_api_id)978 resources = rs["items"]979 self.assertEqual(3, len(resources))980 paths = [res["path"] for res in resources]981 self.assertIn("/", paths)982 self.assertIn("/pets", paths)983 self.assertIn("/pets/{petId}", paths)984 # clean up985 client.delete_rest_api(restApiId=rest_api_id)986 def test_step_function_integrations(self):987 client = aws_stack.create_external_boto_client("apigateway")988 sfn_client = aws_stack.create_external_boto_client("stepfunctions")989 lambda_client = aws_stack.create_external_boto_client("lambda")990 state_machine_name = "test"991 state_machine_def = {992 "Comment": "Hello World example",993 "StartAt": "step1",994 "States": {995 "step1": {"Type": "Task", "Resource": "__tbd__", "End": True},996 },997 }998 # create state machine999 fn_name = "test-stepfunctions-apigw"1000 testutil.create_lambda_function(1001 handler_file=TEST_LAMBDA_PYTHON_ECHO,1002 func_name=fn_name,1003 runtime=LAMBDA_RUNTIME_PYTHON36,1004 )1005 role_arn = aws_stack.role_arn("sfn_role")1006 # create state machine definition1007 definition = clone(state_machine_def)1008 lambda_arn_1 = aws_stack.lambda_function_arn(fn_name)1009 definition["States"]["step1"]["Resource"] = lambda_arn_11010 definition = json.dumps(definition)1011 # create state machine1012 result = sfn_client.create_state_machine(1013 name=state_machine_name, definition=definition, roleArn=role_arn1014 )1015 sm_arn = result["stateMachineArn"]1016 # create REST API and method1017 rest_api = client.create_rest_api(name="test", description="test")1018 resources = client.get_resources(restApiId=rest_api["id"])1019 root_resource_id = resources["items"][0]["id"]1020 client.put_method(1021 restApiId=rest_api["id"],1022 resourceId=root_resource_id,1023 httpMethod="POST",1024 authorizationType="NONE",1025 )1026 def _prepare_method_integration(1027 integr_kwargs=None, resp_templates=None, action="StartExecution", overwrite=False1028 ):1029 if integr_kwargs is None:1030 integr_kwargs = {}1031 if resp_templates is None:1032 resp_templates = {}1033 if overwrite:1034 client.delete_integration(1035 restApiId=rest_api["id"],1036 resourceId=resources["items"][0]["id"],1037 httpMethod="POST",1038 )1039 uri = f"arn:aws:apigateway:{aws_stack.get_region()}:states:action/{action}"1040 client.put_integration(1041 restApiId=rest_api["id"],1042 resourceId=root_resource_id,1043 httpMethod="POST",1044 integrationHttpMethod="POST",1045 type="AWS",1046 uri=uri,1047 **integr_kwargs,1048 )1049 if resp_templates:1050 client.put_integration_response(1051 restApiId=rest_api["id"],1052 resourceId=root_resource_id,1053 selectionPattern="",1054 responseTemplates=resp_templates,1055 httpMethod="POST",1056 statusCode="200",1057 )1058 # STEP 1: test integration with request template1059 _prepare_method_integration(1060 integr_kwargs={1061 "requestTemplates": {1062 "application/json": """1063 #set($data = $util.escapeJavaScript($input.json('$')))1064 {"input": $data, "stateMachineArn": "%s"}1065 """1066 % sm_arn1067 }1068 }1069 )1070 # invoke stepfunction via API GW, assert results1071 client.create_deployment(restApiId=rest_api["id"], stageName="dev")1072 url = path_based_url(api_id=rest_api["id"], stage_name="dev", path="/")1073 test_data = {"test": "test-value"}1074 resp = requests.post(url, data=json.dumps(test_data))1075 self.assertEqual(200, resp.status_code)1076 self.assertIn("executionArn", resp.content.decode())1077 self.assertIn("startDate", resp.content.decode())1078 # STEP 2: test integration without request template1079 _prepare_method_integration(overwrite=True)1080 test_data_1 = {1081 "input": json.dumps(test_data),1082 "name": "MyExecution",1083 "stateMachineArn": sm_arn,1084 }1085 # invoke stepfunction via API GW, assert results1086 resp = requests.post(url, data=json.dumps(test_data_1))1087 self.assertEqual(200, resp.status_code)1088 self.assertIn("executionArn", resp.content.decode())1089 self.assertIn("startDate", resp.content.decode())1090 # STEP 3: test integration with synchronous execution1091 _prepare_method_integration(overwrite=True, action="StartSyncExecution")1092 # invoke stepfunction via API GW, assert results1093 test_data_1["name"] += "1"1094 resp = requests.post(url, data=json.dumps(test_data_1))1095 self.assertEqual(200, resp.status_code)1096 content = json.loads(to_str(resp.content.decode()))1097 self.assertEqual("SUCCEEDED", content.get("status"))1098 self.assertEqual(test_data, json.loads(content.get("output")))1099 # STEP 4: test integration with synchronous execution and response templates1100 resp_templates = {APPLICATION_JSON: "$input.path('$.output')"}1101 _prepare_method_integration(1102 resp_templates=resp_templates, overwrite=True, action="StartSyncExecution"1103 )1104 # invoke stepfunction via API GW, assert results1105 test_data_1["name"] += "2"1106 resp = requests.post(url, data=json.dumps(test_data_1))1107 self.assertEqual(200, resp.status_code)1108 self.assertEqual(test_data, json.loads(to_str(resp.content.decode())))1109 _prepare_method_integration(overwrite=True, action="DeleteStateMachine")1110 # Remove state machine with API GW1111 resp = requests.post(url, data=json.dumps({"stateMachineArn": sm_arn}))1112 self.assertEqual(200, resp.status_code)1113 # Clean up1114 lambda_client.delete_function(FunctionName=fn_name)1115 client.delete_rest_api(restApiId=rest_api["id"])1116 def test_api_gateway_http_integration_with_path_request_parameter(self):1117 client = aws_stack.create_external_boto_client("apigateway")1118 test_port = get_free_tcp_port()1119 backend_url = "http://localhost:%s/person/{id}" % (test_port)1120 # start test HTTP backend1121 proxy = self.start_http_backend(test_port)1122 # create rest api1123 api_rest = client.create_rest_api(name="test")1124 api_id = api_rest["id"]1125 parent_response = client.get_resources(restApiId=api_id)1126 parent_id = parent_response["items"][0]["id"]1127 resource_1 = client.create_resource(restApiId=api_id, parentId=parent_id, pathPart="person")1128 resource_1_id = resource_1["id"]1129 resource_2 = client.create_resource(1130 restApiId=api_id, parentId=resource_1_id, pathPart="{id}"1131 )1132 resource_2_id = resource_2["id"]1133 client.put_method(1134 restApiId=api_id,1135 resourceId=resource_2_id,1136 httpMethod="GET",1137 authorizationType="NONE",1138 apiKeyRequired=False,1139 requestParameters={"method.request.path.id": True},1140 )1141 client.put_integration(1142 restApiId=api_id,1143 resourceId=resource_2_id,1144 httpMethod="GET",1145 integrationHttpMethod="GET",1146 type="HTTP",1147 uri=backend_url,1148 timeoutInMillis=3000,1149 contentHandling="CONVERT_TO_BINARY",1150 requestParameters={"integration.request.path.id": "method.request.path.id"},1151 )1152 client.create_deployment(restApiId=api_id, stageName="test")1153 def _test_invoke(url):1154 result = requests.get(url)1155 content = json.loads(to_str(result.content))1156 self.assertEqual(200, result.status_code)1157 self.assertRegex(1158 content["headers"].get(HEADER_LOCALSTACK_REQUEST_URL),1159 "http://.*localhost.*/person/123",1160 )1161 for use_hostname in [True, False]:1162 for use_ssl in [True, False] if use_hostname else [False]:1163 url = self._get_invoke_endpoint(1164 api_id,1165 stage="test",1166 path="/person/123",1167 use_hostname=use_hostname,1168 use_ssl=use_ssl,1169 )1170 _test_invoke(url)1171 # clean up1172 client.delete_rest_api(restApiId=api_id)1173 proxy.stop()1174 def _get_invoke_endpoint(1175 self, api_id, stage="test", path="/", use_hostname=False, use_ssl=False1176 ):1177 path = path or "/"1178 path = path if path.startswith(path) else f"/{path}"1179 proto = "https" if use_ssl else "http"1180 if use_hostname:1181 return f"{proto}://{api_id}.execute-api.{LOCALHOST_HOSTNAME}:{config.EDGE_PORT}/{stage}{path}"1182 return (1183 f"{proto}://localhost:{config.EDGE_PORT}/restapis/{api_id}/{stage}/_user_request_{path}"1184 )1185 def test_api_gateway_s3_get_integration(self):1186 apigw_client = aws_stack.create_external_boto_client("apigateway")1187 s3_client = aws_stack.create_external_boto_client("s3")1188 try:1189 bucket_name = f"test-bucket-{short_uid()}"1190 apigateway_name = f"test-api-{short_uid()}"1191 object_name = "test.json"1192 object_content = '{ "success": "true" }'1193 object_content_type = "application/json"1194 api = apigw_client.create_rest_api(name=apigateway_name)1195 api_id = api["id"]1196 s3_client.create_bucket(Bucket=bucket_name)1197 s3_client.put_object(1198 Bucket=bucket_name,1199 Key=object_name,1200 Body=object_content,1201 ContentType=object_content_type,1202 )1203 self.connect_api_gateway_to_s3(bucket_name, object_name, api_id, "GET")1204 apigw_client.create_deployment(restApiId=api_id, stageName="test")1205 url = path_based_url(api_id, "test", f"/{object_name}")1206 result = requests.get(url)1207 self.assertEqual(200, result.status_code)1208 self.assertEqual(object_content, result.text)1209 self.assertEqual(object_content_type, result.headers["content-type"])1210 finally:1211 # clean up1212 apigw_client.delete_rest_api(restApiId=api_id)1213 s3_client.delete_object(Bucket=bucket_name, Key=object_name)1214 s3_client.delete_bucket(Bucket=bucket_name)1215 def test_api_mock_integration_response_params(self):1216 # apigw_client = aws_stack.create_external_boto_client('apigateway')1217 resps = [1218 {1219 "statusCode": "204",1220 "httpMethod": "OPTIONS",1221 "responseParameters": {1222 "method.response.header.Access-Control-Allow-Methods": "'POST,OPTIONS'",1223 "method.response.header.Vary": "'Origin'",1224 },1225 }1226 ]1227 api_id = self.create_api_gateway_and_deploy(1228 integration_type="MOCK", integration_responses=resps1229 )1230 url = path_based_url(api_id=api_id, stage_name=self.TEST_STAGE_NAME, path="/")1231 result = requests.options(url)1232 self.assertLess(result.status_code, 400)1233 self.assertEqual("Origin", result.headers.get("vary"))1234 self.assertEqual("POST,OPTIONS", result.headers.get("Access-Control-Allow-Methods"))1235 def test_api_gateway_update_resource_path_part(self):1236 apigw_client = aws_stack.connect_to_service("apigateway")1237 api = apigw_client.create_rest_api(name="test-api", description="")1238 api_id = api["id"]1239 root_res_id = apigw_client.get_resources(restApiId=api_id)["items"][0]["id"]1240 api_resource = apigw_client.create_resource(1241 restApiId=api_id, parentId=root_res_id, pathPart="test"1242 )1243 response = apigw_client.update_resource(1244 restApiId=api_id,...

Full Screen

Full Screen

owner_update_1_7_18.py

Source:owner_update_1_7_18.py Github

copy

Full Screen

1#!/usr/bin/env python32# -*- coding: utf-8 -*-3import re4import argparse5import json6import copy7import urllib.request8import urllib.parse9SCHEME_LOCALUNIT = "personium-localunit"10UNIT_PREFIX = "u0"11CATEGORY_AD = "ad"12CELL_INDEX = f"{UNIT_PREFIX}_{CATEGORY_AD}.cell"13def convert_old_normalized_uri(uri):14 """Function for converting old-style localunit shema url to current-style.15 Before v1.7.17, normalized URI is created by16 replacing path-based URL's schema with `personium-localunit:/`17 Args:18 uri (str): old-style normalized uri19 Returns:20 str or None: Result of converting. None if the input uri is not old-style normalized uri.21 """22 UNIT_URL_DUMMY = "https://impossibleurl.example.com/"23 SCHEME_UNIT_URL_OLD = "personium-localunit:/"24 # uri is not old normalized style25 if not uri.startswith(SCHEME_UNIT_URL_OLD):26 return None27 path_based_url = re.sub(f"^{SCHEME_UNIT_URL_OLD}", UNIT_URL_DUMMY, uri)28 if not re.match(f"^{UNIT_URL_DUMMY}([^/?&#]*)/?", path_based_url):29 return None30 return convert_scheme_http_to_localunit(path_based_url, UNIT_URL_DUMMY, True)31def convert_scheme_http_to_localunit(uri, unit_url, path_based):32 """Function for scheme conversion from http(s).33 Converting url scheme from http(s) to personium-localunit.34 Args:35 uri (str): http(s) schema uri36 unit_url (str): Unit url used to determine localunit.37 path_based (bool): Flag whether input uri is path_based.38 Returns:39 str or None: Result of converting. None if the input uri is None or empty.40 """41 if not uri:42 return None43 if not path_based:44 raise NotImplementedError("non path-based url is not supported")45 if not uri.startswith(unit_url):46 return uri47 result = re.sub(r":/(?P<cell>[^/?&#]*)/?", r":\g<cell>:/",48 uri.replace(unit_url, f"{SCHEME_LOCALUNIT}:/"))49 return result50def main():51 """main function of this script52 Parse arguments and run script.53 """54 parser = argparse.ArgumentParser(55 description="Update entities in ElasticSearch")56 parser.add_argument("host", metavar="host", type=str,57 help="elasticsearch host (ex: es.example.com)")58 parser.add_argument("--port", default=9200,59 help="elasticsearch port (default: 9200)")60 args = parser.parse_args()61 node_url = f"http://{args.host}:{args.port}"62 search_url = f"{node_url}/{CELL_INDEX}/_search"63 req = urllib.request.Request(search_url)64 num_cells = 065 cnt_fetched = 066 cnt_error = 067 cnt_processed = 068 cells_per_request = 1069 with urllib.request.urlopen(req) as res:70 body = res.read()71 body_json = json.loads(body)72 num_cells = body_json['hits']['total']73 print(f"total {num_cells} cells found")74 # process little by little ( per cells_per_request cells )75 for i in range(0, num_cells // cells_per_request + 1):76 list_cells = []77 search_query = urllib.parse.urlencode({78 "size": cells_per_request,79 "from": i * cells_per_request,80 "sort": "_id:asc"81 })82 req = urllib.request.Request(f"{search_url}?{search_query}")83 with urllib.request.urlopen(req) as res:84 body = res.read()85 body_json = json.loads(body)86 list_cells = body_json['hits']['hits']87 cnt_fetched += len(list_cells)88 for target_cell in list_cells:89 hidden_json = copy.deepcopy(target_cell['_source']['h'])90 if 'Owner' not in hidden_json:91 continue92 new_owner = convert_old_normalized_uri(hidden_json['Owner'])93 if not new_owner:94 # skip due to illegal or empty owner95 continue96 hidden_json['Owner'] = new_owner97 update_query = urllib.parse.urlencode({98 "routing": target_cell['_routing'],99 })100 update_body = json.dumps(101 {"doc": {"h": hidden_json}}).encode("utf-8")102 update_url = f"{node_url}/{CELL_INDEX}/{target_cell['_type']}/{target_cell['_id']}/_update?{update_query}"103 headers = {104 "Content-Type": "application/json"105 }106 req = urllib.request.Request(107 update_url, update_body, headers, method="POST")108 try:109 with urllib.request.urlopen(req) as res:110 body = res.read()111 except urllib.error.HTTPError as err:112 print(113 f"Processing cell(_id:{target_cell['_id']}) was failed. {err.reason}")114 cnt_error += 1115 continue116 cnt_processed += 1117 result = [{118 "_id": target_cell['_id'],119 "from": target_cell['_source']['h']['Owner'],120 "to": new_owner121 }]122 print(result)123 print(f"total {cnt_processed} / {cnt_fetched} cells are successfully processed. ( skipped {cnt_fetched-cnt_processed-cnt_error} , failed {cnt_error})")124 refresh_url = f"{node_url}/{CELL_INDEX}/_refresh"125 req = urllib.request.Request(refresh_url)126 print(f"refreshing elasticsearch index: {CELL_INDEX}")127 try:128 with urllib.request.urlopen(req) as res:129 body = res.read()130 except urllib.error.URLError as err:131 print(err)132if __name__ == '__main__':...

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run localstack automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

... View sample repo... View DocumentationSign up Free
›
_

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

...
Sign up with GoogleSign up with Email
Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful