from __future__ import annotations
|
|
import json
|
import sys
|
import tempfile
|
import unittest
|
from pathlib import Path
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
from generate_validation_inputs import write_validation_inputs # noqa: E402
|
from run_risk_rule_engine import run # noqa: E402
|
|
|
class RiskRuleEngineTests(unittest.TestCase):
|
def run_case(self, root: Path, name: str) -> dict:
|
source = root / "inputs" / name
|
return run(source / "observations.geojson", source / "zones.geojson", source / "rules.json", root / f"output-{name}", 10)
|
|
def test_normal_case_has_expected_scores_and_artifacts(self):
|
with tempfile.TemporaryDirectory() as directory:
|
root = Path(directory)
|
write_validation_inputs(root / "inputs")
|
metadata = self.run_case(root, "normal")
|
payload = json.loads((root / "output-normal" / "risk_scores.geojson").read_text(encoding="utf-8"))
|
scores = {feature["properties"]["object_id"]: feature["properties"]["risk_score"] for feature in payload["features"]}
|
self.assertEqual(scores, {"normal-outside": 0.0, "normal-caution": 25.0, "normal-restricted": 90.0})
|
self.assertEqual(metadata["summary"]["risk_level_counts"], {"low": 2, "medium": 0, "high": 0, "critical": 1})
|
for name in ("risk_score.tif", "risk_preview.png", "risk_scores.geojson", "risk_scores.csv", "risk_summary.json", "run_metadata.json"):
|
self.assertTrue((root / "output-normal" / name).is_file())
|
|
def test_difficult_case_captures_boundary_overlap_and_score_cap(self):
|
with tempfile.TemporaryDirectory() as directory:
|
root = Path(directory)
|
write_validation_inputs(root / "inputs")
|
self.run_case(root, "difficult")
|
payload = json.loads((root / "output-difficult" / "risk_scores.geojson").read_text(encoding="utf-8"))
|
scores = {feature["properties"]["object_id"]: feature["properties"] for feature in payload["features"]}
|
self.assertEqual(scores["boundary-overlap"]["risk_score"], 90.0)
|
self.assertEqual(set(scores["boundary-overlap"]["hit_rule_ids"]), {"restricted_zone", "caution_zone"})
|
self.assertEqual(scores["low-confidence-restricted"]["risk_score"], 100.0)
|
self.assertTrue(scores["low-confidence-restricted"]["score_capped"])
|
self.assertEqual(scores["low-confidence-outside"]["risk_score"], 20.0)
|
|
def test_rejects_missing_crs(self):
|
with tempfile.TemporaryDirectory() as directory:
|
root = Path(directory)
|
write_validation_inputs(root / "inputs")
|
observations = root / "inputs" / "normal" / "observations.geojson"
|
payload = json.loads(observations.read_text(encoding="utf-8"))
|
payload.pop("crs", None)
|
observations.write_text(json.dumps(payload), encoding="utf-8")
|
with self.assertRaisesRegex(ValueError, "CRS"):
|
self.run_case(root, "normal")
|
|
|
if __name__ == "__main__":
|
unittest.main()
|
