Mac mini GitHub Actions Runner：接入与恢复验收

name: Mac mini queue check
run-name: queue-${{ inputs.case_name }}-${{ github.run_number }}

on:
  workflow_dispatch:
    inputs:
      case_name:
        description: Queue acceptance case
        required: true
        type: choice
        default: A
        options: [A, B, C]

permissions: {}

concurrency:
  group: mac-mini-queue-check
  queue: max
  cancel-in-progress: false

jobs:
  observe:
    if: >-
      github.event_name == 'workflow_dispatch' &&
      github.event.repository.private == true &&
      github.ref == format('refs/heads/{0}',
      github.event.repository.default_branch)
    runs-on: [self-hosted, macOS, ARM64, mac-mini-ci]
    timeout-minutes: 5
    env:
      QUEUE_CASE: ${{ inputs.case_name }}
    steps:
      - name: Observe queue entry and completion
        timeout-minutes: 4
        shell: bash
        run: |
          set -euo pipefail
          test "$(uname -s)" = Darwin
          test "$(uname -m)" = arm64
          printf 'case=%s start=' "$QUEUE_CASE"
          date -u +'%Y-%m-%dT%H:%M:%SZ'
          case "$QUEUE_CASE" in
            A) sleep 180 ;;
            B|C) sleep 5 ;;
            *) exit 2 ;;
          esac
          printf 'case=%s finish=' "$QUEUE_CASE"
          date -u +'%Y-%m-%dT%H:%M:%SZ'

"""Offline snapshot triage, not GitHub scheduling or a security sandbox."""
import json
import sys
from pathlib import Path

REQUIRED = frozenset({"self-hosted", "macos", "arm64", "mac-mini-ci"})


def unique_object(pairs):
    """Reject duplicate decoded member names in every JSON object."""
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("duplicate JSON field: " + repr(key))
        result[key] = value
    return result


def parse_snapshot(raw):
    return json.loads(raw, object_pairs_hook=unique_object)


def check(snapshot):
    if not isinstance(snapshot, dict):
        raise ValueError("snapshot must be an object")
    required = {"private", "trusted", "event", "ref", "default_branch", "runners"}
    if not required <= snapshot.keys():
        raise ValueError("missing snapshot fields")
    for key in ("private", "trusted"):
        if type(snapshot[key]) is not bool:
            raise ValueError(key + " must be boolean")
    for key in ("event", "ref", "default_branch"):
        if not isinstance(snapshot[key], str) or not snapshot[key]:
            raise ValueError(key + " must be a nonempty string")
    if not isinstance(snapshot["runners"], list):
        raise ValueError("runners must be an array")
    policy = []
    if not snapshot["private"]:
        policy.append("repository_not_private")
    if not snapshot["trusted"]:
        policy.append("code_not_trusted")
    if snapshot["event"] != "workflow_dispatch":
        policy.append("event_not_manual")
    expected_ref = "refs/heads/" + snapshot["default_branch"]
    if snapshot["ref"] != expected_ref:
        policy.append("ref_not_default_branch")
    rows = []
    seen = set()
    for runner in snapshot["runners"]:
        if not isinstance(runner, dict):
            raise ValueError("runner must be an object")
        fields = {"name", "status", "busy", "allowed", "labels"}
        if not fields <= runner.keys():
            raise ValueError("missing runner fields")
        name = runner["name"]
        if not isinstance(name, str) or not name or name in seen:
            raise ValueError("runner name must be nonempty and unique")
        seen.add(name)
        if runner["status"] not in ("online", "offline"):
            raise ValueError("status must be online or offline")
        for key in ("busy", "allowed"):
            if type(runner[key]) is not bool:
                raise ValueError(key + " must be boolean")
        labels = runner["labels"]
        if not isinstance(labels, list) or not labels:
            raise ValueError("labels must be a nonempty array")
        if any(not isinstance(x, str) or not x for x in labels):
            raise ValueError("label must be a nonempty string")
        missing = sorted(REQUIRED - {x.lower() for x in labels})
        reasons = list(policy)
        if not runner["allowed"]:
            reasons.append("access_not_allowed")
        if missing:
            reasons.append("missing_labels:" + ",".join(missing))
        if runner["status"] != "online":
            reasons.append("offline")
        if runner["busy"]:
            reasons.append("busy")
        rows.append({"name": name, "candidate": not reasons, "reasons": reasons})
    return {
        "scope": "synthetic snapshot only; no scheduler or live API validation",
        "candidates": [row["name"] for row in rows if row["candidate"]],
        "runners": rows,
    }


def main(argv=None):
    args = sys.argv[1:] if argv is None else argv
    if len(args) != 1:
        print("usage: python3 route_check.py snapshot.json", file=sys.stderr)
        return 2
    try:
        raw = Path(args[0]).read_text(encoding="utf-8")
        report = check(parse_snapshot(raw))
    except (OSError, ValueError, TypeError) as error:
        print("input error: " + str(error), file=sys.stderr)
        return 2
    print(json.dumps(report, indent=2, ensure_ascii=False))
    return 0 if report["candidates"] else 1


if __name__ == "__main__":
    raise SystemExit(main())

{
  "private": true,
  "trusted": true,
  "event": "workflow_dispatch",
  "ref": "refs/heads/main",
  "default_branch": "main",
  "runners": [
    {
      "name": "mini-idle",
      "status": "online",
      "busy": false,
      "allowed": true,
      "labels": ["self-hosted", "macOS", "ARM64", "mac-mini-ci"]
    },
    {
      "name": "mini-busy",
      "status": "online",
      "busy": true,
      "allowed": true,
      "labels": ["self-hosted", "macOS", "ARM64", "mac-mini-ci"]
    },
    {
      "name": "intel-offline",
      "status": "offline",
      "busy": false,
      "allowed": true,
      "labels": ["self-hosted", "macOS", "X64", "mac-mini-ci"]
    }
  ]
}

import copy
import json
import tempfile
import unittest
from contextlib import redirect_stderr, redirect_stdout
from io import StringIO
from pathlib import Path

from route_check import check, main, parse_snapshot

BASE = json.loads(Path(__file__).with_name("snapshot.json").read_text())


class RoutingTests(unittest.TestCase):
    def setUp(self):
        self.data = copy.deepcopy(BASE)

    def test_fixture(self):
        result = check(self.data)
        self.assertEqual(result["candidates"], ["mini-idle"])
        self.assertEqual(result["runners"][1]["reasons"], ["busy"])
        self.assertEqual(result["runners"][2]["reasons"],
                         ["missing_labels:arm64", "offline"])

    def test_labels_case_insensitive(self):
        self.data["runners"][0]["labels"] = [
            "SELF-HOSTED", "MACOS", "arm64", "MAC-MINI-CI"]
        self.assertEqual(check(self.data)["candidates"], ["mini-idle"])

    def test_all_labels_required(self):
        self.data["runners"][0]["labels"].remove("mac-mini-ci")
        self.assertEqual(check(self.data)["candidates"], [])

    def test_busy(self):
        self.data["runners"][0]["busy"] = True
        self.assertEqual(check(self.data)["candidates"], [])

    def test_offline(self):
        self.data["runners"][0]["status"] = "offline"
        self.assertEqual(check(self.data)["candidates"], [])

    def test_access(self):
        self.data["runners"][0]["allowed"] = False
        self.assertEqual(check(self.data)["candidates"], [])

    def test_public_repo(self):
        self.data["private"] = False
        self.assertEqual(check(self.data)["candidates"], [])

    def test_untrusted(self):
        self.data["trusted"] = False
        self.assertEqual(check(self.data)["candidates"], [])

    def test_pr_event(self):
        self.data["event"] = "pull_request"
        self.assertEqual(check(self.data)["candidates"], [])

    def test_other_branch(self):
        self.data["ref"] = "refs/heads/feature"
        self.assertEqual(check(self.data)["candidates"], [])

    def test_other_default_branch(self):
        self.data["default_branch"] = "trunk"
        self.data["ref"] = "refs/heads/trunk"
        self.assertEqual(check(self.data)["candidates"], ["mini-idle"])

    def test_string_boolean_rejected(self):
        self.data["runners"][0]["busy"] = "false"
        with self.assertRaises(ValueError):
            check(self.data)

    def test_invalid_status(self):
        self.data["runners"][0]["status"] = "idle"
        with self.assertRaises(ValueError):
            check(self.data)

    def test_missing_field(self):
        del self.data["trusted"]
        with self.assertRaises(ValueError):
            check(self.data)

    def test_duplicate_name(self):
        self.data["runners"][1]["name"] = "mini-idle"
        with self.assertRaises(ValueError):
            check(self.data)

    def test_empty_runners(self):
        self.data["runners"] = []
        self.assertEqual(check(self.data)["candidates"], [])

    def test_invalid_label(self):
        self.data["runners"][0]["labels"] = [False]
        with self.assertRaises(ValueError):
            check(self.data)

    def test_cli_success(self):
        with redirect_stdout(StringIO()) as out:
            code = main([str(Path(__file__).with_name("snapshot.json"))])
        self.assertEqual(code, 0)
        self.assertEqual(json.loads(out.getvalue())["candidates"], ["mini-idle"])

    def test_cli_no_candidate(self):
        self.data["trusted"] = False
        with tempfile.TemporaryDirectory() as tmp:
            file = Path(tmp) / "snapshot.json"
            file.write_text(json.dumps(self.data))
            with redirect_stdout(StringIO()):
                self.assertEqual(main([str(file)]), 1)

    def test_cli_bad_json(self):
        with tempfile.TemporaryDirectory() as tmp:
            file = Path(tmp) / "bad.json"
            file.write_text("{")
            with redirect_stderr(StringIO()):
                self.assertEqual(main([str(file)]), 2)

    def assert_cli_input_error(self, raw, field):
        with tempfile.TemporaryDirectory() as tmp:
            file = Path(tmp) / "snapshot.json"
            file.write_text(raw, encoding="utf-8")
            with redirect_stdout(StringIO()) as out, redirect_stderr(StringIO()) as err:
                self.assertEqual(main([str(file)]), 2)
        self.assertEqual(out.getvalue(), "")
        self.assertIn("duplicate JSON field: " + repr(field), err.getvalue())

    def test_duplicate_top_level_boolean_rejected(self):
        for pair in ('"trusted": false, "trusted": true',
                     '"trusted": true, "trusted": false'):
            with self.subTest(pair=pair):
                raw = json.dumps(self.data).replace('"trusted": true', pair, 1)
                self.assert_cli_input_error(raw, "trusted")

    def test_duplicate_runner_boolean_rejected(self):
        for pair in ('"busy": true, "busy": false',
                     '"busy": false, "busy": true'):
            with self.subTest(pair=pair):
                raw = json.dumps(self.data).replace('"busy": false', pair, 1)
                self.assert_cli_input_error(raw, "busy")

    def test_duplicate_same_value_rejected(self):
        raw = json.dumps(self.data)
        raw = raw.replace('"trusted": true', '"trusted": true, "trusted": true', 1)
        self.assert_cli_input_error(raw, "trusted")

    def test_escaped_duplicate_name_rejected(self):
        raw = json.dumps(self.data)
        raw = raw.replace('"busy": false', r'"busy": true, "b\u0075sy": false', 1)
        self.assert_cli_input_error(raw, "busy")

    def test_duplicate_unchecked_nested_field_rejected(self):
        self.data["note"] = {"source": "manual"}
        raw = json.dumps(self.data)
        raw = raw.replace('"source": "manual"', '"source": "manual", "source": "copy"', 1)
        self.assert_cli_input_error(raw, "source")

    def test_same_field_in_separate_objects_is_valid(self):
        self.assertEqual(parse_snapshot(json.dumps(self.data)), self.data)
        self.assertEqual(check(parse_snapshot(json.dumps(self.data)))["candidates"],
                         ["mini-idle"])


if __name__ == "__main__":
    unittest.main()