File size: 3,206 Bytes
85dd36a 9205f9c 142642e 9205f9c 85dd36a 238719b 9205f9c 4fed430 9205f9c 4fed430 9205f9c 4fed430 9205f9c 4fed430 9205f9c 238719b 85dd36a 142642e 9205f9c 142642e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 |
import json
import time
from http import HTTPStatus
from unittest.mock import patch
from src.io.lambda_helpers import get_parsed_bbox_points, get_parsed_request_body, get_response
from src.utilities.type_hints import RawRequestInput
from src.utilities import utilities
from tests import TEST_EVENTS_FOLDER
@patch.object(time, "time")
def test_get_response(time_mocked):
time_diff = 108
end_run = 1000
time_mocked.return_value = end_run
start_time = end_run-time_diff
aws_request_id = "test_invoke_id"
with open(TEST_EVENTS_FOLDER / "get_response.json") as tst_json:
inputs_outputs = json.load(tst_json)
response_type = "200"
body_response = inputs_outputs[response_type]["input"]
output = get_response(HTTPStatus.OK.value, start_time, aws_request_id, body_response)
assert json.loads(output) == inputs_outputs[response_type]["output"]
response_type = "400"
response_400 = get_response(HTTPStatus.BAD_REQUEST.value, start_time, aws_request_id, {})
assert response_400 == inputs_outputs[response_type]["output"]
response_type = "422"
response_422 = get_response(HTTPStatus.UNPROCESSABLE_ENTITY.value, start_time, aws_request_id, {})
assert response_422 == inputs_outputs[response_type]["output"]
response_type = "500"
response_500 = get_response(HTTPStatus.INTERNAL_SERVER_ERROR.value, start_time, aws_request_id, {})
assert response_500 == inputs_outputs[response_type]["output"]
def test_get_parsed_bbox_points():
with open(TEST_EVENTS_FOLDER / "get_parsed_bbox_points.json") as tst_json:
inputs_outputs = json.load(tst_json)
for k, input_output in inputs_outputs.items():
print(f"k:{k}.")
raw_body = get_parsed_request_body(**input_output["input"])
output = get_parsed_bbox_points(raw_body)
assert output == input_output["output"]
def test_get_parsed_request_body():
input_event = {
"event": {
"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
}
}
expected_output_dict = {
"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 = get_parsed_request_body(input_event["event"])
assert output == RawRequestInput.model_validate(input_event["event"])
input_event_str = json.dumps(input_event["event"])
output = get_parsed_request_body(input_event_str)
assert output == RawRequestInput.model_validate(expected_output_dict)
event = {"body": utilities.base64_encode(input_event_str).decode("utf-8")}
output = get_parsed_request_body(event)
assert output == RawRequestInput.model_validate(expected_output_dict)
|