Log in
benchmarksPublic

Viewing files at commit c85a464b4b25Commit detailsView latest

Browse files
tests/test_platform_lifecycle.py 1327 lines · 53.5 KB · python
1
"""Platform lane CLI: skip-row honesty without credentials.
2
3
These tests never touch the network or GitHub: they exercise exactly the
4
credential-less path (skip rows, exit 3) plus pure helpers, the scenario
5
spec, and (via the scripted in-memory transport shared with the race tests)
6
the non-race check-instant paths: the harness-api driver must actually post
7
the required status before merging, and the cli driver — which cannot post
8
statuses — must emit honest skip rows instead of rows that claim the
9
protection variant held. The absent token env var name is invented so a
10
developer's real GITHUB_TOKEN can never leak into the subprocess path under
11
test.
12
"""
13
14
from __future__ import annotations
15
16
import argparse
17
import json
18
import os
19
import subprocess
20
import sys
21
import tempfile
22
import unittest
23
from pathlib import Path
24
25
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
26
27
import platform_lifecycle as lane  # noqa: E402
28
from oakbench.platform_api import GitHubClient  # noqa: E402
29
from oakbench.rows import LANE_REQUIRED_KEYS, validate_rows  # noqa: E402
30
from oakbench.runner import runner_fields, stamp_row  # noqa: E402
31
32
try:  # discovery imports this module as tests.test_platform_lifecycle
33
    from tests.test_platform_race import FakeGitHubTransport
34
except ImportError:  # direct invocation: python3 tests/test_platform_lifecycle.py
35
    from test_platform_race import FakeGitHubTransport
36
37
ROOT = Path(__file__).resolve().parents[1]
38
SCRIPT = ROOT / "scripts" / "platform_lifecycle.py"
39
ABSENT_TOKEN_ENV = "OAKBENCH_TEST_TOKEN_THAT_IS_ABSENT"
40
41
PR_ANATOMY_OPERATIONS = [
42
    "branch.publish",
43
    "integration.open",
44
    "integration.visible.settle",
45
    "integration.mergeable.settle",
46
    "integration.merge.request",
47
    "integration.main.settle",
48
    "integration.sync.local",
49
    "integration.total",
50
]
51
52
BRANCH_TRIAGE_OPERATIONS = [
53
    "triage.seed",
54
    "triage.review",
55
    "triage.plan",
56
    "triage.apply",
57
    "triage.oracle",
58
    "triage.total",
59
]
60
61
62
def run_lane(*argv: str) -> subprocess.CompletedProcess[str]:
63
    env = os.environ.copy()
64
    env.pop(ABSENT_TOKEN_ENV, None)
65
    return subprocess.run(
66
        [sys.executable, str(SCRIPT), *argv],
67
        capture_output=True,
68
        text=True,
69
        env=env,
70
        timeout=120,
71
    )
72
73
74
def read_rows(results: Path) -> list[dict]:
75
    raw = results / "latest.platform.jsonl"
76
    return [json.loads(line) for line in raw.read_text().splitlines() if line.strip()]
77
78
79
class MissingTokenSkipTest(unittest.TestCase):
80
    def test_cli_step_row_keeps_full_stdout_for_machine_parse(self) -> None:
81
        payload_len = 25_000
82
        with tempfile.TemporaryDirectory() as tmp:
83
            row = lane.cli_step_row(
84
                {"bench_id": "test"},
85
                "branch_fleet_n10",
86
                "fleet.classify",
87
                [
88
                    sys.executable,
89
                    "-c",
90
                    (
91
                        "import json; "
92
                        f"print(json.dumps({{'payload': 'x' * {payload_len}}}))"
93
                    ),
94
                ],
95
                Path(tmp),
96
                0,
97
                os.environ.copy(),
98
                admitted=100,
99
                full_stdout_bytes=1_000_000,
100
            )
101
102
        self.assertEqual(row["returncode"], 0)
103
        self.assertTrue(row["stdout_truncated"])
104
        self.assertLess(row["token_estimate"]["stdout_admitted_chars"], payload_len)
105
        self.assertEqual(len(json.loads(row["_stdout"])["payload"]), payload_len)
106
107
    def test_pr_single_anatomy_without_token_exits_3_with_skip_rows(self) -> None:
108
        with tempfile.TemporaryDirectory() as tmp:
109
            results = Path(tmp) / "results"
110
            proc = run_lane(
111
                "--platform", "github",
112
                "--driver", "harness-api",
113
                "--scenario", "pr_single_anatomy",
114
                "--runs", "2",
115
                "--results", str(results),
116
                "--token-env", ABSENT_TOKEN_ENV,
117
            )
118
            self.assertEqual(proc.returncode, 3, proc.stderr)
119
            rows = read_rows(results)
120
121
        self.assertEqual(len(rows), 2 * len(PR_ANATOMY_OPERATIONS))
122
        self.assertEqual(
123
            [row["operation"] for row in rows[: len(PR_ANATOMY_OPERATIONS)]],
124
            PR_ANATOMY_OPERATIONS,
125
        )
126
        self.assertEqual(sorted({row["run"] for row in rows}), [0, 1])
127
        for row in rows:
128
            self.assertTrue(row["skipped"])
129
            self.assertEqual(row["returncode"], 77)
130
            self.assertEqual(
131
                row["skip_reason"], f"github_token_missing:{ABSENT_TOKEN_ENV}"
132
            )
133
            self.assertEqual(row["platform"], "github")
134
            self.assertEqual(row["driver"], "harness-api")
135
            self.assertEqual(row["platform_semantics_version"], "platform-v1")
136
            self.assertEqual(row["platform_semantics_class"], "integration-default")
137
            self.assertEqual(row["protection_variant"], "protection-none")
138
            self.assertEqual(
139
                row["platform_comparison_key"],
140
                "platform-v1:integration-default:protection-none",
141
            )
142
            self.assertTrue(row["semantic_contract_match"])
143
            self.assertIsInstance(row["runner_id"], str)
144
            self.assertIn("runner_class", row)
145
            if row["operation"].endswith(".settle"):
146
                self.assertIsNone(row["settled"])
147
                self.assertEqual(row["poll_iterations"], 0)
148
                self.assertEqual(row["poll_interval_s"], 2.0)
149
                self.assertEqual(row["poll_quantization_ms"], 2000.0)
150
        # The durable rows satisfy the platform lane contract.
151
        self.assertEqual(validate_rows(rows, "platform"), [])
152
153
    def test_capability_probe_without_token_exits_3(self) -> None:
154
        with tempfile.TemporaryDirectory() as tmp:
155
            results = Path(tmp) / "results"
156
            proc = run_lane(
157
                "--platform", "github",
158
                "--driver", "harness-api",
159
                "--scenario", "platform_capability_probe",
160
                "--results", str(results),
161
                "--token-env", ABSENT_TOKEN_ENV,
162
            )
163
            self.assertEqual(proc.returncode, 3, proc.stderr)
164
            rows = read_rows(results)
165
        self.assertEqual([row["operation"] for row in rows], ["capability.probe"])
166
        self.assertEqual(
167
            rows[0]["skip_reason"], f"github_token_missing:{ABSENT_TOKEN_ENV}"
168
        )
169
170
    def test_unimplemented_protection_variant_skips(self) -> None:
171
        with tempfile.TemporaryDirectory() as tmp:
172
            results = Path(tmp) / "results"
173
            env = os.environ.copy()
174
            env[ABSENT_TOKEN_ENV] = "fake-token-value"
175
            proc = subprocess.run(
176
                [
177
                    sys.executable, str(SCRIPT),
178
                    "--platform", "github",
179
                    "--driver", "harness-api",
180
                    "--scenario", "pr_single_anatomy",
181
                    "--repo", "acme/disposable",
182
                    "--protection-variant", "check-slow60s",
183
                    "--results", str(results),
184
                    "--token-env", ABSENT_TOKEN_ENV,
185
                ],
186
                capture_output=True, text=True, env=env, timeout=120,
187
            )
188
            self.assertEqual(proc.returncode, 3, proc.stderr)
189
            rows = read_rows(results)
190
        for row in rows:
191
            self.assertEqual(
192
                row["skip_reason"], "protection_variant_not_implemented:check-slow60s"
193
            )
194
            self.assertFalse(row["semantic_contract_match"])
195
196
197
class ReadinessReasonTest(unittest.TestCase):
198
    def test_gh_cli_missing(self) -> None:
199
        reason = lane.readiness_skip_reason(
200
            "github", "cli", "a/b", "GITHUB_TOKEN",
201
            requires_remote=True, which=lambda _: None, environ={"GITHUB_TOKEN": "x"},
202
        )
203
        self.assertEqual(reason, "gh_cli_missing")
204
205
    def test_github_token_missing_names_the_env_var(self) -> None:
206
        reason = lane.readiness_skip_reason(
207
            "github", "cli", "a/b", "MY_PAT",
208
            requires_remote=True, which=lambda _: "/usr/bin/gh", environ={},
209
        )
210
        self.assertEqual(reason, "github_token_missing:MY_PAT")
211
212
    def test_github_repo_not_configured(self) -> None:
213
        reason = lane.readiness_skip_reason(
214
            "github", "harness-api", "", "MY_PAT",
215
            requires_remote=True, which=lambda _: None, environ={"MY_PAT": "x"},
216
        )
217
        self.assertEqual(reason, "github_repo_not_configured")
218
219
    def test_oak_cli_missing_then_oak_remote_missing(self) -> None:
220
        missing_bin = lane.readiness_skip_reason(
221
            "oak", "cli", "oak/repo", "GITHUB_TOKEN",
222
            requires_remote=True, which=lambda _: None, environ={},
223
        )
224
        self.assertEqual(missing_bin, "oak_cli_missing")
225
        missing_remote = lane.readiness_skip_reason(
226
            "oak", "cli", "", "GITHUB_TOKEN",
227
            requires_remote=True, which=lambda _: "/usr/local/bin/oak", environ={},
228
        )
229
        self.assertEqual(missing_remote, "oak_remote_missing")
230
231
    def test_oak_credentials_missing_is_preflight_visible(self) -> None:
232
        with tempfile.TemporaryDirectory() as tmp:
233
            reason = lane.readiness_skip_reason(
234
                "oak", "cli", "oak/repo", "GITHUB_TOKEN",
235
                requires_remote=True,
236
                which=lambda _: "/usr/local/bin/oak",
237
                environ={"HOME": tmp},
238
            )
239
        self.assertEqual(reason, "oak_credentials_missing")
240
241
    def test_ready_github_returns_none(self) -> None:
242
        reason = lane.readiness_skip_reason(
243
            "github", "cli", "a/b", "MY_PAT",
244
            requires_remote=True, which=lambda _: "/usr/bin/gh", environ={"MY_PAT": "x"},
245
        )
246
        self.assertIsNone(reason)
247
248
249
class ScenarioSpecTest(unittest.TestCase):
250
    def setUp(self) -> None:
251
        self.spec = lane.load_platform_spec(ROOT / "scenarios" / "platform.yaml")
252
253
    def test_immutable_scenario_names(self) -> None:
254
        names = [s["name"] for s in self.spec["scenarios"]]
255
        self.assertEqual(
256
            names,
257
            [
258
                "platform_capability_probe",
259
                "pr_single_anatomy",
260
                "poll_until_merged_i10s",
261
                "integration_race_n1",
262
                "integration_race_n10",
263
                "branch_triage_n4",
264
                "branch_fleet_n10",
265
                "branch_fleet_n100",
266
                "branch_fleet_n1000",
267
            ],
268
        )
269
270
    def test_pr_single_anatomy_operation_vocabulary(self) -> None:
271
        scenario = lane.scenario_by_name(self.spec, "pr_single_anatomy")
272
        self.assertEqual(scenario["operations"], PR_ANATOMY_OPERATIONS)
273
274
    def test_poll_scenario_encodes_its_interval(self) -> None:
275
        scenario = lane.scenario_by_name(self.spec, "poll_until_merged_i10s")
276
        self.assertEqual(scenario["poll_interval_s"], 10)
277
        self.assertIn("integration.merged.settle", scenario["operations"])
278
        settings = lane.scenario_settings(self.spec, scenario, admitted=100)
279
        self.assertEqual(settings.interval_s, 10.0)
280
281
    def test_branch_triage_scenario_declared(self) -> None:
282
        scenario = lane.scenario_by_name(self.spec, "branch_triage_n4")
283
        self.assertIsNotNone(scenario)
284
        assert scenario is not None
285
        self.assertEqual(scenario["operations"], BRANCH_TRIAGE_OPERATIONS)
286
        self.assertEqual(lane.branch_triage_count(scenario), 4)
287
        self.assertTrue(lane.is_branch_triage_scenario(scenario))
288
289
    def test_branch_fleet_scenarios_declared(self) -> None:
290
        for name, count in (
291
            ("branch_fleet_n10", 10),
292
            ("branch_fleet_n100", 100),
293
            ("branch_fleet_n1000", 1000),
294
        ):
295
            scenario = lane.scenario_by_name(self.spec, name)
296
            self.assertIsNotNone(scenario)
297
            assert scenario is not None
298
            self.assertEqual(
299
                scenario["operations"],
300
                [
301
                    "fleet.seed",
302
                    "fleet.classify",
303
                    "fleet.plan",
304
                    "fleet.apply",
305
                    "fleet.sync",
306
                    "fleet.oracle",
307
                    "fleet.total",
308
                ],
309
            )
310
            self.assertEqual(lane.branch_fleet_count(scenario), count)
311
            self.assertTrue(lane.is_branch_fleet_scenario(scenario))
312
        self.assertEqual(lane.branch_fleet_conflict_count(10), 3)
313
314
    def test_branch_fleet_large_live_tier_requires_opt_in(self) -> None:
315
        n100 = lane.scenario_by_name(self.spec, "branch_fleet_n100")
316
        n1000 = lane.scenario_by_name(self.spec, "branch_fleet_n1000")
317
        assert n100 is not None
318
        assert n1000 is not None
319
        self.assertIsNone(
320
            lane.branch_fleet_live_skip_reason(
321
                n100, "harness-api", environ={}
322
            )
323
        )
324
        self.assertIsNone(
325
            lane.branch_fleet_live_skip_reason(
326
                n1000, lane.BRANCH_TRIAGE_FAKE_DRIVER, environ={}
327
            )
328
        )
329
        self.assertEqual(
330
            lane.branch_fleet_live_skip_reason(
331
                n1000, "harness-api", environ={}
332
            ),
333
            "branch_fleet_large_live_requires_opt_in:"
334
            "OAK_BENCH_BRANCH_FLEET_LIVE_LARGE=1",
335
        )
336
        self.assertIsNone(
337
            lane.branch_fleet_live_skip_reason(
338
                n1000,
339
                "harness-api",
340
                environ={"OAK_BENCH_BRANCH_FLEET_LIVE_LARGE": "1"},
341
            )
342
        )
343
344
    def test_branch_fleet_worker_count_env_is_bounded(self) -> None:
345
        self.assertEqual(
346
            lane.branch_fleet_seed_worker_count(
347
                100, {lane.BRANCH_FLEET_SEED_WORKERS_ENV: "8"}
348
            ),
349
            8,
350
        )
351
        self.assertEqual(
352
            lane.branch_fleet_seed_worker_count(
353
                100, {lane.BRANCH_FLEET_SEED_WORKERS_ENV: "999"}
354
            ),
355
            lane.BRANCH_FLEET_WORKERS_MAX,
356
        )
357
        self.assertEqual(
358
            lane.branch_fleet_seed_worker_count(
359
                3, {lane.BRANCH_FLEET_SEED_WORKERS_ENV: "999"}
360
            ),
361
            3,
362
        )
363
        self.assertEqual(
364
            lane.branch_fleet_seed_worker_count(
365
                100, {lane.BRANCH_FLEET_SEED_WORKERS_ENV: "not-an-int"}
366
            ),
367
            lane.BRANCH_FLEET_SEED_WORKERS_DEFAULT,
368
        )
369
370
    def test_oak_branch_fleet_mergeability_mapping(self) -> None:
371
        self.assertIs(
372
            lane.oak_branch_fleet_mergeable(
373
                {
374
                    "recommended_action": "validate_then_merge",
375
                    "vcs_merge_safe": True,
376
                }
377
            ),
378
            True,
379
        )
380
        self.assertIs(lane.oak_branch_fleet_mergeable({"mergeability": "clean"}), True)
381
        self.assertIs(
382
            lane.oak_branch_fleet_mergeable(
383
                {
384
                    "recommended_action": "resolve",
385
                    "vcs_merge_safe": False,
386
                }
387
            ),
388
            False,
389
        )
390
        self.assertIs(
391
            lane.oak_branch_fleet_mergeable({"mergeability": "conflicts"}),
392
            False,
393
        )
394
        self.assertIsNone(lane.oak_branch_fleet_mergeable({}))
395
396
    def test_oak_branch_fleet_cleanup_uses_valid_remote_close_shape(self) -> None:
397
        commands = lane.oak_branch_fleet_cleanup_commands(
398
            [{"name": "bench-clean"}, {"name": "bench-conflict"}],
399
            ["bench-main-side"],
400
        )
401
        self.assertEqual(
402
            commands,
403
            [
404
                ["oak", "close", "--remote", "--reason", "stale", "--json", "bench-clean"],
405
                ["oak", "close", "--remote", "--reason", "stale", "--json", "bench-conflict"],
406
                ["oak", "close", "--remote", "--reason", "stale", "--json", "bench-main-side"],
407
            ],
408
        )
409
410
    def test_oak_branch_fleet_seed_steps_collapse_commit_push(self) -> None:
411
        specs = lane.branch_fleet_specs(
412
            bench_id="bench",
413
            scenario_name="branch_fleet_n10",
414
            run_index=0,
415
            branch_count=10,
416
        )
417
        steps, advancers = lane.oak_branch_fleet_seed_steps(
418
            specs, {"bench_id": "bench"}, 0
419
        )
420
        commands = [command for step in steps for command in step.commands]
421
422
        self.assertEqual(len(advancers), 3)
423
        self.assertEqual(len(commands), 42)
424
        self.assertNotIn(["oak", "push"], commands)
425
        self.assertEqual(commands.count(["oak", "commit", "--no-verify", "--push"]), 13)
426
        self.assertEqual(commands.count(["oak", "merge"]), 3)
427
        self.assertEqual(
428
            lane.oak_branch_fleet_cleanup_commands(specs, advancers),
429
            lane.oak_branch_fleet_cleanup_commands_for_branches(
430
                [str(spec["name"]) for spec in specs] + advancers
431
            ),
432
        )
433
434
    def test_oak_branch_fleet_live_skips_when_oak_lacks_commit_push(self) -> None:
435
        scenario = lane.scenario_by_name(self.spec, "branch_fleet_n10")
436
        assert scenario is not None
437
        args = argparse.Namespace(
438
            platform="oak",
439
            driver="cli",
440
            repo="oak/disposable-branch-fleet-test",
441
            token_env=ABSENT_TOKEN_ENV,
442
        )
443
        original = lane.oak_cli_commit_push_skip_reason
444
        try:
445
            lane.oak_cli_commit_push_skip_reason = lambda _env: (
446
                "oak_commit_push_not_supported:requires_current_oak"
447
            )
448
            with tempfile.TemporaryDirectory() as tmp:
449
                rows = lane.run_scenario_with_credentials(
450
                    args,
451
                    scenario,
452
                    {**fake_meta("cli", "protection-none"), "platform": "oak"},
453
                    0,
454
                    Path(tmp),
455
                    fake_settings(),
456
                )
457
        finally:
458
            lane.oak_cli_commit_push_skip_reason = original
459
460
        self.assertEqual([row["operation"] for row in rows], scenario["operations"])
461
        self.assertTrue(all(row.get("skipped") for row in rows))
462
        self.assertTrue(
463
            all(
464
                row["skip_reason"] == "oak_commit_push_not_supported:requires_current_oak"
465
                for row in rows
466
            )
467
        )
468
469
    def test_oak_branch_fleet_seed_chunks_cover_branch_steps_once(self) -> None:
470
        specs = lane.branch_fleet_specs(
471
            bench_id="bench",
472
            scenario_name="branch_fleet_n10",
473
            run_index=0,
474
            branch_count=10,
475
        )
476
        steps, _advancers = lane.oak_branch_fleet_seed_steps(
477
            specs, {"bench_id": "bench"}, 0
478
        )
479
480
        chunks = lane.oak_branch_fleet_seed_step_chunks(steps, worker_count=4)
481
        flattened = [step.branch for _worker_id, _offset, chunk in chunks for step in chunk]
482
483
        self.assertEqual(flattened, [str(spec["name"]) for spec in specs])
484
        self.assertEqual([offset for _worker_id, offset, _chunk in chunks], [0, 9, 18, 24])
485
        self.assertTrue(all(step.phase == "branch" for _wid, _off, chunk in chunks for step in chunk))
486
487
    def test_oak_branch_fleet_seed_row_progress_and_diagnostics(self) -> None:
488
        specs = lane.branch_fleet_specs(
489
            bench_id="bench",
490
            scenario_name="branch_fleet_n3",
491
            run_index=0,
492
            branch_count=3,
493
        )
494
        steps, advancers = lane.oak_branch_fleet_seed_steps(
495
            specs, {"bench_id": "bench"}, 0
496
        )
497
        progress: list[str] = []
498
        cleanup: list[str] = []
499
        ran: list[list[str]] = []
500
501
        def runner(
502
            command: list[str], cwd: Path | None, env: dict[str, str]
503
        ) -> subprocess.CompletedProcess[str]:
504
            ran.append(command)
505
            return subprocess.CompletedProcess(command, 0, "", "")
506
507
        with tempfile.TemporaryDirectory() as tmp:
508
            row = lane.oak_branch_fleet_seed_row(
509
                {"bench_id": "bench"},
510
                "branch_fleet_n3",
511
                steps,
512
                Path(tmp),
513
                0,
514
                {},
515
                admitted=20_000,
516
                branch_count=3,
517
                conflict_count=len(advancers),
518
                progress=progress.append,
519
                cleanup_branches=cleanup,
520
                runner=runner,
521
            )
522
523
        self.assertEqual(row["returncode"], 0)
524
        self.assertEqual(row["seed_branch_count"], 3)
525
        self.assertEqual(row["seed_conflict_count"], 1)
526
        self.assertEqual(row["seed_command_count"], 13)
527
        self.assertEqual(row["seed_completed_count"], 4)
528
        self.assertEqual(row["seed_completed_branch_count"], 3)
529
        self.assertEqual(row["seed_completed_conflict_count"], 1)
530
        self.assertEqual(row["seed_completed_command_count"], 13)
531
        self.assertEqual(row["tool_call_count"], 13)
532
        self.assertEqual(row["last_completed_branch"], advancers[-1])
533
        self.assertIsNone(row["failed_command_index"])
534
        self.assertIsNone(row["failed_branch"])
535
        self.assertEqual(ran, row["command"])
536
        self.assertEqual(cleanup, [str(spec["name"]) for spec in specs] + advancers)
537
        self.assertIn("[branch-fleet seed] seeded 3/3", "\n".join(progress))
538
        self.assertIn("[branch-fleet seed] advanced conflicts 1/1", "\n".join(progress))
539
540
    def test_oak_branch_fleet_seed_retries_transient_502(self) -> None:
541
        specs = lane.branch_fleet_specs(
542
            bench_id="bench",
543
            scenario_name="branch_fleet_n1",
544
            run_index=0,
545
            branch_count=1,
546
        )
547
        steps, advancers = lane.oak_branch_fleet_seed_steps(
548
            specs, {"bench_id": "bench"}, 0
549
        )
550
        self.assertEqual(advancers, [])
551
        attempts: dict[tuple[str, ...], int] = {}
552
        progress: list[str] = []
553
554
        def runner(
555
            command: list[str], cwd: Path | None, env: dict[str, str]
556
        ) -> subprocess.CompletedProcess[str]:
557
            key = tuple(command)
558
            attempts[key] = attempts.get(key, 0) + 1
559
            if command == ["oak", "commit", "--no-verify", "--push"] and attempts[key] == 1:
560
                return subprocess.CompletedProcess(command, 6, "abc123\n", "HTTP 502 Bad Gateway")
561
            return subprocess.CompletedProcess(command, 0, "", "")
562
563
        with tempfile.TemporaryDirectory() as tmp:
564
            row = lane.oak_branch_fleet_seed_row(
565
                {"bench_id": "bench"},
566
                "branch_fleet_n1",
567
                steps,
568
                Path(tmp),
569
                0,
570
                {},
571
                admitted=20_000,
572
                branch_count=1,
573
                conflict_count=0,
574
                progress=progress.append,
575
                cleanup_branches=[],
576
                runner=runner,
577
                retry_delay_s=0.0,
578
                retry_sleep=lambda _seconds: None,
579
            )
580
581
        self.assertEqual(row["returncode"], 0)
582
        self.assertEqual(row["seed_command_count"], 3)
583
        self.assertEqual(row["tool_call_count"], 4)
584
        self.assertEqual(row["seed_retry_count"], 1)
585
        self.assertEqual(row["seed_retries"][0]["command_index"], 2)
586
        self.assertEqual(row["seed_completed_branch_count"], 1)
587
        self.assertIsNone(row["failed_command_index"])
588
        self.assertIn("retrying transient failure 1/2", "\n".join(progress))
589
590
    def test_oak_branch_fleet_seed_timeout_becomes_failed_row(self) -> None:
591
        specs = lane.branch_fleet_specs(
592
            bench_id="bench",
593
            scenario_name="branch_fleet_n1",
594
            run_index=0,
595
            branch_count=1,
596
        )
597
        steps, _advancers = lane.oak_branch_fleet_seed_steps(
598
            specs, {"bench_id": "bench"}, 0
599
        )
600
601
        def runner(
602
            command: list[str], cwd: Path | None, env: dict[str, str]
603
        ) -> subprocess.CompletedProcess[str]:
604
            if command[0] == "/bin/zsh":
605
                raise subprocess.TimeoutExpired(command, timeout=300, output="out", stderr="err")
606
            return subprocess.CompletedProcess(command, 0, "", "")
607
608
        with tempfile.TemporaryDirectory() as tmp:
609
            row = lane.oak_branch_fleet_seed_row(
610
                {"bench_id": "bench"},
611
                "branch_fleet_n1",
612
                steps,
613
                Path(tmp),
614
                0,
615
                {},
616
                admitted=20_000,
617
                branch_count=1,
618
                conflict_count=0,
619
                progress=lambda _message: None,
620
                cleanup_branches=[],
621
                runner=runner,
622
            )
623
624
        self.assertEqual(row["returncode"], 124)
625
        self.assertEqual(row["failure_reason"], "seed_command_failed")
626
        self.assertEqual(row["failed_command_index"], 1)
627
        self.assertEqual(row["failed_branch"], str(specs[0]["name"]))
628
        self.assertIn("seed_command_timeout:300s", row["failed_command_stderr"])
629
        self.assertEqual(row["seed_completed_count"], 0)
630
        self.assertEqual(row["tool_call_count"], 2)
631
632
    def test_oak_branch_fleet_seed_failure_records_branch_and_aborts(self) -> None:
633
        scenario = lane.scenario_by_name(self.spec, "branch_fleet_n10")
634
        assert scenario is not None
635
        original_run_untimed = lane.run_untimed
636
637
        def fake_run_untimed(
638
            command: list[str],
639
            cwd: Path | None,
640
            env: dict[str, str],
641
            timeout: float = 300.0,
642
        ) -> subprocess.CompletedProcess[str]:
643
            if len(command) >= 2 and command[:2] == ["oak", "clone"]:
644
                Path(command[-1]).mkdir(parents=True, exist_ok=True)
645
                return subprocess.CompletedProcess(command, 0, "", "")
646
            if command == ["oak", "commit", "--no-verify", "--push"]:
647
                return subprocess.CompletedProcess(command, 55, "", "push failed")
648
            return subprocess.CompletedProcess(command, 0, "", "")
649
650
        args = argparse.Namespace(
651
            platform="oak",
652
            driver="cli",
653
            repo="oak/disposable-branch-fleet-test",
654
            token_env=ABSENT_TOKEN_ENV,
655
        )
656
        try:
657
            lane.run_untimed = fake_run_untimed
658
            with tempfile.TemporaryDirectory() as tmp:
659
                rows = lane.run_oak_cli_branch_fleet_scenario(
660
                    args,
661
                    scenario,
662
                    {**fake_meta("cli", "protection-none"), "platform": "oak"},
663
                    0,
664
                    Path(tmp),
665
                    fake_settings(),
666
                )
667
        finally:
668
            lane.run_untimed = original_run_untimed
669
670
        seed = rows[0]
671
        self.assertEqual(seed["operation"], "fleet.seed")
672
        self.assertEqual(seed["returncode"], 55)
673
        self.assertEqual(seed["failure_reason"], "seed_command_failed")
674
        self.assertEqual(seed["failed_command_index"], 2)
675
        self.assertTrue(str(seed["failed_branch"]).endswith("-clea-0000"))
676
        self.assertEqual(seed["seed_completed_count"], 0)
677
        self.assertEqual(seed["seed_completed_branch_count"], 0)
678
        self.assertEqual(seed["seed_completed_conflict_count"], 0)
679
        self.assertEqual(seed["seed_completed_command_count"], 12)
680
        self.assertTrue(seed["seed_parallel"])
681
        self.assertEqual(seed["seed_worker_count"], 4)
682
        self.assertEqual(seed["seed_worker_clone_count"], 5)
683
        self.assertEqual(seed["seed_worker_success_count"], 0)
684
        self.assertEqual(seed["branches_seeded"], 0)
685
        self.assertEqual(seed["failed_command_stderr"], "push failed")
686
        for row in rows[1:-1]:
687
            self.assertTrue(row.get("skipped"), row["operation"])
688
            self.assertEqual(row["skip_reason"], "prior_step_failed:fleet.seed")
689
        total = rows[-1]
690
        self.assertEqual(total["operation"], "fleet.total")
691
        self.assertFalse(total.get("skipped"))
692
        self.assertEqual(total["returncode"], 1)
693
        self.assertEqual(total["branch_fleet_setup_operations"], ["fleet.seed"])
694
        self.assertEqual(total["branch_fleet_workflow_elapsed_ms"], 0.0)
695
        self.assertEqual(total["branch_fleet_workflow_returncode"], 77)
696
697
    def test_oak_branch_fleet_parallel_seed_succeeds_and_cleans_same_branches(self) -> None:
698
        scenario = {
699
            "name": "branch_fleet_n4",
700
            "branch_count": 4,
701
            "operations": ["fleet.seed", "fleet.total"],
702
        }
703
        cleanup_seen: list[str] = []
704
        created_seen: list[str] = []
705
        original_run_untimed = lane.run_untimed
706
707
        def fake_run_untimed(
708
            command: list[str],
709
            cwd: Path | None,
710
            env: dict[str, str],
711
            timeout: float = 300.0,
712
        ) -> subprocess.CompletedProcess[str]:
713
            if len(command) >= 2 and command[:2] == ["oak", "clone"]:
714
                Path(command[-1]).mkdir(parents=True, exist_ok=True)
715
                return subprocess.CompletedProcess(command, 0, "", "")
716
            if len(command) >= 4 and command[:4] == ["oak", "switch", "--clean", "-c"]:
717
                created_seen.append(command[-1])
718
            if len(command) >= 2 and command[:2] == ["oak", "close"]:
719
                cleanup_seen.append(command[-1])
720
            return subprocess.CompletedProcess(command, 0, "", "")
721
722
        args = argparse.Namespace(
723
            platform="oak",
724
            driver="cli",
725
            repo="oak/disposable-branch-fleet-test",
726
            token_env=ABSENT_TOKEN_ENV,
727
        )
728
        meta = {**fake_meta("cli", "protection-none"), "platform": "oak"}
729
        try:
730
            lane.run_untimed = fake_run_untimed
731
            with tempfile.TemporaryDirectory() as tmp:
732
                rows = lane.run_oak_cli_branch_fleet_scenario(
733
                    args,
734
                    scenario,
735
                    meta,
736
                    0,
737
                    Path(tmp),
738
                    fake_settings(),
739
                )
740
        finally:
741
            lane.run_untimed = original_run_untimed
742
743
        seed = rows[0]
744
        total = rows[-1]
745
        self.assertEqual(seed["returncode"], 0)
746
        self.assertTrue(seed["seed_parallel"])
747
        self.assertEqual(seed["seed_worker_count"], 4)
748
        self.assertEqual(seed["seed_worker_clone_count"], 5)
749
        self.assertEqual(seed["seed_worker_success_count"], 4)
750
        self.assertEqual(seed["seed_command_count"], 16)
751
        self.assertEqual(seed["seed_completed_branch_count"], 4)
752
        self.assertEqual(seed["seed_completed_conflict_count"], 1)
753
        self.assertEqual(seed["branches_seeded"], 4)
754
        self.assertEqual(total["branch_fleet_setup_operations"], ["fleet.seed"])
755
        self.assertTrue(total["branch_fleet_setup_excluded_from_workflow"])
756
        self.assertEqual(total["cleanup_failure_count"], 0)
757
        self.assertEqual(len(created_seen), 5)
758
        self.assertEqual(set(cleanup_seen), set(created_seen))
759
760
    def test_oak_branch_fleet_apply_uses_checkout_free_merge(self) -> None:
761
        self.assertEqual(
762
            lane.oak_branch_fleet_apply_commands(["bench-clean-1", "bench-clean-2"]),
763
            [["oak", "merge", "bench-clean-1"], ["oak", "merge", "bench-clean-2"]],
764
        )
765
766
    def test_oak_branch_fleet_cleanup_fields_surface_failures(self) -> None:
767
        self.assertEqual(
768
            lane.oak_branch_fleet_cleanup_fields([]),
769
            {"cleanup_failure_count": 0},
770
        )
771
        failures = [{"branch": "bench-leaked", "returncode": 6, "stderr": "network"}]
772
        self.assertEqual(
773
            lane.oak_branch_fleet_cleanup_fields(failures),
774
            {"cleanup_failure_count": 1, "cleanup_failures": failures},
775
        )
776
777
    def test_oak_branch_fleet_cleanup_parallel_records_worker_context(self) -> None:
778
        commands = lane.oak_branch_fleet_cleanup_commands_for_branches(
779
            ["bench-a", "bench-b", "bench-c"]
780
        )
781
        calls: list[tuple[str, str]] = []
782
783
        def runner(
784
            command: list[str], cwd: Path | None, env: dict[str, str]
785
        ) -> subprocess.CompletedProcess[str]:
786
            calls.append((cwd.name if cwd else "", command[-1]))
787
            if command[-1] == "bench-b":
788
                return subprocess.CompletedProcess(command, 6, "", "network")
789
            return subprocess.CompletedProcess(command, 0, "", "")
790
791
        with tempfile.TemporaryDirectory() as tmp:
792
            workdirs = [Path(tmp) / "w0", Path(tmp) / "w1"]
793
            for workdir in workdirs:
794
                workdir.mkdir()
795
            failures = lane.oak_branch_fleet_run_cleanup(
796
                commands,
797
                workdirs,
798
                {},
799
                worker_count=2,
800
                runner=runner,
801
            )
802
803
        self.assertEqual(len(calls), 3)
804
        self.assertEqual(
805
            failures,
806
            [
807
                {
808
                    "branch": "bench-b",
809
                    "command": commands[1],
810
                    "command_index": 1,
811
                    "worker_id": 0,
812
                    "cwd": str(workdirs[0]),
813
                    "returncode": 6,
814
                    "stdout": "",
815
                    "stderr": "network",
816
                }
817
            ],
818
        )
819
820
    def test_oak_branch_fleet_cleanup_records_runner_exceptions(self) -> None:
821
        commands = lane.oak_branch_fleet_cleanup_commands_for_branches(
822
            ["bench-timeout", "bench-error"]
823
        )
824
825
        def runner(
826
            command: list[str], cwd: Path | None, env: dict[str, str]
827
        ) -> subprocess.CompletedProcess[str]:
828
            if command[-1] == "bench-timeout":
829
                raise subprocess.TimeoutExpired(command, timeout=300, output="out")
830
            raise OSError("runner failed")
831
832
        with tempfile.TemporaryDirectory() as tmp:
833
            workdir = Path(tmp) / "w0"
834
            workdir.mkdir()
835
            failures = lane.oak_branch_fleet_run_cleanup(
836
                commands,
837
                [workdir],
838
                {},
839
                worker_count=1,
840
                runner=runner,
841
            )
842
843
        self.assertEqual(len(failures), 2)
844
        self.assertEqual(failures[0]["branch"], "bench-timeout")
845
        self.assertEqual(failures[0]["command"], commands[0])
846
        self.assertEqual(failures[0]["cwd"], str(workdir))
847
        self.assertEqual(failures[0]["returncode"], 124)
848
        self.assertEqual(failures[0]["stdout"], "out")
849
        self.assertIn("cleanup_command_timeout:300s", failures[0]["stderr"])
850
        self.assertEqual(failures[1]["branch"], "bench-error")
851
        self.assertEqual(failures[1]["command"], commands[1])
852
        self.assertEqual(failures[1]["returncode"], 1)
853
        self.assertIn("cleanup_command_exception:OSError", failures[1]["stderr"])
854
855
    def test_branch_fleet_oracle_fails_on_apply_failure(self) -> None:
856
        specs = [
857
            {"name": "bench-clean", "kind": "clean", "expected_action": "merge"},
858
            {
859
                "name": "bench-conflict",
860
                "kind": "conflicting",
861
                "expected_action": "leave_conflict_open",
862
            },
863
        ]
864
        plan = {
865
            "actions": [
866
                {"branch": "bench-clean", "classification": "clean", "action": "merge"},
867
                {
868
                    "branch": "bench-conflict",
869
                    "classification": "conflicting",
870
                    "action": "leave_conflict_open",
871
                },
872
            ],
873
            "merge_order": ["bench-clean"],
874
        }
875
        oracle = lane.branch_fleet_oracle(
876
            specs,
877
            plan,
878
            {
879
                "apply_succeeded": False,
880
                "merged_branches": [],
881
                "final_survivors": ["bench-conflict"],
882
            },
883
        )
884
        self.assertFalse(oracle["apply_succeeded"])
885
        self.assertTrue(oracle["action_plan_correct"])
886
        self.assertTrue(oracle["merge_order_correct"])
887
        self.assertFalse(oracle["final_survivors_correct"])
888
889
    def test_oak_branch_fleet_sync_survivors_parse_actual_remote_branch_list(self) -> None:
890
        specs = [
891
            {"name": "bench-clean", "kind": "clean", "expected_action": "merge"},
892
            {
893
                "name": "bench-conflict",
894
                "kind": "conflicting",
895
                "expected_action": "leave_conflict_open",
896
            },
897
        ]
898
        stdout = json.dumps(
899
            [
900
                {"name": "bench-conflict", "status": "open"},
901
                {"name": "unrelated-disposable-branch", "status": "open"},
902
            ]
903
        )
904
905
        self.assertEqual(
906
            lane.branch_fleet_survivors_from_sync_stdout(stdout, specs),
907
            ["bench-conflict"],
908
        )
909
910
        oracle = lane.branch_fleet_oracle(
911
            specs,
912
            {
913
                "actions": [
914
                    {"branch": "bench-clean", "classification": "clean", "action": "merge"},
915
                    {
916
                        "branch": "bench-conflict",
917
                        "classification": "conflicting",
918
                        "action": "leave_conflict_open",
919
                    },
920
                ],
921
                "merge_order": ["bench-clean"],
922
            },
923
            {
924
                "apply_succeeded": True,
925
                "merged_branches": ["bench-clean"],
926
                "final_survivors": [],
927
            },
928
        )
929
        self.assertFalse(oracle["final_survivors_correct"])
930
931
    def test_oak_branch_fleet_sync_survivors_reject_wrong_json_schema(self) -> None:
932
        specs = [{"name": "bench-clean", "kind": "clean", "expected_action": "merge"}]
933
934
        bad_payloads = [
935
            "",
936
            "   ",
937
            "{}",
938
            json.dumps({"error": "remote failed"}),
939
            json.dumps({"branches": {"name": "bench-clean"}}),
940
            json.dumps([{"status": "open"}]),
941
        ]
942
        for payload in bad_payloads:
943
            with self.subTest(payload=payload):
944
                with self.assertRaises(ValueError):
945
                    lane.branch_fleet_survivors_from_sync_stdout(payload, specs)
946
947
    def test_branch_fleet_workflow_summary_excludes_seed_time(self) -> None:
948
        fields = lane.branch_fleet_workflow_summary_fields(
949
            [
950
                {"operation": "fleet.seed", "elapsed_ms": 100.0, "returncode": 0, "tool_call_count": 10},
951
                {"operation": "fleet.classify", "elapsed_ms": 11.0, "returncode": 0, "tool_call_count": 1},
952
                {"operation": "fleet.plan", "elapsed_ms": 2.5, "returncode": 0, "tool_call_count": 0},
953
                {"operation": "fleet.apply", "elapsed_ms": 20.0, "returncode": 0, "tool_call_count": 7},
954
                {"operation": "fleet.sync", "elapsed_ms": 3.0, "returncode": 0, "tool_call_count": 1},
955
                {"operation": "fleet.oracle", "elapsed_ms": 0.5, "returncode": 1, "tool_call_count": 0},
956
            ]
957
        )
958
        self.assertEqual(fields["branch_fleet_setup_operations"], ["fleet.seed"])
959
        self.assertEqual(
960
            fields["branch_fleet_workflow_operations"],
961
            ["fleet.classify", "fleet.plan", "fleet.apply", "fleet.sync", "fleet.oracle"],
962
        )
963
        self.assertTrue(fields["branch_fleet_setup_excluded_from_workflow"])
964
        self.assertEqual(fields["branch_fleet_setup_elapsed_ms"], 100.0)
965
        self.assertEqual(fields["branch_fleet_workflow_elapsed_ms"], 37.0)
966
        self.assertEqual(fields["branch_fleet_workflow_failure_count"], 1)
967
        self.assertEqual(fields["branch_fleet_workflow_skip_count"], 0)
968
        self.assertEqual(fields["branch_fleet_workflow_returncode"], 1)
969
        self.assertEqual(fields["branch_fleet_workflow_tool_call_count"], 9)
970
971
    def test_branch_fleet_workflow_summary_all_skipped_is_not_success(self) -> None:
972
        fields = lane.branch_fleet_workflow_summary_fields(
973
            [
974
                {"operation": "fleet.seed", "elapsed_ms": 100.0, "returncode": 1},
975
                {"operation": "fleet.classify", "elapsed_ms": 0.0, "returncode": 77},
976
                {"operation": "fleet.plan", "elapsed_ms": 0.0, "returncode": 77},
977
                {"operation": "fleet.apply", "elapsed_ms": 0.0, "returncode": 77},
978
                {"operation": "fleet.sync", "elapsed_ms": 0.0, "returncode": 77},
979
                {"operation": "fleet.oracle", "elapsed_ms": 0.0, "returncode": 77},
980
            ]
981
        )
982
        self.assertEqual(fields["branch_fleet_workflow_failure_count"], 0)
983
        self.assertEqual(fields["branch_fleet_workflow_skip_count"], 5)
984
        self.assertEqual(fields["branch_fleet_workflow_returncode"], 77)
985
986
    def test_semantics_config_declares_variants_and_classes(self) -> None:
987
        semantics = lane.load_semantics(ROOT / "config" / "platform_semantics.json")
988
        self.assertEqual(semantics["schema_version"], 1)
989
        self.assertEqual(semantics["semantics_version"], "platform-v1")
990
        self.assertIn("integration-default", semantics["classes"])
991
        self.assertIn("minimal-merge", semantics["classes"])
992
        self.assertEqual(
993
            semantics["protection_variants"],
994
            [
995
                "protection-none",
996
                "check-instant",
997
                "check-slow60s",
998
                "required-review",
999
                "strict-uptodate",
1000
                "merge-queue",
1001
            ],
1002
        )
1003
        self.assertTrue(
1004
            lane.compute_semantic_contract_match(semantics, "integration-default", "protection-none")
1005
        )
1006
        self.assertFalse(
1007
            lane.compute_semantic_contract_match(semantics, "integration-default", "merge-queue")
1008
        )
1009
1010
1011
FAKE_REPO = "acme/race-disposable"  # FakeGitHubTransport is scripted for this repo
1012
CHECKS_GREEN = "integration.checks.green.settle"
1013
1014
1015
def fake_args(driver: str, variant: str) -> argparse.Namespace:
1016
    return argparse.Namespace(
1017
        platform="github",
1018
        driver=driver,
1019
        repo=FAKE_REPO,
1020
        token_env=ABSENT_TOKEN_ENV,
1021
        protection_variant=variant,
1022
        semantics_class="integration-default",
1023
    )
1024
1025
1026
def fake_meta(driver: str, variant: str, match: bool = True) -> dict:
1027
    return {
1028
        "bench_id": "20260612T000000Z",
1029
        "profile": "platform",
1030
        "platform": "github",
1031
        "driver": driver,
1032
        "platform_semantics_version": "platform-v1",
1033
        "platform_semantics_class": "integration-default",
1034
        "protection_variant": variant,
1035
        "platform_comparison_key": (
1036
            f"platform-v1:integration-default:{variant}"
1037
        ),
1038
        "semantic_contract_match": match,
1039
    }
1040
1041
1042
def fake_settings() -> lane.Settings:
1043
    return lane.Settings(interval_s=0.5, settle_timeout_s=30.0, timeout_s=120.0, admitted=20_000)
1044
1045
1046
def anatomy_scenario() -> dict:
1047
    spec = lane.load_platform_spec(ROOT / "scenarios" / "platform.yaml")
1048
    scenario = lane.scenario_by_name(spec, "pr_single_anatomy")
1049
    assert scenario is not None
1050
    return scenario
1051
1052
1053
class NonRaceCheckInstantApiDriverTest(unittest.TestCase):
1054
    """pr_single_anatomy + check-instant on the harness-api driver: the
1055
    required status is posted (and settles green) BEFORE the merge request,
1056
    and the run emits the integration.checks.green.settle row."""
1057
1058
    def setUp(self) -> None:
1059
        self.transport = FakeGitHubTransport()
1060
        client = GitHubClient(token="fake-token", transport=self.transport)
1061
        self.sleeps: list[float] = []
1062
        self.rows = lane.run_github_api_scenario(
1063
            fake_args("harness-api", "check-instant"),
1064
            anatomy_scenario(),
1065
            fake_meta("harness-api", "check-instant"),
1066
            0,
1067
            fake_settings(),
1068
            client=client,
1069
            sleep=self.sleeps.append,
1070
        )
1071
1072
    def test_checks_green_row_inserted_before_merge_request(self) -> None:
1073
        operations = [row["operation"] for row in self.rows]
1074
        expected = list(PR_ANATOMY_OPERATIONS)
1075
        expected.insert(expected.index("integration.merge.request"), CHECKS_GREEN)
1076
        self.assertEqual(operations, expected)
1077
1078
    def test_checks_green_row_is_measured_and_settled(self) -> None:
1079
        row = next(r for r in self.rows if r["operation"] == CHECKS_GREEN)
1080
        self.assertEqual(row["returncode"], 0)
1081
        self.assertTrue(row["settled"])
1082
        self.assertEqual(row["status_context"], "oakbench/check-instant")
1083
        self.assertNotIn("structural", row)
1084
        for r in self.rows:
1085
            self.assertEqual(r["returncode"], 0, r["operation"])
1086
1087
    def test_status_posted_before_merge(self) -> None:
1088
        kinds = [
1089
            (request["method"], request["url"]) for request in self.transport.requests
1090
        ]
1091
        status_posts = [
1092
            i for i, (method, url) in enumerate(kinds)
1093
            if method == "POST" and "/statuses/" in url
1094
        ]
1095
        merges = [
1096
            i for i, (method, url) in enumerate(kinds)
1097
            if method == "PUT" and url.endswith("/merge")
1098
        ]
1099
        self.assertEqual(len(status_posts), 1)
1100
        self.assertEqual(len(merges), 1)
1101
        self.assertLess(status_posts[0], merges[0])
1102
        body = json.loads(self.transport.requests[status_posts[0]]["body"].decode())
1103
        self.assertEqual(body["state"], "success")
1104
        self.assertEqual(body["context"], "oakbench/check-instant")
1105
1106
    def test_rows_validate_under_platform_contract(self) -> None:
1107
        runner = runner_fields()
1108
        stamped = [stamp_row(row, runner) for row in self.rows]
1109
        self.assertEqual(validate_rows(stamped, "platform"), [])
1110
1111
1112
class NonRaceCheckInstantCliDriverTest(unittest.TestCase):
1113
    """The cli driver cannot post commit statuses: under check-instant every
1114
    operation row of the run is an honest skip, never a merge that pretends
1115
    the protection variant held."""
1116
1117
    def test_all_operations_skip_with_the_exact_reason(self) -> None:
1118
        with tempfile.TemporaryDirectory() as tmp:
1119
            rows = lane.run_github_cli_scenario(
1120
                fake_args("cli", "check-instant"),
1121
                anatomy_scenario(),
1122
                fake_meta("cli", "check-instant", match=False),
1123
                0,
1124
                Path(tmp),
1125
                fake_settings(),
1126
            )
1127
        self.assertEqual([row["operation"] for row in rows], PR_ANATOMY_OPERATIONS)
1128
        for row in rows:
1129
            self.assertTrue(row["skipped"], row["operation"])
1130
            self.assertEqual(row["returncode"], 77)
1131
            self.assertEqual(
1132
                row["skip_reason"], "protection_variant_requires_api_driver:check-instant"
1133
            )
1134
        runner = runner_fields()
1135
        self.assertEqual(
1136
            validate_rows([stamp_row(row, runner) for row in rows], "platform"), []
1137
        )
1138
1139
    def test_main_marks_cli_check_instant_contract_unmatched(self) -> None:
1140
        # End-to-end through main(): the row meta itself must record that the
1141
        # cli driver never executes check-instant (semantic_contract_match
1142
        # false), even on the credential-less skip path.
1143
        with tempfile.TemporaryDirectory() as tmp:
1144
            results = Path(tmp) / "results"
1145
            proc = run_lane(
1146
                "--platform", "github",
1147
                "--driver", "cli",
1148
                "--scenario", "pr_single_anatomy",
1149
                "--protection-variant", "check-instant",
1150
                "--results", str(results),
1151
                "--token-env", ABSENT_TOKEN_ENV,
1152
            )
1153
            self.assertEqual(proc.returncode, 3, proc.stderr)
1154
            rows = read_rows(results)
1155
        self.assertEqual(len(rows), len(PR_ANATOMY_OPERATIONS))
1156
        for row in rows:
1157
            self.assertTrue(row["skipped"])
1158
            self.assertEqual(row["protection_variant"], "check-instant")
1159
            self.assertFalse(row["semantic_contract_match"])
1160
1161
1162
class ProtectionNoneUnchangedTest(unittest.TestCase):
1163
    def test_api_driver_protection_none_posts_no_status_and_adds_no_row(self) -> None:
1164
        transport = FakeGitHubTransport()
1165
        client = GitHubClient(token="fake-token", transport=transport)
1166
        rows = lane.run_github_api_scenario(
1167
            fake_args("harness-api", "protection-none"),
1168
            anatomy_scenario(),
1169
            fake_meta("harness-api", "protection-none"),
1170
            0,
1171
            fake_settings(),
1172
            client=client,
1173
            sleep=lambda _s: None,
1174
        )
1175
        self.assertEqual([row["operation"] for row in rows], PR_ANATOMY_OPERATIONS)
1176
        self.assertEqual(
1177
            [r for r in transport.requests if "/statuses/" in r["url"]], []
1178
        )
1179
        for row in rows:
1180
            self.assertEqual(row["returncode"], 0, row["operation"])
1181
1182
    def test_variant_executability_table(self) -> None:
1183
        self.assertTrue(
1184
            lane.protection_variant_executable("github", "harness-api", "check-instant")
1185
        )
1186
        self.assertFalse(
1187
            lane.protection_variant_executable("github", "cli", "check-instant")
1188
        )
1189
        # protection-none behavior unchanged on both drivers.
1190
        self.assertTrue(
1191
            lane.protection_variant_executable("github", "cli", "protection-none")
1192
        )
1193
        self.assertTrue(
1194
            lane.protection_variant_executable("github", "harness-api", "protection-none")
1195
        )
1196
1197
    def test_platform_rows_comparable_requires_same_variant(self) -> None:
1198
        github = fake_meta("harness-api", "protection-none")
1199
        oak = {
1200
            **fake_meta("cli", "protection-none"),
1201
            "platform": "oak",
1202
        }
1203
        check_instant = fake_meta("harness-api", "check-instant")
1204
        unmatched = fake_meta("cli", "check-instant", match=False)
1205
1206
        self.assertTrue(lane.platform_rows_comparable(github, oak))
1207
        self.assertFalse(lane.platform_rows_comparable(github, check_instant))
1208
        self.assertFalse(lane.platform_rows_comparable(check_instant, unmatched))
1209
1210
1211
class BranchTriageFakeProviderTest(unittest.TestCase):
1212
    def test_fake_provider_rows_check_action_plan_and_survivors(self) -> None:
1213
        scenario = lane.scenario_by_name(
1214
            lane.load_platform_spec(ROOT / "scenarios" / "platform.yaml"),
1215
            "branch_triage_n4",
1216
        )
1217
        assert scenario is not None
1218
        meta = fake_meta("fake-provider", "protection-none", match=False)
1219
        rows = lane.run_branch_triage_fake_scenario(scenario, meta, 0)
1220
        self.assertEqual([row["operation"] for row in rows], BRANCH_TRIAGE_OPERATIONS)
1221
        for row in rows:
1222
            self.assertEqual(row["returncode"], 0, row["operation"])
1223
            self.assertEqual(row["branch_triage_provider"], "fake")
1224
            self.assertEqual(row["tool_call_count"], 0)
1225
            self.assertIsNone(row["estimated_tokens_total"])
1226
1227
        plan = next(row for row in rows if row["operation"] == "triage.plan")
1228
        self.assertEqual(plan["branches_reviewed"], 4)
1229
        self.assertEqual(plan["stale_detected_count"], 1)
1230
        self.assertEqual(plan["duplicate_detected_count"], 1)
1231
        self.assertEqual(plan["conflict_detected_count"], 1)
1232
        self.assertEqual(
1233
            [action["action"] for action in plan["action_plan"]],
1234
            ["merge", "refresh_then_merge", "close_duplicate", "keep_open_manual_resolution"],
1235
        )
1236
1237
        apply = next(row for row in rows if row["operation"] == "triage.apply")
1238
        self.assertEqual(apply["branches_closed"], 3)
1239
        self.assertEqual(apply["final_survivors"], ["triage-conflicting-3"])
1240
        oracle = next(row for row in rows if row["operation"] == "triage.oracle")
1241
        self.assertTrue(oracle["action_plan_correct"])
1242
        self.assertTrue(oracle["merge_order_correct"])
1243
        self.assertTrue(oracle["final_survivors_correct"])
1244
1245
        runner = runner_fields()
1246
        self.assertEqual(
1247
            validate_rows([stamp_row(row, runner) for row in rows], "platform"), []
1248
        )
1249
1250
    def test_fake_provider_subprocess_runs_without_token(self) -> None:
1251
        with tempfile.TemporaryDirectory() as tmp:
1252
            results = Path(tmp) / "results"
1253
            proc = run_lane(
1254
                "--platform", "github",
1255
                "--driver", "fake-provider",
1256
                "--scenario", "branch_triage_n4",
1257
                "--results", str(results),
1258
                "--token-env", ABSENT_TOKEN_ENV,
1259
            )
1260
            self.assertEqual(proc.returncode, 0, proc.stderr)
1261
            rows = read_rows(results)
1262
        self.assertEqual([row["operation"] for row in rows], BRANCH_TRIAGE_OPERATIONS)
1263
        self.assertFalse(any(row.get("skipped") for row in rows))
1264
        self.assertFalse(rows[0]["semantic_contract_match"])
1265
1266
    def test_real_driver_without_credentials_emits_skip_rows(self) -> None:
1267
        with tempfile.TemporaryDirectory() as tmp:
1268
            results = Path(tmp) / "results"
1269
            proc = run_lane(
1270
                "--platform", "github",
1271
                "--driver", "harness-api",
1272
                "--scenario", "branch_triage_n4",
1273
                "--results", str(results),
1274
                "--token-env", ABSENT_TOKEN_ENV,
1275
            )
1276
            self.assertEqual(proc.returncode, 3, proc.stderr)
1277
            rows = read_rows(results)
1278
        self.assertEqual([row["operation"] for row in rows], BRANCH_TRIAGE_OPERATIONS)
1279
        for row in rows:
1280
            self.assertTrue(row["skipped"])
1281
            self.assertEqual(row["skip_reason"], f"github_token_missing:{ABSENT_TOKEN_ENV}")
1282
1283
    def test_real_driver_with_credentials_skips_until_provider_exists(self) -> None:
1284
        with tempfile.TemporaryDirectory() as tmp:
1285
            results = Path(tmp) / "results"
1286
            env = os.environ.copy()
1287
            env[ABSENT_TOKEN_ENV] = "fake-token-value"
1288
            proc = subprocess.run(
1289
                [
1290
                    sys.executable, str(SCRIPT),
1291
                    "--platform", "github",
1292
                    "--driver", "harness-api",
1293
                    "--scenario", "branch_triage_n4",
1294
                    "--repo", "acme/disposable",
1295
                    "--results", str(results),
1296
                    "--token-env", ABSENT_TOKEN_ENV,
1297
                ],
1298
                capture_output=True, text=True, env=env, timeout=120,
1299
            )
1300
            self.assertEqual(proc.returncode, 3, proc.stderr)
1301
            rows = read_rows(results)
1302
        for row in rows:
1303
            self.assertTrue(row["skipped"])
1304
            self.assertEqual(
1305
                row["skip_reason"],
1306
                "branch_triage_real_provider_not_implemented:github:harness-api",
1307
            )
1308
1309
1310
class LaneRegistrationTest(unittest.TestCase):
1311
    def test_platform_lane_registered_with_fairness_keys(self) -> None:
1312
        self.assertIn("platform", LANE_REQUIRED_KEYS)
1313
        for key in (
1314
            "platform",
1315
            "driver",
1316
            "platform_semantics_version",
1317
            "platform_semantics_class",
1318
            "protection_variant",
1319
            "platform_comparison_key",
1320
            "semantic_contract_match",
1321
            "runner_id",
1322
        ):
1323
            self.assertIn(key, LANE_REQUIRED_KEYS["platform"])
1324
1325
1326
if __name__ == "__main__":
1327
    unittest.main()