Jens/geointel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
8343c9b6be02c3bd7ce0821809a1f3bf693e70e3
geointel/backend/tests/test_segmentation_configured_models.py
T
JensandClaude Opus 5 e2f586c029 consume only artifacts the runtime produced 
tile_manifest_path arrives in the detection and segmentation request and was
read straight off disk, and a manifest entry may name an absolute tile path.
That makes an API field an unbounded reference to the host filesystem, and it
contradicts the rule the persistence model rests on: only a governed,
runtime-produced artifact may be consumed, and a file outside the storage root
is not one.

Both the manifest and every tile it names now resolve under STORAGE_ROOT.
Resolution happens before the comparison, so ".." cannot climb out and a
sibling that merely shares a name prefix does not pass.
GEOINTEL_ALLOW_EXTERNAL_ARTIFACT_PATHS opts out for provisioning workflows that
stage tiles before ingest.

The check honours the Settings the caller is operating under rather than the
process-wide ones, because every analysis path already threads its own.

The affected tests write manifests into tmp_path, so they now declare tmp_path
as the storage root — which is what a deployment does, and makes the fixtures
more honest than they were.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-22 16:16:25 +02:00

552 lines
19 KiB
Python
Raw Blame History

from __future__ import annotations
from hashlib import sha256
import json
from pathlib import Path
from uuid import uuid4
import pytest
from geoalchemy2.shape import to_shape
from app.core.config import Settings
from app.models import Dataset, Project, Segmentation, SourceRegistry, SourceSnapshot
from app.services.detection_georeferencing import pixel_points_to_epsg4326_polygon
from app.services.model_registry_service import ModelRegistryService
from app.services.runtime_model_provenance_service import RuntimeModelProvenanceService
from app.services.segmentation_service import SegmentationService
ROOT = Path(__file__).resolve().parents[2]
class FakeSession:
def __init__(self, objects=None) -> None:
self.objects = objects or {}
self.added = []
self.commits = 0
self.refreshes = []
def get(self, model, item_id):
return self.objects.get((model, item_id))
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)
class AvailableSegAdapter:
def __init__(self, settings: Settings) -> None:
self.settings = settings
@staticmethod
def dependencies_available() -> bool:
return True
def load_model(self, model_path: Path):
return object()
def predict_tile(self, model, tile_path: Path, confidence_threshold: float) -> list[dict]:
return [
{
"class_name": "building",
"confidence": 0.91,
"points": [[10.0, 20.0], [30.0, 20.0], [30.0, 40.0], [10.0, 40.0]],
"bbox": [10.0, 20.0, 30.0, 40.0],
"properties": {"class_id": 0},
}
]
class ClassAgnosticSamAdapter(AvailableSegAdapter):
def predict_tile(self, model, tile_path: Path, confidence_threshold: float) -> list[dict]:
return [
{
"class_name": "segment",
"confidence": None,
"points": [[5.0, 5.0], [25.0, 5.0], [25.0, 25.0], [5.0, 25.0]],
"bbox": [5.0, 5.0, 25.0, 25.0],
"properties": {"class_id": -1},
}
]
class MissingDependencySegAdapter(AvailableSegAdapter):
@staticmethod
def dependencies_available() -> bool:
return False
class NeverLoadSegAdapter(AvailableSegAdapter):
load_calls = 0
def load_model(self, model_path: Path):
type(self).load_calls += 1
raise AssertionError("unmanifested weights must not reach adapter.load_model")
def _project_and_dataset(dataset_type: str = "raster"):
project_id = uuid4()
dataset_id = uuid4()
source_id = uuid4()
snapshot_id = uuid4()
checksum = "a" * 64
project = Project(id=project_id, name="Mol")
source = SourceRegistry(
id=source_id,
source_key="digitaal_vlaanderen_orthophoto",
display_name="Governed orthophoto test source",
classification="contextual",
authority_name="Digitaal Vlaanderen",
authority_scope_json={"zone": "Flanders", "role": "imagery"},
)
snapshot = SourceSnapshot(
id=snapshot_id,
source_registry_id=source_id,
snapshot_key="configured-segmentation-orthophoto",
checksum_sha256=checksum,
ingest_status="ingested",
freshness_status="current",
)
dataset = Dataset(
id=dataset_id,
project_id=project_id,
name="ortho.tif",
dataset_type=dataset_type,
source="digitaal_vlaanderen_orthophoto",
source_name="digitaal_vlaanderen_orthophoto",
storage_path="storage/uploads/ortho.tif",
checksum_sha256=checksum,
source_registry_id=source_id,
source_snapshot_id=snapshot_id,
data_contract_key="geointel.raster.geotiff",
data_contract_version="1.0.0",
validation_status="passed",
provenance_status="complete",
lineage_status="not_applicable",
quarantine_status="not_quarantined",
status="ready",
)
dataset.source_registry = source
dataset.source_snapshot = snapshot
db = FakeSession(objects={(Project, project_id): project, (Dataset, dataset_id): dataset})
return db, project_id, dataset_id
def _settings(tmp_path: Path, **overrides) -> Settings:
values = {
# The runtime only consumes artifacts under the storage root, so a
# test that writes tiles into tmp_path must say that is the root.
"storage_root": str(tmp_path),
"yolo_seg_enabled": True,
"yolo_seg_model_path": str(tmp_path / "seg.pt"),
"sam_enabled": True,
"sam_model_path": str(tmp_path / "sam.pt"),
"yolo_max_tiles": 4,
}
values.update(overrides)
return Settings(**values)
def _write_model_sidecar(
model_path: Path,
*,
model_id: str,
framework: str,
source_version: str | None,
db: FakeSession | None = None,
) -> None:
"""Create explicit local test evidence; no production code creates sidecars."""
model_sha256 = sha256(model_path.read_bytes()).hexdigest()
source_registry_id = uuid4()
source_snapshot_id = uuid4()
resolved_source_version = source_version or "test-v1"
if db is not None:
source_registry = SourceRegistry(
id=source_registry_id,
source_key="model",
display_name="Governed test model artifact",
classification="experimental",
authority_name="GeoIntel test fixture",
freshness_status="current",
ingest_status="configured",
)
source_snapshot = SourceSnapshot(
id=source_snapshot_id,
source_registry_id=source_registry_id,
snapshot_key=f"model-{model_id}-{resolved_source_version}",
source_version=resolved_source_version,
checksum_sha256=model_sha256,
freshness_status="current",
ingest_status="ingested",
)
db.objects[(SourceRegistry, source_registry_id)] = source_registry
db.objects[(SourceSnapshot, source_snapshot_id)] = source_snapshot
payload = {
"schema_version": RuntimeModelProvenanceService.MANIFEST_SCHEMA_VERSION,
"data_contract": {"key": "geointel.model.pytorch", "version": "1.0.0"},
"model": {
"model_id": model_id,
"task_type": "segmentation",
"sha256": model_sha256,
"model_format": "pytorch",
"framework": framework,
"class_mapping": {"0": "segment"},
"source_version": resolved_source_version,
},
"source": {
"source_registry_id": str(source_registry_id),
"source_snapshot_id": str(source_snapshot_id),
"source_registry_key": "model",
"source_snapshot_checksum_sha256": model_sha256,
},
"lineage": {
"upstream_asset_ids": ["test-training-corpus"],
"upstream_checksums_sha256": ["a" * 64],
"transformations": [
{"name": "test-training", "version": "1.0.0", "checksum_sha256": "b" * 64}
],
},
"metadata": {"training_manifest_sha256": "c" * 64},
"imported_at": "2026-08-01T10:00:00+00:00",
}
payload["metadata"]["runtime_manifest_sha256"] = RuntimeModelProvenanceService.manifest_self_checksum(payload)
RuntimeModelProvenanceService.manifest_path_for_model(model_path).write_text(
json.dumps(payload, sort_keys=True),
encoding="utf-8",
)
def _write_configured_model_sidecars(
tmp_path: Path,
settings: Settings,
*,
include_yolo: bool = True,
include_sam: bool = True,
db: FakeSession | None = None,
) -> None:
if include_yolo:
_write_model_sidecar(
tmp_path / "seg.pt",
model_id=settings.yolo_seg_model_id,
framework="ultralytics/pytorch",
source_version=settings.yolo_seg_model_version,
db=db,
)
if include_sam:
_write_model_sidecar(
tmp_path / "sam.pt",
model_id=settings.sam_model_id,
framework="ultralytics/sam",
source_version=settings.sam_model_version,
db=db,
)
def _manifest(tmp_path: Path, tile_count: int = 1) -> Path:
tiles = []
for index in range(tile_count):
tile_path = tmp_path / f"tile_{index:04d}.tif"
tile_path.write_bytes(b"fixture")
tiles.append(
{
"path": str(tile_path),
"pixel_window": [0, 0, 100, 100],
"bounds": [4.0, 51.0, 5.0, 52.0],
"transform": [4.0, 0.01, 0.0, 52.0, 0.0, -0.01],
"crs": "EPSG:4326",
"index": index,
}
)
manifest_path = tmp_path / "manifest.json"
manifest_path.write_text(
json.dumps(
{
"tile_set_id": "tiles-fixture",
"source_dataset_id": str(uuid4()),
"source_raster_id": str(uuid4()),
"crs": "EPSG:4326",
"bounds": [4.0, 51.0, 5.0, 52.0],
"tile_size": 100,
"overlap": 0,
"count": tile_count,
"tiles": tiles,
}
),
encoding="utf-8",
)
return manifest_path
def test_segmentation_models_report_not_configured_when_disabled(tmp_path: Path) -> None:
settings = _settings(tmp_path, yolo_seg_enabled=False, sam_enabled=False)
models = {
model.model_id: model
for model in ModelRegistryService.list_segmentation_model_capabilities(settings=settings)
}
assert models["yolo-seg-configured"].configured is False
assert models["yolo-seg-configured"].status == "not_configured"
assert models["sam-configured"].configured is False
assert models["sam-configured"].status == "not_configured"
def test_segmentation_models_report_dependency_unavailable(tmp_path: Path) -> None:
(tmp_path / "seg.pt").write_bytes(b"weights")
(tmp_path / "sam.pt").write_bytes(b"weights")
settings = _settings(tmp_path)
models = {
model.model_id: model
for model in ModelRegistryService.list_segmentation_model_capabilities(
settings=settings,
yolo_seg_adapter_class=MissingDependencySegAdapter,
sam_adapter_class=MissingDependencySegAdapter,
)
}
assert models["yolo-seg-configured"].status == "dependency_unavailable"
assert models["sam-configured"].status == "dependency_unavailable"
def test_segmentation_models_require_runtime_provenance_sidecars(tmp_path: Path) -> None:
(tmp_path / "seg.pt").write_bytes(b"unmanifested yolo segmentation weights")
(tmp_path / "sam.pt").write_bytes(b"unmanifested sam weights")
settings = _settings(tmp_path)
models = {
model.model_id: model
for model in ModelRegistryService.list_segmentation_model_capabilities(
settings=settings,
yolo_seg_adapter_class=AvailableSegAdapter,
sam_adapter_class=ClassAgnosticSamAdapter,
)
}
assert models["yolo-seg-configured"].configured is False
assert models["yolo-seg-configured"].status == "contract_incomplete"
assert models["sam-configured"].configured is False
assert models["sam-configured"].status == "contract_incomplete"
def test_segmentation_models_report_configured_with_local_weights(tmp_path: Path) -> None:
(tmp_path / "seg.pt").write_bytes(b"weights")
(tmp_path / "sam.pt").write_bytes(b"weights")
settings = _settings(tmp_path)
_write_configured_model_sidecars(tmp_path, settings)
models = {
model.model_id: model
for model in ModelRegistryService.list_segmentation_model_capabilities(
settings=settings,
yolo_seg_adapter_class=AvailableSegAdapter,
sam_adapter_class=ClassAgnosticSamAdapter,
)
}
assert models["yolo-seg-configured"].configured is True
assert models["yolo-seg-configured"].status == "configured"
assert models["sam-configured"].configured is True
assert models["sam-configured"].status == "configured"
def test_segmentation_dependency_check_uses_real_imports_not_find_spec() -> None:
source = (ROOT / "backend" / "app" / "services" / "segmentation_adapter.py").read_text(encoding="utf-8")
assert 'find_spec("ultralytics")' not in source
assert "import ultralytics" in source
assert "import torch" in source
def test_configured_segmentation_requires_tile_manifest(tmp_path: Path) -> None:
(tmp_path / "seg.pt").write_bytes(b"weights")
db, project_id, dataset_id = _project_and_dataset()
settings = _settings(tmp_path)
_write_configured_model_sidecars(tmp_path, settings, include_sam=False, db=db)
with pytest.raises(Exception) as exc_info:
SegmentationService.run_segmentation(
db=db,
project_id=project_id,
dataset_id=dataset_id,
model_id="yolo-seg-configured",
confidence_threshold=0.5,
settings=settings,
yolo_seg_adapter_class=AvailableSegAdapter,
sam_adapter_class=ClassAgnosticSamAdapter,
)
assert getattr(exc_info.value, "code", None) == "SEGMENTATION_TILE_MANIFEST_REQUIRED"
def test_configured_segmentation_fails_before_adapter_load_without_sidecar(tmp_path: Path) -> None:
(tmp_path / "seg.pt").write_bytes(b"unmanifested weights")
db, project_id, dataset_id = _project_and_dataset()
settings = _settings(tmp_path)
NeverLoadSegAdapter.load_calls = 0
response = SegmentationService.run_segmentation(
db=db,
project_id=project_id,
dataset_id=dataset_id,
model_id="yolo-seg-configured",
confidence_threshold=0.5,
tile_manifest_path=str(_manifest(tmp_path)),
settings=settings,
yolo_seg_adapter_class=NeverLoadSegAdapter,
sam_adapter_class=ClassAgnosticSamAdapter,
)
assert response.status == "failed"
assert response.error_code == "SEGMENTATION_MODEL_UNAVAILABLE"
assert NeverLoadSegAdapter.load_calls == 0
def test_configured_segmentation_rejects_unbound_model_snapshot_before_adapter_load(tmp_path: Path) -> None:
(tmp_path / "seg.pt").write_bytes(b"structurally valid but unbound weights")
db, project_id, dataset_id = _project_and_dataset()
settings = _settings(tmp_path)
# The sidecar passes catalog validation but its source registry/snapshot
# was never registered in this production-session fixture.
_write_configured_model_sidecars(tmp_path, settings, include_sam=False)
NeverLoadSegAdapter.load_calls = 0
response = SegmentationService.run_segmentation(
db=db,
project_id=project_id,
dataset_id=dataset_id,
model_id="yolo-seg-configured",
confidence_threshold=0.5,
tile_manifest_path=str(_manifest(tmp_path)),
settings=settings,
yolo_seg_adapter_class=NeverLoadSegAdapter,
sam_adapter_class=ClassAgnosticSamAdapter,
)
assert response.status == "failed"
assert response.error_code == "MODEL_PROVENANCE_SOURCE_REGISTRY_NOT_FOUND"
assert NeverLoadSegAdapter.load_calls == 0
def test_configured_yolo_seg_run_persists_georeferenced_masks(tmp_path: Path) -> None:
(tmp_path / "seg.pt").write_bytes(b"weights")
db, project_id, dataset_id = _project_and_dataset()
settings = _settings(tmp_path)
_write_configured_model_sidecars(tmp_path, settings, include_sam=False, db=db)
manifest_path = _manifest(tmp_path)
response = SegmentationService.run_segmentation(
db=db,
project_id=project_id,
dataset_id=dataset_id,
model_id="yolo-seg-configured",
confidence_threshold=0.5,
tile_manifest_path=str(manifest_path),
settings=settings,
yolo_seg_adapter_class=AvailableSegAdapter,
sam_adapter_class=ClassAgnosticSamAdapter,
)
assert response.status == "success"
assert response.segmentation_count == 1
persisted = [item for item in db.added if isinstance(item, Segmentation)]
assert len(persisted) == 1
segmentation = persisted[0]
assert segmentation.class_name == "building"
assert segmentation.confidence == pytest.approx(0.91)
geometry = to_shape(segmentation.geometry)
assert geometry.geom_type == "MultiPolygon"
min_x, min_y, max_x, max_y = geometry.bounds
assert 4.0 <= min_x <= 5.0
assert 51.0 <= min_y <= 52.0
assert max_x <= 5.0
assert max_y <= 52.0
assert segmentation.area_m2 is not None and segmentation.area_m2 > 0
assert segmentation.provenance_json["inference"] == "local"
assert segmentation.provenance_json["model_id"] == "yolo-seg-configured"
assert segmentation.provenance_json["runtime_model_provenance"]["data_contract_key"] == "geointel.model.pytorch"
def test_configured_sam_run_is_class_agnostic(tmp_path: Path) -> None:
(tmp_path / "sam.pt").write_bytes(b"weights")
db, project_id, dataset_id = _project_and_dataset()
settings = _settings(tmp_path)
_write_configured_model_sidecars(tmp_path, settings, include_yolo=False, db=db)
manifest_path = _manifest(tmp_path)
response = SegmentationService.run_segmentation(
db=db,
project_id=project_id,
dataset_id=dataset_id,
model_id="sam-configured",
confidence_threshold=0.5,
tile_manifest_path=str(manifest_path),
settings=settings,
yolo_seg_adapter_class=AvailableSegAdapter,
sam_adapter_class=ClassAgnosticSamAdapter,
)
assert response.status == "success"
assert response.segmentation_count == 1
persisted = [item for item in db.added if isinstance(item, Segmentation)]
assert persisted[0].class_name == "segment"
assert persisted[0].confidence is None
def test_unconfigured_segmentation_run_fails_closed(tmp_path: Path) -> None:
db, project_id, dataset_id = _project_and_dataset()
settings = _settings(tmp_path, yolo_seg_enabled=False)
manifest_path = _manifest(tmp_path)
response = SegmentationService.run_segmentation(
db=db,
project_id=project_id,
dataset_id=dataset_id,
model_id="yolo-seg-configured",
confidence_threshold=0.5,
tile_manifest_path=str(manifest_path),
settings=settings,
yolo_seg_adapter_class=AvailableSegAdapter,
sam_adapter_class=ClassAgnosticSamAdapter,
)
assert response.status == "failed"
assert response.error_code == "SEGMENTATION_MODEL_UNAVAILABLE"
assert not [item for item in db.added if isinstance(item, Segmentation)]
def test_pixel_points_to_epsg4326_polygon_uses_tile_transform() -> None:
tile = {
"transform": [4.0, 0.01, 0.0, 52.0, 0.0, -0.01],
"bounds": [4.0, 51.0, 5.0, 52.0],
"pixel_window": [0, 0, 100, 100],
}
polygon = pixel_points_to_epsg4326_polygon(
points=[[0.0, 0.0], [100.0, 0.0], [100.0, 100.0], [0.0, 100.0]],
tile=tile,
crs="EPSG:4326",
)
min_x, min_y, max_x, max_y = polygon.bounds
assert min_x == pytest.approx(4.0)
assert max_x == pytest.approx(5.0)
assert min_y == pytest.approx(51.0)
assert max_y == pytest.approx(52.0)
def test_pixel_points_to_epsg4326_polygon_rejects_degenerate_input() -> None:
tile = {"transform": [4.0, 0.01, 0.0, 52.0, 0.0, -0.01]}
with pytest.raises(Exception) as exc_info:
pixel_points_to_epsg4326_polygon(points=[[0.0, 0.0], [1.0, 1.0]], tile=tile)
assert getattr(exc_info.value, "code", None) == "SEGMENTATION_INVALID_MASK"
Reference in New Issue View Git Blame Copy Permalink