114 lines
5.2 KiB
Python
114 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.machinery
|
|
import importlib.util
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
def load_rpctl():
|
|
path = Path(__file__).with_name("rpctl")
|
|
loader = importlib.machinery.SourceFileLoader("rpctl_module", str(path))
|
|
spec = importlib.util.spec_from_loader(loader.name, loader)
|
|
module = importlib.util.module_from_spec(spec)
|
|
loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class RpkiProverProfileTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.rpctl = load_rpctl()
|
|
|
|
def test_profile_aliases_and_algorithms(self) -> None:
|
|
self.assertEqual(self.rpctl.normalize_profile("rpki-prover"), "rpki-prover-incremental")
|
|
self.assertEqual(self.rpctl.normalize_profile("rpki-prover-no-incremental"), "rpki-prover-full")
|
|
self.assertNotIn("--no-incremental-validation", self.rpctl.rpki_prover_arguments("rpki-prover-incremental"))
|
|
self.assertIn("--no-incremental-validation", self.rpctl.rpki_prover_arguments("rpki-prover-full"))
|
|
incremental = self.rpctl.parse_command_summary(
|
|
f"docker run {self.rpctl.RPKI_PROVER_IMAGE_TAG} --once"
|
|
)
|
|
full = self.rpctl.parse_command_summary(
|
|
f"docker run {self.rpctl.RPKI_PROVER_IMAGE_TAG} --once --no-incremental-validation"
|
|
)
|
|
self.assertEqual(incremental, "algorithm=incremental")
|
|
self.assertEqual(full, "algorithm=full-every-iteration")
|
|
|
|
def test_vrp_count_preserves_raw_and_unique(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
csv_path = Path(temp) / "vrps.csv"
|
|
csv_path.write_text(
|
|
"ASN,IP Prefix,Max Length,Trust Anchor\n"
|
|
"AS64496,192.0.2.0/24,24,TA-A\n"
|
|
"AS64496,192.0.2.0/24,24,TA-B\n"
|
|
"AS64497,2001:db8::/32,48,TA-A\n",
|
|
encoding="utf-8",
|
|
)
|
|
raw, unique = self.rpctl.collect_rpki_prover_vrps(csv_path)
|
|
self.assertEqual(raw, 3)
|
|
self.assertEqual(len(unique), 2)
|
|
|
|
def test_run_metadata_marks_vaps_not_applicable(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp)
|
|
old_ctl_root = self.rpctl.ctl_root
|
|
self.rpctl.ctl_root = lambda: root
|
|
try:
|
|
self.rpctl.save_config({"rpkiProverImage": {"tag": self.rpctl.RPKI_PROVER_IMAGE_TAG}})
|
|
run_root = root / "runs" / "rpki-prover-incremental" / "test"
|
|
run_dir = run_root / "profiles" / "rpki-prover-incremental" / "runs" / "run_0001"
|
|
run_dir.mkdir(parents=True)
|
|
(run_dir / "start-utc.txt").write_text("2026-07-17T00:00:00Z\n", encoding="utf-8")
|
|
(run_dir / "end-utc.txt").write_text("2026-07-17T00:01:00Z\n", encoding="utf-8")
|
|
(run_dir / "vrps.csv").write_text(
|
|
"ASN,IP Prefix,Max Length,Trust Anchor\nAS64496,192.0.2.0/24,24,TA-A\n",
|
|
encoding="utf-8",
|
|
)
|
|
meta = self.rpctl.write_rpki_prover_run_metadata(
|
|
run_root, "rpki-prover-incremental", 1, "snapshot", 60_000, 1234, 1200, 0
|
|
)
|
|
self.assertEqual(meta["counts"]["vrpsRaw"], 1)
|
|
self.assertEqual(meta["counts"]["vrpsUnique"], 1)
|
|
self.assertEqual(meta["counts"]["vaps"], "n/a")
|
|
self.assertFalse(meta["vapsAvailable"])
|
|
self.rpctl.write_rpki_prover_group_report(run_root, "rpki-prover-incremental")
|
|
report = (run_root / "reports" / "per_run_metrics.csv").read_text(encoding="utf-8")
|
|
self.assertIn("vrps_unique", report)
|
|
self.assertIn("n/a", report)
|
|
finally:
|
|
self.rpctl.ctl_root = old_ctl_root
|
|
|
|
def test_missing_image_configuration_fails_before_docker(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp)
|
|
old_ctl_root = self.rpctl.ctl_root
|
|
self.rpctl.ctl_root = lambda: root
|
|
try:
|
|
self.rpctl.save_config({})
|
|
with self.assertRaisesRegex(SystemExit, "not initialized"):
|
|
self.rpctl.require_rpki_prover_image()
|
|
finally:
|
|
self.rpctl.ctl_root = old_ctl_root
|
|
|
|
def test_versions_include_prover_profiles(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp)
|
|
old_ctl_root = self.rpctl.ctl_root
|
|
self.rpctl.ctl_root = lambda: root
|
|
try:
|
|
(root / "versions.json").write_text(json.dumps({"profiles": ["fort-latest-release"]}), encoding="utf-8")
|
|
self.rpctl.save_config({"rpkiProverImage": {"tag": self.rpctl.RPKI_PROVER_IMAGE_TAG}})
|
|
run_root = root / "experiment"
|
|
self.rpctl.write_experiment_versions(run_root)
|
|
versions = json.loads((run_root / "versions.json").read_text(encoding="utf-8"))
|
|
self.assertIn("rpki-prover-incremental", versions["profiles"])
|
|
self.assertIn("rpki-prover-full", versions["profiles"])
|
|
finally:
|
|
self.rpctl.ctl_root = old_ctl_root
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|