tests/test_lambda_app.py

import json
import time
import unittest
from unittest.mock import patch

from samgis import IS_AWS_LAMBDA

if IS_AWS_LAMBDA:
    try:
        from awslambdaric.lambda_context import LambdaContext

        from samgis.io import wrappers_helpers
        from wrappers import lambda_wrapper
        from tests.local_tiles_http_server import LocalTilesHttpServer


        class TestLambdaApp(unittest.TestCase):
            @patch.object(time, "time")
            @patch.object(lambda_wrapper, "samexporter_predict")
            @patch.object(lambda_wrapper, "get_parsed_bbox_points")
            @patch.object(lambda_wrapper, "get_parsed_request_body")
            def test_lambda_handler_500(
                    self,
                    get_parsed_request_body_mocked,
                    get_parsed_bbox_points_mocked,
                    samexporter_predict_mocked,
                    time_mocked
            ):
                from wrappers.lambda_wrapper import lambda_handler

                time_mocked.return_value = 0
                get_parsed_request_body_mocked.value = {}
                get_parsed_bbox_points_mocked.return_value = {"bbox": "bbox_object", "prompt": "prompt_object",
                                                              "zoom": 1}
                samexporter_predict_mocked.side_effect = ValueError("I raise a value error!")

                event = {"body": {}, "version": 1.0}
                lambda_context = LambdaContext(
                    invoke_id="test_invoke_id",
                    client_context=None,
                    cognito_identity=None,
                    epoch_deadline_time_in_ms=time.time()
                )
                expected_response_500 = '{"statusCode": 500, "header": {"Content-Type": "application/json"}, '
                expected_response_500 += '"body": "{\\"duration_run\\": 0, \\"message\\": \\"Internal server error\\", '
                expected_response_500 += '\\"request_id\\": \\"test_invoke_id\\"}", "isBase64Encoded": false}'

                assert lambda_handler(event, lambda_context) == expected_response_500


        @patch.object(time, "time")
        @patch.object(lambda_wrapper, "get_parsed_request_body")
        def test_lambda_handler_400(self, get_parsed_request_body_mocked, time_mocked):
            from wrappers.lambda_wrapper import lambda_handler

            time_mocked.return_value = 0
            get_parsed_request_body_mocked.return_value = {}

            event = {"body": {}, "version": 1.0}
            lambda_context = LambdaContext(
                invoke_id="test_invoke_id",
                client_context=None,
                cognito_identity=None,
                epoch_deadline_time_in_ms=time.time()
            )

            assert lambda_handler(event, lambda_context) == (
                '{"statusCode": 400, "header": {"Content-Type": "application/json"}, '
                '"body": "{\\"duration_run\\": 0, \\"message\\": \\"Bad Request\\", '
                '\\"request_id\\": \\"test_invoke_id\\"}", "isBase64Encoded": false}')


        @patch.object(time, "time")
        def test_lambda_handler_422(self, time_mocked):
            from wrappers.lambda_wrapper import lambda_handler

            time_mocked.return_value = 0
            event = {"body": {}, "version": 1.0}
            lambda_context = LambdaContext(
                invoke_id="test_invoke_id",
                client_context=None,
                cognito_identity=None,
                epoch_deadline_time_in_ms=time.time()
            )

            response_422 = lambda_handler(event, lambda_context)
            expected_response_422 = '{"statusCode": 422, "header": {"Content-Type": "application/json"}, '
            expected_response_422 += '"body": "{\\"duration_run\\": 0, \\"message\\": \\"Missing required parameter\\", '
            expected_response_422 += '\\"request_id\\": \\"test_invoke_id\\"}", "isBase64Encoded": false}'

            assert response_422 == expected_response_422


        @patch.object(time, "time")
        @patch.object(lambda_wrapper, "samexporter_predict")
        @patch.object(lambda_wrapper, "get_response")
        @patch.object(lambda_wrapper, "get_parsed_bbox_points")
        @patch.object(lambda_wrapper, "get_parsed_request_body")
        def test_lambda_handler_200_mocked(
                self,
                get_parsed_request_body_mocked,
                get_parsed_bbox_points_mocked,
                get_response_mocked,
                samexporter_predict_mocked,
                time_mocked
        ):
            from wrappers.lambda_wrapper import lambda_handler
            from tests import TEST_EVENTS_FOLDER

            time_mocked.return_value = 0
            get_parsed_request_body_mocked.value = {}
            get_parsed_bbox_points_mocked.return_value = {"bbox": "bbox_object", "prompt": "prompt_object", "zoom": 1}

            response_type = "200"
            with open(TEST_EVENTS_FOLDER / "get_response.json") as tst_json_get_response:
                get_response_io = json.load(tst_json_get_response)

            input_200 = {
                "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": "OpenStreetMap.Mapnik",
                "debug": True
            }

            samexporter_predict_output = get_response_io[response_type]["input"]
            samexporter_predict_mocked.return_value = samexporter_predict_output
            samexporter_predict_mocked.side_effect = None
            get_response_mocked.return_value = get_response_io[response_type]["output"]

            event = {"body": input_200, "version": 1.0}

            lambda_context = LambdaContext(
                invoke_id="test_invoke_id",
                client_context=None,
                cognito_identity=None,
                epoch_deadline_time_in_ms=time.time()
            )

            response_200 = lambda_handler(event, lambda_context)
            expected_response_200 = get_response_io[response_type]["output"]
            print(f"types: response_200:{type(response_200)}, expected:{type(expected_response_200)}.")
            assert response_200 == expected_response_200


        @patch.object(wrappers_helpers, "get_url_tile")
        def test_lambda_handler_200_real_single_multi_point(self, get_url_tile_mocked):
            import xyzservices
            import shapely

            from wrappers.lambda_wrapper import lambda_handler
            from tests import LOCAL_URL_TILE, TEST_EVENTS_FOLDER

            local_tile_provider = xyzservices.TileProvider(name="local_tile_provider", url=LOCAL_URL_TILE,
                                                           attribution="")
            get_url_tile_mocked.return_value = local_tile_provider
            fn_name = "lambda_handler"
            invoke_id = "test_invoke_id"

            for json_filename in [
                "single_point",
                "multi_prompt",
                "single_rectangle"
            ]:
                with open(TEST_EVENTS_FOLDER / f"{fn_name}_{json_filename}.json") as tst_json:
                    inputs_outputs = json.load(tst_json)
                    lambda_context = LambdaContext(
                        invoke_id=invoke_id,
                        client_context=None,
                        cognito_identity=None,
                        epoch_deadline_time_in_ms=time.time()
                    )
                    expected_response_dict = inputs_outputs["output"]
                    listen_port = 8000
                    expected_response_body = json.loads(expected_response_dict["body"])

                    with LocalTilesHttpServer.http_server("localhost", listen_port, directory=TEST_EVENTS_FOLDER):
                        input_event = inputs_outputs["input"]
                        input_event_body = json.loads(input_event["body"])
                        input_event["body"] = json.dumps(input_event_body)
                        response = lambda_handler(event=input_event, context=lambda_context)

                    response_dict = json.loads(response)
                    assert response_dict["statusCode"] == 200
                    body_dict = json.loads(response_dict["body"])
                    assert body_dict["n_predictions"] == 1
                    assert body_dict["request_id"] == invoke_id
                    assert body_dict["message"] == "ok"
                    assert body_dict["n_shapes_geojson"] == expected_response_body["n_shapes_geojson"]

                    output_geojson = shapely.from_geojson(body_dict["geojson"])
                    print("output_geojson::", type(output_geojson))
                    assert isinstance(output_geojson, shapely.GeometryCollection)
                    assert len(output_geojson.geoms) == expected_response_body["n_shapes_geojson"]


        def test_debug(self):
            from wrappers.lambda_wrapper import lambda_handler

            input_event = {
                'bbox': {
                    'ne': {'lat': 46.302592089330524, 'lng': 9.49493408203125},
                    'sw': {'lat': 46.14011755129237, 'lng': 9.143371582031252}},
                'prompt': [
                    {'id': 166, 'type': 'point', 'data': {'lat': 46.18244521829928, 'lng': 9.418544769287111},
                     'label': 1}
                ],
                'zoom': 12, 'source_type': 'OpenStreetMap'
            }
            lambda_context = LambdaContext(
                invoke_id="test_invoke_id",
                client_context=None,
                cognito_identity=None,
                epoch_deadline_time_in_ms=time.time()
            )
            response = lambda_handler(event=input_event, context=lambda_context)
            print(response)
    except ModuleNotFoundError as mnfe:
        print("missing awslambdaric...")
        raise mnfe


if __name__ == '__main__':
    if IS_AWS_LAMBDA:
        unittest.main()