import json import time import unittest from pathlib import Path from unittest.mock import patch from fastapi.testclient import TestClient from samgis_web.web import web_helpers try: import app except ImportError: import sys from pathlib import Path sys.path.append(str(Path(__file__).parent.parent)) import app PROJECT_ROOT_FOLDER = Path(globals().get("__file__", "./_")).absolute().parent.parent TEST_ROOT_FOLDER = PROJECT_ROOT_FOLDER / "tests" TEST_EVENTS_FOLDER = TEST_ROOT_FOLDER / "events" # in case of OpenStreetMap Mapnik the tile url is "https://tile.openstreetmap.org/{z}/{x}/{y}.png" LOCAL_URL_TILE = "http://localhost:8000/lambda_handler/{z}/{x}/{y}.png" infer_lisa = "/infer_lisa" response_status_code = "response.status_code:{}." response_body_loaded = "response.body_loaded:{}." client = TestClient(app.app) source = { 'url': 'https://tile.openstreetmap.org/{z}/{x}/{y}.png', 'max_zoom': 19, 'html_attribution': '© OpenStreetMap contributors', 'attribution': '(C) OpenStreetMap contributors', 'name': 'OpenStreetMap.Mapnik' } string_prompt = "You are a skilled gis analyst with a lot of expertise in photogrammetry, remote sensing and geomorphology field. " string_prompt += "You need to identify the the southernmost island in this photogrammetric image. Please output segmentation mask." event = { "bbox":{ "ne":{"lat":39.11194810263149,"lng":15.185852050781252}, "sw":{"lat":38.226314067139185,"lng":13.488464355468752} }, "string_prompt": string_prompt, "zoom":10, "source_type": "OpenStreetMap" } class TestFastapiApp(unittest.TestCase): def test_fastapi_handler_health_200(self): response = client.get("/health") assert response.status_code == 200 body = response.json() assert body == {"msg": "still alive..."} # def test_index(self): # import subprocess # # subprocess.run(["pnpm", "build"], cwd=PROJECT_ROOT_FOLDER / "static") # subprocess.run(["pnpm", "tailwindcss", "-i", "./src/input.css", "-o", "./dist/output.css"], # cwd=PROJECT_ROOT_FOLDER / "static") # response = client.get("/") # assert response.status_code == 200 # html_body = response.read().decode("utf-8") # assert "html" in html_body # assert "head" in html_body # assert "body" in html_body def test_404(self): response = client.get("/404") assert response.status_code == 404 def test_infer_samgis_lisa_422(self): response = client.post(infer_lisa, json={}) app.app_logger.info(response_status_code.format(response.status_code)) assert response.status_code == 422 body_loaded = response.json() app.app_logger.info(response_body_loaded.format(body_loaded)) assert body_loaded == {"msg": "Error - Unprocessable Entity"} def test_infer_samgis_lisa_middleware_422(self): from copy import deepcopy local_event = deepcopy(event) local_event["source_type"] = "source_fake" response = client.post(infer_lisa, json=local_event) app.app_logger.info(response_status_code.format(response.status_code)) assert response.status_code == 422 body_loaded = response.json() app.app_logger.info(response_body_loaded.format(body_loaded)) assert body_loaded == {"msg": "Error - Unprocessable Entity"} def test_infer_samgis_lisa_middleware_500(self): from copy import deepcopy local_event = deepcopy(event) local_event["zoom"] = -1 response = client.post(infer_lisa, json=local_event) app.app_logger.info(response_status_code.format(response.status_code)) assert response.status_code == 500 body_loaded = response.json() app.app_logger.info(response_body_loaded.format(body_loaded)) assert body_loaded == {'msg': 'Error - Internal Server Error'} @patch.object(time, "time") @patch.object(app.lisa_module, "lisa_predict") def test_infer_lisa_samgis_mocked_200(self, lisa_predict_mocked, time_mocked): self.maxDiff = None time_mocked.return_value = 0 lisa_predict_output = { "n_predictions": 1, "geojson": "{\"type\": \"FeatureCollection\", \"features\": [{\"id\": \"0\", \"type\": \"Feature\", " + "\"properties\": {\"raster_val\": 255.0}, \"geometry\": {\"type\": \"Polygon\", " + "\"coordinates\": [[[148.359375, -40.4469470596005], [148.447265625, -40.4469470596005], " + "[148.447265625, -40.51379915504414], [148.359375, -40.4469470596005]]]}}]}", "n_shapes_geojson": 2 } lisa_predict_mocked.return_value = lisa_predict_output response = client.post(infer_lisa, json=event) app.app_logger.info(response_status_code.format(response.status_code)) assert response.status_code == 200 response_json = response.json() body_loaded = json.loads(response_json["body"]) app.app_logger.debug(response_body_loaded.format(body_loaded)) self.assertDictEqual(body_loaded, {'duration_run': 0, 'output': lisa_predict_output}) @patch.object(web_helpers, "get_url_tile") def test_infer_lisa_samgis_real_200(self, get_url_tile_mocked): import shapely import xyzservices from samgis_web.utilities.local_tiles_http_server import LocalTilesHttpServer from tests import LOCAL_URL_TILE, TEST_EVENTS_FOLDER listen_port = 8000 local_tile_provider = xyzservices.TileProvider(name="local_tile_provider", url=LOCAL_URL_TILE, attribution="") get_url_tile_mocked.return_value = local_tile_provider with LocalTilesHttpServer.http_server("localhost", listen_port, directory=TEST_EVENTS_FOLDER): response = client.post(infer_lisa, json=event) app.app_logger.info(response_status_code.format(response.status_code)) assert response.status_code == 200 response = response.json() body_string = response["body"] body_loaded = json.loads(body_string) app.app_logger.debug(response_body_loaded.format(body_loaded)) assert "duration_run" in body_loaded assert float(body_loaded["duration_run"]) > 0 output = body_loaded["output"] assert "n_shapes_geojson" in output geojson = output["geojson"] output_geojson = shapely.from_geojson(geojson) app.app_logger.debug("output_geojson::{}.".format(output_geojson)) assert isinstance(output_geojson, shapely.GeometryCollection) assert len(output_geojson.geoms) > 0 first_polygon = output_geojson.geoms[0] assert isinstance(first_polygon, shapely.geometry.Polygon) assert len(output_geojson.geoms) > 1 if __name__ == "__main__": unittest.main()