309 lines
10 KiB
Python
309 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from types import SimpleNamespace
|
|
from urllib.parse import parse_qs, urlparse
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from geoalchemy2.shape import from_shape
|
|
from pyproj import Transformer
|
|
from shapely.geometry import MultiPolygon, Polygon
|
|
|
|
from app.core.config import Settings
|
|
from app.core.errors import AppError
|
|
from app.models import Area, Dataset, Project
|
|
from app.schemas.official_vector import OfficialVectorAcquireRequest
|
|
from app.services.dataset_service import DatasetService
|
|
from app.services.official_vector_acquisition_service import (
|
|
OfficialVectorAcquisitionService,
|
|
_TO_LAMBERT72,
|
|
)
|
|
|
|
|
|
class FakeQuery:
|
|
def __init__(self, result=None):
|
|
self.result = result
|
|
|
|
def filter(self, *_args):
|
|
return self
|
|
|
|
def order_by(self, *_args):
|
|
return self
|
|
|
|
def all(self):
|
|
return self.result if isinstance(self.result, list) else []
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, rows=None, query_result=None):
|
|
self.rows = rows or {}
|
|
self.query_result = query_result
|
|
|
|
def get(self, model, row_id):
|
|
return self.rows.get((model, row_id))
|
|
|
|
def query(self, _model):
|
|
return FakeQuery(self.query_result)
|
|
|
|
|
|
class JsonResponse:
|
|
def __init__(self, payload, content_type="application/geo+json"):
|
|
self.content = json.dumps(payload).encode("utf-8")
|
|
self.content_type = content_type
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def read(self, size=-1):
|
|
return self.content if size < 0 else self.content[:size]
|
|
|
|
def getheader(self, name):
|
|
return self.content_type if name.lower() == "content-type" else None
|
|
|
|
|
|
def request(product_key: str, bbox: tuple[float, float, float, float], area_id=None):
|
|
return OfficialVectorAcquireRequest(
|
|
bbox={
|
|
"min_x": bbox[0],
|
|
"min_y": bbox[1],
|
|
"max_x": bbox[2],
|
|
"max_y": bbox[3],
|
|
"crs": "EPSG:4326",
|
|
},
|
|
area_id=area_id,
|
|
product_key=product_key,
|
|
force_refresh=True,
|
|
)
|
|
|
|
|
|
def area(project_id, name: str, bounds: tuple[float, float, float, float]):
|
|
min_x, min_y, max_x, max_y = bounds
|
|
geometry = MultiPolygon(
|
|
[
|
|
Polygon(
|
|
[
|
|
(min_x, min_y),
|
|
(max_x, min_y),
|
|
(max_x, max_y),
|
|
(min_x, max_y),
|
|
(min_x, min_y),
|
|
]
|
|
)
|
|
]
|
|
)
|
|
return Area(
|
|
id=uuid4(),
|
|
project_id=project_id,
|
|
name=name,
|
|
geometry=from_shape(geometry, srid=4326),
|
|
)
|
|
|
|
|
|
def test_regional_product_registry_is_explicit_and_source_specific() -> None:
|
|
products = {
|
|
item["key"]: item
|
|
for item in OfficialVectorAcquisitionService.list_products()
|
|
}
|
|
|
|
assert products["spw_picc_buildings"]["coverage_zones"] == ["wallonia"]
|
|
assert products["spw_picc_roads"]["geometry_types"] == [
|
|
"LineString",
|
|
"MultiLineString",
|
|
]
|
|
assert products["spw_picc_waterways"]["collection"] == "28"
|
|
assert products["spw_picc_water_surfaces"]["collection"] == "30"
|
|
assert products["urbis_buildings"]["coverage_zones"] == ["brussels"]
|
|
assert products["urbis_buildings"]["license_note"] == "Buildings are published under CC0."
|
|
assert "FPS Finance" in products["urbis_cadastral_parcels"]["license_note"]
|
|
|
|
|
|
def test_spw_arcgis_paging_is_bounded_stable_and_clipped() -> None:
|
|
product = OfficialVectorAcquisitionService._product("spw_picc_buildings")
|
|
scope = Polygon(
|
|
[(4.55, 50.58), (4.56, 50.58), (4.56, 50.59), (4.55, 50.59), (4.55, 50.58)]
|
|
)
|
|
scope_metric = Polygon(
|
|
[
|
|
_TO_LAMBERT72.transform(x, y)
|
|
for x, y in scope.exterior.coords
|
|
]
|
|
)
|
|
offsets = []
|
|
|
|
def feature(object_id: int, min_x: float):
|
|
return {
|
|
"type": "Feature",
|
|
"id": object_id,
|
|
"geometry": {
|
|
"type": "Polygon",
|
|
"coordinates": [[
|
|
[min_x, 50.581],
|
|
[min_x + 0.002, 50.581],
|
|
[min_x + 0.002, 50.583],
|
|
[min_x, 50.583],
|
|
[min_x, 50.581],
|
|
]],
|
|
},
|
|
"properties": {"OBJECTID": object_id, "GEOREF_ID": f"wallonia-{object_id}"},
|
|
}
|
|
|
|
def opener(raw_request, timeout):
|
|
assert timeout == 180
|
|
query = parse_qs(urlparse(raw_request.full_url).query)
|
|
assert query["orderByFields"] == ["OBJECTID"]
|
|
assert query["f"] == ["geojson"]
|
|
offset = int(query["resultOffset"][0])
|
|
offsets.append(offset)
|
|
return JsonResponse(
|
|
{
|
|
"type": "FeatureCollection",
|
|
"features": [feature(offset + 1, 4.551 + offset * 0.0001)],
|
|
"exceededTransferLimit": offset == 0,
|
|
}
|
|
)
|
|
|
|
features, transfer = OfficialVectorAcquisitionService._fetch_features(
|
|
product,
|
|
scope,
|
|
scope_metric,
|
|
"wallonia",
|
|
Settings(_env_file=None, OFFICIAL_VECTOR_PAGE_SIZE=1),
|
|
opener,
|
|
)
|
|
|
|
assert offsets == [0, 1]
|
|
assert transfer["page_count"] == 2
|
|
assert transfer["reference_truncated"] is False
|
|
assert {item["properties"]["source_feature_id"] for item in features} == {
|
|
"11:wallonia-1",
|
|
"11:wallonia-2",
|
|
}
|
|
assert all(item["properties"]["coverage_scope"] == "wallonia" for item in features)
|
|
assert all(item["properties"]["clipped_area_ha"] > 0 for item in features)
|
|
|
|
|
|
def test_regional_products_require_the_persisted_authoritative_coverage_area() -> None:
|
|
project_id = uuid4()
|
|
db = FakeSession({(Project, project_id): Project(id=project_id, name="Belgium")})
|
|
|
|
with pytest.raises(AppError) as exc_info:
|
|
OfficialVectorAcquisitionService.acquire(
|
|
db,
|
|
project_id,
|
|
request("spw_picc_buildings", (4.55, 50.58, 4.56, 50.59)),
|
|
settings=Settings(_env_file=None),
|
|
)
|
|
|
|
assert exc_info.value.code == "OFFICIAL_VECTOR_COVERAGE_NOT_READY"
|
|
|
|
|
|
def test_urbis_wfs_transforms_lambert72_and_persists_through_dataset_service(
|
|
monkeypatch,
|
|
) -> None:
|
|
project_id, dataset_id = uuid4(), uuid4()
|
|
brussels = area(project_id, "Brussels-Capital Region", (4.25, 50.75, 4.5, 50.95))
|
|
db = FakeSession(
|
|
{
|
|
(Project, project_id): Project(id=project_id, name="Belgium"),
|
|
(Area, brussels.id): brussels,
|
|
},
|
|
query_result=[brussels],
|
|
)
|
|
to_lambert = Transformer.from_crs("EPSG:4326", "EPSG:31370", always_xy=True)
|
|
min_x, min_y = to_lambert.transform(4.35, 50.84)
|
|
max_x, max_y = to_lambert.transform(4.351, 50.841)
|
|
captured = {}
|
|
|
|
def opener(raw_request, timeout):
|
|
assert timeout == 180
|
|
query = parse_qs(urlparse(raw_request.full_url).query)
|
|
assert query["typeNames"] == ["urbisvector:Buildings"]
|
|
assert query["srsName"] == ["EPSG:31370"]
|
|
assert query["sortBy"] == ["INSPIRE_ID"]
|
|
return JsonResponse(
|
|
{
|
|
"type": "FeatureCollection",
|
|
"numberMatched": 1,
|
|
"numberReturned": 1,
|
|
"features": [
|
|
{
|
|
"type": "Feature",
|
|
"id": "Buildings.1",
|
|
"geometry": {
|
|
"type": "MultiPolygon",
|
|
"coordinates": [[[
|
|
[min_x, min_y],
|
|
[max_x, min_y],
|
|
[max_x, max_y],
|
|
[min_x, max_y],
|
|
[min_x, min_y],
|
|
]]],
|
|
},
|
|
"properties": {
|
|
"INSPIRE_ID": "https://databrussels.be/id/building/1",
|
|
"AREA": 75,
|
|
},
|
|
}
|
|
],
|
|
},
|
|
"application/json",
|
|
)
|
|
|
|
def persist(_db, **kwargs):
|
|
captured.update(kwargs)
|
|
dataset = Dataset(
|
|
id=dataset_id,
|
|
project_id=project_id,
|
|
area_id=brussels.id,
|
|
name=kwargs["filename"],
|
|
dataset_type="vector",
|
|
source=kwargs["source"],
|
|
dataset_role=kwargs["dataset_role"],
|
|
source_name=kwargs["source_name"],
|
|
reference_layer_name=kwargs["reference_layer_name"],
|
|
temporal_series_key=kwargs["temporal_series_key"],
|
|
observed_at=kwargs["observed_at"],
|
|
source_version=kwargs["source_version"],
|
|
source_metadata=kwargs["source_metadata"],
|
|
provenance_metadata=kwargs["provenance_metadata"],
|
|
metadata_json={"feature_count": 1},
|
|
status="ready",
|
|
)
|
|
db.rows[(Dataset, dataset_id)] = dataset
|
|
return SimpleNamespace(id=dataset_id)
|
|
|
|
monkeypatch.setattr(DatasetService, "import_vector_bytes", persist)
|
|
result = OfficialVectorAcquisitionService.acquire(
|
|
db,
|
|
project_id,
|
|
request(
|
|
"urbis_buildings",
|
|
(4.349, 50.839, 4.352, 50.842),
|
|
area_id=brussels.id,
|
|
),
|
|
settings=Settings(_env_file=None),
|
|
opener=opener,
|
|
)
|
|
|
|
assert result["output_dataset_id"] == str(dataset_id)
|
|
assert captured["source_name"] == "urbis"
|
|
assert captured["reference_layer_name"] == "buildings"
|
|
assert captured["source_metadata"]["coverage_zones"] == ["brussels"]
|
|
assert captured["source_metadata"]["selection_aggregation"]["metric_key"] == (
|
|
"building_footprint_area"
|
|
)
|
|
collection = json.loads(captured["content"])
|
|
geometry = collection["features"][0]["geometry"]
|
|
assert geometry["type"] in {"Polygon", "MultiPolygon"}
|
|
first_coordinate = (
|
|
geometry["coordinates"][0][0][0]
|
|
if geometry["type"] == "MultiPolygon"
|
|
else geometry["coordinates"][0][0]
|
|
)
|
|
assert 4.34999 <= first_coordinate[0] <= 4.35101
|
|
assert 50.83999 <= first_coordinate[1] <= 50.84101
|