104 lines
3.4 KiB
Python
104 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from app.api.routes.selection_partitions import select_vector_partitions
|
|
from app.core.errors import AppError
|
|
from app.models import Dataset
|
|
from app.schemas.selection_partitions import VectorPartitionSelectionRequest
|
|
from app.services.vector_feature_service import VectorFeatureService
|
|
|
|
|
|
class DatasetQuery:
|
|
def __init__(self, datasets):
|
|
self.datasets = datasets
|
|
|
|
def filter(self, *_args):
|
|
return self
|
|
|
|
def all(self):
|
|
return self.datasets
|
|
|
|
|
|
class DatasetSession:
|
|
def __init__(self, datasets):
|
|
self.datasets = datasets
|
|
|
|
def query(self, model):
|
|
assert model is Dataset
|
|
return DatasetQuery(self.datasets)
|
|
|
|
|
|
def make_dataset(project_id, dataset_id, *, source_name="grb", product_key="buildings"):
|
|
return Dataset(
|
|
id=dataset_id,
|
|
project_id=project_id,
|
|
name=f"{source_name}-{product_key}",
|
|
dataset_type="vector",
|
|
source="official",
|
|
dataset_role="reference",
|
|
source_name=source_name,
|
|
reference_layer_name=product_key,
|
|
source_metadata={"product_key": product_key, "theme": product_key},
|
|
provenance_metadata={},
|
|
metadata_json={},
|
|
status="ready",
|
|
)
|
|
|
|
|
|
def test_vector_partition_route_combines_one_governed_product(monkeypatch) -> None:
|
|
project_id = uuid4()
|
|
dataset_ids = [uuid4(), uuid4()]
|
|
db = DatasetSession([make_dataset(project_id, dataset_id) for dataset_id in dataset_ids])
|
|
captured = {}
|
|
|
|
def select_features(_db, **kwargs):
|
|
captured.update(kwargs)
|
|
return {
|
|
"selection_bbox": kwargs["bbox"],
|
|
"feature_count": 1,
|
|
"total_feature_count": 3,
|
|
"limit": kwargs["limit"],
|
|
"truncated": False,
|
|
"geojson": {"type": "FeatureCollection", "features": []},
|
|
"summary": None,
|
|
}
|
|
|
|
monkeypatch.setattr(VectorFeatureService, "select_features_by_bbox", select_features)
|
|
payload = VectorPartitionSelectionRequest(
|
|
dataset_ids=dataset_ids,
|
|
bbox={"min_x": 5.0, "min_y": 51.0, "max_x": 5.3, "max_y": 51.2},
|
|
)
|
|
response = select_vector_partitions(project_id, payload, db)
|
|
|
|
assert captured["dataset_ids"] == dataset_ids
|
|
assert captured["deduplicate_source_features"] is True
|
|
assert response["data"]["partition_count"] == 2
|
|
assert response["data"]["dataset_ids"] == dataset_ids
|
|
|
|
|
|
def test_vector_partition_request_has_a_bounded_fan_out() -> None:
|
|
with pytest.raises(ValidationError):
|
|
VectorPartitionSelectionRequest(
|
|
dataset_ids=[uuid4() for _ in range(17)],
|
|
bbox={"min_x": 5.0, "min_y": 51.0, "max_x": 5.3, "max_y": 51.2},
|
|
)
|
|
|
|
|
|
def test_vector_partition_route_rejects_mixed_source_products() -> None:
|
|
project_id = uuid4()
|
|
datasets = [
|
|
make_dataset(project_id, uuid4(), source_name="grb", product_key="buildings"),
|
|
make_dataset(project_id, uuid4(), source_name="spw_picc", product_key="picc_buildings"),
|
|
]
|
|
payload = VectorPartitionSelectionRequest(
|
|
dataset_ids=[dataset.id for dataset in datasets],
|
|
bbox={"min_x": 5.0, "min_y": 51.0, "max_x": 5.3, "max_y": 51.2},
|
|
)
|
|
with pytest.raises(AppError) as exc_info:
|
|
select_vector_partitions(project_id, payload, DatasetSession(datasets))
|
|
assert getattr(exc_info.value, "code", None) == "VECTOR_PARTITION_SOURCE_MISMATCH"
|