chore: Refactor TenantIdValidationError handling for consistency (#8474)

* Refactor TenantIdValidationError handling for consistency

* Add unit test for TenantIdValidationError handling in API Gateway service

---------

Co-authored-by: Chengjun Li <>
Co-authored-by: vicheey <181402101+vicheey@users.noreply.github.com>

Author: licjun

samcli/local/apigw/local_apigw_service.py

@@ -8,7 +8,7 @@
 from time import time
 from typing import Any, Dict, List, Optional, Tuple, Union
 
-from flask import Flask, Request, jsonify, make_response, request
+from flask import Flask, Request, request
 from werkzeug.datastructures import Headers
 from werkzeug.routing import BaseConverter
 from werkzeug.serving import WSGIRequestHandler
@@ -740,8 +740,7 @@ def _request_handler(self, **kwargs):
             # invoke the route's Lambda function
             lambda_response = self._invoke_lambda_function(route.function_name, route_lambda_event, tenant_id)
         except TenantIdValidationError as e:
-            response_data = jsonify({"message": str(e)})
-            endpoint_service_error = make_response(response_data, 400)  # HTTP 400 Bad Request
+            endpoint_service_error = ServiceErrorResponses.tenant_id_validation_error(str(e))
         except FunctionNotFound:
             endpoint_service_error = ServiceErrorResponses.lambda_not_found_response()
         except UnsupportedInlineCodeError:

samcli/local/apigw/service_error_responses.py

@@ -19,6 +19,7 @@ class ServiceErrorResponses:
     HTTP_STATUS_CODE_502 = 502
     HTTP_STATUS_CODE_403 = 403
     HTTP_STATUS_CODE_401 = 401
+    HTTP_STATUS_CODE_400 = 400
 
     @staticmethod
     def lambda_authorizer_unauthorized() -> Response:
@@ -110,3 +111,21 @@ def container_creation_failed(message):
         """
         response_data = jsonify({"message": message})
         return make_response(response_data, ServiceErrorResponses.HTTP_STATUS_CODE_501)
+
+    @staticmethod
+    def tenant_id_validation_error(message: str) -> Response:
+        """
+        Constructs a Flask Response for when tenant ID validation fails
+
+        Parameters
+        ----------
+        message : str
+            The error message describing the validation failure
+
+        Returns
+        -------
+        Response
+            A Flask Response object with HTTP 400 status
+        """
+        response_data = jsonify({"message": message})
+        return make_response(response_data, ServiceErrorResponses.HTTP_STATUS_CODE_400)

tests/unit/local/apigw/test_local_apigw_service.py

@@ -27,7 +27,7 @@
 from samcli.local.apigw.service_error_responses import ServiceErrorResponses
 from samcli.local.docker.exceptions import DockerContainerCreationFailedException
 from samcli.local.lambdafn.exceptions import FunctionNotFound
-from samcli.commands.local.lib.exceptions import UnsupportedInlineCodeError
+from samcli.commands.local.lib.exceptions import TenantIdValidationError, UnsupportedInlineCodeError
 
 
 class TestApiGatewayService(TestCase):
@@ -460,6 +460,29 @@ def test_request_handles_error_when_invoke_function_with_inline_code(
 
         self.assertEqual(response, not_implemented_response_mock)
 
+    @patch.object(LocalApigwService, "get_request_methods_endpoints")
+    @patch("samcli.local.apigw.local_apigw_service.ServiceErrorResponses")
+    @patch("samcli.local.apigw.local_apigw_service.LocalApigwService._generate_lambda_event")
+    def test_request_handles_error_when_tenant_id_validation_fails(
+        self, generate_mock, service_error_responses_patch, request_mock
+    ):
+        generate_mock.return_value = {}
+        tenant_id_error_response_mock = Mock()
+        self.api_service._get_current_route = MagicMock()
+        self.api_service._get_current_route.return_value.payload_format_version = "2.0"
+        self.api_service._get_current_route.return_value.authorizer_object = None
+        self.api_service._get_current_route.methods = []
+
+        service_error_responses_patch.tenant_id_validation_error.return_value = tenant_id_error_response_mock
+
+        self.lambda_runner.invoke.side_effect = TenantIdValidationError(
+            "The invoked function is enabled with tenancy configuration. Add a valid tenant ID in your request and try again."
+        )
+        request_mock.return_value = ("test", "test")
+        response = self.api_service._request_handler()
+
+        self.assertEqual(response, tenant_id_error_response_mock)
+
     @patch.object(LocalApigwService, "get_request_methods_endpoints")
     @patch("samcli.local.apigw.local_apigw_service.ServiceErrorResponses")
     @patch("samcli.local.apigw.local_apigw_service.LocalApigwService._generate_lambda_event")