178 lines
7.0 KiB
Python
178 lines
7.0 KiB
Python
import json
|
|
|
|
import pytest
|
|
|
|
from app.core.datasource_defaults import DEFAULT_DATASOURCES
|
|
from app.services.collectors.registry import collector_registry
|
|
from app.services import earth_boundaries
|
|
|
|
|
|
def write_geojson(path, name="Test"):
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"type": "FeatureCollection",
|
|
"features": [
|
|
{
|
|
"type": "Feature",
|
|
"properties": {"name": name},
|
|
"geometry": {
|
|
"type": "Polygon",
|
|
"coordinates": [[[-1, -1], [1, -1], [1, 1], [-1, 1], [-1, -1]]],
|
|
},
|
|
}
|
|
],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def patch_paths(monkeypatch, tmp_path):
|
|
repo = tmp_path
|
|
source_dir = repo / "data/earth-boundary-sources"
|
|
boundary_dir = repo / "frontend/public/earth/data/boundaries/v1"
|
|
pmtiles = repo / "frontend/public/earth/data/boundaries/earth-boundaries-china-pov-v1.pmtiles"
|
|
legacy = repo / "frontend/public/earth/data/countries-admin0.min.geojson"
|
|
config = repo / "config/earth-boundary-sources.local.json"
|
|
example = repo / "config/earth-boundary-sources.example.json"
|
|
policy = repo / "config/earth-boundary-pov-policy.china-v1.json"
|
|
for path in (source_dir, boundary_dir, pmtiles.parent, legacy.parent, config.parent):
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
policy.write_text('{"productionTileFormat":"pmtiles+mvt"}\n', encoding="utf-8")
|
|
example.write_text('{"collectorConfigs":{}}\n', encoding="utf-8")
|
|
monkeypatch.setattr(earth_boundaries, "REPO_ROOT", repo)
|
|
monkeypatch.setattr(earth_boundaries, "SOURCE_OUTPUT_DIR", source_dir)
|
|
monkeypatch.setattr(earth_boundaries, "SOURCE_MANIFEST_PATH", source_dir / "manifest.json")
|
|
monkeypatch.setattr(earth_boundaries, "BUILD_RESULT_PATH", source_dir / "build-result.json")
|
|
monkeypatch.setattr(earth_boundaries, "BUILD_JOB_PATH", source_dir / "build-job.json")
|
|
monkeypatch.setattr(earth_boundaries, "BOUNDARY_OUTPUT_DIR", boundary_dir)
|
|
monkeypatch.setattr(earth_boundaries, "BOUNDARY_MANIFEST_PATH", boundary_dir / "manifest.json")
|
|
monkeypatch.setattr(earth_boundaries, "PMTILES_ARTIFACT_PATH", pmtiles)
|
|
monkeypatch.setattr(earth_boundaries, "LEGACY_GEOJSON_PATH", legacy)
|
|
monkeypatch.setattr(earth_boundaries, "LOCAL_CONFIG_PATH", config)
|
|
monkeypatch.setattr(earth_boundaries, "EXAMPLE_CONFIG_PATH", example)
|
|
monkeypatch.setattr(earth_boundaries, "POV_POLICY_PATH", policy)
|
|
return {
|
|
"repo": repo,
|
|
"config": config,
|
|
"legacy": legacy,
|
|
"pmtiles": pmtiles,
|
|
"manifest": boundary_dir / "manifest.json",
|
|
}
|
|
|
|
|
|
def test_boundary_status_uses_legacy_provider_when_pmtiles_missing(monkeypatch, tmp_path):
|
|
paths = patch_paths(monkeypatch, tmp_path)
|
|
write_geojson(paths["legacy"])
|
|
|
|
status = earth_boundaries.get_boundary_status()
|
|
|
|
assert status["provider"] == "legacy-geojson"
|
|
assert status["fallback_available"] is True
|
|
assert status["high_precision_ready"] is False
|
|
|
|
|
|
def test_boundary_status_prefers_high_precision_when_manifest_and_pmtiles_exist(monkeypatch, tmp_path):
|
|
paths = patch_paths(monkeypatch, tmp_path)
|
|
write_geojson(paths["legacy"])
|
|
paths["pmtiles"].write_bytes(b"pmtiles")
|
|
paths["manifest"].write_text('{"tileProvider":"pmtiles-mvt"}\n', encoding="utf-8")
|
|
|
|
status = earth_boundaries.get_boundary_status()
|
|
|
|
assert status["provider"] == "pmtiles-mvt"
|
|
assert status["high_precision_ready"] is True
|
|
|
|
|
|
def test_save_boundary_config_writes_local_config(monkeypatch, tmp_path):
|
|
paths = patch_paths(monkeypatch, tmp_path)
|
|
payload = {"collectorConfigs": {"earth_admin0_boundaries": {"endpoint": "file:///tmp/a.geojson"}}}
|
|
|
|
status = earth_boundaries.save_boundary_config(payload)
|
|
|
|
assert paths["config"].exists()
|
|
assert status["config_source"] == "local"
|
|
assert status["config"] == payload
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_reports_missing_tools_after_source_artifacts(monkeypatch, tmp_path):
|
|
paths = patch_paths(monkeypatch, tmp_path)
|
|
source_files = {}
|
|
for source in earth_boundaries.BOUNDARY_SOURCE_KINDS:
|
|
source_path = paths["repo"] / f"{source}.geojson"
|
|
write_geojson(source_path, name=source)
|
|
source_files[source] = source_path
|
|
paths["config"].write_text(
|
|
json.dumps(
|
|
{
|
|
"collectorConfigs": {
|
|
source: {
|
|
"sourceKind": kind,
|
|
"endpoint": str(source_files[source]),
|
|
"method": "GET",
|
|
}
|
|
for source, kind in earth_boundaries.BOUNDARY_SOURCE_KINDS.items()
|
|
}
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setattr(earth_boundaries.shutil, "which", lambda _tool: None)
|
|
progress_events = []
|
|
|
|
status = await earth_boundaries.build_boundary_assets(
|
|
lambda progress, phase, message, **_extra: progress_events.append((progress, phase, message))
|
|
)
|
|
|
|
assert status["provider"] == "geojson-high-precision"
|
|
assert status["high_precision_ready"] is True
|
|
assert (paths["repo"] / "data/earth-boundary-sources/manifest.json").exists()
|
|
assert paths["manifest"].exists()
|
|
assert any(phase == "download" for _progress, phase, _message in progress_events)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_boundary_build_job_records_geojson_fallback_success(monkeypatch, tmp_path):
|
|
paths = patch_paths(monkeypatch, tmp_path)
|
|
monkeypatch.setattr(earth_boundaries, "_build_task", None)
|
|
monkeypatch.setattr(earth_boundaries, "_build_job_state", {})
|
|
source_files = {}
|
|
for source in earth_boundaries.BOUNDARY_SOURCE_KINDS:
|
|
source_path = paths["repo"] / f"{source}.geojson"
|
|
write_geojson(source_path, name=source)
|
|
source_files[source] = source_path
|
|
paths["config"].write_text(
|
|
json.dumps(
|
|
{
|
|
"collectorConfigs": {
|
|
source: {
|
|
"sourceKind": kind,
|
|
"endpoint": str(source_files[source]),
|
|
"method": "GET",
|
|
}
|
|
for source, kind in earth_boundaries.BOUNDARY_SOURCE_KINDS.items()
|
|
}
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setattr(earth_boundaries.shutil, "which", lambda _tool: None)
|
|
|
|
response = await earth_boundaries.start_boundary_build_job()
|
|
await earth_boundaries._build_task
|
|
status = earth_boundaries.get_boundary_build_status()
|
|
|
|
assert response["accepted"] is True
|
|
assert status["job"]["status"] == "succeeded"
|
|
assert status["job"]["result"]["provider"] == "geojson-high-precision"
|
|
|
|
|
|
def test_earth_boundary_collectors_are_not_registered_as_datasources():
|
|
removed = set(earth_boundaries.BOUNDARY_SOURCE_KINDS) | {"earth_boundary_tiles"}
|
|
|
|
assert removed.isdisjoint(DEFAULT_DATASOURCES)
|
|
for source in removed:
|
|
assert collector_registry.get(source) is None
|