[test] add local server as tiles xyz provider
Browse files
samgis/io/wrappers_helpers.py
CHANGED
|
@@ -174,8 +174,8 @@ def get_url_tile(source_type: str):
|
|
| 174 |
case XYZTerrainProvidersNames.NEXTZEN_TERRAIN_TILES_NAME:
|
| 175 |
app_logger.info("nextzen_terrain_rgb:", nextzen_terrain_rgb)
|
| 176 |
return nextzen_terrain_rgb
|
| 177 |
-
|
| 178 |
-
|
| 179 |
except ValueError as ve:
|
| 180 |
from pydantic_core import ValidationError
|
| 181 |
|
|
|
|
| 174 |
case XYZTerrainProvidersNames.NEXTZEN_TERRAIN_TILES_NAME:
|
| 175 |
app_logger.info("nextzen_terrain_rgb:", nextzen_terrain_rgb)
|
| 176 |
return nextzen_terrain_rgb
|
| 177 |
+
case _:
|
| 178 |
+
return providers.query_name(source_type)
|
| 179 |
except ValueError as ve:
|
| 180 |
from pydantic_core import ValidationError
|
| 181 |
|
samgis/utilities/utilities.py
CHANGED
|
@@ -17,7 +17,8 @@ def _is_base64(sb: str or bytes):
|
|
| 17 |
|
| 18 |
try:
|
| 19 |
sb_bytes = _prepare_base64_input(sb)
|
| 20 |
-
|
|
|
|
| 21 |
except ValueError:
|
| 22 |
return False
|
| 23 |
|
|
|
|
| 17 |
|
| 18 |
try:
|
| 19 |
sb_bytes = _prepare_base64_input(sb)
|
| 20 |
+
decoded = base64.b64decode(sb_bytes, validate=True)
|
| 21 |
+
return base64.b64encode(decoded).decode("utf-8") == sb_bytes.decode("utf-8")
|
| 22 |
except ValueError:
|
| 23 |
return False
|
| 24 |
|
scripts/extract-openapi-fastapi.py
CHANGED
|
@@ -3,14 +3,15 @@ import argparse
|
|
| 3 |
import json
|
| 4 |
import logging
|
| 5 |
import sys
|
|
|
|
| 6 |
import yaml
|
| 7 |
from uvicorn.importer import import_from_string
|
| 8 |
|
|
|
|
| 9 |
|
| 10 |
parser = argparse.ArgumentParser(prog="extract-openapi-fastapi.py")
|
| 11 |
parser.add_argument("app", help='App import string. Eg. "main:app"', default="main:app")
|
| 12 |
parser.add_argument("--app-dir", help="Directory containing the app", default=None)
|
| 13 |
-
parser.add_argument("--out", help="Output file ending in .json or .yaml", default="openapi.yaml")
|
| 14 |
|
| 15 |
|
| 16 |
if __name__ == "__main__":
|
|
@@ -26,11 +27,11 @@ if __name__ == "__main__":
|
|
| 26 |
openapi = app.openapi()
|
| 27 |
version = openapi.get("openapi", "unknown version")
|
| 28 |
|
| 29 |
-
logging.info(f"writing openapi spec v{version}")
|
| 30 |
-
|
| 31 |
-
|
| 32 |
-
|
| 33 |
-
|
| 34 |
-
|
| 35 |
|
| 36 |
logging.info(f"spec written to {args.out} #")
|
|
|
|
| 3 |
import json
|
| 4 |
import logging
|
| 5 |
import sys
|
| 6 |
+
|
| 7 |
import yaml
|
| 8 |
from uvicorn.importer import import_from_string
|
| 9 |
|
| 10 |
+
from samgis import PROJECT_ROOT_FOLDER
|
| 11 |
|
| 12 |
parser = argparse.ArgumentParser(prog="extract-openapi-fastapi.py")
|
| 13 |
parser.add_argument("app", help='App import string. Eg. "main:app"', default="main:app")
|
| 14 |
parser.add_argument("--app-dir", help="Directory containing the app", default=None)
|
|
|
|
| 15 |
|
| 16 |
|
| 17 |
if __name__ == "__main__":
|
|
|
|
| 27 |
openapi = app.openapi()
|
| 28 |
version = openapi.get("openapi", "unknown version")
|
| 29 |
|
| 30 |
+
logging.info(f"writing openapi spec v{version}...")
|
| 31 |
+
output_dir_path = PROJECT_ROOT_FOLDER / "docs" / "specs"
|
| 32 |
+
with open(output_dir_path / "output.json", "w") as f:
|
| 33 |
+
json.dump(openapi, f)
|
| 34 |
+
with open(output_dir_path / "output.yaml", "w") as f:
|
| 35 |
+
yaml.dump(openapi, f, sort_keys=False)
|
| 36 |
|
| 37 |
logging.info(f"spec written to {args.out} #")
|
tests/test_fastapi_app.py
CHANGED
|
@@ -6,7 +6,9 @@ from unittest.mock import patch
|
|
| 6 |
from fastapi.testclient import TestClient
|
| 7 |
|
| 8 |
from samgis import PROJECT_ROOT_FOLDER
|
|
|
|
| 9 |
from tests import TEST_EVENTS_FOLDER
|
|
|
|
| 10 |
from wrappers import fastapi_wrapper
|
| 11 |
from wrappers.fastapi_wrapper import app
|
| 12 |
|
|
@@ -125,14 +127,22 @@ class TestFastapiApp(unittest.TestCase):
|
|
| 125 |
print("response.body:", body)
|
| 126 |
assert body == {'msg': 'Error - Internal Server Error'}
|
| 127 |
|
|
|
|
| 128 |
@patch.object(time, "time")
|
| 129 |
-
def test_infer_samgis_real_200(self, time_mocked):
|
| 130 |
import shapely
|
|
|
|
|
|
|
| 131 |
|
| 132 |
time_mocked.return_value = 0
|
|
|
|
| 133 |
|
| 134 |
-
|
| 135 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
| 136 |
assert response.status_code == 200
|
| 137 |
body_string = response.json()["body"]
|
| 138 |
body_loaded = json.loads(body_string)
|
|
|
|
| 6 |
from fastapi.testclient import TestClient
|
| 7 |
|
| 8 |
from samgis import PROJECT_ROOT_FOLDER
|
| 9 |
+
from samgis.io import wrappers_helpers
|
| 10 |
from tests import TEST_EVENTS_FOLDER
|
| 11 |
+
from tests.local_tiles_http_server import LocalTilesHttpServer
|
| 12 |
from wrappers import fastapi_wrapper
|
| 13 |
from wrappers.fastapi_wrapper import app
|
| 14 |
|
|
|
|
| 127 |
print("response.body:", body)
|
| 128 |
assert body == {'msg': 'Error - Internal Server Error'}
|
| 129 |
|
| 130 |
+
@patch.object(wrappers_helpers, "get_url_tile")
|
| 131 |
@patch.object(time, "time")
|
| 132 |
+
def test_infer_samgis_real_200(self, time_mocked, get_url_tile_mocked):
|
| 133 |
import shapely
|
| 134 |
+
import xyzservices
|
| 135 |
+
from tests import LOCAL_URL_TILE, TEST_EVENTS_FOLDER
|
| 136 |
|
| 137 |
time_mocked.return_value = 0
|
| 138 |
+
listen_port = 8000
|
| 139 |
|
| 140 |
+
local_tile_provider = xyzservices.TileProvider(name="local_tile_provider", url=LOCAL_URL_TILE, attribution="")
|
| 141 |
+
get_url_tile_mocked.return_value = local_tile_provider
|
| 142 |
+
|
| 143 |
+
with LocalTilesHttpServer.http_server("localhost", listen_port, directory=TEST_EVENTS_FOLDER):
|
| 144 |
+
response = client.post("/infer_samgis", json=event)
|
| 145 |
+
print("response.status_code:", response.status_code)
|
| 146 |
assert response.status_code == 200
|
| 147 |
body_string = response.json()["body"]
|
| 148 |
body_loaded = json.loads(body_string)
|