[test] add get_parsed_bbox_points test case, refactor test structure

src/app.py

@@ -1,120 +1,11 @@
-import json
 import time
 from http import HTTPStatus
-from typing import Dict, List
 
-from aws_lambda_powertools.event_handler import content_types
 from aws_lambda_powertools.utilities.typing import LambdaContext
-from aws_lambda_powertools.utilities.parser import BaseModel
 
 from src import app_logger
-from src.io.coordinates_pixel_conversion import get_latlng_to_pixel_coordinates
+from src.io.lambda_helpers import get_parsed_request_body, get_parsed_bbox_points, get_response
 from src.prediction_api.predictors import samexporter_predict
-from src.utilities.constants import CUSTOM_RESPONSE_MESSAGES, DEFAULT_LOG_LEVEL
-from src.utilities.utilities import base64_decode
-
-
-list_float = List[float]
-llist_float = List[list_float]
-
-
-class LatLngDict(BaseModel):
-    lat: float
-    lng: float
-
-
-class RawBBox(BaseModel):
-    ne: LatLngDict
-    sw: LatLngDict
-
-
-class RawPrompt(BaseModel):
-    type: str
-    data: LatLngDict
-    label: int = 0
-
-
-class RawRequestInput(BaseModel):
-    bbox: RawBBox
-    prompt: RawPrompt
-    zoom: int | float
-    source_type: str = "Satellite"
-
-
-class ParsedPrompt(BaseModel):
-    type: str
-    data: llist_float
-    label: int = 0
-
-
-class ParsedRequestInput(BaseModel):
-    bbox: llist_float
-    prompt: ParsedPrompt
-    zoom: int | float
-
-
-def get_response(status: int, start_time: float, request_id: str, response_body: Dict = None) -> str:
-    """
-    Return a response for frontend clients.
-
-    Args:
-        status: status response
-        start_time: request start time (float)
-        request_id: str
-        response_body: dict we embed into our response
-
-    Returns:
-        str: json response
-
-    """
-    app_logger.debug(f"response_body:{response_body}.")
-    response_body["duration_run"] = time.time() - start_time
-    response_body["message"] = CUSTOM_RESPONSE_MESSAGES[status]
-    response_body["request_id"] = request_id
-
-    response = {
-        "statusCode": status,
-        "header": {"Content-Type": content_types.APPLICATION_JSON},
-        "body": json.dumps(response_body),
-        "isBase64Encoded": False
-    }
-    app_logger.debug(f"response type:{type(response)} => {response}.")
-    return json.dumps(response)
-
-
-def get_parsed_bbox_points(request_input: RawRequestInput) -> Dict:
-    app_logger.info(f"try to parsing input request {request_input}...")
-    bbox = request_input["bbox"]
-    app_logger.debug(f"request bbox: {type(bbox)}, value:{bbox}.")
-    ne = bbox["ne"]
-    sw = bbox["sw"]
-    app_logger.debug(f"request ne: {type(ne)}, value:{ne}.")
-    app_logger.debug(f"request sw: {type(sw)}, value:{sw}.")
-    ne_latlng = [float(ne["lat"]), float(ne["lng"])]
-    sw_latlng = [float(sw["lat"]), float(sw["lng"])]
-    bbox = [ne_latlng, sw_latlng]
-    zoom = int(request_input["zoom"])
-    for prompt in request_input["prompt"]:
-        app_logger.debug(f"current prompt: {type(prompt)}, value:{prompt}.")
-        data = prompt["data"]
-        if prompt["type"] == "point":
-            current_point = get_latlng_to_pixel_coordinates(ne, sw, data, zoom, "point")
-            app_logger.debug(f"current prompt: {type(current_point)}, value:{current_point}.")
-            new_prompt_data = [current_point['x'], current_point['y']]
-            app_logger.debug(f"new_prompt_data: {type(new_prompt_data)}, value:{new_prompt_data}.")
-            prompt["data"] = new_prompt_data
-        else:
-            raise ValueError("valid prompt type is only 'point'")
-
-    app_logger.debug(f"bbox => {bbox}.")
-    app_logger.debug(f'request_input-prompt updated => {request_input["prompt"]}.')
-
-    app_logger.info(f"unpacking elaborated {request_input}...")
-    return {
-        "bbox": bbox,
-        "prompt": request_input["prompt"],
-        "zoom": zoom
-    }
 
 
 def lambda_handler(event: dict, context: LambdaContext):
@@ -125,7 +16,7 @@ def lambda_handler(event: dict, context: LambdaContext):
         app_logger.info(f"event version: {event['version']}.")
 
     try:
-        body = get_parsed_request_body(context, event)
+        body = get_parsed_request_body(event, context)
 
         try:
             prompt_latlng = body["prompt"]
@@ -144,28 +35,3 @@ def lambda_handler(event: dict, context: LambdaContext):
 
     app_logger.info(f"response_dumped:{response}...")
     return response
-
-
-def get_parsed_request_body(context, event):
-    app_logger.info(f"event:{json.dumps(event)}...")
-    app_logger.info(f"context:{context}...")
-    try:
-        body = event["body"]
-    except Exception as e_constants1:
-        app_logger.error(f"e_constants1:{e_constants1}.")
-        body = event
-    app_logger.debug(f"body, #1: {type(body)}, {body}...")
-    if isinstance(body, str):
-        body_decoded_str = base64_decode(body)
-        app_logger.debug(f"body_decoded_str: {type(body_decoded_str)}, {body_decoded_str}...")
-        body = json.loads(body_decoded_str)
-    app_logger.info(f"body, #2: {type(body)}, {body}...")
-    try:
-        log_level = 'DEBUG' if body['debug'] else DEFAULT_LOG_LEVEL
-        app_logger.warning(f"set logger level to DEBUG")
-        app_logger.setLevel(log_level)
-    except KeyError:
-        app_logger.warning(f"can't set log level, reset it...")
-        app_logger.setLevel(DEFAULT_LOG_LEVEL)
-    app_logger.warning(f"logger level is {app_logger.log_level}.")
-    return body

src/io/lambda_helpers.py

@@ -0,0 +1,100 @@
+import json
+import time
+from typing import Dict
+
+from aws_lambda_powertools.event_handler import content_types
+
+from src import app_logger
+from src.io.coordinates_pixel_conversion import get_latlng_to_pixel_coordinates
+from src.utilities.constants import CUSTOM_RESPONSE_MESSAGES, DEFAULT_LOG_LEVEL
+from src.utilities.type_hints import RawRequestInput
+from src.utilities.utilities import base64_decode, LogArgumentsDecorator
+
+
+def get_response(status: int, start_time: float, request_id: str, response_body: Dict = None) -> str:
+    """
+    Return a response for frontend clients.
+
+    Args:
+        status: status response
+        start_time: request start time (float)
+        request_id: str
+        response_body: dict we embed into our response
+
+    Returns:
+        str: json response
+
+    """
+    app_logger.debug(f"response_body:{response_body}.")
+    response_body["duration_run"] = time.time() - start_time
+    response_body["message"] = CUSTOM_RESPONSE_MESSAGES[status]
+    response_body["request_id"] = request_id
+
+    response = {
+        "statusCode": status,
+        "header": {"Content-Type": content_types.APPLICATION_JSON},
+        "body": json.dumps(response_body),
+        "isBase64Encoded": False
+    }
+    app_logger.debug(f"response type:{type(response)} => {response}.")
+    return json.dumps(response)
+
+
+def get_parsed_bbox_points(request_input: RawRequestInput) -> Dict:
+    app_logger.info(f"try to parsing input request {request_input}...")
+    bbox = request_input["bbox"]
+    app_logger.debug(f"request bbox: {type(bbox)}, value:{bbox}.")
+    ne = bbox["ne"]
+    sw = bbox["sw"]
+    app_logger.debug(f"request ne: {type(ne)}, value:{ne}.")
+    app_logger.debug(f"request sw: {type(sw)}, value:{sw}.")
+    ne_latlng = [float(ne["lat"]), float(ne["lng"])]
+    sw_latlng = [float(sw["lat"]), float(sw["lng"])]
+    bbox = [ne_latlng, sw_latlng]
+    zoom = int(request_input["zoom"])
+    for prompt in request_input["prompt"]:
+        app_logger.debug(f"current prompt: {type(prompt)}, value:{prompt}.")
+        data = prompt["data"]
+        if prompt["type"] == "point":
+            current_point = get_latlng_to_pixel_coordinates(ne, sw, data, zoom, "point")
+            app_logger.debug(f"current prompt: {type(current_point)}, value:{current_point}.")
+            new_prompt_data = [current_point['x'], current_point['y']]
+            app_logger.debug(f"new_prompt_data: {type(new_prompt_data)}, value:{new_prompt_data}.")
+            prompt["data"] = new_prompt_data
+        else:
+            raise ValueError("valid prompt type is only 'point'")
+
+    app_logger.debug(f"bbox => {bbox}.")
+    app_logger.debug(f'request_input-prompt updated => {request_input["prompt"]}.')
+
+    app_logger.info(f"unpacking elaborated {request_input}...")
+    return {
+        "bbox": bbox,
+        "prompt": request_input["prompt"],
+        "zoom": zoom
+    }
+
+
+def get_parsed_request_body(event, context):
+    app_logger.info(f"event:{json.dumps(event)}...")
+    app_logger.info(f"context:{context}...")
+    try:
+        body = event["body"]
+    except Exception as e_constants1:
+        app_logger.error(f"e_constants1:{e_constants1}.")
+        body = event
+    app_logger.debug(f"body, #1: {type(body)}, {body}...")
+    if isinstance(body, str):
+        body_decoded_str = base64_decode(body)
+        app_logger.debug(f"body_decoded_str: {type(body_decoded_str)}, {body_decoded_str}...")
+        body = json.loads(body_decoded_str)
+    app_logger.info(f"body, #2: {type(body)}, {body}...")
+    try:
+        log_level = 'DEBUG' if body['debug'] else DEFAULT_LOG_LEVEL
+        app_logger.warning(f"set logger level to DEBUG")
+        app_logger.setLevel(log_level)
+    except KeyError:
+        app_logger.warning(f"can't set log level, reset it...")
+        app_logger.setLevel(DEFAULT_LOG_LEVEL)
+    app_logger.warning(f"logger level is {app_logger.log_level}.")
+    return body

src/utilities/type_hints.py

@@ -1,11 +1,48 @@
 """custom type hints"""
+from pydantic import BaseModel
 from typing import TypedDict
 
 
 ts_dict_str2 = dict[str, str]
 ts_dict_str3 = dict[str, str, any]
 ts_ddict1 = dict[str, dict[str, any], dict, dict, any]
-# ts_list_float = list[float, float]
+list_float = list[float]
+llist_float = list[list_float]
+
+
+class LatLngDict(BaseModel):
+    lat: float
+    lng: float
+
+
+class RawBBox(BaseModel):
+    ne: LatLngDict
+    sw: LatLngDict
+
+
+class RawPrompt(BaseModel):
+    type: str
+    data: LatLngDict
+    label: int = 0
+
+
+class RawRequestInput(BaseModel):
+    bbox: RawBBox
+    prompt: RawPrompt
+    zoom: int | float
+    source_type: str = "Satellite"
+
+
+class ParsedPrompt(BaseModel):
+    type: str
+    data: llist_float
+    label: int = 0
+
+
+class ParsedRequestInput(BaseModel):
+    bbox: llist_float
+    prompt: ParsedPrompt
+    zoom: int | float
 
 
 class PixelCoordinate(TypedDict):

tests/events/get_parsed_bbox_points.json

@@ -0,0 +1,208 @@
+{
+  "europe": {
+    "input": {
+      "request_input": {
+        "bbox": {
+          "ne": {
+            "lat": 38.03932961278458,
+            "lng": 15.36808069832851
+          },
+          "sw": {
+            "lat": 37.455509218936974,
+            "lng": 14.632807441554068
+          }
+        },
+        "prompt": [
+          {
+            "type": "point",
+            "data": {
+              "lat": 37.0,
+              "lng": 15.0
+            },
+            "label": 0
+          }
+        ],
+        "zoom": 10,
+        "source_type": "Satellite",
+        "debug": true
+      }
+    },
+    "output": {
+      "bbox": [
+        [
+          38.03932961278458,
+          15.36808069832851
+        ],
+        [
+          37.455509218936974,
+          14.632807441554068
+        ]
+      ],
+      "prompt": [
+        {
+          "type": "point",
+          "data": [
+            267,
+            954
+          ],
+          "label": 0
+        }
+      ],
+      "zoom": 10
+    }
+  },
+  "north_america": {
+    "input": {
+      "request_input": {
+        "bbox": {
+          "ne": {
+            "lat": 44.918201144476456,
+            "lng": -111.7859533595829
+          },
+          "sw": {
+            "lat": 39.35135789075441,
+            "lng": -123.03595335958288
+          }
+        },
+        "prompt": [
+          {
+            "type": "point",
+            "data": {
+              "lat": 41.142370385846974,
+              "lng": -112.64967251996322
+            },
+            "label": 0
+          }
+        ],
+        "zoom": 7,
+        "source_type": "Satellite"
+      }
+    },
+    "output": {
+      "bbox": [
+        [
+          44.918201144476456,
+          -111.7859533595829
+        ],
+        [
+          39.35135789075441,
+          -123.03595335958288
+        ]
+      ],
+      "prompt": [
+        {
+          "type": "point",
+          "data": [
+            946,
+            471
+          ],
+          "label": 0
+        }
+      ],
+      "zoom": 7
+    }
+  },
+  "oceania": {
+    "input": {
+      "request_input": {
+        "bbox": {
+          "ne": {
+            "lat": -1.4939713066293112,
+            "lng": 155.30273437500003
+          },
+          "sw": {
+            "lat": -52.32191088594772,
+            "lng": 65.30273437500001
+          }
+        },
+        "prompt": [
+          {
+            "type": "point",
+            "data": {
+              "lat": -42.51390169585491,
+              "lng": 147.0746141732014
+            },
+            "label": 0
+          }
+        ],
+        "zoom": 4,
+        "source_type": "Satellite"
+      }
+    },
+    "output": {
+      "bbox": [
+        [
+          -1.4939713066293112,
+          155.30273437500003
+        ],
+        [
+          -52.32191088594772,
+          65.30273437500001
+        ]
+      ],
+      "prompt": [
+        {
+          "type": "point",
+          "data": [
+            930,
+            518
+          ],
+          "label": 0
+        }
+      ],
+      "zoom": 4
+    }
+  },
+  "south_america": {
+    "input": {
+      "request_input": {
+        "bbox": {
+          "ne": {
+            "lat": -24.80663308621806,
+            "lng": -63.98446722241735
+          },
+          "sw": {
+            "lat": -31.42861774444052,
+            "lng": -75.23446722241735
+          }
+        },
+        "prompt": [
+          {
+            "type": "point",
+            "data": {
+              "lat": -29.911045956544207,
+              "lng": -64.78226803685988
+            },
+            "label": 0
+          }
+        ],
+        "zoom": 7,
+        "source_type": "Satellite"
+      }
+    },
+    "output": {
+      "bbox": [
+        [
+          -24.80663308621806,
+          -63.98446722241735
+        ],
+        [
+          -31.42861774444052,
+          -75.23446722241735
+        ]
+      ],
+      "prompt": [
+        {
+          "type": "point",
+          "data": [
+            952,
+            524
+          ],
+          "label": 0
+        }
+      ],
+      "zoom": 7
+    }
+  }
+}
+

tests/io/test_coordinates_pixel_conversion.py

@@ -1,36 +1,13 @@
-import json
-from unittest import TestCase
+from tests.io.test_utilities import fn_reading_json_inputs_outputs__test
 
-from src.io.coordinates_pixel_conversion import get_latlng2pixel_projection, get_point_latlng_to_pixel_coordinates, \
-    get_latlng_to_pixel_coordinates
 
-from tests import TEST_EVENTS_FOLDER
+def test_get_latlng2pixel_projection():
+    fn_reading_json_inputs_outputs__test(name_fn="get_latlng2pixel_projection")
 
 
-names_fn_dict = {
-    "get_latlng2pixel_projection": get_latlng2pixel_projection,
-    "get_point_latlng_to_pixel_coordinates": get_point_latlng_to_pixel_coordinates,
-    "get_latlng_to_pixel_coordinates": get_latlng_to_pixel_coordinates
-}
+def test_get_point_latlng_to_pixel_coordinates():
+    fn_reading_json_inputs_outputs__test(name_fn="get_point_latlng_to_pixel_coordinates")
 
 
-def test_fn_reading_json_inputs_outputs(name_fn):
-    fn = names_fn_dict[name_fn]
-
-    with open(TEST_EVENTS_FOLDER / f"{name_fn}.json") as tst_json:
-        inputs_outputs = json.load(tst_json)
-        for k, input_output in inputs_outputs.items():
-            print(f"k:{k}.")
-            output = fn(**input_output["input"])
-            assert output == input_output["output"]
-
-
-class Test(TestCase):
-    def test_get_latlng2pixel_projection(self):
-        test_fn_reading_json_inputs_outputs("get_latlng2pixel_projection")
-
-    def test_get_point_latlng_to_pixel_coordinates(self):
-        test_fn_reading_json_inputs_outputs("get_point_latlng_to_pixel_coordinates")
-
-    def test_get_latlng_to_pixel_coordinates(self):
-        test_fn_reading_json_inputs_outputs("get_latlng_to_pixel_coordinates")
+def test_get_latlng_to_pixel_coordinates():
+    fn_reading_json_inputs_outputs__test(name_fn="get_latlng_to_pixel_coordinates")

tests/io/test_lambda_helpers.py

@@ -0,0 +1,5 @@
+from tests.io.test_utilities import fn_reading_json_inputs_outputs__test
+
+
+def test_get_parsed_bbox_points():
+    fn_reading_json_inputs_outputs__test(name_fn="get_parsed_bbox_points")

tests/io/test_utilities.py

@@ -0,0 +1,26 @@
+import json
+
+from src.io.coordinates_pixel_conversion import get_latlng2pixel_projection, get_point_latlng_to_pixel_coordinates, \
+    get_latlng_to_pixel_coordinates
+from src.io.lambda_helpers import get_parsed_bbox_points
+
+from tests import TEST_EVENTS_FOLDER
+
+
+names_fn_dict = {
+    "get_latlng2pixel_projection": get_latlng2pixel_projection,
+    "get_point_latlng_to_pixel_coordinates": get_point_latlng_to_pixel_coordinates,
+    "get_latlng_to_pixel_coordinates": get_latlng_to_pixel_coordinates,
+    "get_parsed_bbox_points": get_parsed_bbox_points
+}
+
+
+def fn_reading_json_inputs_outputs__test(name_fn):
+    fn = names_fn_dict[name_fn]
+
+    with open(TEST_EVENTS_FOLDER / f"{name_fn}.json") as tst_json:
+        inputs_outputs = json.load(tst_json)
+        for k, input_output in inputs_outputs.items():
+            print(f"k:{k}.")
+            output = fn(**input_output["input"])
+            assert output == input_output["output"]