147 lines
5.5 KiB
Python
147 lines
5.5 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import zipfile
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SCRIPTS = ROOT / "scripts"
|
|
if str(SCRIPTS) not in sys.path:
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
|
|
|
|
def load_script(name: str):
|
|
path = SCRIPTS / name
|
|
module_name = f"test_{path.stem}"
|
|
spec = importlib.util.spec_from_file_location(module_name, path)
|
|
assert spec is not None
|
|
assert spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[module_name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def population_archive() -> bytes:
|
|
text = "\n".join(
|
|
(
|
|
"CD_REFNIS|CD_SECTOR|TOTAL|TX_DESCR_SECTOR_NL|TX_DESCR_NL",
|
|
"13025|13025A00-|120|Mol centrum|Mol",
|
|
"13008|13008A00-|240|Geel centrum|Geel",
|
|
"11002|11002A00-|360|Antwerpen centrum|Antwerpen",
|
|
)
|
|
)
|
|
buffer = io.BytesIO()
|
|
with zipfile.ZipFile(buffer, "w") as archive:
|
|
archive.writestr("population.csv", text)
|
|
return buffer.getvalue()
|
|
|
|
|
|
def test_population_operator_filters_to_the_approved_scope() -> None:
|
|
module = load_script("provision_mol_population_history.py")
|
|
regional = module.GEOGRAPHIC_SCOPES["kempen-transport-region"]
|
|
mol = module.GEOGRAPHIC_SCOPES["mol"]
|
|
|
|
regional_rows = module.population_rows(population_archive(), regional)
|
|
mol_rows = module.population_rows(population_archive(), mol)
|
|
|
|
assert set(regional_rows) == {"13025A00-", "13008A00-"}
|
|
assert regional_rows["13008A00-"]["municipality"] == "Geel"
|
|
assert regional_rows["13008A00-"]["nis_code"] == "13008"
|
|
assert set(mol_rows) == {"13025A00-"}
|
|
assert module.series_key(regional) == "statbel:population-statistical-sector:kempen-transport-region"
|
|
assert module.series_key(mol) == "statbel:population-statistical-sector:mol"
|
|
|
|
|
|
def test_population_operator_resolves_the_persisted_scope_boundary(tmp_path: Path) -> None:
|
|
module = load_script("provision_mol_population_history.py")
|
|
scope = module.GEOGRAPHIC_SCOPES["kempen-transport-region"]
|
|
scope_dir = tmp_path / scope.key
|
|
scope_dir.mkdir(parents=True)
|
|
boundary = scope_dir / "boundary.geojson"
|
|
boundary.write_text('{"type":"FeatureCollection","features":[]}', encoding="utf-8")
|
|
manifest = scope_dir / "kempen_transport_region_scope_manifest.json"
|
|
manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"status": "complete",
|
|
"scope_key": scope.key,
|
|
"boundary_filename": boundary.name,
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
args = argparse.Namespace(boundary_path=None, scope_output_root=tmp_path)
|
|
|
|
assert module.resolve_boundary_path(args, scope) == boundary
|
|
|
|
|
|
def test_regional_coordinator_builds_explicit_population_and_forest_commands(tmp_path: Path) -> None:
|
|
module = load_script("provision_regional_timeseries.py")
|
|
scope = module.GEOGRAPHIC_SCOPES["kempen-transport-region"]
|
|
args = argparse.Namespace(
|
|
output_root=tmp_path / "time-series",
|
|
fetch_only=False,
|
|
force=False,
|
|
skip_population=False,
|
|
skip_landuse=False,
|
|
base_url="http://backend:8000",
|
|
population_years="2021,2025",
|
|
landuse_years="2013,2025",
|
|
request_timeout=300,
|
|
import_timeout=3600,
|
|
max_landuse_features=500000,
|
|
)
|
|
|
|
commands = dict(module.build_operator_commands(args, scope, tmp_path / "boundary.geojson"))
|
|
|
|
assert set(commands) == {"population", "forest"}
|
|
assert commands["population"][0] == sys.executable
|
|
assert "--scope" in commands["population"]
|
|
assert "kempen-transport-region" in commands["population"]
|
|
assert scope.project_name in commands["population"]
|
|
assert commands["forest"][0] == sys.executable
|
|
assert "--max-features" in commands["forest"]
|
|
assert ",".join(scope.nis_codes) in commands["forest"]
|
|
assert "--force" not in commands["population"]
|
|
assert "--fetch-only" not in commands["forest"]
|
|
|
|
|
|
def test_regional_forest_provenance_does_not_claim_one_municipality() -> None:
|
|
module = load_script("provision_official_landuse_timeseries.py")
|
|
|
|
regional = module.scope_identity("Kempen (28 gemeenten)", "13001,13008,13025")
|
|
municipal = module.scope_identity("Mol", "13025")
|
|
|
|
assert regional == {
|
|
"scope_display_name": "Kempen (28 gemeenten)",
|
|
"member_nis_codes": ["13001", "13008", "13025"],
|
|
"municipality": None,
|
|
"nis_code": None,
|
|
}
|
|
assert municipal["municipality"] == "Mol"
|
|
assert municipal["nis_code"] == "13025"
|
|
|
|
|
|
def test_regional_timeseries_operator_is_packaged_and_release_checked() -> 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")
|
|
|
|
assert "COPY scripts/provision_regional_timeseries.py" in dockerfile
|
|
assert "py_compile scripts/provision_regional_timeseries.py" in readiness
|
|
|
|
|
|
def test_end_user_dataset_sources_are_human_readable() -> None:
|
|
display = (ROOT / "frontend/src/lib/datasetDisplay.ts").read_text(encoding="utf-8")
|
|
workspace = (ROOT / "frontend/src/components/map/MapWorkspace.tsx").read_text(encoding="utf-8")
|
|
|
|
assert "department_omgeving_land_use: 'Departement Omgeving'" in display
|
|
assert "statbel: 'Statbel'" in display
|
|
assert "getDatasetSourceDisplayName(activeThemeDataset)" in workspace
|
|
assert "dataset ? getDatasetSourceDisplayName(dataset)" in workspace
|