xavierk
6609cfa6d9
A version tag push builds every Void template through packaging/void/repo.sh (podman, pinned Void glibc container), signs the repository, and publishes the complete glibc x86_64 set to the tag's release and to the rolling latest release that the client registration URL serves. Gates, each failing the job before anything publishes: the tag must equal v plus __version__ exactly, XBPS_SIGNING_KEY must be set, the staged repository must hold every class of file and nothing but glibc x86_64 data, every package must carry its signature, and the signing key must have the pinned fingerprint. The key lives in a 0600 temp file that an always-run step removes. Nothing commits, pushes, or bumps templates. The offline tests run the workflow's own gate, verify, build and publish scripts against a stateful stand-in for the Gitea release API.
765 lines
29 KiB
Python
765 lines
29 KiB
Python
"""Offline guards for the signed xbps release workflow (ticket #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 against
|
|
``src/vocalinux/version.py``, requires the ``XBPS_SIGNING_KEY`` secret, builds
|
|
and signs only through ``packaging/void/repo.sh`` (podman, pinned Void glibc
|
|
container), verifies the staged repository, and publishes it 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, verify, build and
|
|
publish steps in a scratch directory, so they assert behaviour rather than
|
|
spelling:
|
|
|
|
- the shell of each step is executed with the environment the workflow
|
|
itself declares (``${{ ... }}`` expressions are rendered from an explicit
|
|
context, so a step wired to the wrong secret or token fails loudly);
|
|
- ``curl`` is replaced by a small 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 repository layout comes from what ``xbps-rindex`` writes
|
|
(``<pkg>.x86_64.xbps`` plus ``<pkg>.x86_64.xbps.sig2`` and ``x86_64-repodata``),
|
|
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 textwrap
|
|
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"
|
|
JOB = "xbps-release"
|
|
|
|
GATE_TAG = "Require the tag to equal the version in version.py"
|
|
GATE_KEY = "Require the repository signing key"
|
|
INSTALL_TOOLS = "Install the tools repo.sh and the publish step need"
|
|
BUILD = "Build and sign the repository"
|
|
VERIFY = "Verify the signed repository is complete"
|
|
PUBLISH = "Publish to the tag release and the latest release"
|
|
CLEANUP = "Remove the signing key"
|
|
STEP_ORDER = (GATE_TAG, GATE_KEY, INSTALL_TOOLS, BUILD, VERIFY, PUBLISH, CLEANUP)
|
|
|
|
API = "https://git.bongbetic.com/api/v1/repos/xavierk/voculinux"
|
|
TOKEN = "test-token-not-a-secret"
|
|
SHA = "0123456789abcdef0123456789abcdef01234567"
|
|
FAKE_KEY = "-----BEGIN RSA PRIVATE KEY-----\nline-one\nline-two\n-----END RSA PRIVATE KEY-----\n"
|
|
|
|
|
|
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,
|
|
) -> 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}
|
|
path = os.environ.get("PATH", "/usr/bin:/bin")
|
|
env["PATH"] = f"{path_prefix}{os.pathsep}{path}" if path_prefix else path
|
|
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")
|
|
return subprocess.run(
|
|
["bash", str(script)],
|
|
cwd=cwd,
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
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; verify precedes publish; key cleanup 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"]
|
|
assert _job()["env"]["XBPS_KEY_FILE"].startswith("/tmp/xbps-sign-key-")
|
|
|
|
|
|
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 }}"
|
|
|
|
|
|
def test_repo_sh_is_the_only_build_and_sign_path() -> None:
|
|
"""No step builds, indexes or signs by hand; only repo.sh does, once."""
|
|
code = _without_comments(_text())
|
|
assert code.count("packaging/void/repo.sh") == 1
|
|
for forbidden in ("xbps-src", "xbps-rindex", "openssl rsa", "xlint", "gpg "):
|
|
assert forbidden not in code, f"{forbidden!r} belongs inside repo.sh, not the workflow"
|
|
assert "repo.sh" in _run_text(BUILD)
|
|
|
|
|
|
def test_pipeline_uses_podman_and_never_docker() -> None:
|
|
code = _without_comments(_text()).lower()
|
|
assert "podman" in _run_text(INSTALL_TOOLS)
|
|
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
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _version_file(root: Path, body: str) -> None:
|
|
target = root / "src" / "vocalinux" / "version.py"
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_text(body, encoding="utf-8")
|
|
|
|
|
|
def _gate_tag(tmp_path: Path, tag: str, version_py: str) -> subprocess.CompletedProcess:
|
|
_version_file(tmp_path, version_py)
|
|
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, f'"""Version."""\n\n__version__ = "{version}"\n')
|
|
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_fails_when_the_version_cannot_be_read(tmp_path: Path) -> None:
|
|
result = _gate_tag(tmp_path, "v0.18.5", '"""Version."""\n__version_info__ = (0, 18, 5)\n')
|
|
assert result.returncode == 1
|
|
assert "__version__" in result.stderr
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Gate 2: the signing key
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
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_signing_key_is_staged_0600_with_newlines_intact(tmp_path: Path) -> None:
|
|
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 == 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"
|
|
|
|
|
|
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()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Build: repo.sh is handed the key file, nothing else
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_build_step_runs_repo_sh_with_the_key_file_only(tmp_path: Path) -> None:
|
|
"""repo.sh gets XBPS_SIGNING_KEY_FILE and never the secret itself."""
|
|
script = tmp_path / "packaging" / "void" / "repo.sh"
|
|
script.parent.mkdir(parents=True)
|
|
script.write_text(
|
|
textwrap.dedent("""\
|
|
#!/bin/sh
|
|
set -eu
|
|
[ -s "$XBPS_SIGNING_KEY_FILE" ] || exit 7
|
|
printf '%s' "$XBPS_SIGNING_KEY_FILE" > called-with
|
|
printf '%s' "${XBPS_SIGNING_KEY:-unset}" > secret-seen
|
|
mkdir -p publish/xbps/repo publish/xbps/keys
|
|
"""),
|
|
encoding="utf-8",
|
|
)
|
|
key_file = tmp_path / "key.pem"
|
|
key_file.write_text(FAKE_KEY, encoding="utf-8")
|
|
result = _run_step(BUILD, tmp_path, extra_env={"XBPS_KEY_FILE": str(key_file)})
|
|
assert result.returncode == 0, result.stderr
|
|
assert (tmp_path / "called-with").read_text(encoding="utf-8") == str(key_file)
|
|
assert (tmp_path / "secret-seen").read_text(encoding="utf-8") == "unset"
|
|
|
|
|
|
def test_build_step_fails_when_repo_sh_fails(tmp_path: Path) -> None:
|
|
"""A failing repo.sh (missing or non-RSA4096 key, build error) ends the job."""
|
|
script = tmp_path / "packaging" / "void" / "repo.sh"
|
|
script.parent.mkdir(parents=True)
|
|
script.write_text("#!/bin/sh\nexit 1\n", encoding="utf-8")
|
|
result = _run_step(BUILD, tmp_path, extra_env={"XBPS_KEY_FILE": str(tmp_path / "k")})
|
|
assert result.returncode != 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Verify: the full signed glibc x86_64 repository, nothing else
|
|
# ---------------------------------------------------------------------------
|
|
|
|
FINGERPRINT = "fa:d1:00:32:ce:b1:b9:fd:73:05:05:76:75:54:37:25"
|
|
SMOKE_SH = REPO_ROOT / "packaging" / "void" / "smoke.sh"
|
|
|
|
|
|
def _smoke_packages() -> List[str]:
|
|
"""The templates the smoke gate builds, in order: the repository's package set."""
|
|
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()
|
|
|
|
|
|
#: One built package per template, versioned the way xbps-src names them.
|
|
PACKAGES = [f"{name}-1.0_1" for name in _smoke_packages()]
|
|
|
|
|
|
def _stage(root: Path, fingerprint: str = FINGERPRINT) -> Path:
|
|
"""Lay out publish/xbps the way repo.sh stages it, plus the smoke gate's list."""
|
|
smoke = root / "packaging" / "void" / "smoke.sh"
|
|
smoke.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy(SMOKE_SH, smoke)
|
|
out = root / "publish" / "xbps"
|
|
repo, keys = out / "repo", out / "keys"
|
|
repo.mkdir(parents=True)
|
|
keys.mkdir()
|
|
for pkg in PACKAGES:
|
|
(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 test_the_pinned_fingerprint_is_what_the_verify_step_checks() -> None:
|
|
assert _job()["env"]["XBPS_KEY_FINGERPRINT"] == FINGERPRINT
|
|
|
|
|
|
def test_verify_accepts_the_complete_signed_repository(tmp_path: Path) -> None:
|
|
_stage(tmp_path)
|
|
result = _run_step(VERIFY, tmp_path)
|
|
assert result.returncode == 0, result.stderr
|
|
assert "complete signed glibc x86_64 repository" in result.stdout
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"missing",
|
|
[
|
|
f"repo/{PACKAGES[0]}.x86_64.xbps",
|
|
f"repo/{PACKAGES[-1]}.x86_64.xbps",
|
|
f"repo/{PACKAGES[0]}.x86_64.xbps.sig2",
|
|
"repo/x86_64-repodata",
|
|
"keys/FINGERPRINT",
|
|
f"keys/{FINGERPRINT}.plist",
|
|
],
|
|
)
|
|
def test_verify_rejects_a_repository_missing_any_class(tmp_path: Path, missing: str) -> None:
|
|
out = _stage(tmp_path)
|
|
(out / missing).unlink()
|
|
result = _run_step(VERIFY, tmp_path)
|
|
assert result.returncode == 1, f"removing {missing} must fail verification"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"extra",
|
|
[
|
|
"repo/voculinux-1.0_1.x86_64-musl.xbps",
|
|
"repo/voculinux-1.0_1.aarch64.xbps",
|
|
"repo/voculinux-1.0_1.noarch.xbps",
|
|
"repo/x86_64-musl-repodata",
|
|
"repo/aarch64-repodata",
|
|
"repo/notes.txt",
|
|
"keys/private.pem",
|
|
],
|
|
)
|
|
def test_verify_rejects_anything_but_glibc_x86_64_data(tmp_path: Path, extra: str) -> None:
|
|
out = _stage(tmp_path)
|
|
(out / extra).write_bytes(b"x")
|
|
result = _run_step(VERIFY, tmp_path)
|
|
assert result.returncode == 1, f"{extra} must not be publishable"
|
|
|
|
|
|
def test_verify_rejects_an_unsigned_package(tmp_path: Path) -> None:
|
|
"""One signature present does not excuse a package without its own."""
|
|
out = _stage(tmp_path)
|
|
(out / "repo" / f"{PACKAGES[-1]}.x86_64.xbps.sig2").write_bytes(b"")
|
|
result = _run_step(VERIFY, tmp_path)
|
|
assert result.returncode == 1
|
|
assert "no signature" in result.stderr
|
|
|
|
|
|
def test_verify_rejects_a_signature_whose_package_is_gone(tmp_path: Path) -> None:
|
|
out = _stage(tmp_path)
|
|
(out / "repo" / f"{PACKAGES[0]}.x86_64.xbps.sig2").write_bytes(b"sig")
|
|
(out / "repo" / "ghost-1.0_1.x86_64.xbps.sig2").write_bytes(b"sig")
|
|
result = _run_step(VERIFY, tmp_path)
|
|
assert result.returncode == 1
|
|
assert "without its package" in result.stderr
|
|
|
|
|
|
def test_verify_rejects_a_repository_signed_by_another_key(tmp_path: Path) -> None:
|
|
"""A throwaway or wrong key must never publish under the pinned fingerprint."""
|
|
other = "00:11:22:33:44:55:66:77:88:99:aa:bb:cc:dd:ee:ff"
|
|
_stage(tmp_path, fingerprint=other)
|
|
result = _run_step(VERIFY, tmp_path)
|
|
assert result.returncode == 1
|
|
assert FINGERPRINT in result.stderr and other in result.stderr
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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 _staged_names(out: Path) -> List[str]:
|
|
return sorted(p.name for sub in ("repo", "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_full_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" / "xbps")
|
|
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_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
|
|
# The tag release is created and filled before latest is touched.
|
|
posts = [e["data"]["tag_name"] for e in log if e["url"] == API + "/releases"]
|
|
assert posts == ["v0.18.5", "latest"]
|
|
|
|
|
|
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" / "xbps")
|
|
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_rerun_keeps_foreign_assets_on_the_tag_release(staged: Path) -> None:
|
|
"""Rerunning a tag replaces only its own files; hand-attached assets survive."""
|
|
gitea = FakeGitea(staged)
|
|
gitea.seed_release("v0.18.5", ["x86_64-repodata", "voculinux-0.18.5.tar.gz"])
|
|
assert gitea.run_publish(staged).returncode == 0
|
|
names = gitea.asset_names("v0.18.5")
|
|
assert "voculinux-0.18.5.tar.gz" in names
|
|
assert names.count("x86_64-repodata") == 1
|
|
assert set(_staged_names(staged / "publish" / "xbps")) <= set(names)
|
|
|
|
|
|
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")
|
|
|
|
|
|
@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_uploads_only_staged_files(staged: Path) -> None:
|
|
"""Nothing is built or created by hand: uploads are exactly the staged set."""
|
|
gitea = FakeGitea(staged)
|
|
assert gitea.run_publish(staged).returncode == 0
|
|
staged_names = set(_staged_names(staged / "publish" / "xbps"))
|
|
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(("repo/", "keys/")), path
|