xavierk
b47a65c91e
Every release is built from its tag alone by packaging/void/build-release.sh, shared by the tag workflow and local dry runs. Signs in a network-less container with the key mounted read-only and publishes to the tag release and the rolling latest release.
778 lines
30 KiB
Python
778 lines
30 KiB
Python
"""Offline guards for the signed xbps release workflow (tickets #28 and #29).
|
|
|
|
Releasing the Void repository is a tag push on the fork's Gitea host. The
|
|
workflow ``.gitea/workflows/xbps-release.yml`` gates the tag, stages the
|
|
signing key on tmpfs, runs ``packaging/void/build-release.sh`` (the one release
|
|
path), and publishes exactly what that script staged to the tag's release and
|
|
to the rolling ``latest`` release the client registration URL serves.
|
|
|
|
These tests parse the workflow and then *run* its gate, key, tools, and publish
|
|
steps in a scratch directory, so they assert behaviour rather than spelling:
|
|
|
|
- each step's shell runs with the environment the workflow itself declares
|
|
(``${{ ... }}`` is rendered from an explicit context, so an unexpected
|
|
expression fails loudly);
|
|
- ``curl`` is replaced by a stateful stand-in for the Gitea release API on
|
|
PATH, so the publish step's requests, their order and its refusal to continue
|
|
past a non-2xx response are observable.
|
|
|
|
Independent sources of truth: the tag/version pair comes from the test (not the
|
|
workflow), the staged layout is what ``build-release.sh`` writes, and the Gitea
|
|
REST contract (release create, asset upload and delete, tag lookup) comes from
|
|
the Gitea API documentation.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional
|
|
from urllib.parse import unquote
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
WORKFLOW = REPO_ROOT / ".gitea" / "workflows" / "xbps-release.yml"
|
|
VOID = REPO_ROOT / "packaging" / "void"
|
|
SMOKE_SH = VOID / "smoke.sh"
|
|
JOB = "xbps-release"
|
|
|
|
GATE_TAG = "Require the tag to equal the version in version.py"
|
|
GATE_KEY = "Require the repository signing key"
|
|
TOOLS = "Check the runner has the release tools"
|
|
BUILD = "Build, sign and stage the repository"
|
|
PUBLISH = "Publish to the tag release and the latest release"
|
|
CLEANUP = "Remove the signing key"
|
|
STEP_ORDER = (GATE_TAG, GATE_KEY, TOOLS, BUILD, PUBLISH, CLEANUP)
|
|
|
|
API = "https://git.bongbetic.com/api/v1/repos/xavierk/voculinux"
|
|
TOKEN = "test-token-not-a-secret"
|
|
SHA = "0123456789abcdef0123456789abcdef01234567"
|
|
FINGERPRINT = "fa:d1:00:32:ce:b1:b9:fd:73:05:05:76:75:54:37:25"
|
|
FAKE_KEY = "-----BEGIN RSA PRIVATE KEY-----\nline-one\nline-two\n-----END RSA PRIVATE KEY-----\n"
|
|
SDIST = "voculinux-0.18.5.tar.gz"
|
|
|
|
|
|
def _text() -> str:
|
|
assert WORKFLOW.is_file(), f"missing signed xbps release workflow: {WORKFLOW}"
|
|
return WORKFLOW.read_text(encoding="utf-8")
|
|
|
|
|
|
def _workflow() -> dict:
|
|
return yaml.safe_load(_text())
|
|
|
|
|
|
def _job() -> dict:
|
|
return _workflow()["jobs"][JOB]
|
|
|
|
|
|
def _step(name: str) -> dict:
|
|
matches = [s for s in _job()["steps"] if s.get("name") == name]
|
|
assert len(matches) == 1, f"workflow must have exactly one step named {name!r}"
|
|
return matches[0]
|
|
|
|
|
|
def _run_text(name: str) -> str:
|
|
return _step(name)["run"]
|
|
|
|
|
|
def _without_comments(text: str) -> str:
|
|
return "\n".join(line for line in text.splitlines() if not line.lstrip().startswith("#"))
|
|
|
|
|
|
def _render(value: str, context: Dict[str, str]) -> str:
|
|
"""Render ``${{ expr }}`` from an explicit context; an unknown one is a bug."""
|
|
|
|
def substitute(match: "re.Match[str]") -> str:
|
|
expression = match.group(1)
|
|
assert expression in context, f"workflow uses an unexpected expression: {expression}"
|
|
return context[expression]
|
|
|
|
return re.sub(r"\$\{\{\s*([\w.]+)\s*\}\}", substitute, value)
|
|
|
|
|
|
def _run_step(
|
|
name: str,
|
|
cwd: Path,
|
|
context: Optional[Dict[str, str]] = None,
|
|
extra_env: Optional[Dict[str, str]] = None,
|
|
path_prefix: Optional[Path] = None,
|
|
path: Optional[str] = None,
|
|
) -> subprocess.CompletedProcess:
|
|
"""Execute one step's ``run`` script with the env the workflow declares."""
|
|
ctx = {"gitea.run_id": "42", "gitea.token": TOKEN, "gitea.sha": SHA}
|
|
ctx.update(context or {})
|
|
step = _step(name)
|
|
env = {k: os.environ[k] for k in ("HOME", "LANG") if k in os.environ}
|
|
base = path if path is not None else os.environ.get("PATH", "/usr/bin:/bin")
|
|
env["PATH"] = f"{path_prefix}{os.pathsep}{base}" if path_prefix else base
|
|
for source in (_job().get("env", {}), step.get("env", {})):
|
|
for key, value in source.items():
|
|
env[key] = _render(str(value), ctx)
|
|
env.update(extra_env or {})
|
|
script = cwd / f"{re.sub(r'[^a-z]+', '-', name.lower())}.sh"
|
|
script.write_text(step["run"], encoding="utf-8")
|
|
bash = shutil.which("bash", path=os.environ.get("PATH", "/usr/bin:/bin")) or "/bin/bash"
|
|
return subprocess.run(
|
|
[bash, str(script)],
|
|
cwd=cwd,
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=60,
|
|
check=False,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Workflow shape
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_triggers_only_on_version_tag_pushes() -> None:
|
|
"""A tag push is a human act: no schedule, no dispatch, no PR trigger."""
|
|
triggers = _workflow()[True] # YAML 1.1 reads the bare key `on` as True
|
|
assert triggers == {"push": {"tags": ["v*"]}}, triggers
|
|
|
|
|
|
def test_grants_only_contents_write() -> None:
|
|
assert _workflow()["permissions"] == {"contents": "write"}
|
|
|
|
|
|
def test_runs_on_the_project_runner_with_no_third_party_actions() -> None:
|
|
"""Checkout is the git fetch the repo's own CI uses; nothing floats on a tag."""
|
|
assert _job()["runs-on"] == "bongbetic-ci"
|
|
uses = [s["uses"] for s in _job()["steps"] if "uses" in s]
|
|
assert not uses, f"the signing job must not run third-party actions: {uses}"
|
|
|
|
|
|
def test_steps_run_in_the_gated_order() -> None:
|
|
"""Both gates precede any container; the cleanup of the key is last."""
|
|
names = [s.get("name") for s in _job()["steps"]]
|
|
assert names[0] == "Checkout"
|
|
assert tuple(names[1:]) == STEP_ORDER, names
|
|
|
|
|
|
def test_signing_key_is_removed_by_an_always_step() -> None:
|
|
last = _job()["steps"][-1]
|
|
assert last["name"] == CLEANUP
|
|
assert last["if"] == "always()"
|
|
assert "rm -f" in last["run"] and "$XBPS_KEY_FILE" in last["run"]
|
|
|
|
|
|
def test_the_key_file_lives_on_tmpfs_under_a_per_run_name() -> None:
|
|
path = _job()["env"]["XBPS_KEY_FILE"]
|
|
assert path.startswith("/dev/shm/xbps-sign-key-"), path
|
|
assert "${{ gitea.run_id }}" in path
|
|
|
|
|
|
def test_signing_secret_is_read_by_exactly_one_step() -> None:
|
|
"""The secret is exposed to the gate that stages it, never to the build."""
|
|
text = _text()
|
|
assert text.count("secrets.XBPS_SIGNING_KEY") == 1
|
|
assert (
|
|
_step(GATE_KEY)["env"]["XBPS_SIGNING_KEY"] == "${{ secrets.XBPS_SIGNING_KEY }}"
|
|
), "the key gate must read the XBPS_SIGNING_KEY secret"
|
|
assert "XBPS_SIGNING_KEY" not in _step(BUILD).get("env", {})
|
|
|
|
|
|
def test_the_tag_reaches_scripts_through_env_never_interpolated() -> None:
|
|
"""`${{ gitea.ref_name }}` inside a run script is shell injection by tag name."""
|
|
for step in _job()["steps"]:
|
|
assert "${{" not in step.get("run", ""), step.get("name")
|
|
assert _step(GATE_TAG)["env"]["TAG"] == "${{ gitea.ref_name }}"
|
|
assert _step(BUILD)["env"]["TAG"] == "${{ gitea.ref_name }}"
|
|
|
|
|
|
def test_the_build_step_is_the_release_path_with_the_key_file_only() -> None:
|
|
"""One path: build-release.sh release, handed the key file and nothing else."""
|
|
code = _without_comments(_text())
|
|
assert code.count("packaging/void/build-release.sh release") == 1
|
|
run = _run_text(BUILD)
|
|
assert '--key-file "$XBPS_KEY_FILE"' in run
|
|
assert "--out-dir publish" in run
|
|
for forbidden in ("xbps-src", "xbps-rindex", "openssl rsa", "xlint", "gpg ", "repo.sh"):
|
|
assert forbidden not in code, f"{forbidden!r} belongs inside build-release.sh"
|
|
|
|
|
|
def test_no_step_installs_packages_and_the_runner_must_bring_its_tools() -> None:
|
|
"""The job checks its tools fail-fast; it does not install them with apt."""
|
|
code = _without_comments(_text())
|
|
assert "apt-get" not in code
|
|
check = _run_text(TOOLS)
|
|
for tool in ("podman", "openssl", "git", "sha256sum", "python3.11", "uv"):
|
|
assert tool in check, f"the tool check must name {tool}"
|
|
assert "0.12.*" in check and "0.13.*" in check, "uv must be checked against the pinned range"
|
|
|
|
|
|
def test_pipeline_uses_podman_and_never_docker() -> None:
|
|
code = _without_comments(_text()).lower()
|
|
assert "podman --version" in code
|
|
assert "docker" not in code
|
|
|
|
|
|
def test_workflow_never_commits_pushes_or_bumps_templates() -> None:
|
|
"""Template bumps are a normal pull request, never a bot commit."""
|
|
code = _without_comments(_text())
|
|
for forbidden in ("git commit", "git push", "git tag", "git add", "srcpkgs", "gh pr"):
|
|
assert forbidden not in code, f"the workflow must not run {forbidden!r}"
|
|
|
|
|
|
def test_workflow_text_names_no_assistant() -> None:
|
|
assert not re.search(r"claude|anthropic", _text(), re.IGNORECASE)
|
|
|
|
|
|
def test_publish_goes_to_the_fork_gitea_api_with_the_job_token() -> None:
|
|
assert _job()["env"]["GITEA_API"] == API
|
|
env = _step(PUBLISH)["env"]
|
|
assert env["GITEA_TOKEN"] == "${{ gitea.token }}"
|
|
assert "Authorization: token ${GITEA_TOKEN}" in _run_text(PUBLISH)
|
|
assert "gh release" not in _text(), "GitHub CLI cannot reach the fork's Gitea releases"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Gate 1: the tag equals the version source of truth, and the template agrees
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _gate_tree(root: Path, version: str, template_version: str, template_tag: str) -> Path:
|
|
"""A scratch repository with the real release script and a chosen version."""
|
|
pkg = root / "packaging" / "void"
|
|
pkg.mkdir(parents=True)
|
|
for name in ("build-release.sh", "smoke.sh", "tool_checksums.txt", "repository-key.plist"):
|
|
shutil.copy2(VOID / name, pkg / name)
|
|
template = (VOID / "srcpkgs" / "voculinux" / "template").read_text(encoding="utf-8")
|
|
template = re.sub(r"^version=.*$", f"version={template_version}", template, flags=re.M)
|
|
template = re.sub(r"^_tag=.*$", f"_tag={template_tag}", template, flags=re.M)
|
|
(pkg / "srcpkgs" / "voculinux").mkdir(parents=True)
|
|
(pkg / "srcpkgs" / "voculinux" / "template").write_text(template, encoding="utf-8")
|
|
(root / "src" / "vocalinux").mkdir(parents=True)
|
|
(root / "src" / "vocalinux" / "version.py").write_text(
|
|
f'"""Version."""\n\n__version__ = "{version}"\n', encoding="utf-8"
|
|
)
|
|
return root
|
|
|
|
|
|
def _gate_tag(tmp_path: Path, tag: str, version: str) -> subprocess.CompletedProcess:
|
|
xbps = version.replace("-", "")
|
|
_gate_tree(tmp_path, version, template_version=xbps, template_tag=f"v{version}")
|
|
return _run_step(GATE_TAG, tmp_path, {"gitea.ref_name": tag})
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("tag", "version", "ok"),
|
|
[
|
|
("v0.18.5", "0.18.5", True),
|
|
("v0.19.0-beta", "0.19.0-beta", True),
|
|
("0.18.5", "0.18.5", False), # no v prefix
|
|
("v0.18.50", "0.18.5", False), # tag contains the version: loose match would pass
|
|
("v0.18", "0.18.5", False), # version contains the tag: loose match would pass
|
|
("v0.18.5-beta", "0.18.5", False),
|
|
("v0.18.5", "0.18.5-beta", False),
|
|
("v0.18.4", "0.18.5", False),
|
|
("vv0.18.5", "0.18.5", False),
|
|
],
|
|
)
|
|
def test_tag_gate_requires_exact_v_plus_version(
|
|
tmp_path: Path, tag: str, version: str, ok: bool
|
|
) -> None:
|
|
result = _gate_tag(tmp_path, tag, version)
|
|
if ok:
|
|
assert result.returncode == 0, result.stderr
|
|
else:
|
|
assert result.returncode == 1, f"{tag} vs {version} must exit 1: {result.stdout}"
|
|
assert "refusing to publish" in result.stderr
|
|
|
|
|
|
def test_tag_gate_refuses_a_committed_template_for_another_release(tmp_path: Path) -> None:
|
|
"""The tag may match version.py, but the template the build tree starts from must too."""
|
|
_gate_tree(tmp_path, "0.19.0-beta", template_version="0.18.5", template_tag="v0.18.5")
|
|
result = _run_step(GATE_TAG, tmp_path, {"gitea.ref_name": "v0.19.0-beta"})
|
|
assert result.returncode == 1
|
|
assert "committed Void template" in result.stderr
|
|
|
|
|
|
def test_tag_gate_fails_when_the_version_cannot_be_read(tmp_path: Path) -> None:
|
|
_gate_tree(tmp_path, "0.18.5", template_version="0.18.5", template_tag="v0.18.5")
|
|
(tmp_path / "src" / "vocalinux" / "version.py").write_text(
|
|
'"""Version."""\n__version_info__ = (0, 18, 5)\n', encoding="utf-8"
|
|
)
|
|
result = _run_step(GATE_TAG, tmp_path, {"gitea.ref_name": "v0.18.5"})
|
|
assert result.returncode == 1
|
|
assert "__version__" in result.stderr
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Gate 2: the signing key is staged on tmpfs, 0600, never echoed
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _tmpfs_dir() -> Optional[Path]:
|
|
"""A fresh directory on /dev/shm when that is tmpfs, or None on a host without one."""
|
|
shm = Path("/dev/shm")
|
|
if not shm.is_dir():
|
|
return None
|
|
fstype = subprocess.run(
|
|
["stat", "-f", "-c", "%T", str(shm)], capture_output=True, text=True, check=False
|
|
).stdout.strip()
|
|
if fstype != "tmpfs":
|
|
return None
|
|
return Path(tempfile.mkdtemp(prefix="xbps-key-test-", dir=str(shm)))
|
|
|
|
|
|
def test_unset_signing_key_exits_1_and_writes_nothing(tmp_path: Path) -> None:
|
|
key_file = tmp_path / "key.pem"
|
|
result = _run_step(
|
|
GATE_KEY,
|
|
tmp_path,
|
|
{"secrets.XBPS_SIGNING_KEY": ""},
|
|
{"XBPS_KEY_FILE": str(key_file)},
|
|
)
|
|
assert result.returncode == 1
|
|
assert "XBPS_SIGNING_KEY" in result.stderr
|
|
assert "unsigned" in result.stderr
|
|
assert not key_file.exists(), "no key material may be written when the secret is empty"
|
|
|
|
|
|
def test_a_key_file_outside_tmpfs_is_refused(tmp_path: Path) -> None:
|
|
"""A key that would land on disk stops the job before the secret is written."""
|
|
if (
|
|
subprocess.run(
|
|
["stat", "-f", "-c", "%T", str(tmp_path)], capture_output=True, text=True, check=False
|
|
).stdout.strip()
|
|
== "tmpfs"
|
|
):
|
|
pytest.skip("the test directory is itself tmpfs")
|
|
key_file = tmp_path / "key.pem"
|
|
result = _run_step(
|
|
GATE_KEY,
|
|
tmp_path,
|
|
{"secrets.XBPS_SIGNING_KEY": FAKE_KEY},
|
|
{"XBPS_KEY_FILE": str(key_file)},
|
|
)
|
|
assert result.returncode == 1
|
|
assert "not tmpfs" in result.stderr
|
|
assert not key_file.exists()
|
|
|
|
|
|
def test_signing_key_is_staged_0600_with_newlines_intact(tmp_path: Path) -> None:
|
|
shm = _tmpfs_dir()
|
|
if shm is None:
|
|
pytest.skip("no tmpfs at /dev/shm on this host")
|
|
try:
|
|
key_file = shm / "key.pem"
|
|
result = _run_step(
|
|
GATE_KEY,
|
|
tmp_path,
|
|
{"secrets.XBPS_SIGNING_KEY": FAKE_KEY},
|
|
{"XBPS_KEY_FILE": str(key_file)},
|
|
)
|
|
assert result.returncode == 0, result.stderr
|
|
assert key_file.read_text(encoding="utf-8") == FAKE_KEY
|
|
assert stat.S_IMODE(key_file.stat().st_mode) == 0o600
|
|
assert FAKE_KEY.splitlines()[1] not in result.stdout + result.stderr, "key was echoed"
|
|
finally:
|
|
shutil.rmtree(shm, ignore_errors=True)
|
|
|
|
|
|
def test_an_existing_key_file_is_never_reused(tmp_path: Path) -> None:
|
|
"""Noclobber: a file already at the key path fails the job rather than being overwritten."""
|
|
shm = _tmpfs_dir()
|
|
if shm is None:
|
|
pytest.skip("no tmpfs at /dev/shm on this host")
|
|
try:
|
|
key_file = shm / "key.pem"
|
|
key_file.write_text("someone else's file\n", encoding="utf-8")
|
|
result = _run_step(
|
|
GATE_KEY,
|
|
tmp_path,
|
|
{"secrets.XBPS_SIGNING_KEY": FAKE_KEY},
|
|
{"XBPS_KEY_FILE": str(key_file)},
|
|
)
|
|
assert result.returncode != 0
|
|
assert key_file.read_text(encoding="utf-8") == "someone else's file\n"
|
|
finally:
|
|
shutil.rmtree(shm, ignore_errors=True)
|
|
|
|
|
|
def test_cleanup_step_removes_the_key_file(tmp_path: Path) -> None:
|
|
key_file = tmp_path / "key.pem"
|
|
key_file.write_text(FAKE_KEY, encoding="utf-8")
|
|
result = _run_step(CLEANUP, tmp_path, extra_env={"XBPS_KEY_FILE": str(key_file)})
|
|
assert result.returncode == 0, result.stderr
|
|
assert not key_file.exists()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tools: a runner without the release tools stops before any work
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_a_runner_missing_the_release_tools_fails_fast(tmp_path: Path) -> None:
|
|
empty = tmp_path / "empty-bin"
|
|
empty.mkdir()
|
|
result = _run_step(TOOLS, tmp_path, path=str(empty))
|
|
assert result.returncode == 1
|
|
assert "lacks" in result.stderr and "podman" in result.stderr
|
|
assert "packaging/void/README.md" in result.stderr, "the message names the runner requirements"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Publish: tag release plus rolling latest release, through a fake Gitea API
|
|
# ---------------------------------------------------------------------------
|
|
|
|
FAKE_CURL = '''#!{python}
|
|
"""Stateful stand-in for curl against the Gitea release API."""
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from urllib.parse import unquote, urlsplit
|
|
|
|
STATE = os.environ["FAKE_GITEA_STATE"]
|
|
BASE = os.environ["FAKE_GITEA_BASE"]
|
|
|
|
|
|
def load():
|
|
with open(STATE, encoding="utf-8") as handle:
|
|
return json.load(handle)
|
|
|
|
|
|
def save(state):
|
|
with open(STATE, "w", encoding="utf-8") as handle:
|
|
json.dump(state, handle)
|
|
|
|
|
|
def main(argv):
|
|
out = method = data = url = None
|
|
headers, forms = [], []
|
|
i = 0
|
|
while i < len(argv):
|
|
arg = argv[i]
|
|
if arg == "-o":
|
|
out, i = argv[i + 1], i + 2
|
|
elif arg == "-w":
|
|
i += 2
|
|
elif arg == "-X":
|
|
method, i = argv[i + 1], i + 2
|
|
elif arg == "-H":
|
|
headers.append(argv[i + 1])
|
|
i += 2
|
|
elif arg == "-d":
|
|
data, i = argv[i + 1], i + 2
|
|
elif arg == "-F":
|
|
forms.append(argv[i + 1])
|
|
i += 2
|
|
elif arg.startswith("-"):
|
|
i += 1
|
|
else:
|
|
url, i = arg, i + 1
|
|
method = method or "GET"
|
|
state = load()
|
|
entry = {{
|
|
"method": method,
|
|
"url": url,
|
|
"auth": [h for h in headers if h.lower().startswith("authorization:")],
|
|
"data": json.loads(data) if data else None,
|
|
"forms": forms,
|
|
}}
|
|
state["log"].append(entry)
|
|
status, body = route(state, method, url, data, forms)
|
|
fail = os.environ.get("FAKE_GITEA_FAIL", "")
|
|
if fail:
|
|
fmethod, fneedle, fcode = fail.split("|")
|
|
if method == fmethod and fneedle in url:
|
|
status, body = int(fcode), {{"message": "injected failure"}}
|
|
save(state)
|
|
if out:
|
|
with open(out, "w", encoding="utf-8") as handle:
|
|
handle.write(json.dumps(body) if body is not None else "")
|
|
sys.stdout.write(str(status))
|
|
return 0
|
|
|
|
|
|
def route(state, method, url, data, forms):
|
|
if not url.startswith(BASE):
|
|
return 599, {{"message": "unexpected host " + url}}
|
|
path = urlsplit(url[len(BASE):]).path
|
|
query = urlsplit(url).query
|
|
releases = state["releases"]
|
|
m = re.fullmatch(r"/releases/tags/(.+)", path)
|
|
if m and method == "GET":
|
|
tag = unquote(m.group(1))
|
|
for rel in releases.values():
|
|
if rel["tag_name"] == tag:
|
|
return 200, rel
|
|
return 404, {{"message": "not found"}}
|
|
if path == "/releases" and method == "POST":
|
|
payload = json.loads(data)
|
|
rid = state["next_id"]
|
|
state["next_id"] += 1
|
|
rel = dict(payload, id=rid, assets=[])
|
|
releases[str(rid)] = rel
|
|
return 201, rel
|
|
m = re.fullmatch(r"/releases/(\\d+)", path)
|
|
if m and method == "PATCH":
|
|
rel = releases[m.group(1)]
|
|
rel.update(json.loads(data))
|
|
return 200, rel
|
|
m = re.fullmatch(r"/releases/(\\d+)/assets", path)
|
|
if m and method == "GET":
|
|
return 200, releases[m.group(1)]["assets"]
|
|
if m and method == "POST":
|
|
rel = releases[m.group(1)]
|
|
name = unquote(re.search(r"(?:^|&)name=([^&]*)", query).group(1))
|
|
if any(a["name"] == name for a in rel["assets"]):
|
|
return 409, {{"message": "asset already exists"}}
|
|
source = re.match(r"attachment=@(.*?);filename=(.*)$", forms[0])
|
|
size = os.path.getsize(source.group(1))
|
|
asset = {{"id": state["next_id"], "name": name, "size": size}}
|
|
state["next_id"] += 1
|
|
rel["assets"].append(asset)
|
|
return 201, asset
|
|
m = re.fullmatch(r"/releases/(\\d+)/assets/(\\d+)", path)
|
|
if m and method == "DELETE":
|
|
rel = releases[m.group(1)]
|
|
rel["assets"] = [a for a in rel["assets"] if str(a["id"]) != m.group(2)]
|
|
return 204, None
|
|
return 404, {{"message": "no such route " + method + " " + path}}
|
|
|
|
|
|
sys.exit(main(sys.argv[1:]))
|
|
'''
|
|
|
|
|
|
class FakeGitea:
|
|
"""A stateful curl stand-in on PATH plus helpers to read what it recorded."""
|
|
|
|
def __init__(self, root: Path) -> None:
|
|
self.bin = root / "fakebin"
|
|
self.bin.mkdir()
|
|
self.state_file = root / "gitea-state.json"
|
|
curl = self.bin / "curl"
|
|
curl.write_text(FAKE_CURL.format(python=sys.executable), encoding="utf-8")
|
|
curl.chmod(0o755)
|
|
self.state_file.write_text(
|
|
json.dumps({"next_id": 100, "releases": {}, "log": []}), encoding="utf-8"
|
|
)
|
|
|
|
@property
|
|
def state(self) -> dict:
|
|
return json.loads(self.state_file.read_text(encoding="utf-8"))
|
|
|
|
def seed_release(self, tag: str, assets: List[str]) -> None:
|
|
state = self.state
|
|
rid = state["next_id"]
|
|
state["next_id"] += 1
|
|
state["releases"][str(rid)] = {
|
|
"id": rid,
|
|
"tag_name": tag,
|
|
"name": tag,
|
|
"body": "old",
|
|
"assets": [{"id": rid * 1000 + n, "name": a, "size": 1} for n, a in enumerate(assets)],
|
|
}
|
|
self.state_file.write_text(json.dumps(state), encoding="utf-8")
|
|
|
|
def release(self, tag: str) -> dict:
|
|
for rel in self.state["releases"].values():
|
|
if rel["tag_name"] == tag:
|
|
return rel
|
|
raise AssertionError(f"no release for tag {tag!r}")
|
|
|
|
def asset_names(self, tag: str) -> List[str]:
|
|
return [a["name"] for a in self.release(tag)["assets"]]
|
|
|
|
def uploads(self, tag: str) -> List[str]:
|
|
"""Asset names POSTed to this tag's release, in upload order."""
|
|
prefix = f"{API}/releases/{self.release(tag)['id']}/assets?name="
|
|
return [
|
|
unquote(e["url"][len(prefix) :])
|
|
for e in self.state["log"]
|
|
if e["method"] == "POST" and e["url"].startswith(prefix)
|
|
]
|
|
|
|
def run_publish(
|
|
self, root: Path, tag: str = "v0.18.5", fail: str = ""
|
|
) -> subprocess.CompletedProcess:
|
|
return _run_step(
|
|
PUBLISH,
|
|
root,
|
|
{"gitea.ref_name": tag},
|
|
{
|
|
"FAKE_GITEA_STATE": str(self.state_file),
|
|
"FAKE_GITEA_BASE": API,
|
|
"FAKE_GITEA_FAIL": fail,
|
|
},
|
|
path_prefix=self.bin,
|
|
)
|
|
|
|
|
|
def _smoke_packages() -> List[str]:
|
|
"""The templates the release builds, in order, from the smoke gate's list."""
|
|
match = re.search(r'^PKGS="([^"]+)"', SMOKE_SH.read_text(encoding="utf-8"), re.M)
|
|
assert match, "smoke.sh must declare its ordered PKGS list"
|
|
return match.group(1).split()
|
|
|
|
|
|
def _stage(root: Path) -> Path:
|
|
"""Lay out publish/ the way build-release.sh stages a release."""
|
|
out = root / "publish"
|
|
sdist = out / "sdist"
|
|
repo = out / "xbps" / "repo"
|
|
keys = out / "xbps" / "keys"
|
|
for directory in (sdist, repo, keys):
|
|
directory.mkdir(parents=True)
|
|
(sdist / SDIST).write_bytes(b"canonical sdist bytes")
|
|
(out / "SDIST_SHA256").write_text("a" * 64 + "\n", encoding="utf-8")
|
|
for name in _smoke_packages():
|
|
pkg = f"{name}-0.18.5_1"
|
|
(repo / f"{pkg}.x86_64.xbps").write_bytes(b"package bytes " + pkg.encode())
|
|
(repo / f"{pkg}.x86_64.xbps.sig2").write_bytes(b"signature " + pkg.encode())
|
|
(repo / "x86_64-repodata").write_bytes(b"signed index")
|
|
(keys / "FINGERPRINT").write_text(
|
|
f"signed-by: Voculinux <xbps@voculinux.com>\nfingerprint: {FINGERPRINT}\n",
|
|
encoding="utf-8",
|
|
)
|
|
(keys / f"{FINGERPRINT}.plist").write_text("<plist></plist>\n", encoding="utf-8")
|
|
return out
|
|
|
|
|
|
def _staged_names(out: Path) -> List[str]:
|
|
return sorted(
|
|
p.name for sub in ("sdist", "xbps/repo", "xbps/keys") for p in (out / sub).iterdir()
|
|
)
|
|
|
|
|
|
@pytest.fixture()
|
|
def staged(tmp_path: Path) -> Path:
|
|
if shutil.which("python3") is None:
|
|
pytest.skip("python3 is required by the publish step")
|
|
_stage(tmp_path)
|
|
return tmp_path
|
|
|
|
|
|
def test_publish_attaches_the_sdist_and_the_repository_to_both_releases(staged: Path) -> None:
|
|
gitea = FakeGitea(staged)
|
|
result = gitea.run_publish(staged)
|
|
assert result.returncode == 0, result.stdout + result.stderr
|
|
expected = _staged_names(staged / "publish")
|
|
assert SDIST in expected
|
|
for tag in ("v0.18.5", "latest"):
|
|
assert sorted(gitea.asset_names(tag)) == expected, tag
|
|
uploaded = gitea.uploads(tag)
|
|
assert sorted(uploaded) == expected, f"{tag}: uploaded {uploaded}"
|
|
assert uploaded[-1] == "x86_64-repodata", "the index goes last, after its packages"
|
|
assert FINGERPRINT in gitea.release("v0.18.5")["body"]
|
|
|
|
|
|
def test_publish_states_the_sdist_digest_in_the_notes(staged: Path) -> None:
|
|
gitea = FakeGitea(staged)
|
|
assert gitea.run_publish(staged).returncode == 0
|
|
assert "sdist sha256: " + "a" * 64 in gitea.release("latest")["body"]
|
|
|
|
|
|
def test_publish_creates_latest_at_the_workflow_commit(staged: Path) -> None:
|
|
gitea = FakeGitea(staged)
|
|
assert gitea.run_publish(staged).returncode == 0
|
|
latest = gitea.release("latest")
|
|
assert latest["target_commitish"] == SHA
|
|
assert latest["tag_name"] == "latest"
|
|
tag_release = gitea.release("v0.18.5")
|
|
assert tag_release["target_commitish"] == SHA
|
|
assert not tag_release["draft"] and not latest["draft"]
|
|
assert "v0.18.5" in latest["body"], "latest says which tag it was built from"
|
|
|
|
|
|
def test_publish_talks_only_to_the_fork_api_authenticated(staged: Path) -> None:
|
|
gitea = FakeGitea(staged)
|
|
assert gitea.run_publish(staged).returncode == 0
|
|
log = gitea.state["log"]
|
|
assert log, "the publish step made no API calls"
|
|
for entry in log:
|
|
assert entry["url"].startswith(API + "/"), entry["url"]
|
|
assert entry["auth"] == [f"Authorization: token {TOKEN}"], entry
|
|
posts = [e["data"]["tag_name"] for e in log if e["url"] == API + "/releases"]
|
|
assert posts == ["v0.18.5", "latest"], "the tag release is filled before latest is touched"
|
|
|
|
|
|
def test_publish_replaces_every_asset_on_an_existing_latest_release(staged: Path) -> None:
|
|
gitea = FakeGitea(staged)
|
|
gitea.seed_release("latest", ["voculinux-0.18.4_1.x86_64.xbps", "x86_64-repodata", "old.txt"])
|
|
assert gitea.run_publish(staged).returncode == 0
|
|
assert sorted(gitea.asset_names("latest")) == _staged_names(staged / "publish")
|
|
assert len([r for r in gitea.state["releases"].values() if r["tag_name"] == "latest"]) == 1
|
|
assert "v0.18.5" in gitea.release("latest")["body"]
|
|
|
|
|
|
def test_publish_replaces_a_same_named_asset_on_a_rerun(staged: Path) -> None:
|
|
"""A rerun of a tag replaces its own files, the sdist included: nothing stale survives."""
|
|
gitea = FakeGitea(staged)
|
|
gitea.seed_release("v0.18.5", [SDIST, "x86_64-repodata"])
|
|
before = {a["name"]: a["id"] for a in gitea.release("v0.18.5")["assets"]}
|
|
assert gitea.run_publish(staged).returncode == 0
|
|
after = {a["name"]: a["id"] for a in gitea.release("v0.18.5")["assets"]}
|
|
assert after[SDIST] != before[SDIST], "the sdist on the tag release must be the staged one"
|
|
assert after["x86_64-repodata"] != before["x86_64-repodata"]
|
|
assert sorted(after) == _staged_names(staged / "publish")
|
|
|
|
|
|
def test_publish_uploads_only_what_the_release_path_staged(staged: Path) -> None:
|
|
"""Nothing is built or uploaded by hand: every upload lives under publish/."""
|
|
gitea = FakeGitea(staged)
|
|
assert gitea.run_publish(staged).returncode == 0
|
|
staged_names = set(_staged_names(staged / "publish"))
|
|
for tag in ("v0.18.5", "latest"):
|
|
assert set(gitea.uploads(tag)) == staged_names
|
|
for entry in gitea.state["log"]:
|
|
for form in entry["forms"]:
|
|
path = re.match(r"attachment=@(.*?);", form).group(1)
|
|
assert path.startswith(("sdist/", "xbps/")), f"not from the staged set: {path}"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"fail",
|
|
[
|
|
"POST|/releases|403", # token may not create releases
|
|
"POST|/assets?|500", # an upload fails midway
|
|
"DELETE|/assets/|500",
|
|
"GET|/releases/tags/|500",
|
|
"PATCH|/releases/|500",
|
|
],
|
|
)
|
|
def test_publish_fails_the_job_on_any_non_2xx_response(staged: Path, fail: str) -> None:
|
|
gitea = FakeGitea(staged)
|
|
gitea.seed_release("latest", ["stale"])
|
|
gitea.seed_release("v0.18.5", ["x86_64-repodata"])
|
|
result = gitea.run_publish(staged, fail=fail)
|
|
assert result.returncode != 0, f"{fail} must fail the job"
|
|
assert "HTTP" in result.stderr or "failed" in result.stderr
|
|
|
|
|
|
def test_publish_does_not_touch_latest_when_the_tag_release_fails(staged: Path) -> None:
|
|
gitea = FakeGitea(staged)
|
|
result = gitea.run_publish(staged, fail="POST|/assets?|500")
|
|
assert result.returncode != 0
|
|
assert not any(r["tag_name"] == "latest" for r in gitea.state["releases"].values())
|
|
|
|
|
|
def test_publish_percent_encodes_the_colon_fingerprint_asset_name(staged: Path) -> None:
|
|
gitea = FakeGitea(staged)
|
|
assert gitea.run_publish(staged).returncode == 0
|
|
plist_urls = [e["url"] for e in gitea.state["log"] if ".plist" in e["url"]]
|
|
assert plist_urls and all(":" not in u.split("name=")[1] for u in plist_urls), plist_urls
|
|
assert f"{FINGERPRINT}.plist" in gitea.asset_names("latest")
|
|
|
|
|
|
def test_publish_refuses_to_run_without_a_staged_sdist_digest(staged: Path) -> None:
|
|
(staged / "publish" / "SDIST_SHA256").unlink()
|
|
gitea = FakeGitea(staged)
|
|
result = gitea.run_publish(staged)
|
|
assert result.returncode != 0
|
|
assert not gitea.state["log"], "nothing may be uploaded without the staged digest"
|