Files
geointel/backend/tests/test_segmentation_result_pagination.py
T

152 lines
4.4 KiB
Python

"""Regression coverage for bounded, stable segmentation result listings."""
from __future__ import annotations
from datetime import UTC, datetime
from types import SimpleNamespace
from uuid import UUID
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.api.routes import segmentation as segmentation_routes
from app.db.session import get_db
from app.schemas.segmentation import SegmentationListResponse
from app.services.segmentation_service import SegmentationService
RUN_ID = UUID("00000000-0000-0000-0000-000000000101")
DATASET_ID = UUID("00000000-0000-0000-0000-000000000102")
PROJECT_ID = UUID("00000000-0000-0000-0000-000000000103")
def _segmentation(index: int) -> SimpleNamespace:
return SimpleNamespace(
id=UUID(int=index + 1),
project_id=PROJECT_ID,
dataset_id=DATASET_ID,
analysis_run_id=RUN_ID,
job_id=None,
model_name="segmentation-test-model",
model_version="1",
class_name="building",
confidence=0.99 - index / 100,
bbox_json=None,
area_m2=float(index + 1),
mask_path=None,
source_tile_path=None,
tile_index=index,
properties_json={},
provenance_json={},
created_at=datetime(2026, 8, 23, tzinfo=UTC),
)
class _Session:
def get(self, _model, identifier):
if identifier == RUN_ID:
return SimpleNamespace(analysis_type="segmentation")
return None
def test_service_returns_one_stable_page_with_complete_metadata(monkeypatch) -> None:
rows = [_segmentation(index) for index in range(5)]
monkeypatch.setattr(
SegmentationService,
"_query_segmentation_rows",
staticmethod(lambda _db, **_filters: rows),
)
result = SegmentationService.list_segmentations(
_Session(),
analysis_run_id=RUN_ID,
dataset_id=DATASET_ID,
limit=2,
offset=1,
)
assert [item.id for item in result.items] == [rows[1].id, rows[2].id]
assert result.total == 5
assert result.limit == 2
assert result.offset == 1
assert result.truncated is True
def test_service_pages_cover_the_stably_ordered_population_once(monkeypatch) -> None:
rows = [_segmentation(index) for index in range(5)]
monkeypatch.setattr(
SegmentationService,
"_query_segmentation_rows",
staticmethod(lambda _db, **_filters: rows),
)
seen = []
for offset in (0, 2, 4):
result = SegmentationService.list_segmentations(
_Session(),
dataset_id=DATASET_ID,
limit=2,
offset=offset,
)
seen.extend(item.id for item in result.items)
assert result.total == len(rows)
assert result.offset == offset
assert seen == [row.id for row in rows]
@pytest.mark.parametrize(
("path", "expected_run_id", "expected_dataset_id"),
[
(f"/api/v1/segmentation/runs/{RUN_ID}/segmentations", RUN_ID, None),
(f"/api/v1/segmentation/datasets/{DATASET_ID}/segmentations", None, DATASET_ID),
],
)
def test_both_listing_routes_forward_the_page_window_and_return_it(
monkeypatch,
path: str,
expected_run_id: UUID | None,
expected_dataset_id: UUID | None,
) -> None:
calls: list[dict] = []
def _list(_db, analysis_run_id=None, **parameters):
calls.append({"analysis_run_id": analysis_run_id, **parameters})
return SegmentationListResponse(
items=[],
total=9,
limit=2,
offset=4,
truncated=True,
)
monkeypatch.setattr(SegmentationService, "list_segmentations", staticmethod(_list))
app = FastAPI()
app.include_router(segmentation_routes.router, prefix="/api/v1")
app.dependency_overrides[get_db] = lambda: object()
response = TestClient(app).get(
path,
params={"limit": 2, "offset": 4, "class_name": "building", "min_confidence": 0.5},
)
assert response.status_code == 200
assert response.json()["data"] == {
"items": [],
"total": 9,
"limit": 2,
"offset": 4,
"truncated": True,
}
assert calls == [
{
"analysis_run_id": expected_run_id,
"limit": 2,
"offset": 4,
"dataset_id": expected_dataset_id,
"class_name": "building",
"min_confidence": 0.5,
}
]