OrcaSlicer/scripts/tests/test_ci_macos_admission.py

#!/usr/bin/env python3
"""Tests for scripts/ci_macos_admission.py (stdlib unittest, no external deps).

Run from the repo root:  python -m unittest discover -s scripts/tests -v
"""

import datetime
import os
import sys
import unittest
import urllib.error

sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))

import ci_macos_admission as admission  # noqa: E402

UTC = datetime.timezone.utc
NOON = datetime.datetime(2026, 10, 10, 12, 0, tzinfo=UTC)


def job(name, status="completed", conclusion="success", labels=("macos-15",)):
    return {"name": name, "status": status, "conclusion": conclusion, "labels": list(labels)}


def gate(arch, status="completed", conclusion="success", priority=False):
    suffix = " (priority)" if priority else ""
    return job(f"build_macos_arch ({arch}) / Wait for a macOS runner{suffix}", status, conclusion,
               labels=("ubuntu-24.04",))


def build(arch, status="in_progress"):
    return job(f"build_macos_arch ({arch}) / Build Deps / Build OrcaSlicer / Build OrcaSlicer",
               status, None if status != "completed" else "success")


CHECK_CACHE = job("build_macos_arch (arm64) / Check Cache", labels=("ubuntu-24.04",))
UNIVERSAL_DONE = job("Build macOS Universal / Build OrcaSlicer")
TESTS_DONE = job("macOS arm64 / Unit Tests")
PUSH = {"id": 1, "event": "push", "head_branch": "main", "status": "in_progress"}
PR = {"id": 2, "event": "pull_request", "head_branch": "topic", "status": "in_progress"}


class MacosDoneTest(unittest.TestCase):
    def test_not_done_before_the_final_jobs_are_listed(self):
        self.assertFalse(admission.macos_done([CHECK_CACHE, build("arm64", "completed")]))

    def test_not_done_while_a_final_job_runs(self):
        running = job("macOS arm64 / Unit Tests", "in_progress", None)
        self.assertFalse(admission.macos_done([UNIVERSAL_DONE, running]))

    def test_done_when_the_final_jobs_finished(self):
        self.assertTrue(admission.macos_done([build("arm64", "completed"), UNIVERSAL_DONE, TESTS_DONE]))

    def test_done_when_the_final_jobs_were_skipped(self):
        # A skipped caller job is listed under its own name, without labels.
        skipped = [job("Build macOS Universal", conclusion="skipped", labels=()),
                   job("macOS arm64", conclusion="skipped", labels=())]
        self.assertTrue(admission.macos_done(skipped))


class RunDemandTest(unittest.TestCase):
    def test_push_reserves_before_its_macos_jobs_are_listed(self):
        self.assertEqual(admission.run_demand(PUSH, []), admission.RESERVE)
        self.assertEqual(admission.run_demand(PUSH, [CHECK_CACHE]), admission.RESERVE)

    def test_push_keeps_its_reserve_between_stages(self):
        jobs = [build("arm64", "completed"), build("x86_64", "completed")]
        self.assertEqual(admission.run_demand(PUSH, jobs), admission.RESERVE)

    def test_push_releases_when_macos_is_done(self):
        jobs = [build("arm64", "completed"), UNIVERSAL_DONE, TESTS_DONE]
        self.assertEqual(admission.run_demand(PUSH, jobs), 0)

    def test_queued_jobs_count_as_busy(self):
        jobs = [build("arm64", "queued")]
        self.assertEqual(admission.run_demand(PR, jobs), 1)

    def test_pull_request_holds_nothing_before_it_is_let_in(self):
        self.assertEqual(admission.run_demand(PR, [CHECK_CACHE, gate("arm64", "in_progress", None)]), 0)

    def test_one_arch_let_in_holds_a_runner_until_it_is_built(self):
        # Including before its first macOS job is listed, and between its stages.
        self.assertEqual(admission.run_demand(PR, [gate("arm64")]), 1)
        self.assertEqual(admission.run_demand(PR, [gate("arm64"), build("arm64")]), 1)

    def test_one_arch_let_in_holds_nothing_once_it_is_built(self):
        # Its other arch may be in the line behind a pull request waiting for this
        # runner, so holding it would deadlock.
        jobs = [gate("arm64"), build("arm64", "completed"), gate("x86_64", "pending", None)]
        self.assertEqual(admission.run_demand(PR, jobs), 0)

    def test_one_arch_let_in_holds_nothing_once_it_failed(self):
        failed = job("build_macos_arch (arm64) / Build Deps / Build Deps", conclusion="failure")
        self.assertEqual(admission.run_demand(PR, [gate("arm64"), failed]), 0)

    def test_pull_request_keeps_its_runners_for_its_later_jobs(self):
        # Both arch builds are done and the universal build and tests are not
        # listed yet: they will need both runners without passing the line.
        jobs = [gate("arm64"), gate("x86_64"), build("arm64", "completed"), build("x86_64", "completed")]
        self.assertEqual(admission.run_demand(PR, jobs), 2)

    def test_pull_request_releases_when_macos_is_done(self):
        jobs = [gate("arm64"), gate("x86_64"), UNIVERSAL_DONE, TESTS_DONE]
        self.assertEqual(admission.run_demand(PR, jobs), 0)

    def test_pull_request_without_the_line_holds_what_it_runs(self):
        # A run started from a workflow without the line.
        self.assertEqual(admission.run_demand(PR, [build("arm64"), build("x86_64", "queued")]), 2)

    def test_finished_and_non_macos_jobs_are_not_counted(self):
        jobs = [build("arm64", "completed"), CHECK_CACHE,
                job("build_linux (ubuntu-24.04) / Check Cache", "in_progress", None, ("ubuntu-24.04",))]
        self.assertEqual(admission.run_demand(PR, jobs), 0)


class PriorityTest(unittest.TestCase):
    WAITING = [gate("arm64", "in_progress", None, priority=True), gate("x86_64", "pending", None, priority=True)]

    def test_normal_line_yields_to_a_waiting_priority_run(self):
        self.assertEqual(admission.run_demand(PR, self.WAITING, yield_to_priority=True), admission.RESERVE)

    def test_priority_line_does_not_count_priority_runs_behind_it(self):
        self.assertEqual(admission.run_demand(PR, self.WAITING), 0)

    def test_a_waiting_normal_run_is_not_yielded_to(self):
        waiting = [gate("arm64", "in_progress", None), gate("x86_64", "pending", None)]
        self.assertEqual(admission.run_demand(PR, waiting, yield_to_priority=True), 0)

    def test_a_priority_run_half_let_in_still_counts_its_waiting_arch(self):
        jobs = [gate("arm64", priority=True), gate("x86_64", "pending", None, priority=True)]
        self.assertEqual(admission.run_demand(PR, jobs), 1)
        self.assertEqual(admission.run_demand(PR, jobs, yield_to_priority=True), 2)

    def test_skipped_priority_jobs_of_other_platforms_are_not_waiting(self):
        jobs = [job("build_linux (ubuntu-24.04) / Wait for a macOS runner (priority)",
                    conclusion="skipped", labels=("ubuntu-24.04",))]
        self.assertEqual(admission.run_demand(PR, jobs, yield_to_priority=True), 0)

    def test_measure_from_each_line(self):
        api = FakeApi({("runs", "pending"): [dict(PR, status="pending")], jobs_path(2): self.WAITING})
        total, lines, _ = admission.measure(api, "o/r", "build_all.yml", 99, NOON)
        self.assertEqual(total, admission.RESERVE)
        self.assertIn("priority, waiting", lines[0])
        total, _, _ = admission.measure(api, "o/r", "build_all.yml", 99, NOON, priority=True)
        self.assertEqual(total, 0)

    def test_own_waiting_run_is_not_counted(self):
        api = FakeApi({("runs", "pending"): [dict(PR, status="pending")], jobs_path(2): self.WAITING})
        total, _, need = admission.measure(api, "o/r", "build_all.yml", 2, NOON, arch="arm64")
        self.assertEqual((total, need), (0, 1))


class FakeApi:
    """Answers GET requests from a dict of path -> list of pages (or one body)."""

    def __init__(self, responses):
        self.responses = responses
        self.calls = []

    def get(self, path, **params):
        self.calls.append((path, params))
        if path.endswith("/runs") and "status" in params:
            runs = self.responses.get(("runs", params["status"]), [])
            start = (params["page"] - 1) * params["per_page"]
            return {"total_count": len(runs), "workflow_runs": runs[start:start + params["per_page"]]}
        if path.endswith("/runs"):
            return {"workflow_runs": self.responses.get(("runs", params.get("event")), [])[:1]}
        return {"jobs": self.responses[path]}


def jobs_path(run_id):
    return f"repos/o/r/actions/runs/{run_id}/jobs"


class MeasureTest(unittest.TestCase):
    def test_sums_in_progress_and_queued_runs(self):
        queued_pr = dict(PR, id=3, status="queued")
        api = FakeApi({
            ("runs", "in_progress"): [PUSH, PR],
            ("runs", "queued"): [queued_pr],
            jobs_path(1): [CHECK_CACHE],
            jobs_path(2): [gate("arm64"), build("arm64")],
            jobs_path(3): [gate("arm64"), gate("x86_64"), build("arm64", "queued")],
        })
        total, lines, _ = admission.measure(api, "o/r", "build_all.yml", 99, NOON)
        self.assertEqual(total, admission.RESERVE + 1 + admission.RESERVE)
        self.assertEqual(len(lines), 3)

    def test_counts_a_pending_run_that_holds_runners(self):
        # One of its jobs waits in a concurrency group while its macOS builds run.
        pending_pr = dict(PR, status="pending")
        api = FakeApi({("runs", "pending"): [pending_pr], jobs_path(2): [gate("arm64"), build("arm64")]})
        total, _, _ = admission.measure(api, "o/r", "build_all.yml", 99, NOON)
        self.assertEqual(total, 1)

    def test_a_push_waiting_behind_another_holds_nothing(self):
        api = FakeApi({("runs", "pending"): [dict(PUSH, status="pending")], jobs_path(1): []})
        total, _, _ = admission.measure(api, "o/r", "build_all.yml", 99, NOON)
        self.assertEqual(total, 0)

    def test_reads_every_page_of_runs(self):
        runs = [dict(PR, id=i) for i in range(150)]
        responses = {("runs", "in_progress"): runs}
        responses.update({jobs_path(i): [gate("arm64")] for i in range(150)})
        total, _, _ = admission.measure(FakeApi(responses), "o/r", "build_all.yml", 999, NOON)
        self.assertEqual(total, 150)

    def test_reserves_for_the_nightly_until_its_run_appears(self):
        due = datetime.datetime(2026, 10, 10, 2, 10, tzinfo=UTC)
        yesterday = {"created_at": "2026-10-09T02:20:00Z"}
        total, lines, _ = admission.measure(
            FakeApi({("runs", "schedule"): [yesterday]}), "OrcaSlicer/OrcaSlicer", "build_all.yml", 99, due)
        self.assertEqual(total, admission.RESERVE)
        self.assertIn("nightly", lines[0])

    def test_no_nightly_reserve_once_its_run_exists(self):
        due = datetime.datetime(2026, 10, 10, 2, 30, tzinfo=UTC)
        today = {"created_at": "2026-10-10T02:21:00Z"}
        total, _, _ = admission.measure(
            FakeApi({("runs", "schedule"): [today]}), "OrcaSlicer/OrcaSlicer", "build_all.yml", 99, due)
        self.assertEqual(total, 0)

    def test_no_nightly_reserve_in_a_fork(self):
        due = datetime.datetime(2026, 10, 10, 2, 10, tzinfo=UTC)
        total, _, _ = admission.measure(FakeApi({}), "fork/OrcaSlicer", "build_all.yml", 99, due)
        self.assertEqual(total, 0)

    def test_counts_a_run_in_both_lists_once(self):
        api = FakeApi({
            ("runs", "in_progress"): [PUSH],
            ("runs", "queued"): [dict(PUSH, status="queued")],
            jobs_path(1): [],
        })
        total, _, _ = admission.measure(api, "o/r", "build_all.yml", 99, NOON)
        self.assertEqual(total, admission.RESERVE)

    def own(self, jobs, arch):
        api = FakeApi({("runs", "in_progress"): [PR], jobs_path(2): jobs})
        total, _, need = admission.measure(api, "o/r", "build_all.yml", 2, NOON, arch=arch)
        return total, need

    def test_first_arch_needs_one(self):
        self.assertEqual(self.own([gate("arm64", "in_progress", None)], "arm64"), (0, 1))

    def test_second_arch_needs_one_while_the_first_builds(self):
        jobs = [gate("arm64"), build("arm64"), gate("x86_64", "in_progress", None)]
        self.assertEqual(self.own(jobs, "x86_64"), (1, 1))

    def test_second_arch_needs_both_once_the_first_is_built(self):
        # Letting it in makes the run hold both for its universal build and tests.
        jobs = [gate("arm64"), build("arm64", "completed"), gate("x86_64", "in_progress", None)]
        self.assertEqual(self.own(jobs, "x86_64"), (0, 2))

    def test_a_run_whose_other_arch_waits_behind_this_one_does_not_block_it(self):
        # The line is A-arm64, B-arm64, A-x86_64. A's arm64 is built and its
        # x86_64 is pending behind B, so A must not hold a runner B waits for.
        other = dict(PR, id=1, status="pending")
        api = FakeApi({
            ("runs", "pending"): [other, dict(PR, status="pending")],
            jobs_path(1): [gate("arm64"), build("arm64", "completed"), gate("x86_64", "pending", None)],
            jobs_path(2): [gate("arm64", "in_progress", None)],
        })
        total, _, need = admission.measure(api, "o/r", "build_all.yml", 2, NOON, arch="arm64")
        self.assertEqual((total, need), (0, 1))

    def test_no_nightly_lookup_outside_its_window(self):
        api = FakeApi({})
        admission.measure(api, "o/r", "build_all.yml", 99, NOON)
        self.assertFalse(any("event" in params for _, params in api.calls))


class WaitTest(unittest.TestCase):
    def run_wait(self, results, limit=5, wait_minutes=60, poll_seconds=120):
        results = iter(results)
        now = [0.0]

        def measure_now():
            result = next(results)
            if isinstance(result, Exception):
                raise result
            total, need = result if isinstance(result, tuple) else (result, 1)
            return total, [], need

        def sleep(seconds):
            now[0] += seconds

        reason = admission.wait(measure_now, limit, wait_minutes, poll_seconds,
                                clock=lambda: now[0], sleep=sleep, log=lambda _: None)
        return reason, now[0]

    def test_lets_in_when_a_runner_fits(self):
        self.assertEqual(self.run_wait([4]), ("a runner is free", 0))

    def test_waits_while_full(self):
        self.assertEqual(self.run_wait([5, 6, 4]), ("a runner is free", 240))

    def test_waits_for_what_this_arch_needs(self):
        self.assertEqual(self.run_wait([(4, 2), (3, 2)]), ("a runner is free", 120))

    def test_lets_in_after_repeated_api_errors(self):
        error = urllib.error.URLError("rate limited")
        reason, _ = self.run_wait([error, error, error])
        self.assertEqual(reason, "the queue could not be read")

    def test_a_good_read_resets_the_error_count(self):
        error = urllib.error.URLError("rate limited")
        reason, _ = self.run_wait([error, error, 5, error, error, 3])
        self.assertEqual(reason, "a runner is free")

    def test_lets_in_at_the_time_limit(self):
        reason, elapsed = self.run_wait([5] * 100, wait_minutes=10, poll_seconds=120)
        self.assertEqual(reason, "it waited 10 minutes")
        self.assertLess(elapsed, 600)


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