xavierk
731b919865
Version 0.19.0 across version.py, the Void template and the synced docs; changelog entry; Gitea project URLs in pyproject.toml; per-backend ladder thresholds measured in #19; the xbps release workflow is manual-only, and packaging/void/publish-release.sh publishes a locally built, verified repository to the Gitea releases.
790 lines
30 KiB
Python
790 lines
30 KiB
Python
"""Offline guards for the signed xbps release workflow (tickets #28 and #29).
|
|
|
|
Releasing the Void repository is a local maintainer run on an x86_64 Void host;
|
|
the workflow ``.gitea/workflows/xbps-release.yml`` is manual dispatch only and is
|
|
kept for a future x86_64 host-mode runner. It 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_is_manual_dispatch_only_and_never_runs_on_a_tag_push() -> None:
|
|
"""A tag push must not start the job: releases are built and published locally.
|
|
|
|
The only runner with the bongbetic-ci label is ARM64 and cannot build the
|
|
glibc x86_64 release, so a tag-triggered run would fail (or worse, publish a
|
|
partial set). No schedule, no push, no pull request trigger.
|
|
"""
|
|
triggers = _workflow()[True] # YAML 1.1 reads the bare key `on` as True
|
|
assert set(triggers) == {"workflow_dispatch"}, triggers
|
|
assert "push" not in triggers and "schedule" not in 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":
|
|
value = argv[i + 1]
|
|
if value.startswith("@"):
|
|
# curl reads a header from a file, which keeps the token off argv.
|
|
with open(value[1:], encoding="utf-8") as handle:
|
|
value = handle.read().strip()
|
|
headers.append(value)
|
|
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"
|