tests/test_app.py

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': '&copy; <a href="https://www.openstreetmap.org/copyright">OpenStreetMap</a> 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()