APIGatewayManagement() - can now be called with an APIgw endpoint URL (…

…getmoto#8013)

moto/apigatewaymanagementapi/responses.py

@@ -3,7 +3,7 @@
 import json
 from typing import Any
 
-from moto.core.responses import BaseResponse
+from moto.core.responses import TYPE_RESPONSE, BaseResponse
 
 from .models import ApiGatewayManagementApiBackend, apigatewaymanagementapi_backends
 
@@ -46,3 +46,16 @@ def post_to_connection(self) -> str:
             connection_id=connection_id,
         )
         return "{}"
+
+    @staticmethod
+    def connect_to_apigateway(  # type: ignore[misc]
+        request: Any, full_url: str, headers: Any
+    ) -> TYPE_RESPONSE:
+        self = ApiGatewayManagementApiResponse()
+        self.setup_class(request, full_url, headers, use_raw_body=True)
+        if request.method == "GET":
+            return 200, {}, self.get_connection()
+        elif request.method == "DELETE":
+            return 200, {}, self.delete_connection()
+        else:
+            return 200, {}, self.post_to_connection()

moto/apigatewaymanagementapi/urls.py

@@ -2,12 +2,15 @@
 
 from .responses import ApiGatewayManagementApiResponse
 
-url_bases = [r"https?://execute-api\.(.+)\.amazonaws\.com"]
+# execute-api.us-east-1.amazonaws.com
+# api_id.execute-api.us-east-1.amazonaws.com
+url_bases = [r"https?://(.+\.)*execute-api\.(.+)\.amazonaws\.com"]
 
 
 response = ApiGatewayManagementApiResponse()
 
 
 url_paths = {
     "{0}/@connections/(?P<connection_id>[^/]+)$": response.dispatch,
+    "{0}/(?P<stage_name>.+/)+@connections/(?P<connection_id>[^/]+)$": response.connect_to_apigateway,
 }

moto/backend_index.py

@@ -8,7 +8,7 @@
     ("apigateway", re.compile("https?://apigateway\\.(.+)\\.amazonaws.com")),
     (
         "apigatewaymanagementapi",
-        re.compile("https?://execute-api\\.(.+)\\.amazonaws\\.com"),
+        re.compile("https?://(.+\\.)*execute-api\\.(.+)\\.amazonaws\\.com"),
     ),
     ("appconfig", re.compile("https?://appconfig\\.(.+)\\.amazonaws\\.com")),
     (

tests/test_apigateway/test_apigateway_integration.py

@@ -112,7 +112,7 @@ def test_aws_integration_dynamodb_multiple_stages():
         f"https://{api_id}.execute-api.us-west-2.amazonaws.com/prod",
         json={"TableName": table_name, "Item": {"name": {"S": "the-key"}}},
     )
-    assert res.status_code == 400
+    assert res.status_code in [400, 404]
 
 
 @mock_aws

tests/test_apigatewaymanagementapi/test_apigatewaymanagementapi.py

@@ -14,8 +14,9 @@
 @mock_aws
 def test_delete_connection():
     if settings.TEST_SERVER_MODE and not is_werkzeug_2_3_x():
-        # URL matching changed between 2.2.x and 2.3.x
-        # 2.3.x has no problem matching the root path '/@connections', but 2.2.x refuses
+        # URL matching changed in 2.2.x - the root path '/@connections' cannot be found
+        # 2.1.x works, 2.2.x is broken, and 2.3.x (and 3.x) works again
+        # We could only skip 2.2.x - but only supporting >= 2.3.x is easier
         raise SkipTest("Can't test this in older werkzeug versions")
     client = boto3.client("apigatewaymanagementapi", region_name="eu-west-1")
     # NO-OP
@@ -54,3 +55,17 @@ def test_post_to_connection():
     if not settings.TEST_SERVER_MODE:
         backend = apigatewaymanagementapi_backends[DEFAULT_ACCOUNT_ID]["ap-southeast-1"]
         assert backend.connections["anything"].data == b"my first bytesmore bytes"
+
+
+@mock_aws
+def test_post_to_connection_using_endpoint_url():
+    if not settings.TEST_DECORATOR_MODE:
+        raise SkipTest("Can only test this with decorators")
+    client = boto3.client(
+        "apigatewaymanagementapi",
+        "us-west-2",
+        endpoint_url="https://someapiid.execute-api.us-west-2.amazonaws.com/stage",
+    )
+
+    client.get_connection(ConnectionId="anything")
+    client.post_to_connection(ConnectionId="n/a", Data=b"some data")