| 1 | """Near-declarative scenario loader: real spec round-trip plus validation."""
|
| 2 |
|
| 3 | from __future__ import annotations
|
| 4 |
|
| 5 | import sys
|
| 6 | import tempfile
|
| 7 | import unittest
|
| 8 | from pathlib import Path
|
| 9 |
|
| 10 | ROOT = Path(__file__).resolve().parents[1]
|
| 11 | sys.path.insert(0, str(ROOT / "scripts"))
|
| 12 |
|
| 13 | from oakbench.scenario_spec import ScenarioSpec, load_scenario_specs
|
| 14 |
|
| 15 |
|
| 16 | class PlatformSpecTest(unittest.TestCase):
|
| 17 | """The existing platform.yaml loads unchanged (parallel read, no migration)."""
|
| 18 |
|
| 19 | def test_loads_platform_yaml(self) -> None:
|
| 20 | specs = load_scenario_specs(ROOT / "scenarios" / "platform.yaml")
|
| 21 | self.assertEqual(
|
| 22 | [spec.id for spec in specs],
|
| 23 | [
|
| 24 | "platform_capability_probe",
|
| 25 | "pr_single_anatomy",
|
| 26 | "poll_until_merged_i10s",
|
| 27 | "integration_race_n1",
|
| 28 | "integration_race_n10",
|
| 29 | "branch_triage_n4",
|
| 30 | "branch_fleet_n10",
|
| 31 | "branch_fleet_n100",
|
| 32 | "branch_fleet_n1000",
|
| 33 | ],
|
| 34 | )
|
| 35 | by_id = {spec.id: spec for spec in specs}
|
| 36 | self.assertEqual(by_id["platform_capability_probe"].operations, ("capability.probe",))
|
| 37 | self.assertEqual(
|
| 38 | by_id["pr_single_anatomy"].operations,
|
| 39 | (
|
| 40 | "branch.publish",
|
| 41 | "integration.open",
|
| 42 | "integration.visible.settle",
|
| 43 | "integration.mergeable.settle",
|
| 44 | "integration.merge.request",
|
| 45 | "integration.main.settle",
|
| 46 | "integration.sync.local",
|
| 47 | "integration.total",
|
| 48 | ),
|
| 49 | )
|
| 50 | for spec in specs:
|
| 51 | self.assertIsInstance(spec, ScenarioSpec)
|
| 52 | self.assertIsNone(spec.fixture)
|
| 53 | self.assertEqual(spec.runtime_class, "standard")
|
| 54 | self.assertEqual(spec.checks, {})
|
| 55 | # Lane-specific keys survive as extras instead of being dropped.
|
| 56 | self.assertEqual(by_id["integration_race_n10"].extras.get("branch_count"), 10)
|
| 57 | self.assertEqual(by_id["branch_triage_n4"].extras.get("branch_count"), 4)
|
| 58 | self.assertEqual(by_id["branch_fleet_n10"].extras.get("branch_count"), 10)
|
| 59 | self.assertEqual(
|
| 60 | by_id["branch_triage_n4"].operations,
|
| 61 | (
|
| 62 | "triage.seed",
|
| 63 | "triage.review",
|
| 64 | "triage.plan",
|
| 65 | "triage.apply",
|
| 66 | "triage.oracle",
|
| 67 | "triage.total",
|
| 68 | ),
|
| 69 | )
|
| 70 | self.assertEqual(
|
| 71 | by_id["branch_fleet_n10"].operations,
|
| 72 | (
|
| 73 | "fleet.seed",
|
| 74 | "fleet.classify",
|
| 75 | "fleet.plan",
|
| 76 | "fleet.apply",
|
| 77 | "fleet.sync",
|
| 78 | "fleet.oracle",
|
| 79 | "fleet.total",
|
| 80 | ),
|
| 81 | )
|
| 82 |
|
| 83 |
|
| 84 | class ValidationTest(unittest.TestCase):
|
| 85 | def _load(self, text: str) -> list[ScenarioSpec]:
|
| 86 | with tempfile.TemporaryDirectory() as tmp:
|
| 87 | path = Path(tmp) / "spec.yaml"
|
| 88 | path.write_text(text)
|
| 89 | return load_scenario_specs(path)
|
| 90 |
|
| 91 | def test_missing_id_raises_with_path_context(self) -> None:
|
| 92 | with self.assertRaises(ValueError) as ctx:
|
| 93 | self._load("scenarios:\n - operations:\n - status.full\n")
|
| 94 | message = str(ctx.exception)
|
| 95 | self.assertIn("spec.yaml", message)
|
| 96 | self.assertIn("id", message)
|
| 97 |
|
| 98 | def test_unknown_fixture_ref_raises_with_context(self) -> None:
|
| 99 | with self.assertRaises(ValueError) as ctx:
|
| 100 | self._load(
|
| 101 | "scenarios:\n"
|
| 102 | " - name: status_on_phantom\n"
|
| 103 | " fixture: no-such-fixture-v9\n"
|
| 104 | " operations:\n"
|
| 105 | " - status.full\n"
|
| 106 | )
|
| 107 | message = str(ctx.exception)
|
| 108 | self.assertIn("spec.yaml", message)
|
| 109 | self.assertIn("status_on_phantom", message)
|
| 110 | self.assertIn("no-such-fixture-v9", message)
|
| 111 |
|
| 112 | def test_known_fixture_ref_and_full_shape(self) -> None:
|
| 113 | specs = self._load(
|
| 114 | "scenarios:\n"
|
| 115 | " - id: status_large\n"
|
| 116 | " description: >-\n"
|
| 117 | " Folded descriptions are already used by checked-in scenario\n"
|
| 118 | " files and must not drop # literal body content.\n"
|
| 119 | " fixture: large-mixed-v1\n"
|
| 120 | " runtime_class: micro\n"
|
| 121 | " operations:\n"
|
| 122 | " - status.full\n"
|
| 123 | " modes:\n"
|
| 124 | " - shared_checkout\n"
|
| 125 | " checks:\n"
|
| 126 | " tree_untouched: true\n"
|
| 127 | )
|
| 128 | self.assertEqual(len(specs), 1)
|
| 129 | spec = specs[0]
|
| 130 | self.assertEqual(spec.id, "status_large")
|
| 131 | self.assertEqual(spec.fixture, "large-mixed-v1")
|
| 132 | self.assertEqual(spec.runtime_class, "micro")
|
| 133 | self.assertEqual(spec.modes, ("shared_checkout",))
|
| 134 | self.assertEqual(spec.checks, {"tree_untouched": True})
|
| 135 | self.assertEqual(
|
| 136 | spec.extras["description"],
|
| 137 | "Folded descriptions are already used by checked-in scenario files and must not drop # literal body content.",
|
| 138 | )
|
| 139 |
|
| 140 | def test_missing_operations_raises(self) -> None:
|
| 141 | with self.assertRaises(ValueError) as ctx:
|
| 142 | self._load("scenarios:\n - name: empty_handed\n")
|
| 143 | self.assertIn("operations", str(ctx.exception))
|
| 144 |
|
| 145 | def test_bad_runtime_class_raises(self) -> None:
|
| 146 | with self.assertRaises(ValueError) as ctx:
|
| 147 | self._load(
|
| 148 | "scenarios:\n"
|
| 149 | " - name: warp_speed\n"
|
| 150 | " runtime_class: instant\n"
|
| 151 | " operations:\n"
|
| 152 | " - status.full\n"
|
| 153 | )
|
| 154 | self.assertIn("runtime_class", str(ctx.exception))
|
| 155 |
|
| 156 | def test_duplicate_ids_raise(self) -> None:
|
| 157 | with self.assertRaises(ValueError) as ctx:
|
| 158 | self._load(
|
| 159 | "scenarios:\n"
|
| 160 | " - name: twin\n"
|
| 161 | " operations:\n"
|
| 162 | " - status.full\n"
|
| 163 | " - name: twin\n"
|
| 164 | " operations:\n"
|
| 165 | " - diff.stat\n"
|
| 166 | )
|
| 167 | self.assertIn("duplicate", str(ctx.exception))
|
| 168 |
|
| 169 |
|
| 170 | if __name__ == "__main__":
|
| 171 | unittest.main()
|