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
87 lines
3.4 KiB
Python
87 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from shapely.geometry import box
|
|
|
|
from app.core.errors import AppError
|
|
from app.services.aoi_operation_executor import AoiOperationExecutor
|
|
from app.services.aoi_operation_service import AoiOperationService
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def test_partition_plan_covers_aoi_without_overlapping_area() -> None:
|
|
aoi = box(0, 0, 25_000, 18_000)
|
|
partitions = AoiOperationService._partition(aoi, 10_000)
|
|
|
|
assert len(partitions) == 6
|
|
assert sum(partition.area for partition in partitions) == pytest.approx(aoi.area)
|
|
assert all(partition.within(aoi) for partition in partitions)
|
|
for index, partition in enumerate(partitions):
|
|
for other in partitions[index + 1 :]:
|
|
assert partition.intersection(other).area == pytest.approx(0.0)
|
|
|
|
|
|
def test_partition_plan_intersects_irregular_aoi_exactly() -> None:
|
|
aoi = box(0, 0, 20_000, 20_000).difference(box(5_000, 5_000, 15_000, 15_000))
|
|
partitions = AoiOperationService._partition(aoi, 8_000)
|
|
|
|
assert sum(partition.area for partition in partitions) == pytest.approx(aoi.area)
|
|
assert all(not partition.intersects(box(5_001, 5_001, 14_999, 14_999)) for partition in partitions)
|
|
|
|
|
|
def test_partition_plan_fails_before_unbounded_fanout(monkeypatch) -> None:
|
|
monkeypatch.setattr(AoiOperationService, "MAX_PARTITIONS", 4)
|
|
|
|
with pytest.raises(AppError) as exc_info:
|
|
AoiOperationService._partition(box(0, 0, 30_000, 30_000), 10_000)
|
|
|
|
assert exc_info.value.code == "AOI_PARTITION_LIMIT_EXCEEDED"
|
|
assert exc_info.value.details["candidate_count"] == 9
|
|
|
|
|
|
def test_executor_retries_only_transient_provider_failures() -> None:
|
|
assert AoiOperationExecutor._retryable(AppError(code="UPSTREAM_UNAVAILABLE", message="down", status_code=503)) is True
|
|
assert AoiOperationExecutor._retryable(AppError(code="INVALID_SCOPE", message="bad", status_code=422)) is False
|
|
|
|
|
|
def test_provider_budget_is_automatic_and_override_can_only_be_stricter() -> None:
|
|
governed = AoiOperationService._partition_side("grb", None)
|
|
assert governed > 0
|
|
assert AoiOperationService._partition_side("grb", governed * 2) == governed
|
|
assert AoiOperationService._partition_side("grb", governed / 2) == governed / 2
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("provider_key", "max_pixels", "resolution_m"),
|
|
[
|
|
("dhmv", 12_000_000, 5.0),
|
|
("flood_hazard", 12_000_000, 5.0),
|
|
("spw_terrain", 12_000_000, 5.0),
|
|
("thematic_raster", 30_000_000, 10.0),
|
|
("walous", 36_000_000, 10.0),
|
|
],
|
|
)
|
|
def test_raster_provider_budget_never_exceeds_decoded_pixel_limit(
|
|
provider_key: str, max_pixels: int, resolution_m: float
|
|
) -> None:
|
|
side_m = AoiOperationService._partition_side(provider_key, None)
|
|
|
|
assert (side_m / resolution_m) ** 2 < max_pixels
|
|
|
|
|
|
def test_migration_and_api_are_registered() -> None:
|
|
migration = (ROOT / "backend/alembic/versions/202607260001_aoi_operations.py").read_text(encoding="utf-8")
|
|
main = (ROOT / "backend/app/main.py").read_text(encoding="utf-8")
|
|
route = (ROOT / "backend/app/api/routes/aoi_operations.py").read_text(encoding="utf-8")
|
|
|
|
assert 'down_revision = "202607160001"' in migration
|
|
assert '"aoi_operations"' in migration
|
|
assert '"aoi_operation_partitions"' in migration
|
|
assert "app.include_router(aoi_operations.router" in main
|
|
assert '"/{operation_id}/execute-next"' in route
|
|
assert '"/{operation_id}/partitions/{partition_id}/checkpoint"' in route
|