"""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"