mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-08 19:56:53 +00:00
kevinveenbirkenbachandClaude Opus 5
5647a2d2bd
cfe20f0726and4915a401d4wrapped three statements more cautiously than the formatter wants. Without this the lint phase rewrites the same diff into the tree on every run. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
429 lines
17 KiB
Python
429 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import subprocess
|
|
import unittest
|
|
import unittest.mock as mock
|
|
from contextlib import redirect_stdout
|
|
from typing import ClassVar
|
|
|
|
from cli.administration.deploy.ci import gh, runs
|
|
from cli.administration.deploy.ci.trigger import __main__ as trigger
|
|
from cli.meta.ci import matrix
|
|
from tests.utils.ci.job_names import deploy_job_name
|
|
from tests.utils.ci.run_name import render
|
|
from utils.github.variant.pools import ARCHITECTURES, DISTROS, FILESYSTEMS
|
|
|
|
|
|
def _job(mode: str, app: str, conclusion: str, *, vpn: bool = False) -> dict:
|
|
return {
|
|
"name": deploy_job_name(mode, app, "0,1", vpn=vpn),
|
|
"status": "completed",
|
|
"conclusion": conclusion,
|
|
}
|
|
|
|
|
|
def _green(app: str) -> dict:
|
|
"""The deploy job that proves the ranking row ``app#0``."""
|
|
return {
|
|
"name": deploy_job_name("docker", app, "0"),
|
|
"status": "completed",
|
|
"conclusion": "success",
|
|
}
|
|
|
|
|
|
_JOBS = [
|
|
_job("docker", "web-app-x", "success"),
|
|
_job("swarm", "web-app-x", "failure"),
|
|
_job("docker", "web-app-y", "failure"),
|
|
_job("swarm", "web-app-y", "success"),
|
|
]
|
|
|
|
_AXES = f"%{DISTROS[0]}:{ARCHITECTURES[0]}"
|
|
"""The distro and architecture glyphs _JOBS carry, spelled back out as ASCII.
|
|
The filesystem is deliberately absent: a title states the assigned kind, not
|
|
the effective one, while the architecture is the machine the job really ran
|
|
on."""
|
|
|
|
_FAILED_TOKENS = (
|
|
f"web-app-x#0,1@swarm+clearnet{_AXES}&direct web-app-y#0,1@compose+clearnet{_AXES}"
|
|
)
|
|
"""What _JOBS failed at, exactly: x in swarm, y in compose, both on variant
|
|
shard 0,1, both on the clearnet and both on the same distro. Only the swarm
|
|
row states its mesh state, because the mesh axis covers no other mode."""
|
|
|
|
_RUN_URL = "https://github.com/o/r/actions/runs/55" # nocheck: url
|
|
_SOURCE_CONFIG = {
|
|
"distros": "arch centos",
|
|
"mode": "swarm",
|
|
"lifecycles": "stable",
|
|
"filesystem": "btrfs",
|
|
"chunk_gate": "false",
|
|
"workspace": "false",
|
|
}
|
|
_SOURCE_RUN = {"jobs": _JOBS, "displayTitle": render(_SOURCE_CONFIG)}
|
|
|
|
|
|
class TestTriggerMain(unittest.TestCase):
|
|
def _run(
|
|
self,
|
|
argv: list[str],
|
|
run: dict | None = None,
|
|
inputs: dict[str, str] | None = None,
|
|
entries: list[dict] | None = None,
|
|
) -> tuple[int, list]:
|
|
calls: list[tuple] = []
|
|
buf = io.StringIO()
|
|
with (
|
|
mock.patch.object(gh, "current_branch", return_value="feature/x"),
|
|
mock.patch.object(gh, "resolve_repo", return_value="o/r"),
|
|
mock.patch.object(runs, "find_last_deploy_run", return_value=run),
|
|
mock.patch.object(runs, "inputs_from_jobs", return_value=inputs or {}),
|
|
mock.patch.object(matrix, "entries_of", return_value=entries or []),
|
|
mock.patch.object(
|
|
runs,
|
|
"dispatch_workflow",
|
|
side_effect=lambda wf, ref, wl="", priority="", config=None, repo=None: (
|
|
calls.append((wf, ref, wl, priority, config, repo))
|
|
),
|
|
),
|
|
redirect_stdout(buf),
|
|
):
|
|
rc = trigger.main(argv)
|
|
return rc, calls
|
|
|
|
def test_default_triggers_all(self) -> None:
|
|
rc, calls = self._run([])
|
|
self.assertEqual(rc, 0)
|
|
self.assertEqual(
|
|
calls, [("entry-manual-steer.yml", "feature/x", "__ALL__", "", {}, "o/r")]
|
|
)
|
|
|
|
def test_priority_is_dispatched_with_the_axis_overrides(self) -> None:
|
|
rc, calls = self._run(
|
|
[
|
|
"--priority",
|
|
"web-app-b#1@compose web-app-a#0@swarm",
|
|
"--workspace",
|
|
"false",
|
|
"--chunk-gate",
|
|
"false",
|
|
"--filesystem",
|
|
"btrfs",
|
|
]
|
|
)
|
|
self.assertEqual(rc, 0)
|
|
self.assertEqual(calls[0][3], "web-app-a#0@swarm web-app-b#1@compose")
|
|
self.assertEqual(
|
|
calls[0][4],
|
|
{"workspace": "false", "chunk_gate": "false", "filesystem": "btrfs"},
|
|
)
|
|
|
|
def test_an_unusable_token_is_refused_before_anything_is_dispatched(self) -> None:
|
|
with mock.patch.object(
|
|
trigger.validate, "problems", return_value=(["bad token"], [])
|
|
):
|
|
rc, calls = self._run(["--priority", "web-app-a#9"])
|
|
self.assertEqual(rc, 1)
|
|
self.assertEqual(calls, [], "a refused line must not reach the workflow")
|
|
|
|
def test_priority_and_failed_cannot_both_build_the_line(self) -> None:
|
|
with self.assertRaises(SystemExit):
|
|
self._run(["--failed", "--priority", "web-app-a#0"])
|
|
|
|
def test_apps_explicit_list(self) -> None:
|
|
rc, calls = self._run(["--apps", "web-app-a web-app-b"])
|
|
self.assertEqual(rc, 0)
|
|
self.assertEqual(calls[0][2], "web-app-a web-app-b")
|
|
self.assertEqual(calls[0][3], "")
|
|
|
|
def test_failed_total_sends_priority_without_whitelist(self) -> None:
|
|
rc, calls = self._run(["--failed"], run={"_jobs": _JOBS})
|
|
self.assertEqual(rc, 0)
|
|
self.assertEqual(calls[0][2], "")
|
|
self.assertEqual(calls[0][3], _FAILED_TOKENS)
|
|
|
|
def test_a_leftover_scope_argument_no_longer_narrows_the_modes(self) -> None:
|
|
for scope in ("swarm", "compose", "total"):
|
|
with self.subTest(scope=scope):
|
|
_rc, calls = self._run(["--failed", scope], run={"_jobs": _JOBS})
|
|
self.assertEqual(calls[0][3], _FAILED_TOKENS)
|
|
|
|
def test_every_failed_mode_of_one_role_comes_back_separately(self) -> None:
|
|
both = [
|
|
_job("docker", "web-app-x", "failure"),
|
|
_job("swarm", "web-app-x", "failure"),
|
|
]
|
|
_rc, calls = self._run(["--failed"], run={"_jobs": both})
|
|
self.assertEqual(
|
|
calls[0][3],
|
|
f"web-app-x#0,1@compose+clearnet{_AXES}"
|
|
f" web-app-x#0,1@swarm+clearnet{_AXES}&direct",
|
|
)
|
|
|
|
def test_a_meshed_swarm_row_comes_back_meshed(self) -> None:
|
|
meshed = [_job("swarm", "web-app-x", "failure", vpn=True)]
|
|
_rc, calls = self._run(["--failed"], run={"_jobs": meshed})
|
|
self.assertEqual(calls[0][3], f"web-app-x#0,1@swarm+clearnet{_AXES}&vpn")
|
|
|
|
def test_never_deployed_priority_roles_join_the_retrigger(self) -> None:
|
|
source = {"jobs": _JOBS, "displayTitle": render(_SOURCE_CONFIG)}
|
|
calls: list = []
|
|
with (
|
|
mock.patch.object(gh, "current_branch", return_value="feature/x"),
|
|
mock.patch.object(gh, "resolve_repo", return_value="o/r"),
|
|
mock.patch.object(gh, "fetch_run", return_value=source),
|
|
mock.patch.object(
|
|
runs,
|
|
"inputs_from_jobs",
|
|
return_value={"priority": "web-app-x web-app-never"},
|
|
),
|
|
mock.patch.object(matrix, "entries_of", return_value=[]),
|
|
mock.patch.object(
|
|
runs,
|
|
"dispatch_workflow",
|
|
side_effect=lambda wf, ref, wl="", priority="", config=None, repo=None: (
|
|
calls.append(priority)
|
|
),
|
|
),
|
|
redirect_stdout(io.StringIO()),
|
|
):
|
|
rc = trigger.main(["--failed", "--run", _RUN_URL])
|
|
self.assertEqual(rc, 0)
|
|
self.assertEqual(
|
|
calls[0],
|
|
f"web-app-never {_FAILED_TOKENS}",
|
|
)
|
|
|
|
def test_an_unreadable_job_log_aborts_instead_of_dropping_the_priority(
|
|
self,
|
|
) -> None:
|
|
source = {
|
|
"jobs": _JOBS,
|
|
"displayTitle": render({**_SOURCE_CONFIG, "priority": "web-app-x"}),
|
|
}
|
|
with (
|
|
mock.patch.object(gh, "current_branch", return_value="feature/x"),
|
|
mock.patch.object(gh, "resolve_repo", return_value="o/r"),
|
|
mock.patch.object(gh, "fetch_run", return_value=source),
|
|
mock.patch.object(runs, "inputs_from_jobs", return_value={}),
|
|
redirect_stdout(io.StringIO()),
|
|
self.assertRaises(SystemExit),
|
|
):
|
|
trigger.main(["--failed", "--run", _RUN_URL])
|
|
|
|
def test_a_never_deployed_priority_role_alone_still_dispatches(self) -> None:
|
|
green = [
|
|
_job("docker", "web-app-x", "success"),
|
|
_job("swarm", "web-app-x", "success"),
|
|
]
|
|
source = {"jobs": green, "displayTitle": render({})}
|
|
calls: list = []
|
|
with (
|
|
mock.patch.object(gh, "current_branch", return_value="feature/x"),
|
|
mock.patch.object(gh, "resolve_repo", return_value="o/r"),
|
|
mock.patch.object(gh, "fetch_run", return_value=source),
|
|
mock.patch.object(
|
|
runs,
|
|
"inputs_from_jobs",
|
|
return_value={"priority": "web-app-x web-app-never"},
|
|
),
|
|
mock.patch.object(matrix, "entries_of", return_value=[]),
|
|
mock.patch.object(
|
|
runs,
|
|
"dispatch_workflow",
|
|
side_effect=lambda wf, ref, wl="", priority="", config=None, repo=None: (
|
|
calls.append(priority)
|
|
),
|
|
),
|
|
redirect_stdout(io.StringIO()),
|
|
):
|
|
rc = trigger.main(["--failed", "--run", _RUN_URL])
|
|
self.assertEqual(rc, 0)
|
|
self.assertEqual(calls[0], "web-app-never")
|
|
|
|
_RANKING: ClassVar[list[dict]] = [
|
|
{
|
|
"apps": f"web-app-{name}",
|
|
"variant": "0",
|
|
"mode": "compose",
|
|
"tor": "false",
|
|
"distro": DISTROS[0],
|
|
"filesystem": FILESYSTEMS[0],
|
|
"priority": "false",
|
|
}
|
|
for name in ("a", "b", "c")
|
|
]
|
|
|
|
def test_the_carried_offset_is_the_floor_for_the_new_one(self) -> None:
|
|
"""Without it the retrigger restarts at the head every time and
|
|
redeploys the stretch its predecessors already proved."""
|
|
_rc, calls = self._run(
|
|
["--failed"],
|
|
run={"_jobs": _JOBS},
|
|
inputs={"offset": "web-app-b#0"},
|
|
entries=self._RANKING,
|
|
)
|
|
self.assertEqual(calls[0][4]["offset"], "web-app-b#0")
|
|
|
|
def test_the_green_stretch_behind_the_carried_offset_moves_it_on(self) -> None:
|
|
_rc, calls = self._run(
|
|
["--failed"],
|
|
run={"_jobs": [*_JOBS, _green("web-app-b"), _green("web-app-c")]},
|
|
inputs={"offset": "web-app-b#0"},
|
|
entries=self._RANKING,
|
|
)
|
|
self.assertEqual(calls[0][4]["offset"], "web-app-c#0")
|
|
|
|
def test_failed_nothing_does_not_dispatch(self) -> None:
|
|
green = [
|
|
_job("docker", "web-app-x", "success"),
|
|
_job("swarm", "web-app-x", "success"),
|
|
]
|
|
rc, calls = self._run(["--failed"], run={"_jobs": green})
|
|
self.assertEqual(rc, 0)
|
|
self.assertEqual(calls, [])
|
|
|
|
def test_failed_with_run_url_uses_that_run(self) -> None:
|
|
calls: list = []
|
|
with (
|
|
mock.patch.object(gh, "current_branch", return_value="feature/x"),
|
|
mock.patch.object(gh, "resolve_repo", return_value="o/r"),
|
|
mock.patch.object(gh, "fetch_run", return_value=_SOURCE_RUN) as fetch,
|
|
mock.patch.object(runs, "find_last_deploy_run") as find_last,
|
|
mock.patch.object(matrix, "entries_of", return_value=[]),
|
|
mock.patch.object(
|
|
runs,
|
|
"dispatch_workflow",
|
|
side_effect=lambda wf, ref, wl="", priority="", config=None, repo=None: (
|
|
calls.append((priority, config))
|
|
),
|
|
),
|
|
redirect_stdout(io.StringIO()),
|
|
):
|
|
rc = trigger.main(["--failed", "--run", _RUN_URL])
|
|
self.assertEqual(rc, 0)
|
|
self.assertEqual(calls[0], (_FAILED_TOKENS, _SOURCE_CONFIG))
|
|
fetch.assert_called_once()
|
|
find_last.assert_not_called()
|
|
|
|
def test_failed_with_bare_run_id_resolves_against_branch_repo(self) -> None:
|
|
calls: list = []
|
|
with (
|
|
mock.patch.object(gh, "current_branch", return_value="feature/x"),
|
|
mock.patch.object(gh, "resolve_repo", return_value="o/r"),
|
|
mock.patch.object(gh, "fetch_run", return_value=_SOURCE_RUN) as fetch,
|
|
mock.patch.object(runs, "find_last_deploy_run") as find_last,
|
|
mock.patch.object(matrix, "entries_of", return_value=[]),
|
|
mock.patch.object(
|
|
runs,
|
|
"dispatch_workflow",
|
|
side_effect=lambda wf, ref, wl="", priority="", config=None, repo=None: (
|
|
calls.append((priority, config))
|
|
),
|
|
),
|
|
redirect_stdout(io.StringIO()),
|
|
):
|
|
rc = trigger.main(["--failed", "--run", "55"])
|
|
self.assertEqual(rc, 0)
|
|
self.assertEqual(calls[0], (_FAILED_TOKENS, _SOURCE_CONFIG))
|
|
fetch.assert_called_once_with("55", repo="o/r")
|
|
find_last.assert_not_called()
|
|
|
|
def test_apps_with_a_run_reproduces_its_configuration(self) -> None:
|
|
calls: list = []
|
|
with (
|
|
mock.patch.object(gh, "current_branch", return_value="feature/x"),
|
|
mock.patch.object(gh, "resolve_repo", return_value="o/r"),
|
|
mock.patch.object(gh, "fetch_run", return_value=_SOURCE_RUN),
|
|
mock.patch.object(matrix, "entries_of", return_value=[]),
|
|
mock.patch.object(
|
|
runs,
|
|
"dispatch_workflow",
|
|
side_effect=lambda wf, ref, wl="", priority="", config=None, repo=None: (
|
|
calls.append((wl, priority, config))
|
|
),
|
|
),
|
|
redirect_stdout(io.StringIO()),
|
|
):
|
|
rc = trigger.main(["--apps", "web-app-a", "--run", _RUN_URL])
|
|
self.assertEqual(rc, 0)
|
|
self.assertEqual(calls[0], ("web-app-a", "", _SOURCE_CONFIG))
|
|
|
|
def test_chunk_gate_override_replaces_the_carried_value(self) -> None:
|
|
calls: list = []
|
|
with (
|
|
mock.patch.object(gh, "current_branch", return_value="feature/x"),
|
|
mock.patch.object(gh, "resolve_repo", return_value="o/r"),
|
|
mock.patch.object(gh, "fetch_run", return_value=_SOURCE_RUN),
|
|
mock.patch.object(matrix, "entries_of", return_value=[]),
|
|
mock.patch.object(
|
|
runs,
|
|
"dispatch_workflow",
|
|
side_effect=lambda wf, ref, wl="", priority="", config=None, repo=None: (
|
|
calls.append(config)
|
|
),
|
|
),
|
|
redirect_stdout(io.StringIO()),
|
|
):
|
|
rc = trigger.main(["--failed", "--run", _RUN_URL, "--chunk-gate", "true"])
|
|
self.assertEqual(rc, 0)
|
|
self.assertEqual(calls[0], {**_SOURCE_CONFIG, "chunk_gate": "true"})
|
|
|
|
def test_chunk_gate_override_without_a_source_run(self) -> None:
|
|
rc, calls = self._run(["--chunk-gate", "false"])
|
|
self.assertEqual(rc, 0)
|
|
self.assertEqual(calls[0][4], {"chunk_gate": "false"})
|
|
|
|
def test_chunk_gate_rejects_other_values(self) -> None:
|
|
with self.assertRaises(SystemExit) as ctx, redirect_stdout(io.StringIO()):
|
|
trigger.main(["--chunk-gate", "maybe"])
|
|
self.assertEqual(ctx.exception.code, 2)
|
|
|
|
def test_failed_no_run_found(self) -> None:
|
|
rc, calls = self._run(["--failed"], run=None)
|
|
self.assertEqual(rc, 1)
|
|
self.assertEqual(calls, [])
|
|
|
|
def test_apps_and_failed_mutually_exclusive(self) -> None:
|
|
with self.assertRaises(SystemExit) as ctx, redirect_stdout(io.StringIO()):
|
|
trigger.main(["--failed", "--apps", "x"])
|
|
self.assertEqual(ctx.exception.code, 2)
|
|
|
|
|
|
class TestBranchRemote(unittest.TestCase):
|
|
def _resolve(self, upstream: str | None, push_default: str | None) -> str:
|
|
remotes = "fork\norigin"
|
|
|
|
def fake_run(args: list[str]) -> str:
|
|
if args[1] == "rev-parse":
|
|
if upstream is None:
|
|
raise subprocess.CalledProcessError(128, args)
|
|
return upstream
|
|
if args[1] == "remote" and len(args) == 2:
|
|
return remotes
|
|
if args[1] == "config":
|
|
if push_default is None:
|
|
raise subprocess.CalledProcessError(1, args)
|
|
return push_default
|
|
raise AssertionError(f"unexpected git call: {args}")
|
|
|
|
with mock.patch.object(gh, "_run", side_effect=fake_run):
|
|
return gh._branch_remote()
|
|
|
|
def test_tracking_remote_wins(self) -> None:
|
|
self.assertEqual(self._resolve("origin/main", "fork"), "origin")
|
|
|
|
def test_push_default_beats_origin_without_upstream(self) -> None:
|
|
self.assertEqual(self._resolve(None, "fork"), "fork")
|
|
|
|
def test_origin_when_no_upstream_and_no_push_default(self) -> None:
|
|
self.assertEqual(self._resolve(None, None), "origin")
|
|
|
|
def test_unknown_push_default_is_ignored(self) -> None:
|
|
self.assertEqual(self._resolve(None, "gone"), "origin")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|