Jens
faeb58ef6d
GeoIntel release gates / Compile, test, contracts and builds (push) Successful in 1m49s
GeoIntel release gates / Python and npm vulnerability policy (push) Successful in 21s
GeoIntel release gates / Production AI image, SBOM and container scan (push) Successful in 5m39s
GeoIntel release gates / Deploy exact gated revision to Unraid (push) Failing after 58m43s
270 lines
9.5 KiB
Python
270 lines
9.5 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import sys
|
|
import zipfile
|
|
from uuid import uuid4
|
|
|
|
import numpy as np
|
|
import pytest
|
|
import rasterio
|
|
from fastapi.testclient import TestClient
|
|
from pyproj import Transformer
|
|
from rasterio.io import MemoryFile
|
|
from rasterio.transform import from_origin
|
|
from shapely.geometry import shape
|
|
|
|
from app.core.config import Settings
|
|
from app.core.errors import AppError
|
|
from app.db.session import get_db
|
|
from app.main import app
|
|
from app.models import Dataset
|
|
from app.schemas.bathymetry import BathymetryRasterSelectionRequest
|
|
from app.services.bathymetry_raster_analysis_service import BathymetryRasterAnalysisService
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SCRIPT_PATH = ROOT / "scripts" / "import_spw_bathymetry.py"
|
|
|
|
|
|
def load_operator():
|
|
name = "test_import_spw_bathymetry_sprint241"
|
|
spec = importlib.util.spec_from_file_location(name, SCRIPT_PATH)
|
|
assert spec is not None
|
|
assert spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
OPERATOR = load_operator()
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, rows):
|
|
self.rows = rows
|
|
|
|
def get(self, model, row_id):
|
|
return self.rows.get((model, row_id))
|
|
|
|
|
|
def bathymetry_tiff(*, nodata_only: bool = False) -> bytes:
|
|
values = np.linspace(72.0, 80.0, 400, dtype="float32").reshape(20, 20)
|
|
values[:, :5] = -9999.0
|
|
if nodata_only:
|
|
values[:] = -9999.0
|
|
with MemoryFile() as memory:
|
|
with memory.open(
|
|
driver="GTiff",
|
|
width=20,
|
|
height=20,
|
|
count=1,
|
|
dtype="float32",
|
|
crs="EPSG:3812",
|
|
transform=from_origin(684_000, 629_000, 0.5, 0.5),
|
|
nodata=-9999.0,
|
|
) as output:
|
|
output.write(values, 1)
|
|
return memory.read()
|
|
|
|
|
|
def selection_payload() -> BathymetryRasterSelectionRequest:
|
|
transformer = Transformer.from_crs("EPSG:3812", "EPSG:4326", always_xy=True)
|
|
min_x, min_y = transformer.transform(684_000, 628_990)
|
|
max_x, max_y = transformer.transform(684_010, 629_000)
|
|
return BathymetryRasterSelectionRequest(
|
|
bbox={
|
|
"min_x": min(min_x, max_x),
|
|
"min_y": min(min_y, max_y),
|
|
"max_x": max(min_x, max_x),
|
|
"max_y": max(min_y, max_y),
|
|
"crs": "EPSG:4326",
|
|
}
|
|
)
|
|
|
|
|
|
def persisted_dataset(path: Path, *, metadata: dict | None = None) -> Dataset:
|
|
path.write_bytes(bathymetry_tiff())
|
|
return Dataset(
|
|
id=uuid4(),
|
|
project_id=uuid4(),
|
|
name="spw_bathymetry_test_3812.tif",
|
|
dataset_type="raster",
|
|
source="SPW official operator archive",
|
|
source_name="spw_bathymetry",
|
|
source_metadata=metadata
|
|
or {
|
|
"product_key": "spw_bathymetry_50cm_mdng",
|
|
"theme": "bathymetry",
|
|
"value_semantics": "bed_elevation",
|
|
"vertical_reference": "mDNG",
|
|
"source_crs": "EPSG:3812",
|
|
"survey_period": "2019-2022",
|
|
},
|
|
storage_path=str(path),
|
|
status="ready",
|
|
)
|
|
|
|
|
|
def test_bathymetry_analysis_returns_real_bed_elevation_and_surface_metrics(tmp_path: Path) -> None:
|
|
dataset = persisted_dataset(tmp_path / "bathymetry.tif")
|
|
db = FakeSession({(Dataset, dataset.id): dataset})
|
|
|
|
result = BathymetryRasterAnalysisService.analyze(
|
|
db,
|
|
dataset.project_id,
|
|
dataset.id,
|
|
selection_payload(),
|
|
settings=Settings(_env_file=None, bathymetry_raster_max_pixels=10_000),
|
|
)
|
|
|
|
metrics = {item["metric_key"]: item for item in result["summary"]["metrics"]}
|
|
assert result["product_key"] == "spw_bathymetry_50cm_mdng"
|
|
assert result["vertical_reference"] == "mDNG"
|
|
assert result["survey_period"] == "2019-2022"
|
|
assert result["selected_cell_count"] == 400
|
|
assert result["valid_cell_count"] == 300
|
|
assert result["coverage_ratio"] == pytest.approx(0.75)
|
|
assert metrics["bed_elevation_mean_m"]["metric_unit"] == "m mDNG"
|
|
assert metrics["surveyed_bed_surface_ha"]["metric_value"] == pytest.approx(0.0075)
|
|
assert metrics["bathymetry_coverage_pct"]["metric_value"] == pytest.approx(75.0)
|
|
assert result["unsupported_metrics"] == [
|
|
"current_water_depth_m",
|
|
"water_volume_m3",
|
|
"vertical_datum_conversion",
|
|
]
|
|
|
|
|
|
def test_bathymetry_analysis_fails_closed_for_metadata_size_and_empty_cells(tmp_path: Path) -> None:
|
|
dataset = persisted_dataset(tmp_path / "bathymetry.tif")
|
|
db = FakeSession({(Dataset, dataset.id): dataset})
|
|
|
|
with pytest.raises(AppError) as size_error:
|
|
BathymetryRasterAnalysisService.analyze(
|
|
db,
|
|
dataset.project_id,
|
|
dataset.id,
|
|
selection_payload(),
|
|
settings=Settings(_env_file=None, bathymetry_raster_max_pixels=100),
|
|
)
|
|
assert size_error.value.code == "BATHYMETRY_SELECTION_TOO_LARGE"
|
|
|
|
dataset.source_metadata = {"theme": "bathymetry"}
|
|
with pytest.raises(AppError) as metadata_error:
|
|
BathymetryRasterAnalysisService.analyze(
|
|
db,
|
|
dataset.project_id,
|
|
dataset.id,
|
|
selection_payload(),
|
|
)
|
|
assert metadata_error.value.code == "INVALID_BATHYMETRY_RASTER_METADATA"
|
|
|
|
dataset.source_metadata = {
|
|
"product_key": "spw_bathymetry_50cm_mdng",
|
|
"theme": "bathymetry",
|
|
"value_semantics": "bed_elevation",
|
|
"vertical_reference": "mDNG",
|
|
"source_crs": "EPSG:3812",
|
|
}
|
|
Path(dataset.storage_path).write_bytes(bathymetry_tiff(nodata_only=True))
|
|
with pytest.raises(AppError) as empty_error:
|
|
BathymetryRasterAnalysisService.analyze(
|
|
db,
|
|
dataset.project_id,
|
|
dataset.id,
|
|
selection_payload(),
|
|
)
|
|
assert empty_error.value.code == "BATHYMETRY_NO_VALID_DATA"
|
|
|
|
|
|
def test_bathymetry_image_and_route_use_persisted_raster_and_canonical_envelope(tmp_path: Path) -> None:
|
|
dataset = persisted_dataset(tmp_path / "bathymetry.tif")
|
|
db = FakeSession({(Dataset, dataset.id): dataset})
|
|
image = BathymetryRasterAnalysisService.render_png(db, dataset.project_id, dataset.id)
|
|
|
|
assert image.startswith(b"\x89PNG\r\n\x1a\n")
|
|
|
|
app.dependency_overrides[get_db] = lambda: db
|
|
try:
|
|
response = TestClient(app).post(
|
|
f"/api/v1/projects/{dataset.project_id}/datasets/{dataset.id}/raster/bathymetry/select",
|
|
json=selection_payload().model_dump(mode="json"),
|
|
)
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert set(payload) == {"data"}
|
|
assert payload["data"]["dataset_id"] == str(dataset.id)
|
|
assert payload["data"]["summary"]["primary_metric_key"] == "bed_elevation_mean_m"
|
|
|
|
|
|
def test_operator_validates_pinned_archive_and_rejects_unsafe_members(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
safe_path = tmp_path / "safe.zip"
|
|
with zipfile.ZipFile(safe_path, "w", compression=zipfile.ZIP_DEFLATED) as archive:
|
|
archive.writestr(OPERATOR.SOURCE_MEMBER, bathymetry_tiff())
|
|
monkeypatch.setattr(OPERATOR, "SOURCE_SHA256", OPERATOR.sha256_file(safe_path))
|
|
|
|
member = OPERATOR.validate_archive(safe_path)
|
|
|
|
assert member.filename == OPERATOR.SOURCE_MEMBER
|
|
|
|
unsafe_path = tmp_path / "unsafe.zip"
|
|
with zipfile.ZipFile(unsafe_path, "w", compression=zipfile.ZIP_DEFLATED) as archive:
|
|
archive.writestr("../escape.txt", "unsafe")
|
|
archive.writestr(OPERATOR.SOURCE_MEMBER, bathymetry_tiff())
|
|
monkeypatch.setattr(OPERATOR, "SOURCE_SHA256", OPERATOR.sha256_file(unsafe_path))
|
|
with pytest.raises(OPERATOR.SpwBathymetryImportError, match="unsafe member"):
|
|
OPERATOR.validate_archive(unsafe_path)
|
|
|
|
|
|
def test_operator_crops_zip_member_to_valid_cog(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
archive_path = tmp_path / "source.zip"
|
|
with zipfile.ZipFile(archive_path, "w", compression=zipfile.ZIP_DEFLATED) as archive:
|
|
archive.writestr(OPERATOR.SOURCE_MEMBER, bathymetry_tiff())
|
|
monkeypatch.setattr(OPERATOR, "SOURCE_SHA256", OPERATOR.sha256_file(archive_path))
|
|
member = OPERATOR.validate_archive(archive_path)
|
|
output_path = tmp_path / "bounded.tif"
|
|
|
|
diagnostics = OPERATOR.crop_source(
|
|
archive_path,
|
|
member,
|
|
shape(
|
|
{
|
|
"type": "Polygon",
|
|
"coordinates": [[
|
|
[selection_payload().bbox.min_x, selection_payload().bbox.min_y],
|
|
[selection_payload().bbox.max_x, selection_payload().bbox.min_y],
|
|
[selection_payload().bbox.max_x, selection_payload().bbox.max_y],
|
|
[selection_payload().bbox.min_x, selection_payload().bbox.max_y],
|
|
[selection_payload().bbox.min_x, selection_payload().bbox.min_y],
|
|
]],
|
|
}
|
|
),
|
|
output_path,
|
|
max_pixels=10_000,
|
|
)
|
|
|
|
with rasterio.open(output_path) as output:
|
|
assert output.crs.to_epsg() == 3812
|
|
assert output.driver == "GTiff"
|
|
assert output.nodata == -9999.0
|
|
assert output.profile["tiled"] is True
|
|
assert diagnostics["valid_cell_count"] == 300
|
|
assert len(diagnostics["output_sha256"]) == 64
|
|
|
|
|
|
def test_operator_is_api_only_and_does_not_claim_depth_or_volume() -> None:
|
|
source = SCRIPT_PATH.read_text(encoding="utf-8")
|
|
|
|
assert "/datasets/upload" in source
|
|
assert "water_depth_available" in source
|
|
assert '"water_volume_available": False' in source
|
|
assert "SessionLocal" not in source
|
|
assert "db.add(" not in source
|