fix(ci): close the admission-controller coverage/docstring gap on main

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

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()

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]

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