| 1 | """integration_race_nN: fake-transport-only tests for the race runner.
|
| 2 |
|
| 3 | These tests never touch the network or GitHub: the transport is a scripted
|
| 4 | in-memory GitHub (refs, contents, PRs, statuses, merges), the budgeter runs
|
| 5 | on a frozen clock so pacing is deterministic and nonzero, and sleeps are
|
| 6 | recorded fakes so the runner never waits on real time. The no-credentials
|
| 7 | subprocess test exercises the skip-row honesty path (exit 3).
|
| 8 | """
|
| 9 |
|
| 10 | from __future__ import annotations
|
| 11 |
|
| 12 | import argparse
|
| 13 | import json
|
| 14 | import os
|
| 15 | import subprocess
|
| 16 | import sys
|
| 17 | import tempfile
|
| 18 | import unittest
|
| 19 | from pathlib import Path
|
| 20 |
|
| 21 | sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
| 22 |
|
| 23 | import platform_lifecycle as lane # noqa: E402
|
| 24 | from oakbench.platform_api import GitHubClient, RateLimitBudgeter # noqa: E402
|
| 25 | from oakbench.rows import validate_rows # noqa: E402
|
| 26 | from oakbench.runner import runner_fields, stamp_row # noqa: E402
|
| 27 |
|
| 28 | ROOT = Path(__file__).resolve().parents[1]
|
| 29 | SCRIPT = ROOT / "scripts" / "platform_lifecycle.py"
|
| 30 | ABSENT_TOKEN_ENV = "OAKBENCH_TEST_TOKEN_THAT_IS_ABSENT"
|
| 31 | REPO = "acme/race-disposable"
|
| 32 |
|
| 33 | RACE_BRANCH_OPERATIONS = [
|
| 34 | "branch.publish",
|
| 35 | "integration.open",
|
| 36 | "integration.visible.settle",
|
| 37 | "integration.mergeable.settle",
|
| 38 | "integration.checks.green.settle",
|
| 39 | "integration.merge.request",
|
| 40 | "integration.merged.settle",
|
| 41 | "integration.main.settle",
|
| 42 | ]
|
| 43 | RACE_LEVEL_OPERATIONS = [
|
| 44 | "race.publish.window",
|
| 45 | "race.integrate.window",
|
| 46 | "race.pacing.injected",
|
| 47 | "race.total",
|
| 48 | ]
|
| 49 | RACE_OPERATIONS = RACE_BRANCH_OPERATIONS + RACE_LEVEL_OPERATIONS
|
| 50 |
|
| 51 |
|
| 52 | def ok(payload: dict, status: int = 200) -> dict:
|
| 53 | return {
|
| 54 | "status": status,
|
| 55 | "headers": {"X-RateLimit-Remaining": "4999", "X-GitHub-Request-Id": "REQ"},
|
| 56 | "body": json.dumps(payload).encode("utf-8"),
|
| 57 | }
|
| 58 |
|
| 59 |
|
| 60 | class FakeGitHubTransport:
|
| 61 | """A scripted in-memory GitHub for one disposable repo.
|
| 62 |
|
| 63 | Simulates the real platform's async mergeability: the first GET of a PR
|
| 64 | (the visible poll) and the first mergeable poll both report mergeable
|
| 65 | null; mergeable resolves after one extra poll. Combined status turns
|
| 66 | green as soon as a success status is posted (check-instant resolves
|
| 67 | immediately under fake transport).
|
| 68 | """
|
| 69 |
|
| 70 | def __init__(self) -> None:
|
| 71 | self.requests: list[dict] = []
|
| 72 | self.main_sha = "main-sha-0"
|
| 73 | self.merged_count = 0
|
| 74 | self.branch_heads: dict[str, str] = {}
|
| 75 | self.prs: dict[int, dict] = {}
|
| 76 | self.statuses: dict[str, str] = {}
|
| 77 | self.next_pr_number = 101
|
| 78 |
|
| 79 | def __call__(self, request: dict) -> dict:
|
| 80 | self.requests.append(request)
|
| 81 | method = request["method"]
|
| 82 | path = request["url"].split("api.github.com", 1)[1]
|
| 83 | body = json.loads(request["body"].decode()) if request.get("body") else {}
|
| 84 | prefix = f"/repos/{REPO}"
|
| 85 | assert path.startswith(prefix), f"unexpected repo path: {path}"
|
| 86 | path = path[len(prefix):]
|
| 87 |
|
| 88 | if method == "GET" and path == "/git/ref/heads/main":
|
| 89 | return ok({"object": {"sha": self.main_sha}})
|
| 90 | if method == "POST" and path == "/git/refs":
|
| 91 | branch = body["ref"].split("refs/heads/", 1)[1]
|
| 92 | self.branch_heads[branch] = body["sha"]
|
| 93 | return ok({"ref": body["ref"], "object": {"sha": body["sha"]}}, status=201)
|
| 94 | if method == "PUT" and path.startswith("/contents/"):
|
| 95 | branch = body["branch"]
|
| 96 | head = f"head-{branch}"
|
| 97 | self.branch_heads[branch] = head
|
| 98 | if branch == "main":
|
| 99 | self.main_sha = head
|
| 100 | return ok({"commit": {"sha": head}}, status=201)
|
| 101 | if method == "POST" and path == "/pulls":
|
| 102 | number = self.next_pr_number
|
| 103 | self.next_pr_number += 1
|
| 104 | head_sha = self.branch_heads[body["head"]]
|
| 105 | self.prs[number] = {
|
| 106 | "head_sha": head_sha,
|
| 107 | "gets": 0,
|
| 108 | "merged": False,
|
| 109 | "conflicting": "conf" in body["head"],
|
| 110 | }
|
| 111 | return ok({"number": number, "head": {"sha": head_sha}}, status=201)
|
| 112 | if method == "GET" and path.startswith("/pulls/"):
|
| 113 | number = int(path.split("/")[2])
|
| 114 | pr = self.prs[number]
|
| 115 | pr["gets"] += 1
|
| 116 | # GET 1 is the visible poll; GET 2 is the first mergeable poll
|
| 117 | # (still null); mergeable resolves on the next poll.
|
| 118 | mergeable = (not pr["conflicting"]) if pr["gets"] >= 3 else None
|
| 119 | return ok(
|
| 120 | {
|
| 121 | "number": number,
|
| 122 | "head": {"sha": pr["head_sha"]},
|
| 123 | "mergeable": mergeable,
|
| 124 | "merged": pr["merged"],
|
| 125 | }
|
| 126 | )
|
| 127 | if method == "POST" and path.startswith("/statuses/"):
|
| 128 | sha = path.split("/statuses/", 1)[1]
|
| 129 | self.statuses[sha] = body["state"]
|
| 130 | return ok({"state": body["state"], "context": body["context"]}, status=201)
|
| 131 | if method == "GET" and path.startswith("/commits/") and path.endswith("/status"):
|
| 132 | sha = path.split("/")[2]
|
| 133 | return ok({"state": self.statuses.get(sha, "pending"), "sha": sha})
|
| 134 | if method == "PUT" and path.endswith("/merge"):
|
| 135 | number = int(path.split("/")[2])
|
| 136 | if self.prs[number].get("conflicting"):
|
| 137 | return ok({"message": "merge conflict"}, status=409)
|
| 138 | self.prs[number]["merged"] = True
|
| 139 | self.merged_count += 1
|
| 140 | self.main_sha = f"main-sha-{self.merged_count}"
|
| 141 | return ok({"merged": True, "sha": self.main_sha})
|
| 142 | if method == "DELETE" and path.startswith("/git/refs/"):
|
| 143 | return {"status": 204, "headers": {}, "body": b""}
|
| 144 | raise AssertionError(f"unscripted request: {method} {path}")
|
| 145 |
|
| 146 |
|
| 147 | def make_args(variant: str) -> argparse.Namespace:
|
| 148 | return argparse.Namespace(
|
| 149 | platform="github",
|
| 150 | driver="harness-api",
|
| 151 | repo=REPO,
|
| 152 | token_env=ABSENT_TOKEN_ENV,
|
| 153 | protection_variant=variant,
|
| 154 | semantics_class="integration-default",
|
| 155 | )
|
| 156 |
|
| 157 |
|
| 158 | def make_meta(variant: str, driver: str = "harness-api") -> dict:
|
| 159 | return {
|
| 160 | "bench_id": "20260612T000000Z",
|
| 161 | "profile": "platform",
|
| 162 | "platform": "github",
|
| 163 | "driver": driver,
|
| 164 | "platform_semantics_version": "platform-v1",
|
| 165 | "platform_semantics_class": "integration-default",
|
| 166 | "protection_variant": variant,
|
| 167 | "platform_comparison_key": f"platform-v1:integration-default:{variant}",
|
| 168 | "semantic_contract_match": True,
|
| 169 | }
|
| 170 |
|
| 171 |
|
| 172 | def race_scenario(branch_count: int) -> dict:
|
| 173 | spec = lane.load_platform_spec(ROOT / "scenarios" / "platform.yaml")
|
| 174 | scenario = lane.scenario_by_name(spec, "integration_race_n1")
|
| 175 | assert scenario is not None
|
| 176 | return {**scenario, "branch_count": branch_count}
|
| 177 |
|
| 178 |
|
| 179 | def branch_fleet_scenario(branch_count: int) -> dict:
|
| 180 | spec = lane.load_platform_spec(ROOT / "scenarios" / "platform.yaml")
|
| 181 | scenario = lane.scenario_by_name(spec, "branch_fleet_n10")
|
| 182 | assert scenario is not None
|
| 183 | return {**scenario, "branch_count": branch_count}
|
| 184 |
|
| 185 |
|
| 186 | def settings() -> lane.Settings:
|
| 187 | return lane.Settings(interval_s=0.5, settle_timeout_s=30.0, timeout_s=120.0, admitted=20_000)
|
| 188 |
|
| 189 |
|
| 190 | def one(rows: list[dict], operation: str) -> dict:
|
| 191 | matches = [row for row in rows if row["operation"] == operation]
|
| 192 | assert len(matches) == 1, f"{operation}: expected exactly 1 row, got {len(matches)}"
|
| 193 | return matches[0]
|
| 194 |
|
| 195 |
|
| 196 | class RaceRunnerCheckInstantTest(unittest.TestCase):
|
| 197 | """Drive the race runner function directly: n=2 under check-instant."""
|
| 198 |
|
| 199 | def setUp(self) -> None:
|
| 200 | self.transport = FakeGitHubTransport()
|
| 201 | client = GitHubClient(token="fake-token", transport=self.transport)
|
| 202 | # Frozen budgeter clock: every pace after the first costs the full
|
| 203 | # 1.0s spacing, so pacing is deterministically nonzero.
|
| 204 | self.budgeter = RateLimitBudgeter(ops_per_hour=3600, clock=lambda: 0.0)
|
| 205 | self.sleeps: list[float] = []
|
| 206 | self.rows = lane.run_github_api_race(
|
| 207 | make_args("check-instant"),
|
| 208 | race_scenario(branch_count=2),
|
| 209 | make_meta("check-instant"),
|
| 210 | 0,
|
| 211 | settings(),
|
| 212 | client=client,
|
| 213 | budgeter=self.budgeter,
|
| 214 | sleep=self.sleeps.append,
|
| 215 | )
|
| 216 |
|
| 217 | def test_per_branch_anatomy_rows_present_for_both_branches(self) -> None:
|
| 218 | for operation in RACE_BRANCH_OPERATIONS:
|
| 219 | branches = [
|
| 220 | row["race_branch"] for row in self.rows if row["operation"] == operation
|
| 221 | ]
|
| 222 | self.assertEqual(branches, [0, 1], operation)
|
| 223 | for row in self.rows:
|
| 224 | self.assertEqual(row["returncode"], 0, row["operation"])
|
| 225 | # Branch names are uniquified per run (timestamped bench_id + pid +
|
| 226 | # token, the workflow_ab disposable-branch convention).
|
| 227 | names = {
|
| 228 | row["race_branch_name"]
|
| 229 | for row in self.rows
|
| 230 | if "race_branch_name" in row
|
| 231 | }
|
| 232 | self.assertEqual(len(names), 2)
|
| 233 | for name in names:
|
| 234 | self.assertIn("20260612t000000z", name.lower())
|
| 235 |
|
| 236 | def test_race_publish_window_covers_each_branch_publish(self) -> None:
|
| 237 | window = one(self.rows, "race.publish.window")
|
| 238 | publishes = [row for row in self.rows if row["operation"] == "branch.publish"]
|
| 239 | self.assertEqual(len(publishes), 2)
|
| 240 | for row in publishes:
|
| 241 | self.assertGreaterEqual(window["elapsed_ms"], row["elapsed_ms"])
|
| 242 | self.assertEqual(window["window_start_operation"], "branch.publish")
|
| 243 | self.assertEqual(window["window_end_operation"], "branch.publish")
|
| 244 | integrate = one(self.rows, "race.integrate.window")
|
| 245 | self.assertEqual(integrate["returncode"], 0)
|
| 246 | self.assertEqual(integrate["window_end_operation"], "integration.main.settle")
|
| 247 |
|
| 248 | def test_pacing_injected_equals_budgeter_total_and_was_slept(self) -> None:
|
| 249 | pacing = one(self.rows, "race.pacing.injected")
|
| 250 | self.assertEqual(pacing["elapsed_ms"], self.budgeter.injected_wait_s_total * 1000)
|
| 251 | # 5 content ops per branch (create_ref, put_file, create_pr,
|
| 252 | # post_status, merge_pr) x 2 branches = 10 paces; the first is free
|
| 253 | # and the frozen clock makes the k-th pace wait (k-1) x 1.0s spacing:
|
| 254 | # 1+2+...+9 = 45.0s injected.
|
| 255 | self.assertEqual(self.budgeter.injected_wait_s_total, 45.0)
|
| 256 | pacing_sleeps = [s for s in self.sleeps if s != settings().interval_s]
|
| 257 | self.assertEqual(sorted(pacing_sleeps), [float(k) for k in range(1, 10)])
|
| 258 | self.assertEqual(pacing["budget_ops_per_hour"], 3600)
|
| 259 |
|
| 260 | def test_race_total_present_and_aggregates(self) -> None:
|
| 261 | total = one(self.rows, "race.total")
|
| 262 | self.assertEqual(total["returncode"], 0)
|
| 263 | self.assertIsInstance(total["elapsed_ms"], float)
|
| 264 | self.assertEqual(
|
| 265 | total["summarized_operations"].count("branch.publish"), 2
|
| 266 | )
|
| 267 | self.assertGreater(total["api_metrics"]["request_count"], 0)
|
| 268 |
|
| 269 | def test_check_instant_posts_status_and_settles_green(self) -> None:
|
| 270 | checks = [
|
| 271 | row for row in self.rows
|
| 272 | if row["operation"] == "integration.checks.green.settle"
|
| 273 | ]
|
| 274 | self.assertEqual(len(checks), 2)
|
| 275 | for row in checks:
|
| 276 | self.assertTrue(row["settled"])
|
| 277 | self.assertEqual(row["returncode"], 0)
|
| 278 | self.assertEqual(row["status_context"], "oakbench/check-instant")
|
| 279 | self.assertNotIn("structural", row)
|
| 280 | status_posts = [
|
| 281 | request for request in self.transport.requests
|
| 282 | if request["method"] == "POST" and "/statuses/" in request["url"]
|
| 283 | ]
|
| 284 | self.assertEqual(len(status_posts), 2)
|
| 285 | for request in status_posts:
|
| 286 | body = json.loads(request["body"].decode())
|
| 287 | self.assertEqual(body["state"], "success")
|
| 288 | self.assertEqual(body["context"], "oakbench/check-instant")
|
| 289 |
|
| 290 |
|
| 291 | class BranchFleetGitHubApiTest(unittest.TestCase):
|
| 292 | def setUp(self) -> None:
|
| 293 | self.transport = FakeGitHubTransport()
|
| 294 | client = GitHubClient(token="fake-token", transport=self.transport)
|
| 295 | self.rows = lane.run_github_api_branch_fleet_scenario(
|
| 296 | make_args("protection-none"),
|
| 297 | branch_fleet_scenario(branch_count=10),
|
| 298 | make_meta("protection-none"),
|
| 299 | 0,
|
| 300 | settings(),
|
| 301 | client=client,
|
| 302 | sleep=lambda _: None,
|
| 303 | )
|
| 304 |
|
| 305 | def test_branch_fleet_rows_cover_full_pipeline(self) -> None:
|
| 306 | self.assertEqual(
|
| 307 | [row["operation"] for row in self.rows],
|
| 308 | [
|
| 309 | "fleet.seed",
|
| 310 | "fleet.classify",
|
| 311 | "fleet.plan",
|
| 312 | "fleet.apply",
|
| 313 | "fleet.sync",
|
| 314 | "fleet.oracle",
|
| 315 | "fleet.total",
|
| 316 | ],
|
| 317 | )
|
| 318 | for row in self.rows:
|
| 319 | self.assertEqual(row["returncode"], 0, row["operation"])
|
| 320 | self.assertEqual(row["branch_fleet_provider"], "github:harness-api")
|
| 321 | self.assertEqual(row["branch_count"], 10)
|
| 322 | classify = one(self.rows, "fleet.classify")
|
| 323 | self.assertEqual(classify["branches_reviewed"], 10)
|
| 324 | self.assertEqual(classify["clean_detected_count"], 7)
|
| 325 | self.assertEqual(classify["conflict_detected_count"], 3)
|
| 326 | apply = one(self.rows, "fleet.apply")
|
| 327 | self.assertEqual(apply["branches_merged"], 7)
|
| 328 | self.assertEqual(apply["branches_left_open"], 3)
|
| 329 | oracle = one(self.rows, "fleet.oracle")
|
| 330 | self.assertTrue(oracle["action_plan_correct"])
|
| 331 | self.assertTrue(oracle["merge_order_correct"])
|
| 332 | self.assertTrue(oracle["final_survivors_correct"])
|
| 333 | total = one(self.rows, "fleet.total")
|
| 334 | self.assertEqual(
|
| 335 | total["branch_fleet_workflow_operations"],
|
| 336 | ["fleet.classify", "fleet.plan", "fleet.apply", "fleet.sync", "fleet.oracle"],
|
| 337 | )
|
| 338 | self.assertEqual(total["branch_fleet_setup_operations"], ["fleet.seed"])
|
| 339 | self.assertTrue(total["branch_fleet_setup_excluded_from_workflow"])
|
| 340 | self.assertGreaterEqual(total["branch_fleet_setup_elapsed_ms"], 0.0)
|
| 341 | self.assertGreaterEqual(total["branch_fleet_workflow_elapsed_ms"], 0.0)
|
| 342 |
|
| 343 | def test_branch_fleet_rows_validate_platform_contract(self) -> None:
|
| 344 | runner = runner_fields()
|
| 345 | self.assertEqual(
|
| 346 | validate_rows([stamp_row(row, runner) for row in self.rows], "platform"),
|
| 347 | [],
|
| 348 | )
|
| 349 |
|
| 350 | def test_merged_branches_are_cleaned_up_untimed(self) -> None:
|
| 351 | deletes = [
|
| 352 | request for request in self.transport.requests
|
| 353 | if request["method"] == "DELETE"
|
| 354 | ]
|
| 355 | self.assertEqual(len(deletes), 10)
|
| 356 |
|
| 357 |
|
| 358 | class RaceRunnerProtectionNoneTest(unittest.TestCase):
|
| 359 | def test_checks_green_is_structural_and_no_status_posted(self) -> None:
|
| 360 | transport = FakeGitHubTransport()
|
| 361 | client = GitHubClient(token="fake-token", transport=transport)
|
| 362 | budgeter = RateLimitBudgeter(ops_per_hour=3600, clock=lambda: 0.0)
|
| 363 | rows = lane.run_github_api_race(
|
| 364 | make_args("protection-none"),
|
| 365 | race_scenario(branch_count=1),
|
| 366 | make_meta("protection-none"),
|
| 367 | 0,
|
| 368 | settings(),
|
| 369 | client=client,
|
| 370 | budgeter=budgeter,
|
| 371 | sleep=lambda _s: None,
|
| 372 | )
|
| 373 | checks = one(rows, "integration.checks.green.settle")
|
| 374 | self.assertTrue(checks["structural"])
|
| 375 | self.assertTrue(checks["settled"])
|
| 376 | self.assertEqual(
|
| 377 | [r for r in transport.requests if "/statuses/" in r["url"]], []
|
| 378 | )
|
| 379 | runner = runner_fields()
|
| 380 | self.assertEqual(
|
| 381 | validate_rows([stamp_row(row, runner) for row in rows], "platform"), []
|
| 382 | )
|
| 383 |
|
| 384 |
|
| 385 | class RaceCliDriverDeferredTest(unittest.TestCase):
|
| 386 | def test_github_cli_race_emits_deferred_skip_rows(self) -> None:
|
| 387 | args = make_args("protection-none")
|
| 388 | args.driver = "cli"
|
| 389 | scenario = race_scenario(branch_count=1)
|
| 390 | rows = lane.run_race_with_credentials(
|
| 391 | args, scenario, make_meta("protection-none", driver="cli"), 0,
|
| 392 | Path(tempfile.gettempdir()), settings(),
|
| 393 | )
|
| 394 | self.assertEqual(
|
| 395 | [row["operation"] for row in rows], RACE_OPERATIONS
|
| 396 | )
|
| 397 | for row in rows:
|
| 398 | self.assertTrue(row["skipped"])
|
| 399 | self.assertEqual(row["skip_reason"], "race_cli_driver_deferred")
|
| 400 | runner = runner_fields()
|
| 401 | self.assertEqual(
|
| 402 | validate_rows([stamp_row(row, runner) for row in rows], "platform"), []
|
| 403 | )
|
| 404 |
|
| 405 |
|
| 406 | class RaceWithoutCredentialsSubprocessTest(unittest.TestCase):
|
| 407 | def test_integration_race_n1_without_token_exits_3_all_skip(self) -> None:
|
| 408 | with tempfile.TemporaryDirectory() as tmp:
|
| 409 | results = Path(tmp) / "results"
|
| 410 | env = os.environ.copy()
|
| 411 | env.pop(ABSENT_TOKEN_ENV, None)
|
| 412 | proc = subprocess.run(
|
| 413 | [
|
| 414 | sys.executable, str(SCRIPT),
|
| 415 | "--platform", "github",
|
| 416 | "--driver", "harness-api",
|
| 417 | "--scenario", "integration_race_n1",
|
| 418 | "--protection-variant", "check-instant",
|
| 419 | "--results", str(results),
|
| 420 | "--token-env", ABSENT_TOKEN_ENV,
|
| 421 | ],
|
| 422 | capture_output=True, text=True, env=env, timeout=120,
|
| 423 | )
|
| 424 | self.assertEqual(proc.returncode, 3, proc.stderr)
|
| 425 | raw = results / "latest.platform.jsonl"
|
| 426 | rows = [json.loads(line) for line in raw.read_text().splitlines() if line.strip()]
|
| 427 | self.assertEqual([row["operation"] for row in rows], RACE_OPERATIONS)
|
| 428 | for row in rows:
|
| 429 | self.assertTrue(row["skipped"])
|
| 430 | self.assertEqual(row["returncode"], 77)
|
| 431 | self.assertEqual(
|
| 432 | row["skip_reason"], f"github_token_missing:{ABSENT_TOKEN_ENV}"
|
| 433 | )
|
| 434 | # check-instant is now an implemented variant: the skip is about
|
| 435 | # credentials, and the fairness contract still matches.
|
| 436 | self.assertEqual(row["protection_variant"], "check-instant")
|
| 437 | self.assertEqual(
|
| 438 | row["platform_comparison_key"],
|
| 439 | "platform-v1:integration-default:check-instant",
|
| 440 | )
|
| 441 | self.assertTrue(row["semantic_contract_match"])
|
| 442 | self.assertEqual(validate_rows(rows, "platform"), [])
|
| 443 |
|
| 444 |
|
| 445 | class RaceScenarioSpecTest(unittest.TestCase):
|
| 446 | def setUp(self) -> None:
|
| 447 | self.spec = lane.load_platform_spec(ROOT / "scenarios" / "platform.yaml")
|
| 448 |
|
| 449 | def test_race_scenarios_declared_with_vocabulary(self) -> None:
|
| 450 | for name, count in (("integration_race_n1", 1), ("integration_race_n10", 10)):
|
| 451 | scenario = lane.scenario_by_name(self.spec, name)
|
| 452 | self.assertIsNotNone(scenario, name)
|
| 453 | self.assertEqual(scenario["operations"], RACE_OPERATIONS, name)
|
| 454 | self.assertTrue(scenario["requires_remote"])
|
| 455 | self.assertEqual(lane.race_branch_count(scenario), count)
|
| 456 | self.assertTrue(lane.is_race_scenario(scenario))
|
| 457 |
|
| 458 | def test_branch_count_parsed_from_name_when_absent(self) -> None:
|
| 459 | self.assertEqual(lane.race_branch_count({"name": "integration_race_n7"}), 7)
|
| 460 | self.assertFalse(lane.is_race_scenario({"name": "pr_single_anatomy"}))
|
| 461 |
|
| 462 | def test_check_instant_is_an_implemented_variant(self) -> None:
|
| 463 | semantics = lane.load_semantics(ROOT / "config" / "platform_semantics.json")
|
| 464 | self.assertEqual(
|
| 465 | semantics["implemented_protection_variants"],
|
| 466 | ["protection-none", "check-instant"],
|
| 467 | )
|
| 468 | self.assertTrue(
|
| 469 | lane.compute_semantic_contract_match(
|
| 470 | semantics, "integration-default", "check-instant"
|
| 471 | )
|
| 472 | )
|
| 473 |
|
| 474 |
|
| 475 | if __name__ == "__main__":
|
| 476 | unittest.main()
|