diff --git a/scripts/ci/audit_codeql_default_setup_rollout.py b/scripts/ci/audit_codeql_default_setup_rollout.py index 17eaa0146c..6637601593 100755 --- a/scripts/ci/audit_codeql_default_setup_rollout.py +++ b/scripts/ci/audit_codeql_default_setup_rollout.py @@ -269,6 +269,7 @@ def load_payload(path: Path | None, stdin: TextIO) -> list[dict[str, Any]]: def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + """Parse CLI arguments for either the file-payload or live-collection mode.""" parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("snapshots_json", nargs="?", type=Path) parser.add_argument("--repository") @@ -277,6 +278,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: def main(argv: list[str] | None = None) -> int: + """Audit CodeQL rollout state from file or live snapshots and print verdicts.""" args = parse_args(argv) try: live_mode = args.repository is not None or args.pr is not None diff --git a/scripts/ci/pr_review_merge_scheduler_core.py b/scripts/ci/pr_review_merge_scheduler_core.py index 04d26fac5a..e97b41074a 100644 --- a/scripts/ci/pr_review_merge_scheduler_core.py +++ b/scripts/ci/pr_review_merge_scheduler_core.py @@ -31,7 +31,7 @@ plan_dispatches, update_state_file, ) -except ModuleNotFoundError: # direct ``python scripts/ci/...`` execution +except ModuleNotFoundError: # pragma: no cover - package import path from review_admission_controller import ( WORKER_BOUNDARIES, AdmissionRequest, @@ -47,6 +47,7 @@ class SchedulerAdmissionGate: """Persist and bound review-worker leases for one scheduler execution.""" def __init__(self, state_path: Path, *, sequence: int, dispatch_budget: int) -> None: + """Bind this gate to one durable state file, run sequence, and worker budget.""" if sequence < 1: raise ValueError("admission sequence must be positive") if dispatch_budget < 0: @@ -68,6 +69,7 @@ def admit(self, component: str, repository: str, pr: dict[str, Any]) -> bool: selected: list[DispatchLease] = [] def lease(state): + """Apply this request to `state` and record any lease it wins.""" plan = plan_dispatches( state, [request], @@ -88,6 +90,7 @@ def reconcile(self, repository: str, prs: Sequence[dict[str, Any]]) -> None: live_prs = {int(pr["number"]): pr for pr in prs} def reconcile_state(state): + """Mark exact-head dispatched leases complete and superseded ones stale.""" records = dict(state.records) latest = dict(state.latest_sequences) for identity, record in tuple(records.items()): @@ -5395,7 +5398,7 @@ def self_test() -> None: from scripts.ci.review_admission_controller import ( self_test as admission_self_test, ) - except ModuleNotFoundError: # direct ``python scripts/ci/...`` execution + except ModuleNotFoundError: # pragma: no cover - package import path from review_admission_controller import self_test as admission_self_test admission_self_test() diff --git a/scripts/ci/review_admission_controller.py b/scripts/ci/review_admission_controller.py index dd26549a29..b8e7c208ab 100644 --- a/scripts/ci/review_admission_controller.py +++ b/scripts/ci/review_admission_controller.py @@ -20,12 +20,15 @@ @dataclass(frozen=True) class WorkerBoundary: + """The credential, permission set, and concurrency namespace one review worker runs under.""" + credential: str permissions: tuple[str, ...] concurrency_namespace: str cancel_in_progress: bool = True def concurrency_group(self, request: AdmissionRequest) -> str: + """Return this worker's `{namespace}-{repository}-{pull_request}` concurrency group.""" return ( f"{self.concurrency_namespace}-{request.repository}-{request.pull_request}" ) @@ -52,6 +55,8 @@ def concurrency_group(self, request: AdmissionRequest) -> str: @dataclass(frozen=True) class AdmissionRequest: + """One validated request to admit a review worker onto a specific PR head.""" + repository: str pull_request: int head_sha: str @@ -68,6 +73,7 @@ def create( component: str, sequence: int, ) -> AdmissionRequest: + """Validate and normalize raw fields into an `AdmissionRequest`.""" if isinstance(pull_request, bool) or not isinstance(pull_request, int): raise TypeError("pull request must be an integer") if isinstance(sequence, bool) or not isinstance(sequence, int): @@ -87,35 +93,45 @@ def create( @property def identity(self) -> str: + """Return the unique key identifying this exact request (including its sequence).""" return f"{self.repository}#{self.pull_request}@{self.head_sha}:{self.component}" @property def stream(self) -> str: + """Return the key identifying this request's PR+component stream across sequences.""" return f"{self.repository}#{self.pull_request}:{self.component}" @dataclass(frozen=True) class RequestRecord: + """An admission request paired with its current lifecycle status.""" + request: AdmissionRequest status: str @dataclass(frozen=True) class DispatchLease: + """A request that has been granted a worker boundary to run under.""" + request: AdmissionRequest boundary: WorkerBoundary @dataclass(frozen=True) class ControllerState: + """The durable admission controller's full state: known records and per-stream sequences.""" + records: dict[str, RequestRecord] latest_sequences: dict[str, int] @classmethod def empty(cls) -> ControllerState: + """Return the initial state with no records and no sequences observed yet.""" return cls({}, {}) def to_json(self) -> str: + """Serialize this state to its canonical, deterministically-ordered JSON form.""" payload = { "latest_sequences": self.latest_sequences, "records": { @@ -130,6 +146,7 @@ def to_json(self) -> str: @classmethod def from_json(cls, value: str) -> ControllerState: + """Parse and fully validate a state snapshot, rejecting any inconsistent JSON.""" payload = json.loads(value) if not isinstance(payload, dict): raise TypeError("durable admission state must be an object") @@ -204,6 +221,7 @@ def _open_regular_nofollow(path: Path, flags: int, mode: int = 0o600) -> int: def _read_state(path: Path) -> ControllerState: + """Read and parse one state file, rejecting a symlink and non-UTF-8 content.""" descriptor = _open_regular_nofollow(path, os.O_RDONLY) try: with os.fdopen(descriptor, encoding="utf-8") as stream: @@ -282,6 +300,8 @@ def update_state_file( @dataclass(frozen=True) class DispatchPlan: + """The result of one admission pass: the updated state, grants, and rejections.""" + state: ControllerState dispatches: tuple[DispatchLease, ...] rejections: dict[str, str] diff --git a/tests/test_codeql_default_setup_rollout.py b/tests/test_codeql_default_setup_rollout.py index 665eed5aa3..8610a39f49 100644 --- a/tests/test_codeql_default_setup_rollout.py +++ b/tests/test_codeql_default_setup_rollout.py @@ -1,7 +1,12 @@ import base64 +import builtins import json +import runpy +import sys from io import StringIO +import pytest + from scripts.ci import audit_codeql_default_setup_rollout as rollout HEAD = "a" * 40 @@ -264,3 +269,192 @@ def request(self, path): assert "head changed" in str(exc) else: raise AssertionError("moving exact-head evidence must fail closed") + + +def test_pagination_rejects_malformed_and_unbounded_evidence(): + with pytest.raises(rollout.EvidenceError, match="malformed pagination"): + rollout._pages(FakeClient({"/items?per_page=100&page=1": {}}), "/items") + + pages = { + f"/items?per_page=100&page={page}": [{}] * 100 + for page in range(1, rollout.MAX_PAGES + 1) + } + with pytest.raises(rollout.EvidenceError, match="pagination exceeded"): + rollout._pages(FakeClient(pages), "/items") + + +@pytest.mark.parametrize( + ("source", "active"), + ( + ( + "steps:\n - name: disabled\n if: ${{ false }}\n" + " uses: github/codeql-action/analyze@pin\n", + False, + ), + ( + "steps:\n - name: disabled\n uses: github/codeql-action/analyze@pin\n" + " with:\n upload: 'never'\n - name: next\n run: true\n", + False, + ), + ( + "steps:\n - name: active\n uses: github/codeql-action/upload-sarif@pin\n", + True, + ), + ), +) +def test_advanced_uploader_detection_honors_only_local_disabling(source, active): + assert rollout._has_active_advanced_upload(source) is active + + +def test_live_snapshot_rejects_ambiguous_or_invalid_workflow_sources(): + repository = "ContextualWisdomLab/xtrmLLMBatchPython" + workflow_path = f"/repos/{repository}/actions/workflows?per_page=100&page=1" + source_path = f"/repos/{repository}/contents/.github/workflows/ci.yml?ref={HEAD}" + + cases = [] + duplicate = live_responses() + duplicate[workflow_path]["workflows"] *= 2 + cases.append((duplicate, "identity is ambiguous")) + + lookup_failure = live_responses() + lookup_failure[source_path] = rollout.GitHubError("HTTP 500") + cases.append((lookup_failure, "source lookup failed")) + + invalid_size = live_responses() + invalid_size[source_path]["size"] = -1 + cases.append((invalid_size, "invalid size")) + + invalid_base64 = live_responses() + invalid_base64[source_path]["content"] = "!" + cases.append((invalid_base64, "source is invalid")) + + size_mismatch = live_responses() + size_mismatch[source_path]["size"] += 1 + cases.append((size_mismatch, "size mismatch")) + + for responses, message in cases: + with pytest.raises(rollout.EvidenceError, match=message): + rollout.collect_live_snapshot(FakeClient(responses), repository, 292) + + +@pytest.mark.parametrize( + ("repository", "pr_number", "message"), + ( + ("Other/example", 1, "must belong"), + ("ContextualWisdomLab/example", 0, "must be positive"), + ), +) +def test_live_snapshot_rejects_invalid_identity(repository, pr_number, message): + with pytest.raises(rollout.EvidenceError, match=message): + rollout.collect_live_snapshot(FakeClient({}), repository, pr_number) + + +def test_live_snapshot_rejects_ambiguous_ruleset_owner_and_missing_states(): + repository = "ContextualWisdomLab/xtrmLLMBatchPython" + pull_path = f"/repos/{repository}/pulls/292" + rulesets_path = f"/repos/{repository}/rulesets?includes_parents=true&per_page=100&page=1" + detail_path = f"/repos/{repository}/rulesets/{rollout.RULESET_ID}?includes_parents=true" + setup_path = f"/repos/{repository}/code-scanning/default-setup" + runs_path = f"/repos/{repository}/actions/runs?head_sha={HEAD}&per_page=100&page=1" + + closed = live_responses() + closed[pull_path] = {"state": "closed", "head": {"sha": HEAD}} + ambiguous_ruleset = live_responses() + ambiguous_ruleset[rulesets_path] *= 2 + ambiguous_owner = live_responses() + ambiguous_owner[detail_path]["rules"][0]["parameters"]["workflows"] *= 2 + missing_setup = live_responses() + missing_setup[setup_path] = {"state": "new-state"} + missing_status = live_responses() + missing_status[runs_path]["workflow_runs"][0].update(status=None, conclusion=None) + + for responses, message in ( + (closed, "not open"), + (ambiguous_ruleset, "ruleset evidence is ambiguous"), + (ambiguous_owner, "ruleset owner is ambiguous"), + (missing_setup, "default-setup state is unavailable"), + (missing_status, "has no status"), + ): + with pytest.raises(rollout.EvidenceError, match=message): + rollout.collect_live_snapshot(FakeClient(responses), repository, 292) + + +def test_exempt_snapshot_revalidates_head_and_classification_edges(): + repository = "ContextualWisdomLab/noema" + pull_path = f"/repos/{repository}/pulls/7" + rulesets_path = f"/repos/{repository}/rulesets?includes_parents=true&per_page=100&page=1" + client = FakeClient( + { + pull_path: {"state": "open", "head": {"sha": HEAD}}, + rulesets_path: [], + } + ) + assert rollout.collect_live_snapshot(client, repository, 7) == { + "name": "noema", + "ruleset_applies": False, + } + assert rollout.classify(snapshot(default_setup_state="unsupported"))[0] == "BLOCK" + assert rollout.classify(snapshot(central_codeql_status="failure"))[0] == "ROLLBACK" + + class MovingExemptClient(FakeClient): + reads = 0 + + def request(self, path): + if path == pull_path: + self.reads += 1 + if self.reads == 2: + return {"state": "open", "head": {"sha": "b" * 40}} + return super().request(path) + + with pytest.raises(rollout.EvidenceError, match="head changed"): + rollout.collect_live_snapshot( + MovingExemptClient(client.responses), repository, 7 + ) + + +def test_payload_file_and_cli_error_paths(tmp_path, monkeypatch, capsys): + payload_path = tmp_path / "snapshots.json" + payload_path.write_text(json.dumps([snapshot()]), encoding="utf-8") + assert rollout.load_payload(payload_path, StringIO()) == [snapshot()] + with pytest.raises(ValueError, match="array of objects"): + rollout.load_payload(None, StringIO("{}")) + + assert rollout.main([str(payload_path), "--repository", "ContextualWisdomLab/x", "--pr", "1"]) == 2 + monkeypatch.setattr(rollout.sys, "stdin", StringIO("{")) + assert rollout.main([]) == 2 + assert "unable to load CodeQL rollout snapshots" in capsys.readouterr().err + + +def test_live_cli_collects_one_snapshot(monkeypatch, capsys): + fake_client = object() + calls = [] + monkeypatch.setattr( + rollout.GitHubClient, + "from_environment", + classmethod(lambda cls: fake_client), + ) + def collect_snapshot(client, repository, pr): + calls.append((client, repository, pr)) + return snapshot() + + monkeypatch.setattr(rollout, "collect_live_snapshot", collect_snapshot) + assert rollout.main( + ["--repository", "ContextualWisdomLab/example", "--pr", "7"] + ) == 0 + assert calls == [(fake_client, "ContextualWisdomLab/example", 7)] + assert "state=VERIFIED" in capsys.readouterr().out + + +def test_direct_script_import_falls_back_to_sibling_module(monkeypatch): + script_path = rollout.Path(rollout.__file__) + real_import = builtins.__import__ + + def import_with_package_missing(name, *args, **kwargs): + if name == "scripts.ci.organization_commercial_readiness_loop": + raise ModuleNotFoundError(name) + return real_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", import_with_package_missing) + monkeypatch.setattr(sys, "path", [str(script_path.parent), *sys.path]) + namespace = runpy.run_path(str(script_path), run_name="rollout_direct_import_test") + assert namespace["GitHubClient"] is not None diff --git a/tests/test_noema_review_gate.py b/tests/test_noema_review_gate.py index 6422ae5012..5fa23dec53 100644 --- a/tests/test_noema_review_gate.py +++ b/tests/test_noema_review_gate.py @@ -1702,6 +1702,42 @@ def open(self, request): assert '{"error":' not in output +@pytest.mark.parametrize( + "attempts", + [ + [{}], + ["not-a-dict"], + ], +) +def test_call_llm_http_error_last_attempt_without_usable_fields_reports_no_attempt_telemetry( + monkeypatch, capsys, attempts +): + """A last attempt with no recognizable fields adds no attempt telemetry.""" + monkeypatch.setenv("NOEMA_LLM_API_URL", "https://llm.example.test/chat") + monkeypatch.setenv("NOEMA_LLM_API_KEY", "secret") + body = json.dumps( + {"error": {"detail": {"model": "github_models/deepseek-v3", "attempts": attempts}}} + ).encode() + + class Opener: + def open(self, request): + raise noema.urllib.error.HTTPError( + request.full_url, 502, "Bad Gateway", {}, io.BytesIO(body) + ) + + monkeypatch.setattr(noema.urllib.request, "build_opener", lambda *_args: Opener()) + + with pytest.raises(noema.NoemaTransportError): + noema.call_llm("owner/repo", 1, make_pr(), "diff", False, "head") + + output = capsys.readouterr().out + assert "served_model=github_models/deepseek-v3" in output + assert "provider_name=" not in output + assert "upstream_phase=" not in output + assert "attempt_number=" not in output + assert "upstream_status=" not in output + + def test_noema_redirect_handler_rejects_redirects(): """Noema must not follow redirects after validating the initial URL.""" handler = noema.NoRedirectHandler() diff --git a/tests/test_pr_review_fix_scheduler.py b/tests/test_pr_review_fix_scheduler.py index 667ced6148..6b9bd91e0c 100644 --- a/tests/test_pr_review_fix_scheduler.py +++ b/tests/test_pr_review_fix_scheduler.py @@ -1467,6 +1467,8 @@ def test_fix_parse_args_and_self_test(monkeypatch): ["--repo", "owner/repo"], ["--repo", "owner/repo", "--base-branch", "main", "--pr-number", "-1"], ["--repo", "owner/repo", "--base-branch", "main", "--max-prs", "0"], + ["--repo", "owner/repo", "--base-branch", "main", "--scan-window-size", "0"], + ["--repo", "owner/repo", "--base-branch", "main", "--rotation-seed", "-1"], ["--repo", "owner/repo", "--base-branch", "main", "--max-dispatches", "0"], ["--repo", "owner/repo", "--base-branch", "main", "--retry-hours", "0"], ["--repo", "owner/repo", "--base-branch", "main", "--autofix-repository", "bad"], diff --git a/tests/test_pr_review_merge_scheduler.py b/tests/test_pr_review_merge_scheduler.py index 4a13bd3bf1..b821fe1ee3 100644 --- a/tests/test_pr_review_merge_scheduler.py +++ b/tests/test_pr_review_merge_scheduler.py @@ -204,6 +204,61 @@ def test_inspect_pr_closes_only_fresh_non_draft_empty_pull_request(monkeypatch): assert calls[-1] == ["gh", "pr", "close", "1", "--repo", "owner/repo"] +def test_inspect_pr_classifies_empty_pull_request_without_closing_in_dry_run(monkeypatch): + head_sha = "a" * 40 + candidate = make_pr( + headRefOid=head_sha, + files={"totalCount": 0, "nodes": []}, + ) + calls = [] + monkeypatch.setattr( + sched, + "_fresh_open_pr_for_cancellation", + lambda _repo, _number: { + "draft": False, + "changed_files": 0, + "head": {"sha": head_sha}, + }, + ) + monkeypatch.setattr(sched, "run", lambda args: calls.append(args) or "") + + decision = inspect(candidate, dry_run=True) + + assert decision.action == "close_empty" + assert calls == [] + + +def test_inspect_pr_closes_empty_pull_request_even_if_the_comment_call_fails(monkeypatch): + head_sha = "a" * 40 + candidate = make_pr( + headRefOid=head_sha, + files={"totalCount": 0, "nodes": []}, + ) + calls = [] + + def fake_run(args): + if args[2] == "comment": + raise RuntimeError("comment API failure") + calls.append(args) + return "" + + monkeypatch.setattr( + sched, + "_fresh_open_pr_for_cancellation", + lambda _repo, _number: { + "draft": False, + "changed_files": 0, + "head": {"sha": head_sha}, + }, + ) + monkeypatch.setattr(sched, "run", fake_run) + + decision = inspect(candidate, dry_run=False) + + assert decision.action == "close_empty" + assert calls == [["gh", "pr", "close", "1", "--repo", "owner/repo"]] + + @pytest.mark.parametrize( "fresh", ( @@ -406,6 +461,11 @@ def test_rotating_pr_window_is_bounded_and_wraps_over_actual_results(): assert sched.rotating_pr_window(prs, offset=100, window_size=50) == prs[100:120] assert sched.rotating_pr_window(prs, offset=150, window_size=50) == prs[:50] assert sched.rotating_pr_window(prs, offset=0, window_size=None) == prs + assert sched.rotating_pr_window([], offset=0, window_size=50) == [] + with pytest.raises(ValueError, match="PR window offset must be non-negative and size must be positive"): + sched.rotating_pr_window(prs, offset=-1, window_size=50) + with pytest.raises(ValueError, match="PR window offset must be non-negative and size must be positive"): + sched.rotating_pr_window(prs, offset=0, window_size=0) def test_rest_fallback_hydrates_only_the_selected_rotating_window(monkeypatch): @@ -4910,6 +4970,37 @@ def fake_read(args): "run_attempt": 1, "created_at": "2026-09-04T01:04:00Z", }, + { + "id": 95, + "workflow_id": 14, + "name": "Weekly Full-Tree Scan", + "event": "schedule", + "head_sha": head_sha, + "status": "completed", + "conclusion": "startup_failure", + "run_attempt": 1, + "created_at": "2026-09-04T01:05:00Z", + }, + { + "id": 96, + "event": "pull_request", + "head_sha": head_sha, + "status": "completed", + "conclusion": "startup_failure", + "run_attempt": 1, + "created_at": "2026-09-04T01:06:00Z", + }, + { + "id": 89, + "workflow_id": 13, + "name": "Dependency Review", + "event": "pull_request", + "head_sha": head_sha, + "status": "completed", + "conclusion": "startup_failure", + "run_attempt": 1, + "created_at": "2026-09-04T01:00:30Z", + }, ] } ) @@ -5074,6 +5165,71 @@ def test_inspect_pr_recovers_startup_failure_before_other_actions(monkeypatch): assert "90" in decision.reason +def test_dispatch_strix_evidence_defers_to_bounded_admission_budget(monkeypatch, tmp_path): + """A fresh Strix dispatch (no existing job) respects the durable admission budget.""" + + def fake_run_with_env(args, *, stdin=None, env=None): + if "/actions/runs" in " ".join(args): + return '{"workflow_runs": []}' + return "" + + monkeypatch.setattr(sched, "run_with_env", fake_run_with_env) + monkeypatch.setenv("GITHUB_ACTIONS", "true") + monkeypatch.setenv("GH_TOKEN", "opencode-app-token") + monkeypatch.setenv("SCHEDULER_REQUIRED_WORKFLOW_REPOSITORY", "ContextualWisdomLab/.github") + pr = make_pr(baseRefOid="b" * 40, headRefOid="a" * 40) + monkeypatch.setattr(sched, "fetch_pr", lambda *_args: [pr]) + + gate = sched.SchedulerAdmissionGate(tmp_path / "admission.json", sequence=1, dispatch_budget=0) + with sched.active_admission_gate(gate): + assert sched.dispatch_strix_evidence( + "ContextualWisdomLab/example", "Strix Security Scan", pr, dry_run=False + ) == "admission_deferred" + + +def test_dispatch_strix_evidence_rerun_defers_to_bounded_admission_budget(monkeypatch, tmp_path): + """Rerunning an existing Strix job also respects the durable admission budget.""" + pr = make_pr(baseRefOid="b" * 40, headRefOid="a" * 40) + monkeypatch.setattr(sched, "matching_actions_job_id", lambda *_args: "202") + + gate = sched.SchedulerAdmissionGate(tmp_path / "admission.json", sequence=1, dispatch_budget=0) + with sched.active_admission_gate(gate): + assert sched.dispatch_strix_evidence( + "ContextualWisdomLab/example", "Strix Security Scan", pr, dry_run=False + ) == "admission_deferred" + + +def test_dispatch_strix_evidence_rerun_rechecks_live_head(monkeypatch): + """Rerunning an existing Strix job rechecks the exact live head first.""" + pr = make_pr(baseRefOid="b" * 40, headRefOid="a" * 40) + monkeypatch.setattr(sched, "matching_actions_job_id", lambda *_args: "202") + monkeypatch.setattr(sched, "fetch_pr", lambda *_args: [make_pr(headRefOid="c" * 40)]) + + assert sched.dispatch_strix_evidence( + "owner/repo", "Strix Security Scan", pr, dry_run=False + ) == "stale_head" + + +def test_dispatch_strix_evidence_rechecks_live_head_before_new_dispatch(monkeypatch): + """A fresh Strix dispatch rechecks the exact live head immediately before dispatching.""" + + def fake_run_with_env(args, *, stdin=None, env=None): + if "/actions/runs" in " ".join(args): + return '{"workflow_runs": []}' + return "" + + monkeypatch.setattr(sched, "run_with_env", fake_run_with_env) + monkeypatch.setenv("GITHUB_ACTIONS", "true") + monkeypatch.setenv("GH_TOKEN", "opencode-app-token") + monkeypatch.setenv("SCHEDULER_REQUIRED_WORKFLOW_REPOSITORY", "ContextualWisdomLab/.github") + pr = make_pr(baseRefOid="b" * 40, headRefOid="a" * 40) + monkeypatch.setattr(sched, "fetch_pr", lambda *_args: [make_pr(headRefOid="c" * 40)]) + + assert sched.dispatch_strix_evidence( + "owner/repo", "Strix Security Scan", pr, dry_run=False + ) == "stale_head" + + def test_missing_evidence_dispatch_uses_central_required_workflow_repository(monkeypatch): calls = [] head_sha = "a" * 40 @@ -5222,6 +5378,19 @@ def test_stacked_pr_waits_when_opencode_dispatch_is_already_active(monkeypatch): assert stacked.reason == "stacked PR onto develop; same-head OpenCode workflow run is already active" +def test_stacked_pr_waits_on_bounded_admission_budget(monkeypatch): + monkeypatch.setattr( + sched, + "dispatch_opencode_review", + lambda repo, workflow, pr, dry_run: "admission_deferred", + ) + + stacked = inspect(make_pr(baseRefName="develop")) + + assert stacked.action == "wait" + assert stacked.reason == "stacked PR onto develop; bounded admission budget is exhausted" + + def test_stacked_pr_waits_when_review_dispatch_budget_is_exhausted(): stacked = inspect(make_pr(baseRefName="develop"), review_dispatch_allowed=False) @@ -6807,6 +6976,14 @@ def test_inspect_pr_blocks_and_waits_for_policy_states(monkeypatch): assert coverage_active.reason == ( "current-head coverage evidence is complete, but a same-head OpenCode workflow run is already active" ) + monkeypatch.setattr( + sched, + "dispatch_opencode_review", + lambda repo, workflow, pr, dry_run: "admission_deferred", + ) + coverage_admission_deferred = inspect(coverage_request) + assert coverage_admission_deferred.action == "wait" + assert "bounded admission budget is exhausted" in coverage_admission_deferred.reason monkeypatch.setattr( sched, "dispatch_opencode_review", @@ -7305,6 +7482,30 @@ def test_draft_pr_review_request_marker_not_checked_when_flag_already_allows(mon assert decision.action == "security_dispatch" +def test_draft_pr_review_only_dispatch_waits_on_bounded_admission_budget(monkeypatch): + """A draft PR's review-only path defers to the same bounded admission budget.""" + monkeypatch.setattr( + sched, + "dispatch_strix_evidence", + lambda repo, workflow, pr, dry_run: "admission_deferred", + ) + decision = inspect(make_pr(isDraft=True), allow_draft_review_dispatch=True) + assert decision.action == "wait" + assert "bounded admission budget is exhausted" in decision.reason + + monkeypatch.setattr( + sched, + "dispatch_opencode_review", + lambda repo, workflow, pr, dry_run: "admission_deferred", + ) + strix_complete_draft = make_pr( + isDraft=True, statusCheckRollup={"contexts": {"nodes": [strix_check()]}} + ) + decision = inspect(strix_complete_draft, allow_draft_review_dispatch=True) + assert decision.action == "wait" + assert "bounded admission budget is exhausted" in decision.reason + + def test_draft_review_request_artifact_name_is_exact_and_stable(): assert sched.draft_review_request_artifact_name("owner/repo", 42, "a" * 40) == ( f"cwl-draft-review-request-owner-repo-42-{'a' * 40}" @@ -7998,6 +8199,14 @@ def followup(updated_pr, **overrides): statusCheckRollup={"contexts": {"nodes": [strix_check(), opencode_check()]}}, ) ) + monkeypatch.setattr( + sched, + "dispatch_strix_evidence", + lambda repo, workflow, pr, dry_run: "admission_deferred", + ) + assert "bounded admission budget is exhausted" in followup( + make_pr(headRefOid="new-head") + ) monkeypatch.setattr( sched, "dispatch_strix_evidence", @@ -8015,6 +8224,17 @@ def followup(updated_pr, **overrides): make_pr(headRefOid="new-head") ) + monkeypatch.setattr( + sched, + "dispatch_opencode_review", + lambda repo, workflow, pr, dry_run: "admission_deferred", + ) + assert "bounded admission budget is exhausted" in followup( + make_pr( + headRefOid="new-head", + statusCheckRollup={"contexts": {"nodes": [strix_check()]}}, + ) + ) monkeypatch.setattr( sched, "dispatch_opencode_review", @@ -8463,6 +8683,14 @@ def test_inspect_pr_handles_approved_reviews_and_dispatch(monkeypatch): busy_strix = inspect(make_pr()) assert busy_strix.action == "wait" assert "target repository already has active Strix evidence" in busy_strix.reason + monkeypatch.setattr( + sched, + "dispatch_strix_evidence", + lambda repo, workflow, pr, dry_run: "admission_deferred", + ) + admission_deferred_strix = inspect(make_pr()) + assert admission_deferred_strix.action == "wait" + assert "bounded admission budget is exhausted" in admission_deferred_strix.reason monkeypatch.setattr( sched, "dispatch_strix_evidence", @@ -8501,6 +8729,19 @@ def test_inspect_pr_handles_approved_reviews_and_dispatch(monkeypatch): stale_already_active.reason == "OpenCode review exceeded the status-check retry threshold, but a same-head workflow run is already active" ) + monkeypatch.setattr( + sched, + "dispatch_opencode_review", + lambda repo, workflow, pr, dry_run: "admission_deferred", + ) + stale_admission_deferred = inspect(stale_opencode, stale_opencode_minutes=0) + assert stale_admission_deferred.action == "wait" + assert "bounded admission budget is exhausted" in stale_admission_deferred.reason + monkeypatch.setattr( + sched, + "dispatch_opencode_review", + lambda repo, workflow, pr, dry_run: "already_running", + ) stale_limited = inspect(stale_opencode, stale_opencode_minutes=0, review_dispatch_allowed=False) assert stale_limited.action == "wait" assert "review dispatch limit reached" in stale_limited.reason @@ -8527,6 +8768,21 @@ def test_inspect_pr_handles_approved_reviews_and_dispatch(monkeypatch): completed_strix_already_active.reason == "current head has completed Strix evidence; same-head OpenCode workflow run is already active" ) + monkeypatch.setattr( + sched, + "dispatch_opencode_review", + lambda repo, workflow, pr, dry_run: "admission_deferred", + ) + completed_strix_admission_deferred = inspect( + make_pr(statusCheckRollup={"contexts": {"nodes": [strix_check()]}}), + ) + assert completed_strix_admission_deferred.action == "wait" + assert "bounded admission budget is exhausted" in completed_strix_admission_deferred.reason + monkeypatch.setattr( + sched, + "dispatch_opencode_review", + lambda repo, workflow, pr, dry_run: "already_running", + ) assert inspect(make_pr(), trigger_reviews=False).reason == "current head has no OpenCode approval" missing_approval_auto = inspect(make_pr(autoMergeRequest={"enabledAt": "now"}), trigger_reviews=False) assert missing_approval_auto.action == "disable_auto_merge" @@ -8741,6 +8997,40 @@ def test_main_limits_review_dispatches_and_branch_updates(monkeypatch, capsys): ) +def test_main_reconciles_the_durable_admission_gate_when_a_state_path_is_given( + monkeypatch, tmp_path +): + """`--admission-state-path` wires a real durable gate into the scan.""" + pr = make_pr(number=1, statusCheckRollup={"contexts": {"nodes": [strix_check()]}}) + dispatched = [] + monkeypatch.setattr(sched, "fetch_open_prs", lambda repo, max_prs: [pr]) + monkeypatch.setattr( + sched, + "dispatch_opencode_review", + lambda repo, workflow, pr, dry_run: dispatched.append(pr["number"]), + ) + monkeypatch.setattr(sched, "cancel_stale_pr_runs", lambda repo, pr, dry_run: []) + + state_path = tmp_path / "admission.json" + assert ( + sched.main( + [ + "--repo", + "owner/repo", + "--base-branch", + "main", + "--project-flow", + "github-flow", + "--admission-state-path", + str(state_path), + ] + ) + == 0 + ) + assert dispatched == [1] + assert state_path.exists() + + def test_main_prioritizes_stacked_prs_without_reordering_each_class(monkeypatch): prs = [ make_pr(number=1, baseRefName="main"), @@ -8859,6 +9149,38 @@ def test_main_rejects_invalid_review_dispatch_limit(): ) +def test_main_rejects_negative_admission_dispatch_budget(): + with pytest.raises(SystemExit, match="--admission-dispatch-budget must not be negative"): + sched.main( + [ + "--repo", + "owner/repo", + "--base-branch", + "main", + "--project-flow", + "github-flow", + "--admission-dispatch-budget", + "-1", + ] + ) + + +def test_main_rejects_non_positive_admission_sequence(): + with pytest.raises(SystemExit, match="--admission-sequence must be positive"): + sched.main( + [ + "--repo", + "owner/repo", + "--base-branch", + "main", + "--project-flow", + "github-flow", + "--admission-sequence", + "0", + ] + ) + + def test_main_rejects_invalid_branch_update_limit(): with pytest.raises(SystemExit, match="--branch-update-limit must be -1 or greater"): sched.main( @@ -10156,6 +10478,15 @@ def fake_api(path): ) is True +def test_admission_gate_rejects_invalid_sequence_and_budget(tmp_path): + """The gate validates its own constructor inputs independent of the CLI.""" + state_path = tmp_path / "admission.json" + with pytest.raises(ValueError, match="admission sequence must be positive"): + sched.SchedulerAdmissionGate(state_path, sequence=0, dispatch_budget=1) + with pytest.raises(ValueError, match="admission dispatch budget must not be negative"): + sched.SchedulerAdmissionGate(state_path, sequence=1, dispatch_budget=-1) + + def test_bounded_admission_persists_leases_and_completes_only_current_head( monkeypatch, tmp_path ): @@ -10258,6 +10589,43 @@ def test_opencode_dispatch_rechecks_live_head_immediately_before_side_effect( assert dispatched == [] +def test_reconcile_marks_lease_stale_when_live_head_has_moved(tmp_path): + """A lease recorded against a superseded head is retired without inspecting evidence.""" + gate = sched.SchedulerAdmissionGate( + tmp_path / "admission.json", sequence=91, dispatch_budget=1 + ) + pr = make_pr(number=7, headRefOid="a" * 40) + assert gate.admit("strix", "ContextualWisdomLab/example", pr) + moved_pr = make_pr(number=7, headRefOid="b" * 40) + gate.reconcile("ContextualWisdomLab/example", [moved_pr]) + + from scripts.ci.review_admission_controller import load_state_file + + record = next(iter(load_state_file(gate.state_path).records.values())) + assert record.status == "stale" + + +def test_reconcile_keeps_lease_dispatched_while_strix_is_still_running(tmp_path): + """A lease for an in-flight, same-head scan is neither completed nor retired.""" + gate = sched.SchedulerAdmissionGate( + tmp_path / "admission.json", sequence=92, dispatch_budget=1 + ) + pr = make_pr( + number=7, + headRefOid="a" * 40, + statusCheckRollup={ + "contexts": {"nodes": [strix_check(status="IN_PROGRESS", conclusion="")]} + }, + ) + assert gate.admit("strix", "ContextualWisdomLab/example", pr) + gate.reconcile("ContextualWisdomLab/example", [pr]) + + from scripts.ci.review_admission_controller import load_state_file + + record = next(iter(load_state_file(gate.state_path).records.values())) + assert record.status == "dispatched" + + def test_reconcile_releases_strix_lease_when_no_run_was_created(tmp_path): gate = sched.SchedulerAdmissionGate( tmp_path / "admission.json", sequence=90, dispatch_budget=1 diff --git a/tests/test_review_admission_controller.py b/tests/test_review_admission_controller.py index fcb2885144..ce83f13918 100644 --- a/tests/test_review_admission_controller.py +++ b/tests/test_review_admission_controller.py @@ -1,9 +1,11 @@ import json +import os import threading from concurrent.futures import ThreadPoolExecutor import pytest +from scripts.ci import review_admission_controller as controller from scripts.ci.review_admission_controller import ( ADMISSION_PERMISSIONS, WORKER_BOUNDARIES, @@ -249,3 +251,190 @@ def test_budget_counts_active_leases_and_stale_heads_cannot_poison_sequence() -> dispatch_budget=1, ) assert retried.dispatches[0].request.sequence == 2 + + +@pytest.mark.parametrize( + ("changes", "error", "message"), + ( + ({"sequence": True}, TypeError, "sequence must be an integer"), + ({"pull_request": 0}, ValueError, "pull request must be positive"), + ({"head_sha": "short"}, ValueError, "head must be a full Git SHA"), + ({"sequence": 0}, ValueError, "sequence must be positive"), + ), +) +def test_request_rejects_each_invalid_scalar(changes, error, message) -> None: + values = { + "repository": "ContextualWisdomLab/example", + "pull_request": 7, + "head_sha": HEAD_1, + "component": "opencode", + "sequence": 1, + } + values.update(changes) + with pytest.raises(error, match=message): + AdmissionRequest.create(**values) + + +@pytest.mark.parametrize( + ("payload", "error", "message"), + ( + ([], TypeError, "must be an object"), + ({"records": []}, TypeError, "invalid collections"), + ( + {"records": {"bad": {"request": {}, "extra": 1}}, "latest_sequences": {}}, + ValueError, + "invalid durable admission record", + ), + ( + { + "records": { + "bad": { + "request": { + "repository": "ContextualWisdomLab/example", + "pull_request": 7, + }, + "status": "queued", + } + }, + "latest_sequences": {}, + }, + ValueError, + "invalid durable admission request", + ), + ), +) +def test_state_json_rejects_malformed_top_level_shapes(payload, error, message) -> None: + with pytest.raises(error, match=message): + ControllerState.from_json(json.dumps(payload)) + + +def test_state_json_rejects_non_string_record_identity(monkeypatch) -> None: + monkeypatch.setattr( + controller.json, + "loads", + lambda _serialized: {"records": {1: {}}, "latest_sequences": {}}, + ) + with pytest.raises(TypeError, match="invalid shape"): + ControllerState.from_json("ignored") + + +def test_state_json_rejects_identity_status_sequence_and_regression() -> None: + item = request("opencode", HEAD_1, 1) + + def encoded(identity=item.identity, status="queued", latest=None, record=item): + return json.dumps( + { + "records": { + identity: { + "request": { + "repository": record.repository, + "pull_request": record.pull_request, + "head_sha": record.head_sha, + "component": record.component, + "sequence": record.sequence, + }, + "status": status, + } + }, + "latest_sequences": latest + if latest is not None + else {item.stream: 1}, + } + ) + + with pytest.raises(ValueError, match="invalid durable admission record"): + ControllerState.from_json(encoded(identity="wrong")) + with pytest.raises(ValueError, match="invalid durable admission record"): + ControllerState.from_json(encoded(status="unknown")) + with pytest.raises(ValueError, match="invalid durable admission sequence"): + ControllerState.from_json(encoded(latest={item.stream: True})) + regressed = request("opencode", HEAD_1, 2) + with pytest.raises(ValueError, match="sequence regressed"): + ControllerState.from_json( + encoded( + identity=regressed.identity, + latest={regressed.stream: 1}, + record=regressed, + ) + ) + with pytest.raises(ValueError, match="sequence is inconsistent"): + ControllerState.from_json(encoded(latest={item.stream: 2})) + + +def test_state_file_rejects_corruption_symlinks_and_nonregular_paths(tmp_path) -> None: + corrupt = tmp_path / "corrupt.json" + corrupt.write_text("{", encoding="utf-8") + with pytest.raises(ValueError, match="corrupt and has no backup"): + load_state_file(corrupt) + + invalid_utf8 = tmp_path / "invalid.json" + invalid_utf8.write_bytes(b"\xff") + with pytest.raises(ValueError, match="not UTF-8"): + controller._read_state(invalid_utf8) + + with pytest.raises(ValueError, match="not a regular file"): + controller._open_regular_nofollow(tmp_path, os.O_RDONLY) + + state_path = tmp_path / "state.json" + backup = tmp_path / "state.json.bak" + backup.symlink_to(corrupt) + with pytest.raises(ValueError, match="backup must not be a symlink"): + load_state_file(state_path) + + atomic_link = tmp_path / "atomic.json" + atomic_link.symlink_to(corrupt) + with pytest.raises(ValueError, match="state path must not be a symlink"): + controller._atomic_write(atomic_link, "{}") + + lock_link = tmp_path / "locked.json.lock" + lock_link.symlink_to(corrupt) + with pytest.raises(ValueError, match="lock must not be a symlink"): + update_state_file(tmp_path / "locked.json", lambda state: state) + + +def test_update_and_dispatch_reject_invalid_transitions(tmp_path) -> None: + with pytest.raises(TypeError, match="must return ControllerState"): + update_state_file(tmp_path / "state.json", lambda state: object()) + with pytest.raises(ValueError, match="budget must not be negative"): + plan_dispatches(ControllerState.empty(), [], live_heads={}, dispatch_budget=-1) + + item = request("opencode", HEAD_2, 2) + lease = DispatchLease(item, WORKER_BOUNDARIES["opencode"]) + with pytest.raises(ValueError, match="active dispatch lease"): + complete_dispatch(ControllerState.empty(), lease, live_head=HEAD_2) + + +def test_new_head_stales_queued_predecessor_and_dispatch_rechecks_live_head() -> None: + old = request("opencode", HEAD_1, 1) + current = request("opencode", HEAD_2, 2) + state = ControllerState( + {old.identity: RequestRecord(old, "queued")}, + {old.stream: 1}, + ) + plan = plan_dispatches( + state, + [current], + live_heads={(current.repository, current.pull_request): HEAD_2}, + dispatch_budget=1, + ) + assert plan.state.records[old.identity].status == "stale" + + class MovingHeads(dict): + reads = 0 + + def get(self, key, default=None): + self.reads += 1 + return HEAD_2 if self.reads == 1 else HEAD_3 + + moved = plan_dispatches( + ControllerState.empty(), + [current], + live_heads=MovingHeads(), + dispatch_budget=1, + ) + assert moved.dispatches == () + assert moved.rejections[current.identity] == "stale_head" + + +def test_controller_self_test_executes_public_smoke_contract() -> None: + controller.self_test()