bjkim1
21ebb16fb7
- SPipe_server_2026-10-30.zip 내용을 풀어서 등록 (압축 파일은 올리지 않음) - README에 데모 v2 서버 업데이트 순서 추가, 저장소에 없는 문서 목차 정리 - 패치내역을 docs/로 이동, _worklog.md 추가 - 임시 파일(backend/temp.*)과 API 키가 든 파일은 제외
70 lines
2.9 KiB
Python
70 lines
2.9 KiB
Python
"""Vector GIS import (/api/gis/vector) with real GDAL files written on the fly."""
|
|
import io
|
|
import os
|
|
import tempfile
|
|
import zipfile
|
|
|
|
import numpy as np
|
|
import pytest
|
|
import shapely
|
|
from fastapi.testclient import TestClient
|
|
from pyogrio import raw as ogr_raw
|
|
|
|
from backend.main import app
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def client():
|
|
with TestClient(app, raise_server_exceptions=False) as c:
|
|
yield c
|
|
|
|
|
|
def _write(path, driver, crs="EPSG:5187"):
|
|
geoms = [shapely.LineString([(200000, 300000), (200100, 300050)]), shapely.LineString([(200100, 300050), (200200, 300000)])]
|
|
with_fields = driver != "DXF" # DXF has a fixed schema (Layer, EntityHandle ...)
|
|
ogr_raw.write(
|
|
path,
|
|
geometry=np.array(shapely.to_wkb(geoms), dtype=object),
|
|
field_data=[np.array([150, 200], dtype=np.int32), np.array(["DCIP", "PE"], dtype=object)] if with_fields else [],
|
|
fields=["DIA", "MAT"] if with_fields else [],
|
|
geometry_type="LineString",
|
|
crs=crs,
|
|
driver=driver,
|
|
)
|
|
|
|
|
|
def test_zipped_shapefile_reprojected_to_wgs84(client):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
shp = os.path.join(tmp, "pipes.shp")
|
|
_write(shp, "ESRI Shapefile")
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w") as z:
|
|
for f in os.listdir(tmp):
|
|
z.write(os.path.join(tmp, f), f)
|
|
r = client.post("/api/gis/vector", files={"file": ("pipes.zip", buf.getvalue(), "application/zip")})
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["crs"] == "EPSG:5187" and body["count"] == 2
|
|
feat = body["featureCollection"]["features"][0]
|
|
assert feat["properties"] == {"DIA": 150, "MAT": "DCIP"}
|
|
lon, lat = feat["geometry"]["coordinates"][0]
|
|
assert 127 < lon < 132 and 33 < lat < 39 # East Belt -> somewhere in Korea
|
|
|
|
|
|
def test_dxf_without_crs_needs_src_epsg(client):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
dxf = os.path.join(tmp, "pipes.dxf")
|
|
_write(dxf, "DXF", crs=None)
|
|
data = open(dxf, "rb").read()
|
|
raw = client.post("/api/gis/vector", files={"file": ("pipes.dxf", data, "application/dxf")}).json()
|
|
assert raw["reprojected"] is False and raw["count"] >= 2
|
|
fixed = client.post("/api/gis/vector", files={"file": ("pipes.dxf", data, "application/dxf")}, data={"srcEpsg": "5187"}).json()
|
|
lon, lat = fixed["featureCollection"]["features"][0]["geometry"]["coordinates"][0]
|
|
assert 127 < lon < 132 and 33 < lat < 39
|
|
|
|
|
|
def test_rejects_unknown_extension_and_bad_epsg(client):
|
|
assert client.post("/api/gis/vector", files={"file": ("a.exe", b"MZ", "application/octet-stream")}).status_code == 400
|
|
assert client.post("/api/gis/vector", files={"file": ("a.dxf", b"0", "x")}, data={"srcEpsg": "abc"}).status_code == 400
|
|
assert client.post("/api/gis/vector", files={"file": ("broken.dxf", b"not a dxf", "x")}).status_code == 400
|