231 lines
9.2 KiB
Python
231 lines
9.2 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from shapely.geometry import Polygon
|
|
from shapely.ops import unary_union
|
|
|
|
from app.core.errors import AppError
|
|
from app.services.storage_service import StorageService
|
|
from app.services.vector_feature_service import VectorFeatureService
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SCRIPTS = ROOT / "scripts"
|
|
|
|
|
|
def load_operator():
|
|
scripts_path = str(SCRIPTS)
|
|
if scripts_path not in sys.path:
|
|
sys.path.insert(0, scripts_path)
|
|
spec = importlib.util.spec_from_file_location(
|
|
"provision_regional_grb_buildings_test",
|
|
SCRIPTS / "provision_regional_grb_buildings.py",
|
|
)
|
|
assert spec is not None
|
|
assert spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def feature(feature_id: str, polygon: Polygon) -> dict:
|
|
return {
|
|
"type": "Feature",
|
|
"id": feature_id,
|
|
"geometry": polygon.__geo_interface__,
|
|
"properties": {"UIDN": feature_id},
|
|
}
|
|
|
|
|
|
def test_partition_assignment_is_deterministic_and_has_no_cross_member_duplicates() -> None:
|
|
operator = load_operator()
|
|
scopes = importlib.import_module("geographic_scopes")
|
|
alpha = scopes.ScopeMember("Alpha", "10001")
|
|
beta = scopes.ScopeMember("Beta", "10002")
|
|
scope = scopes.GeographicScope(
|
|
key="test-region",
|
|
display_name="Test region",
|
|
project_name="Test project",
|
|
project_region="Test",
|
|
area_name="Test operation boundary",
|
|
authority_name="Test authority",
|
|
authority_url="https://example.test/scope",
|
|
scope_type="policy_region",
|
|
limitation_message="Test limitation.",
|
|
members=(alpha, beta),
|
|
)
|
|
alpha_boundary = Polygon([(0, 0), (1, 0), (1, 1), (0, 1)])
|
|
beta_boundary = Polygon([(1, 0), (2, 0), (2, 1), (1, 1)])
|
|
members = [(alpha, alpha_boundary), (beta, beta_boundary)]
|
|
region = unary_union([alpha_boundary, beta_boundary])
|
|
alpha_building = feature("GBG.alpha", Polygon([(0.1, 0.1), (0.2, 0.1), (0.2, 0.2), (0.1, 0.2)]))
|
|
beta_building = feature("GBG.beta", Polygon([(1.2, 0.1), (1.3, 0.1), (1.3, 0.2), (1.2, 0.2)]))
|
|
crossing = feature("GBG.crossing", Polygon([(0.8, 0.3), (1.1, 0.3), (1.1, 0.5), (0.8, 0.5)]))
|
|
page = ({"type": "FeatureCollection", "features": [alpha_building, beta_building, crossing]}, "https://example.test/grb")
|
|
|
|
alpha_features, alpha_summary = operator.build_partition_features(
|
|
[page], member=alpha, members=members, regional_boundary=region, scope=scope, max_features=10
|
|
)
|
|
beta_features, beta_summary = operator.build_partition_features(
|
|
[page], member=beta, members=members, regional_boundary=region, scope=scope, max_features=10
|
|
)
|
|
|
|
assert {item["id"] for item in alpha_features} == {"GBG.alpha", "GBG.crossing"}
|
|
assert {item["id"] for item in beta_features} == {"GBG.beta"}
|
|
assert alpha_summary["reference_truncated"] is False
|
|
assert beta_summary["reference_truncated"] is False
|
|
assert alpha_features[1]["properties"]["partition_assignment"] == "maximum_boundary_intersection"
|
|
|
|
|
|
def test_interior_buildings_skip_regional_owner_scan(monkeypatch) -> None:
|
|
operator = load_operator()
|
|
scopes = importlib.import_module("geographic_scopes")
|
|
member = scopes.ScopeMember("Alpha", "10001")
|
|
boundary = Polygon([(0, 0), (1, 0), (1, 1), (0, 1)])
|
|
scope = scopes.GeographicScope(
|
|
key="single",
|
|
display_name="Single",
|
|
project_name="Single",
|
|
project_region="Single",
|
|
area_name="Single boundary",
|
|
authority_name="Test",
|
|
authority_url="https://example.test",
|
|
scope_type="municipality",
|
|
limitation_message="Test.",
|
|
members=(member,),
|
|
)
|
|
monkeypatch.setattr(
|
|
operator,
|
|
"assign_owner_nis",
|
|
lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("interior feature used slow owner scan")),
|
|
)
|
|
|
|
features, _ = operator.build_partition_features(
|
|
[({"type": "FeatureCollection", "features": [feature("GBG.inside", Polygon([(0.1, 0.1), (0.2, 0.1), (0.2, 0.2), (0.1, 0.2)]))]}, "https://example.test/grb")],
|
|
member=member,
|
|
members=[(member, boundary)],
|
|
regional_boundary=boundary,
|
|
scope=scope,
|
|
max_features=10,
|
|
)
|
|
|
|
assert [item["id"] for item in features] == ["GBG.inside"]
|
|
|
|
|
|
def test_combined_artifact_streams_partitions_and_rejects_duplicate_source_ids(tmp_path: Path) -> None:
|
|
operator = load_operator()
|
|
scope = importlib.import_module("geographic_scopes").KEMPEN_TRANSPORT_REGION_SCOPE
|
|
first = tmp_path / "first.geojson"
|
|
second = tmp_path / "second.geojson"
|
|
first.write_text(json.dumps({"type": "FeatureCollection", "features": [feature("GBG.1", Polygon([(4, 51), (4.01, 51), (4.01, 51.01), (4, 51.01)]))]}), encoding="utf-8")
|
|
second.write_text(json.dumps({"type": "FeatureCollection", "features": [feature("GBG.2", Polygon([(4.1, 51), (4.11, 51), (4.11, 51.01), (4.1, 51.01)]))]}), encoding="utf-8")
|
|
combined = tmp_path / "combined.geojson"
|
|
|
|
summary = operator.write_combined_artifact(
|
|
combined,
|
|
scope=scope,
|
|
observed_date=operator.date(2026, 7, 14),
|
|
partition_paths=[first, second],
|
|
expected_feature_count=2,
|
|
)
|
|
|
|
payload = json.loads(combined.read_text(encoding="utf-8"))
|
|
assert summary["feature_count"] == 2
|
|
assert summary["sha256"] == operator.sha256_file(combined)
|
|
assert [item["id"] for item in payload["features"]] == ["GBG.1", "GBG.2"]
|
|
|
|
second.write_text(json.dumps({"type": "FeatureCollection", "features": [feature("GBG.1", Polygon([(4.1, 51), (4.11, 51), (4.11, 51.01), (4.1, 51.01)]))]}), encoding="utf-8")
|
|
with pytest.raises(RuntimeError, match="Duplicate regional source feature"):
|
|
operator.write_combined_artifact(
|
|
combined,
|
|
scope=scope,
|
|
observed_date=operator.date(2026, 7, 14),
|
|
partition_paths=[first, second],
|
|
expected_feature_count=2,
|
|
)
|
|
|
|
|
|
class FakeDb:
|
|
def __init__(self) -> None:
|
|
self.rows = []
|
|
self.flush_count = 0
|
|
self.expunge_count = 0
|
|
|
|
def add(self, row) -> None:
|
|
self.rows.append(row)
|
|
|
|
def flush(self) -> None:
|
|
self.flush_count += 1
|
|
|
|
def expunge(self, row) -> None:
|
|
assert row in self.rows
|
|
self.expunge_count += 1
|
|
|
|
|
|
def test_partition_persistence_batches_rows_and_guards_source_identity(tmp_path: Path) -> None:
|
|
first = tmp_path / "one.geojson"
|
|
second = tmp_path / "two.geojson"
|
|
first.write_text(json.dumps({"type": "FeatureCollection", "features": [feature("GBG.1", Polygon([(4, 51), (4.01, 51), (4.01, 51.01), (4, 51.01)]))]}), encoding="utf-8")
|
|
second.write_text(json.dumps({"type": "FeatureCollection", "features": [feature("GBG.2", Polygon([(4.1, 51), (4.11, 51), (4.11, 51.01), (4.1, 51.01)]))]}), encoding="utf-8")
|
|
db = FakeDb()
|
|
|
|
persisted = VectorFeatureService.persist_geojson_partitions(
|
|
db,
|
|
uuid4(),
|
|
[first, second],
|
|
feature_class="buildings",
|
|
batch_size=1,
|
|
)
|
|
|
|
assert persisted == 2
|
|
assert db.flush_count == 2
|
|
assert db.expunge_count == 2
|
|
assert {row.source_feature_id for row in db.rows} == {"GBG.1", "GBG.2"}
|
|
|
|
second.write_text(first.read_text(encoding="utf-8"), encoding="utf-8")
|
|
with pytest.raises(AppError) as error:
|
|
VectorFeatureService.persist_geojson_partitions(FakeDb(), uuid4(), [first, second])
|
|
assert error.value.code == "DUPLICATE_SOURCE_FEATURE"
|
|
|
|
|
|
def test_storage_service_copies_large_artifacts_without_loading_them_as_upload_bytes(tmp_path: Path, monkeypatch) -> None:
|
|
source = tmp_path / "source.geojson"
|
|
source.write_bytes((b"0123456789abcdef" * 1024 * 1024) + b"tail")
|
|
storage_root = tmp_path / "storage"
|
|
monkeypatch.setattr(StorageService, "_base_dir", staticmethod(lambda: storage_root))
|
|
|
|
metadata = StorageService.persist_dataset_file_from_path(
|
|
"project",
|
|
"dataset",
|
|
"vector",
|
|
"regional.geojson",
|
|
source,
|
|
"application/geo+json",
|
|
)
|
|
|
|
stored = Path(metadata["storage_path"])
|
|
assert stored.read_bytes() == source.read_bytes()
|
|
assert metadata["size_bytes"] == source.stat().st_size
|
|
assert metadata["checksum_sha256"] == load_operator().sha256_file(source)
|
|
|
|
|
|
def test_regional_operator_is_packaged_documented_and_uses_service_boundaries() -> None:
|
|
dockerfile = (ROOT / "deploy/unraid/Dockerfile.all-in-one").read_text(encoding="utf-8")
|
|
readiness = (ROOT / "scripts/run_readiness_check.sh").read_text(encoding="utf-8")
|
|
operator = (SCRIPTS / "provision_regional_grb_buildings.py").read_text(encoding="utf-8")
|
|
dataset_service = (ROOT / "backend/app/services/dataset_service.py").read_text(encoding="utf-8")
|
|
|
|
assert "provision_regional_grb_buildings.py" in dockerfile
|
|
assert "py_compile scripts/provision_regional_grb_buildings.py" in readiness
|
|
assert "DatasetService.import_partitioned_vector_artifact" in operator
|
|
assert "VectorFeatureService.persist_geojson_partitions" in dataset_service
|
|
assert "insert into vector_features" not in operator.lower()
|
|
assert "db.add(VectorFeature" not in operator
|