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
478 lines
18 KiB
Python
478 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from geoalchemy2.shape import from_shape
|
|
from shapely.geometry import MultiPolygon, box, mapping
|
|
|
|
from app.db.session import get_db
|
|
from app.main import app
|
|
from app.models import (
|
|
AnalysisRun,
|
|
Dataset,
|
|
Job,
|
|
Metric,
|
|
Project,
|
|
QualityCheck,
|
|
Segmentation,
|
|
SourceRegistry,
|
|
SourceSnapshot,
|
|
VectorFeature,
|
|
)
|
|
from app.services.model_registry_service import ModelRegistryService
|
|
from app.services.segmentation_service import SegmentationService
|
|
|
|
|
|
class FakeQuery:
|
|
def __init__(self, rows):
|
|
self.rows = list(rows)
|
|
|
|
def filter(self, *criteria):
|
|
for criterion in criteria:
|
|
left = getattr(criterion, "left", None)
|
|
right = getattr(criterion, "right", None)
|
|
operator = getattr(criterion, "operator", None)
|
|
name = getattr(left, "name", None)
|
|
value = getattr(right, "value", right)
|
|
if name and operator:
|
|
if operator.__name__ == "eq":
|
|
self.rows = [row for row in self.rows if getattr(row, name) == value]
|
|
elif operator.__name__ == "ge":
|
|
self.rows = [row for row in self.rows if getattr(row, name) >= value]
|
|
return self
|
|
|
|
def order_by(self, *_args):
|
|
return self
|
|
|
|
def all(self):
|
|
return list(self.rows)
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, objects=None, query_rows=None) -> None:
|
|
self.objects = objects or {}
|
|
self.query_rows = query_rows or {}
|
|
self.added = []
|
|
self.commits = 0
|
|
self.refreshes = []
|
|
|
|
def get(self, model, item_id):
|
|
return self.objects.get((model, item_id))
|
|
|
|
def query(self, model):
|
|
return FakeQuery(self.query_rows.get(model, []))
|
|
|
|
def add(self, item) -> None:
|
|
self.added.append(item)
|
|
if getattr(item, "id", None) is not None:
|
|
self.objects[(item.__class__, item.id)] = item
|
|
|
|
def commit(self) -> None:
|
|
self.commits += 1
|
|
|
|
def refresh(self, item) -> None:
|
|
self.refreshes.append(item)
|
|
|
|
|
|
def _project_and_dataset(dataset_type: str = "raster"):
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
project = Project(id=project_id, name="Geel")
|
|
dataset = Dataset(
|
|
id=dataset_id,
|
|
project_id=project_id,
|
|
name="source.tif",
|
|
dataset_type=dataset_type,
|
|
source="test",
|
|
storage_path="storage/uploads/source.tif",
|
|
)
|
|
db = FakeSession(objects={(Project, project_id): project, (Dataset, dataset_id): dataset})
|
|
return db, project_id, dataset_id
|
|
|
|
|
|
def _segmentation(project_id, dataset_id, analysis_run_id, class_name="vegetation", confidence=0.81, geom=None):
|
|
segmentation_id = uuid4()
|
|
return Segmentation(
|
|
id=segmentation_id,
|
|
project_id=project_id,
|
|
dataset_id=dataset_id,
|
|
analysis_run_id=analysis_run_id,
|
|
job_id=uuid4(),
|
|
model_name="fixture-segmenter",
|
|
model_version="fixture-v1",
|
|
class_name=class_name,
|
|
confidence=confidence,
|
|
geometry=from_shape(geom or MultiPolygon([box(4.0, 51.0, 4.1, 51.1)]), srid=4326),
|
|
bbox_json={"min_x": 4.0, "min_y": 51.0, "max_x": 4.1, "max_y": 51.1},
|
|
area_m2=123.4,
|
|
mask_path=f"storage/masks/{project_id}/{analysis_run_id}/tile_0/mask_{segmentation_id}.png",
|
|
source_tile_path="storage/tiles/tile_0000.tif",
|
|
tile_index=0,
|
|
properties_json={"source": "unit-test-fixture"},
|
|
provenance_json={"fixture_mode": True},
|
|
)
|
|
|
|
|
|
def _authoritative_reference(dataset: Dataset) -> Dataset:
|
|
"""Model segmentation QA references as governed authoritative GRB evidence."""
|
|
|
|
source_id = uuid4()
|
|
snapshot_id = uuid4()
|
|
checksum = "a" * 64
|
|
source = SourceRegistry(
|
|
id=source_id,
|
|
source_key="grb",
|
|
display_name="GRB segmentation QA fixture",
|
|
classification="authoritative",
|
|
authority_name="Digitaal Vlaanderen",
|
|
authority_scope_json={"zone": "Flanders"},
|
|
usage_policy_json={"ground_truth_allowed": True, "validation_authority": {"building_validation": "primary"}},
|
|
)
|
|
snapshot = SourceSnapshot(
|
|
id=snapshot_id,
|
|
source_registry_id=source_id,
|
|
snapshot_key=f"segmentation-qa-{dataset.id}",
|
|
checksum_sha256=checksum,
|
|
ingest_status="ingested",
|
|
freshness_status="current",
|
|
)
|
|
dataset.source = "grb"
|
|
dataset.source_name = "grb"
|
|
dataset.dataset_role = "reference"
|
|
dataset.status = "ready"
|
|
dataset.checksum_sha256 = checksum
|
|
dataset.source_registry_id = source_id
|
|
dataset.source_snapshot_id = snapshot_id
|
|
dataset.data_contract_key = "geointel.vector.geojson"
|
|
dataset.data_contract_version = "1.0.0"
|
|
dataset.validation_status = "passed"
|
|
dataset.provenance_status = "complete"
|
|
dataset.lineage_status = "complete"
|
|
dataset.quarantine_status = "not_quarantined"
|
|
dataset.source_registry = source
|
|
dataset.source_snapshot = snapshot
|
|
return dataset
|
|
|
|
|
|
def test_model_registry_returns_segmentation_states() -> None:
|
|
models = {model.model_id: model for model in ModelRegistryService.list_model_capabilities(task_type="segmentation")}
|
|
|
|
assert set(models) == {"segmentation-placeholder", "fixture-segmenter", "yolo-seg-configured", "sam-configured"}
|
|
assert models["segmentation-placeholder"].task_type == "segmentation"
|
|
assert models["segmentation-placeholder"].status == "not_configured"
|
|
assert models["fixture-segmenter"].configured is True
|
|
assert "fixture" in models["fixture-segmenter"].limitation_message.lower()
|
|
assert models["yolo-seg-configured"].status == "not_configured"
|
|
assert models["sam-configured"].status == "not_configured"
|
|
|
|
|
|
def test_unavailable_segmentation_model_creates_failed_run_without_segmentations() -> None:
|
|
db, project_id, dataset_id = _project_and_dataset()
|
|
|
|
result = SegmentationService.run_segmentation(
|
|
db=db,
|
|
project_id=project_id,
|
|
dataset_id=dataset_id,
|
|
model_id="segmentation-placeholder",
|
|
confidence_threshold=0.5,
|
|
)
|
|
|
|
runs = [item for item in db.added if isinstance(item, AnalysisRun)]
|
|
jobs = [item for item in db.added if isinstance(item, Job)]
|
|
segmentations = [item for item in db.added if isinstance(item, Segmentation)]
|
|
|
|
assert result.status == "failed"
|
|
assert result.error_code == "SEGMENTATION_MODEL_UNAVAILABLE"
|
|
assert result.segmentation_count == 0
|
|
assert runs[0].analysis_type == "segmentation"
|
|
assert runs[0].status == "failed"
|
|
assert jobs[0].job_type == "segmentation.run"
|
|
assert jobs[0].status == "failed"
|
|
assert segmentations == []
|
|
|
|
|
|
def test_fixture_segmenter_requires_explicit_mode_and_persists_segmentations() -> None:
|
|
db, project_id, dataset_id = _project_and_dataset()
|
|
|
|
with pytest.raises(Exception) as exc_info:
|
|
SegmentationService.run_segmentation(
|
|
db=db,
|
|
project_id=project_id,
|
|
dataset_id=dataset_id,
|
|
model_id="fixture-segmenter",
|
|
confidence_threshold=0.5,
|
|
parameters_json={"fixture_segmentations": []},
|
|
)
|
|
assert getattr(exc_info.value, "code", None) == "FIXTURE_MODE_REQUIRED"
|
|
|
|
geometry = mapping(box(4.0, 51.0, 4.1, 51.1))
|
|
result = SegmentationService.run_segmentation(
|
|
db=db,
|
|
project_id=project_id,
|
|
dataset_id=dataset_id,
|
|
model_id="fixture-segmenter",
|
|
confidence_threshold=0.5,
|
|
class_filter=["vegetation"],
|
|
parameters_json={
|
|
"fixture_mode": True,
|
|
"fixture_segmentations": [
|
|
{
|
|
"class_name": "vegetation",
|
|
"confidence": 0.88,
|
|
"geometry": geometry,
|
|
"bbox_json": {"min_x": 4.0, "min_y": 51.0, "max_x": 4.1, "max_y": 51.1},
|
|
"source_tile_path": "storage/tiles/tile_0000.tif",
|
|
"tile_index": 0,
|
|
"properties_json": {"source": "unit-test-fixture"},
|
|
"provenance_json": {"crs": "EPSG:4326"},
|
|
}
|
|
],
|
|
},
|
|
)
|
|
|
|
segmentations = [item for item in db.added if isinstance(item, Segmentation)]
|
|
assert result.status == "success"
|
|
assert result.segmentation_count == 1
|
|
assert segmentations[0].project_id == project_id
|
|
assert segmentations[0].dataset_id == dataset_id
|
|
assert segmentations[0].analysis_run_id == result.analysis_run_id
|
|
assert segmentations[0].model_name == "fixture-segmenter"
|
|
assert segmentations[0].class_name == "vegetation"
|
|
assert segmentations[0].confidence == 0.88
|
|
assert segmentations[0].mask_path.endswith(f"mask_{segmentations[0].id}.png")
|
|
|
|
|
|
def test_non_raster_dataset_request_is_rejected() -> None:
|
|
db, project_id, dataset_id = _project_and_dataset(dataset_type="vector")
|
|
|
|
with pytest.raises(Exception) as exc_info:
|
|
SegmentationService.run_segmentation(
|
|
db=db,
|
|
project_id=project_id,
|
|
dataset_id=dataset_id,
|
|
model_id="segmentation-placeholder",
|
|
confidence_threshold=0.5,
|
|
)
|
|
|
|
assert getattr(exc_info.value, "code", None) == "INVALID_DATASET_TYPE"
|
|
|
|
|
|
def test_invalid_empty_fixture_geometry_is_rejected() -> None:
|
|
db, project_id, dataset_id = _project_and_dataset()
|
|
|
|
with pytest.raises(Exception) as exc_info:
|
|
SegmentationService.run_segmentation(
|
|
db=db,
|
|
project_id=project_id,
|
|
dataset_id=dataset_id,
|
|
model_id="fixture-segmenter",
|
|
confidence_threshold=0.5,
|
|
parameters_json={
|
|
"fixture_mode": True,
|
|
"fixture_segmentations": [
|
|
{"class_name": "vegetation", "confidence": 0.9, "geometry": {"type": "Polygon", "coordinates": []}},
|
|
],
|
|
},
|
|
)
|
|
|
|
assert getattr(exc_info.value, "code", None) == "INVALID_FIXTURE_GEOMETRY"
|
|
|
|
|
|
def test_segmentation_geojson_feature_collection_shape_and_provenance() -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
analysis_run_id = uuid4()
|
|
segmentation = _segmentation(project_id, dataset_id, analysis_run_id)
|
|
db = FakeSession(
|
|
objects={(AnalysisRun, analysis_run_id): AnalysisRun(id=analysis_run_id, project_id=project_id, analysis_type="segmentation", status="success", parameters_json={})},
|
|
query_rows={Segmentation: [segmentation]},
|
|
)
|
|
|
|
feature_collection = SegmentationService.segmentations_to_geojson(db, analysis_run_id=analysis_run_id)
|
|
|
|
assert feature_collection["type"] == "FeatureCollection"
|
|
feature = feature_collection["features"][0]
|
|
assert feature["geometry"]["type"] == "MultiPolygon"
|
|
assert feature["properties"]["segmentation_id"] == str(segmentation.id)
|
|
assert feature["properties"]["class_name"] == "vegetation"
|
|
assert feature["properties"]["confidence"] == 0.81
|
|
assert feature["properties"]["area_m2"] == 123.4
|
|
assert feature["properties"]["model_name"] == "fixture-segmenter"
|
|
assert feature["properties"]["analysis_run_id"] == str(analysis_run_id)
|
|
assert feature["properties"]["dataset_id"] == str(dataset_id)
|
|
assert feature["properties"]["job_id"] == str(segmentation.job_id)
|
|
assert feature["properties"]["source_tile_path"] == "storage/tiles/tile_0000.tif"
|
|
assert feature["properties"]["tile_index"] == 0
|
|
assert feature["properties"]["mask_path"] == segmentation.mask_path
|
|
assert feature["properties"]["bbox_json"] == segmentation.bbox_json
|
|
assert feature["properties"]["provenance_json"] == {"fixture_mode": True}
|
|
|
|
|
|
def test_segmentation_qa_persists_quality_check_and_metrics() -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
reference_dataset_id = uuid4()
|
|
analysis_run_id = uuid4()
|
|
segmentation = _segmentation(project_id, dataset_id, analysis_run_id, geom=box(0, 0, 1, 1))
|
|
reference_dataset = _authoritative_reference(Dataset(
|
|
id=reference_dataset_id,
|
|
project_id=project_id,
|
|
name="reference.geojson",
|
|
dataset_type="vector",
|
|
source="test",
|
|
dataset_role="reference",
|
|
))
|
|
reference_feature = VectorFeature(
|
|
id=uuid4(),
|
|
dataset_id=reference_dataset_id,
|
|
feature_class="vegetation",
|
|
geometry=from_shape(box(0, 0, 1, 1), srid=4326),
|
|
)
|
|
db = FakeSession(
|
|
objects={
|
|
(AnalysisRun, analysis_run_id): AnalysisRun(id=analysis_run_id, project_id=project_id, dataset_id=dataset_id, analysis_type="segmentation", status="success", parameters_json={}),
|
|
(Dataset, dataset_id): Dataset(id=dataset_id, project_id=project_id, name="fixture.tif", dataset_type="raster", source="test"),
|
|
(Dataset, reference_dataset_id): reference_dataset,
|
|
},
|
|
query_rows={Segmentation: [segmentation], VectorFeature: [reference_feature]},
|
|
)
|
|
|
|
result = SegmentationService.compare_segmentations_with_reference(
|
|
db=db,
|
|
analysis_run_id=analysis_run_id,
|
|
reference_dataset_id=reference_dataset_id,
|
|
iou_threshold=0.5,
|
|
)
|
|
|
|
quality_checks = [item for item in db.added if isinstance(item, QualityCheck)]
|
|
metrics = [item for item in db.added if isinstance(item, Metric)]
|
|
assert result["matches"] == 1
|
|
assert result["precision"] == 1.0
|
|
assert result["recall"] == 1.0
|
|
assert result["f1_score"] == 1.0
|
|
assert quality_checks[0].check_type == "segmentations_vs_reference"
|
|
assert quality_checks[0].analysis_run_id == analysis_run_id
|
|
assert [metric.metric_key for metric in metrics] == [
|
|
"precision",
|
|
"recall",
|
|
"f1",
|
|
"mean_iou",
|
|
"false_positive_count",
|
|
"false_negative_count",
|
|
# Threshold-independent, as detection QA already reported. The two
|
|
# panels previously answered different questions about the same run.
|
|
"average_precision",
|
|
"best_f1",
|
|
"best_f1_threshold",
|
|
]
|
|
|
|
|
|
def test_segmentation_qa_no_match_case_persists_zero_scores() -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
reference_dataset_id = uuid4()
|
|
analysis_run_id = uuid4()
|
|
segmentation = _segmentation(project_id, dataset_id, analysis_run_id, geom=box(0, 0, 1, 1))
|
|
reference_dataset = _authoritative_reference(Dataset(
|
|
id=reference_dataset_id,
|
|
project_id=project_id,
|
|
name="reference.geojson",
|
|
dataset_type="vector",
|
|
source="test",
|
|
dataset_role="reference",
|
|
))
|
|
reference_feature = VectorFeature(
|
|
id=uuid4(),
|
|
dataset_id=reference_dataset_id,
|
|
feature_class="vegetation",
|
|
geometry=from_shape(box(10, 10, 11, 11), srid=4326),
|
|
)
|
|
db = FakeSession(
|
|
objects={
|
|
(AnalysisRun, analysis_run_id): AnalysisRun(id=analysis_run_id, project_id=project_id, dataset_id=dataset_id, analysis_type="segmentation", status="success", parameters_json={}),
|
|
(Dataset, dataset_id): Dataset(id=dataset_id, project_id=project_id, name="fixture.tif", dataset_type="raster", source="test"),
|
|
(Dataset, reference_dataset_id): reference_dataset,
|
|
},
|
|
query_rows={Segmentation: [segmentation], VectorFeature: [reference_feature]},
|
|
)
|
|
|
|
result = SegmentationService.compare_segmentations_with_reference(
|
|
db=db,
|
|
analysis_run_id=analysis_run_id,
|
|
reference_dataset_id=reference_dataset_id,
|
|
iou_threshold=0.5,
|
|
)
|
|
|
|
assert result["matches"] == 0
|
|
assert result["false_positives"] == 1
|
|
assert result["false_negatives"] == 1
|
|
assert result["precision"] == 0.0
|
|
assert result["recall"] == 0.0
|
|
assert result["f1_score"] == 0.0
|
|
|
|
|
|
def test_segmentation_models_api_uses_envelope() -> None:
|
|
response = TestClient(app).get("/api/v1/segmentation/models")
|
|
|
|
assert response.status_code == 200
|
|
assert "data" in response.json()
|
|
assert {model["model_id"] for model in response.json()["data"]["models"]} == {
|
|
"segmentation-placeholder",
|
|
"fixture-segmenter",
|
|
"yolo-seg-configured",
|
|
"sam-configured",
|
|
}
|
|
|
|
|
|
def test_segmentation_geojson_api_uses_canonical_envelope(monkeypatch) -> None:
|
|
analysis_run_id = uuid4()
|
|
|
|
monkeypatch.setattr(
|
|
"app.api.routes.segmentation.SegmentationService.segmentations_to_geojson",
|
|
lambda *_args, **_kwargs: {"type": "FeatureCollection", "features": []},
|
|
)
|
|
app.dependency_overrides[get_db] = lambda: FakeSession()
|
|
try:
|
|
response = TestClient(app).get(f"/api/v1/segmentation/runs/{analysis_run_id}/geojson")
|
|
finally:
|
|
app.dependency_overrides.pop(get_db, None)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"data": {"type": "FeatureCollection", "features": []}}
|
|
|
|
|
|
def test_sprint9_migration_declares_segmentation_foundation() -> None:
|
|
migration_path = Path(__file__).parents[1] / "alembic" / "versions" / "202606120900_sprint9_segmentation_foundation.py"
|
|
migration_text = migration_path.read_text(encoding="utf-8")
|
|
|
|
for required_text in (
|
|
"segmentations",
|
|
"project_id",
|
|
"dataset_id",
|
|
"job_id",
|
|
"analysis_run_id",
|
|
"model_name",
|
|
"model_version",
|
|
"class_name",
|
|
"confidence",
|
|
"MultiPolygon",
|
|
"bbox_json",
|
|
"area_m2",
|
|
"mask_path",
|
|
"source_tile_path",
|
|
"tile_index",
|
|
"properties_json",
|
|
"provenance_json",
|
|
"ix_segmentations_project_id",
|
|
"ix_segmentations_dataset_id",
|
|
"ix_segmentations_analysis_run_id",
|
|
"ix_segmentations_job_id",
|
|
"ix_segmentations_class_name",
|
|
"ix_segmentations_geometry",
|
|
'postgresql_using="gist"',
|
|
):
|
|
assert required_text in migration_text
|