mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-08 11:46:35 +00:00
kevinveenbirkenbachandClaude Opus 5
677a63931c
Resolves 68 conflicted files. The governing rule was our structure with their
network semantics: the tor boolean becomes a network mode (clearnet | tor |
multi) throughout the variant layer, while this branch's architecture and mesh
axes, its extracted modules and its i18n content stay.
Where the two sides covered different ground, both are kept rather than one
picked:
- ca_injected answers a ca_client on the deployment's TLS mode OR on an SSO
issuer that resolves self-signed. 3a3e9ed2c9 records why: an issuer is not
an object store, so keeping only one side drops the other's coverage.
- get_merged_domains keeps the loud refusal when svc-net-tor is deployed
without a provisioned node, now gated on their tor_serves_host; their
node_onion_of returns an empty string there, which would have served every
onion-exclusive app on clearnet in silence.
- sso_cookie_domains stays the single source for oauth2-proxy's cookie scope.
Their per-network branch omitted the issuer domain on the onion arm, which
breaks the callback leg.
- The clearnet CSP batch carries --clearnet-timeout again; their batch dict
hardcoded zero while the role still passes the flag.
- The dashboard keeps its i18n content dir, image, version and design
resolution; their side reduced the logo sources to the bare service
provider values.
Their side wins where it is measurably better: the onion mirror rewrite is
gated per vhost rather than per app, the Playwright onion decision reads the
pass's own domain instead of the role's canonical one, and the matomo
reinjection and dashboard config render per network sibling.
Eight files were broken by the merge without ever carrying a conflict marker,
because git marks only the lines both sides touched:
- utils/cache/domains.py lost the onionize block that sat between two hunks
- roles/sys-front-inj-all/templates/body_filter.lua.j2 called it
- web-app-bluesky, keycloak sso_proxy: the get_entity_name filter was renamed
- trigger/__main__.py, playwright_coverage.py and its test: the deleted tor
module
- cli/administration/deploy/ci/timings.py read Label.tor
- plugins/filter/csp_filters.py called _aligned_url with four of six arguments
- tests/unit/python/utils/cache/test_domains.py referenced a renamed class
- tests/.../sys-ctl-hlth-csp/files/test_script.py mocked a function main no
longer calls
Verified: the variant layer (155 tests, 65 subtests), ca_injected (18),
the CI layer (175), the domains cache including the restored guard (32),
retrigger inputs (9), timings and the CSP health script (53). make
quality-high clears autoformat; its remaining findings are tracked separately.
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",
|
|
"network": "clearnet",
|
|
"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()
|