Christian Manivong
06825cb504
CI / Tests, lint (push) Successful in 11s
Moving bump.py from base-os took its tests up to the one that checks base-os's own dependents.toml and cut everything after it: an odd new reference, a repo without the file yet, a dependent without a workflow, a compose pin with the registry variable, a repo the bot may not see. They are back; base-os's list check stays in base-os.
236 lines
9.2 KiB
Python
236 lines
9.2 KiB
Python
"""Moving a pin in the repos that use an image (#1308).
|
|
|
|
When an image improves, its repo opens (or updates) one pull request in every
|
|
repo that pins it, rewriting only the pin, and starts that repo's own nightly
|
|
run, which rebuilds on the new pin and verifies itself.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
|
|
import pytest
|
|
|
|
|
|
from image_tools.bump import Dependent, bump_all, load_dependents, open_or_update_pr, rewrite_pin
|
|
|
|
OLD = "registry.netork.io/netork/base-os:13-20261001.1@sha256:" + "a" * 64
|
|
NEW = "registry.netork.io/netork/base-os:13-20261020.1@sha256:" + "b" * 64
|
|
|
|
|
|
class TestRewritePin:
|
|
def test_every_from_line_moves(self):
|
|
text = f"FROM {OLD} AS build\nRUN x\nFROM {OLD}\n"
|
|
|
|
assert (
|
|
rewrite_pin(text, "netork/base-os", NEW) == f"FROM {NEW} AS build\nRUN x\nFROM {NEW}\n"
|
|
)
|
|
|
|
def test_a_compose_image_line_moves(self):
|
|
text = f"services:\n cache:\n image: {OLD}\n"
|
|
|
|
assert (
|
|
rewrite_pin(text, "netork/base-os", NEW) == f"services:\n cache:\n image: {NEW}\n"
|
|
)
|
|
|
|
def test_another_image_stays(self):
|
|
other = "registry.netork.io/netork/base-python:3.12.15-13-20261001.1@sha256:" + "c" * 64
|
|
text = f"FROM {other}\n"
|
|
|
|
assert rewrite_pin(text, "netork/base-os", NEW) == text
|
|
|
|
def test_an_image_whose_name_only_starts_alike_stays(self):
|
|
text = "FROM registry.netork.io/netork/base-os-extra:1@sha256:" + "d" * 64 + "\n"
|
|
|
|
assert rewrite_pin(text, "netork/base-os", NEW) == text
|
|
|
|
|
|
class FakeGitea:
|
|
def __init__(self, files: dict[str, str], prs: list[dict] | None = None, branches=()):
|
|
self.files = dict(files)
|
|
self.prs = list(prs or [])
|
|
self.branches = set(branches)
|
|
self.calls: list[tuple[str, str]] = []
|
|
self.missing_workflows: set[str] = set()
|
|
self.dispatched: list[str] = []
|
|
|
|
def __call__(self, method: str, path: str, body: dict | None = None):
|
|
self.calls.append((method, path))
|
|
if method == "GET" and path.count("/") == 3: # /repos/<owner>/<repo>
|
|
return {"full_name": path.split("/repos/", 1)[1]}
|
|
if method == "GET" and "/contents/" in path:
|
|
name = path.split("/contents/", 1)[1].split("?", 1)[0]
|
|
content = base64.b64encode(self.files[name].encode()).decode()
|
|
return {"content": content, "sha": f"sha-{name}"}
|
|
if method == "PUT" and "/contents/" in path:
|
|
name = path.split("/contents/", 1)[1]
|
|
self.files[name] = base64.b64decode(body["content"]).decode()
|
|
return {}
|
|
if method == "POST" and path.endswith("/branches"):
|
|
self.branches.add(body["new_branch_name"])
|
|
return {}
|
|
if method == "GET" and "/branches/" in path:
|
|
name = path.rsplit("/branches/", 1)[1]
|
|
if name not in self.branches:
|
|
raise FileNotFoundError(name)
|
|
return {}
|
|
if method == "GET" and "/pulls?" in path:
|
|
return [p for p in self.prs if p["state"] == "open"]
|
|
if method == "POST" and path.endswith("/pulls"):
|
|
pr = {"number": 7, "state": "open", "head": {"ref": body["head"]}, "body": body["body"]}
|
|
self.prs.append(pr)
|
|
return pr
|
|
if method == "POST" and path.endswith("/dispatches"):
|
|
workflow = path.split("/workflows/", 1)[1].split("/", 1)[0]
|
|
if workflow in self.missing_workflows:
|
|
raise FileNotFoundError(path)
|
|
self.dispatched.append(workflow)
|
|
return {}
|
|
if method == "PATCH" and "/pulls/" in path:
|
|
self.prs[0]["body"] = body["body"]
|
|
return self.prs[0]
|
|
raise AssertionError(f"unexpected {method} {path}")
|
|
|
|
|
|
DEP = Dependent(repo="NetOrk/base-python", files=("Dockerfile",), workflow="nightly.yml")
|
|
|
|
|
|
class TestPullRequest:
|
|
def test_it_opens_one_pr_with_only_the_pin_changed(self):
|
|
api = FakeGitea({"Dockerfile": f"FROM {OLD}\nRUN echo hi\n"})
|
|
|
|
pr = open_or_update_pr(api, DEP, "netork/base-os", NEW, "13-20261020.1", "report")
|
|
|
|
assert pr["number"] == 7
|
|
assert api.files["Dockerfile"] == f"FROM {NEW}\nRUN echo hi\n"
|
|
assert "chore/base-os-13-20261020.1" in api.branches
|
|
assert api.prs[0]["body"] == "report"
|
|
|
|
def test_a_second_run_updates_the_open_pr(self):
|
|
branch = "chore/base-os-13-20261020.1"
|
|
api = FakeGitea(
|
|
{"Dockerfile": f"FROM {NEW}\n"},
|
|
prs=[{"number": 7, "state": "open", "head": {"ref": branch}, "body": "old"}],
|
|
branches={branch},
|
|
)
|
|
|
|
open_or_update_pr(api, DEP, "netork/base-os", NEW, "13-20261020.1", "new report")
|
|
|
|
assert len(api.prs) == 1
|
|
assert api.prs[0]["body"] == "new report"
|
|
assert not [c for c in api.calls if c[0] == "POST" and c[1].endswith("/branches")]
|
|
|
|
def test_a_file_without_the_pin_is_left_alone(self):
|
|
api = FakeGitea({"Dockerfile": "FROM something-else\n"})
|
|
|
|
assert open_or_update_pr(api, DEP, "netork/base-os", NEW, "13-20261020.1", "r") is None
|
|
assert not [c for c in api.calls if c[0] in ("PUT", "POST")]
|
|
|
|
|
|
def test_the_dependents_are_listed_in_the_repo(tmp_path):
|
|
path = tmp_path / "dependents.toml"
|
|
path.write_text(
|
|
'[[dependent]]\nrepo = "NetOrk/netork"\nfiles = ["docker/ui/Dockerfile"]\n'
|
|
'workflow = "ui-nightly.yml"\n'
|
|
)
|
|
|
|
assert load_dependents(path) == [
|
|
Dependent("NetOrk/netork", ("docker/ui/Dockerfile",), "ui-nightly.yml")
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("bad", ["", "base-os", "netork/base-os:1"])
|
|
def test_an_odd_new_reference_is_refused(bad):
|
|
with pytest.raises(ValueError):
|
|
rewrite_pin("FROM x\n", "netork/base-os", bad)
|
|
|
|
|
|
def test_a_repo_without_the_file_yet_is_skipped():
|
|
"""The service repos start empty; a missing file is no pin to move, not a crash."""
|
|
|
|
class Empty(FakeGitea):
|
|
def __call__(self, method, path, body=None):
|
|
if method == "GET" and "/contents/" in path:
|
|
self.calls.append((method, path))
|
|
raise FileNotFoundError(path)
|
|
return super().__call__(method, path, body)
|
|
|
|
api = Empty({})
|
|
|
|
assert open_or_update_pr(api, DEP, "netork/base-os", NEW, "13-20261020.1", "r") is None
|
|
assert not [c for c in api.calls if c[0] in ("PUT", "POST")]
|
|
|
|
|
|
def test_a_dependent_without_a_workflow_gets_the_pr_only():
|
|
"""engine and satellite have no nightly run; their pull request is the trigger."""
|
|
from image_tools.bump import dispatch
|
|
|
|
calls = []
|
|
dispatch(lambda *a: calls.append(a), Dependent("NetOrk/netork", ("x",), ""), {"base_ref": NEW})
|
|
|
|
assert calls == []
|
|
|
|
|
|
def test_a_compose_pin_with_the_registry_variable_moves():
|
|
"""netork's compose files name the registry as ${REGISTRY_HOST:-registry.netork.io}
|
|
(tests/test_compose_image_names.py there); the variable stays, the pin moves."""
|
|
old = (
|
|
"${REGISTRY_HOST:-registry.netork.io}/netork/apt-cacher-ng:3.7.5-1-13-20261020.1-2026102017@sha256:"
|
|
+ "a" * 64
|
|
)
|
|
new = (
|
|
"registry.netork.io/netork/apt-cacher-ng:3.7.5-1-13-20261021.1-2026102103@sha256:"
|
|
+ "b" * 64
|
|
)
|
|
text = f" apt-cacher-ng:\n image: {old}\n"
|
|
|
|
moved = rewrite_pin(text, "netork/apt-cacher-ng", new)
|
|
|
|
assert moved == (
|
|
" apt-cacher-ng:\n image: ${REGISTRY_HOST:-registry.netork.io}/netork/apt-cacher-ng:"
|
|
"3.7.5-1-13-20261021.1-2026102103@sha256:" + "b" * 64 + "\n"
|
|
)
|
|
|
|
|
|
def test_a_repo_the_bot_may_not_see_is_an_error_not_a_skip():
|
|
"""Gitea answers 404 for the repo and its files alike when the token's account
|
|
has no access; on 2026-10-22 that read as "no pin to move" everywhere."""
|
|
|
|
class Invisible(FakeGitea):
|
|
def __call__(self, method, path, body=None):
|
|
self.calls.append((method, path))
|
|
raise FileNotFoundError(path)
|
|
|
|
with pytest.raises(PermissionError, match="NetOrk/"):
|
|
open_or_update_pr(Invisible({}), DEP, "netork/base-os", NEW, "13-20261020.1", "r")
|
|
|
|
|
|
class TestEveryDependent:
|
|
"""2026-10-22: netork's dependents entry named ui-nightly.yml, which does not
|
|
exist yet; the dispatch's 404 ended the run with a traceback after the last
|
|
pull request. A dependent that cannot be started is named, the others go
|
|
on, and the run fails at the end, as for a repo the token cannot see."""
|
|
|
|
DEPS = [
|
|
Dependent(repo="NetOrk/netork", files=("Dockerfile",), workflow="ui-nightly.yml"),
|
|
Dependent(repo="NetOrk/valkey", files=("Dockerfile",), workflow="nightly.yml"),
|
|
]
|
|
|
|
def test_a_missing_workflow_is_named_and_the_rest_goes_on(self):
|
|
api = FakeGitea({"Dockerfile": f"FROM {OLD}\n"})
|
|
api.missing_workflows = {"ui-nightly.yml"}
|
|
lines: list[str] = []
|
|
|
|
ok = bump_all(api, self.DEPS, "netork/base-os", NEW, "13-20261020.1", "r", lines.append)
|
|
|
|
assert not ok
|
|
assert api.dispatched == ["nightly.yml"]
|
|
assert any("NetOrk/netork" in ln and "ui-nightly.yml" in ln for ln in lines)
|
|
assert any(ln.startswith("NetOrk/valkey: PR #7") for ln in lines)
|
|
|
|
def test_all_well_is_ok(self):
|
|
api = FakeGitea({"Dockerfile": f"FROM {OLD}\n"})
|
|
|
|
assert bump_all(api, self.DEPS[1:], "netork/base-os", NEW, "13-20261020.1", "r", print)
|
|
assert api.dispatched == ["nightly.yml"]
|