🐛 fix: GTC-3062 implementation replacement on redirect (#176) (#178)

terraform/modules/content_delivery_network/lambda_functions/redirect_s3_404/src/lambda_function.py

@@ -1,12 +1,28 @@
 # mypy: ignore-errors
+from urllib.parse import urlparse, urlunparse
 
 
 def handler(event, context):
-    """
-    This function updates the HTTP status code in the response to 307, to redirect to another
-    path (cache behavior) that has a different origin configured. Note the following:
+    """This function updates the HTTP status code in the response to 307, to
+    redirect to another path (cache behavior) that has a different origin
+    configured.
+
+    Note the following:
     1. The function is triggered in an origin response
     2. The response status from the origin server is an error status code (404)
+
+    The pattern for the incoming request uri:
+
+    /{dataset}/{version}/{implementation}/{z}/{x}/{y}.(png|pbf)
+
+    results in a redirect response like:
+
+    /{dataset}/{version}/dynamic/{z}/{x}/{y}.(png|pbf)?implementation={implementation}
+
+    *Implementation Note: The request URI and redirect URL are relative (with a leading '/'). When python `splits` the string
+    of a relative URL, the first element of the list is the empty string (''). Therefore,
+
+    /{dataset}/{version}/{implementation}/{z}/{x}/{y}.(png|pbf) has seven (7) elements after splitting.
     """
 
     response = event["Records"][0]["cf"]["response"]
@@ -17,38 +33,56 @@ def handler(event, context):
     # custom origin is tile cache app. URL is passed via custom header set in cloud front
     # (env variables are not support for Lambda@Edge)
 
-    if int(response["status"]) == 404 and is_tile(request["uri"]):
+    parsed_url = urlparse(request["uri"])
+    path_parts = parsed_url.path.split("/")
 
-        implementation = get_implementation(request["uri"])
+    if int(response["status"]) == 404 and is_tile(path_parts):
+        implementation = replace_implementation_in_path(path_parts)
+        querystring = add_implementation_to_query_params(
+            implementation, request["querystring"]
+        )
+        updated_url = urlunparse(
+            parsed_url._replace(path="/".join(path_parts), query=querystring)
+        )
+        update_headers_for_redirect(headers, updated_url)
+        return build_redirect_response(headers)
 
-        redirect_path = request["uri"].replace(implementation, "dynamic")
+    return response
 
-        if request["querystring"]:
-            querystring = f"{request['querystring']}&implementation={implementation}"
-        else:
-            querystring = f"implementation={implementation}"
 
-        redirect_path += f"?{querystring}"
+def is_tile(uri):
+    """The resource is a tile if its last path element ends in .png or .pbf."""
+    print("REQUEST URI", "/".join(uri))
+    return len(uri) == 7 and uri[6][-4:] in [".png", ".pbf"]
 
-        headers["location"] = [{"key": "Location", "value": redirect_path}]
-        headers["content-type"] = [{"key": "Content-Type", "value": "application/json"}]
-        headers["content-encoding"] = [{"key": "Content-Encoding", "value": "UTF-8"}]
 
-        response = {
-            "status": "307",
-            "statusDescription": "Temporary Redirect",
-            "headers": headers,
-        }
+def replace_implementation_in_path(path_parts):
+    """Replace the implementation path segment with "dynamic" and return the
+    original implementation."""
+    implementation = path_parts[3]
+    path_parts[3] = "dynamic"
+    return implementation
 
-    return response
 
+def add_implementation_to_query_params(implementation, query_string):
+    implementation_param = f"implementation={implementation}"
+    if query_string:
+        querystring = f"{query_string}&{implementation_param}"
+    else:
+        querystring = implementation_param
+    return querystring
 
-def is_tile(uri):
-    print("REQUEST URI", uri)
-    parts = uri.split("/")
-    return len(parts) == 7 and parts[6][-4:] in [".png", ".pbf"]
+
+def update_headers_for_redirect(headers, updated_url):
+    headers["location"] = [{"key": "Location", "value": updated_url}]
+    headers["content-type"] = [{"key": "Content-Type", "value": "application/json"}]
+    headers["content-encoding"] = [{"key": "Content-Encoding", "value": "UTF-8"}]
 
 
-def get_implementation(uri):
-    parts = uri.split("/")
-    return parts[3]
+def build_redirect_response(headers):
+    response = {
+        "status": "307",
+        "statusDescription": "Temporary Redirect",
+        "headers": headers,
+    }
+    return response

tests/lambda/test_redirect_s3_404.py

@@ -0,0 +1,165 @@
+from terraform.modules.content_delivery_network.lambda_functions.redirect_s3_404.src.lambda_function import (
+    handler,
+)
+
+
+def create_event(status="404", uri="not important for this test", querystring=""):
+    """Helper method to create the base event dictionary with customizable
+    status, URI, and query string."""
+    return {
+        "Records": [
+            {
+                "cf": {
+                    "response": {
+                        "status": status,
+                        "headers": {
+                            "content-type": [
+                                {"key": "Content-Type", "value": "application/json"}
+                            ]
+                        },
+                    },
+                    "request": {
+                        "uri": uri,
+                        "querystring": querystring,
+                    },
+                }
+            }
+        ]
+    }
+
+
+class TestRedirectOnlyTileRequestsThatAreNotFound:
+    def test_handler_does_not_modify_response_if_status_is_something_other_than_404(
+        self,
+    ):
+        event = create_event(status="200")
+
+        response = handler(event, {})
+
+        assert response == {
+            "status": "200",
+            "headers": {
+                "content-type": [{"key": "Content-Type", "value": "application/json"}]
+            },
+        }
+
+    def test_handler_creates_a_redirect_response_if_status_is_404_and_is_a_png_tile(
+        self,
+    ):
+        event = create_event(
+            uri="/sbtn_natural_forests_map/v202310/natural_forest/10/20/30.png"
+        )
+
+        response = handler(event, {})
+
+        assert (
+            response.items()
+            >= {
+                "status": "307",
+                "statusDescription": "Temporary Redirect",
+            }.items()
+        )
+
+    def test_handler_creates_a_redirect_response_if_status_is_404_and_is_a_pbf_tile(
+        self,
+    ):
+        event = create_event(
+            uri="/sbtn_natural_forests_map/v202310/natural_forest/10/20/30.pbf"
+        )
+
+        response = handler(event, {})
+
+        assert (
+            response.items()
+            >= {
+                "status": "307",
+                "statusDescription": "Temporary Redirect",
+            }.items()
+        )
+
+    def test_handler_does_not_modify_response_if_request_is_a_resource_other_than_a_tile(
+        self,
+    ):
+        event = create_event(
+            uri="/sbtn_natural_forests_map/v202310/natural_forest/10/20/30.txt"
+        )
+
+        response = handler(event, {})
+
+        assert response == {
+            "status": "404",
+            "headers": {
+                "content-type": [{"key": "Content-Type", "value": "application/json"}]
+            },
+        }
+
+
+class TestRedirectsToADynamicTileResource:
+    def test_original_implementation_is_replaced_with_dynamic(self):
+        implementation = "natural_forest"
+        event = create_event(
+            uri=f"/sbtn_natural_forests_map/v202310/{implementation}/10/20/30.png"
+        )
+
+        response = handler(event, {})
+
+        assert response["headers"]["location"][0]["key"] == "Location"
+        assert (
+            "/sbtn_natural_forests_map/v202310/dynamic/10/20/30.png"
+            in response["headers"]["location"][0]["value"]
+        )
+
+
+class TestAddsOriginalImplementationToTheExistingQueryParams:
+    def test_original_implementation_is_added_to_the_list_of_query_parameters(self):
+        implementation = "natural_forest"
+        event = create_event(
+            uri=f"/sbtn_natural_forests_map/v202310/{implementation}/10/20/30.png",
+            querystring="some_param=30",
+        )
+
+        response = handler(event, {})
+
+        assert response["headers"]["location"][0]["key"] == "Location"
+        assert (
+            "?some_param=30&implementation=natural_forest"
+            in response["headers"]["location"][0]["value"]
+        )
+
+    def test_original_implementation_is_added_as_a_query_parameter(self):
+        implementation = "natural_forest"
+        event = create_event(
+            uri=f"/sbtn_natural_forests_map/v202310/{implementation}/10/20/30.png"
+        )
+
+        response = handler(event, {})
+
+        assert response["headers"]["location"][0]["key"] == "Location"
+        assert (
+            "?implementation=natural_forest"
+            in response["headers"]["location"][0]["value"]
+        )
+
+
+class TestStandardHeaderInfoIsAddedToRedirect:
+    def test_content_type_is_set(self):
+        event = create_event(
+            uri="/sbtn_natural_forests_map/v202310/default/10/20/30.png"
+        )
+
+        response = handler(event, {})
+
+        assert response["headers"]["content-type"] == [
+            {"key": "Content-Type", "value": "application/json"}
+        ]
+
+    def test_content_encoding_is_set(self):
+        event = create_event(
+            uri="/sbtn_natural_forests_map/v202310/default/10/20/30.png"
+        )
+
+        response = handler(event, {})
+
+        assert response["headers"]["content-encoding"] == [
+            {"key": "Content-Encoding", "value": "UTF-8"}
+        ]