alessandro trinca tornidor
commited on
Commit
•
2c1de55
1
Parent(s):
2d51c41
[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)
|