orisys-sdk/build/release/pypi.py

"""Upload the certified wheels to TestPyPI, then (on confirmation) to PyPI.

``python build/run.py release pypi`` runs:

1. certified-wheels gate (same contract as ``release github``)
2. ``twine check`` on every wheel - the wheel metadata is generated, so its
   long description and classifiers are worth validating before an upload that
   cannot be undone
3. refuse if the version already exists on the target index (read-only probe)
4. upload to TestPyPI, then verify by fetching it back from the index
5. stop there unless ``--confirm-pypi`` is given; then repeat 3-4 against PyPI

PyPI uploads are permanent: a version number can never be reused, and a bad
artifact cannot be replaced. Hence the TestPyPI stage, the read-only existence
probe, and ``--apply`` being required even to reach TestPyPI.
"""

from __future__ import annotations

import json
import shutil
import subprocess
import sys
import tempfile
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from pathlib import Path

_BUILD_ROOT = Path(__file__).resolve().parents[1]
if str(_BUILD_ROOT) not in sys.path:
    sys.path.insert(0, str(_BUILD_ROOT))

from common.paths import REPO_ROOT  # noqa: E402
from release.publish import GateResult, PublishGateError, load_gate  # noqa: E402

INDEX_JSON = {
    "testpypi": "https://test.pypi.org/pypi/orisys/json",
    "pypi": "https://pypi.org/pypi/orisys/json",
}
INDEX_SIMPLE = {
    "testpypi": "https://test.pypi.org/simple/",
    "pypi": "https://pypi.org/simple/",
}


@dataclass
class Options:
    version: str
    apply: bool = False
    confirm_pypi: bool = False
    repository: str = "testpypi"
    verify_install: bool = True
    skip_check: bool = False


@dataclass
class IndexState:
    versions: list[str] = field(default_factory=list)
    error: str | None = None

    @property
    def known(self) -> bool:
        return self.error is None


def twine_check_argv(wheels: list[Path]) -> list[str]:
    return [sys.executable, "-m", "twine", "check", *[str(w) for w in wheels]]


def twine_upload_argv(wheels: list[Path], repository: str | None) -> list[str]:
    argv = [sys.executable, "-m", "twine", "upload"]
    if repository:
        argv += ["--repository", repository]
    argv += [str(wheel) for wheel in wheels]
    return argv


def credentials_hint() -> str | None:
    """Return guidance when twine has no obvious credentials, else ``None``.

    twine itself reads ``~/.pypirc`` or ``TWINE_USERNAME``/``TWINE_PASSWORD``;
    failing here is much better than failing halfway through an upload.
    """
    import os

    if os.environ.get("TWINE_PASSWORD") or os.environ.get("TWINE_USERNAME"):
        return None
    pypirc = Path.home() / ".pypirc"
    if pypirc.is_file():
        return None
    return (
        "no PyPI credentials found: create ~/.pypirc with [testpypi] and [pypi] "
        "sections, or set TWINE_USERNAME=__token__ and TWINE_PASSWORD=<token>. "
        "Never commit the token."
    )


def index_state(index: str, *, timeout: float = 20.0) -> IndexState:
    """Read the published versions of orisys from an index (read-only)."""
    url = INDEX_JSON[index]
    request = urllib.request.Request(
        url, headers={"User-Agent": "orisys-release-publish", "Accept": "application/json"}
    )
    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:
            payload = json.load(response)
    except urllib.error.HTTPError as exc:
        if exc.code == 404:
            return IndexState(versions=[])
        return IndexState(error=f"{index} returned HTTP {exc.code}")
    except Exception as exc:  # network, DNS, TLS, JSON
        return IndexState(error=f"{index} lookup failed: {type(exc).__name__}: {exc}")

    releases = payload.get("releases")
    if not isinstance(releases, dict):
        return IndexState(error=f"unexpected {index} response (no 'releases' mapping)")
    return IndexState(
        versions=sorted(name for name, files in releases.items() if files)
    )


def index_files(index: str, version: str, *, timeout: float = 20.0) -> list[str] | None:
    """Filenames published for ``version`` on ``index``, or ``None`` if absent."""
    url = INDEX_JSON[index]
    request = urllib.request.Request(
        url, headers={"User-Agent": "orisys-release-publish", "Accept": "application/json"}
    )
    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:
            payload = json.load(response)
    except Exception:
        return None
    files = (payload.get("releases") or {}).get(version)
    if not isinstance(files, list):
        return None
    return sorted(item.get("filename", "") for item in files)


def _upload(wheels: list[Path], repository: str | None) -> int:
    argv = twine_upload_argv(wheels, repository)
    print(f"[run ] {' '.join(str(part) for part in argv)}")
    completed = subprocess.run(argv, check=False)
    return completed.returncode


def _fetch_back(index: str, version: str, wheel: Path) -> int:
    """Download the freshly uploaded wheel back out of the index."""
    with tempfile.TemporaryDirectory(prefix="orisys-fetch-") as tmp:
        target = Path(tmp) / "out"
        argv = [
            sys.executable,
            "-m",
            "pip",
            "download",
            "--no-deps",
            "--no-cache-dir",
            "--index-url",
            INDEX_SIMPLE[index],
            "--dest",
            str(target),
            f"orisys=={version}",
        ]
        if index == "testpypi":
            # TestPyPI mirrors only what was uploaded there; deps live on PyPI.
            # --no-deps keeps this a pure "is the artifact fetchable" probe.
            argv += ["--extra-index-url", INDEX_SIMPLE["pypi"]]
        print(f"[run ] {' '.join(argv)}")
        completed = subprocess.run(argv, check=False)
        if completed.returncode != 0:
            return completed.returncode

        fetched = sorted(target.glob(f"orisys-{version}*"))
        if not fetched:
            print(f"[FAIL] pip download produced nothing for orisys=={version}")
            return 1
        print(f"[ok  ] fetched {[p.name for p in fetched]} from {index}")
        if wheel.name not in {path.name for path in fetched}:
            print(
                f"[warn] {wheel.name} was not the file fetched "
                "(the index may serve a different platform first)"
            )
        return 0


def publish(options: Options) -> int:
    version = options.version
    mode = "APPLY" if options.apply else "DRY RUN"
    print(f"=== release pypi: orisys {version} ({mode}) ===")

    # --- 1. gate -------------------------------------------------------
    try:
        gate: GateResult = load_gate(REPO_ROOT, version)
    except PublishGateError as exc:
        print(f"[FAIL] {exc}")
        return 1
    print(f"[ok  ] preflight report: {gate.report_path}")
    for wheel in gate.wheels:
        print(f"       [ok] {wheel.name} sha256={wheel.sha256[:12]}")
    for warning in gate.warnings:
        print(f"[warn] {warning}")

    wheels = [REPO_ROOT / "SDK" / version / record.name for record in gate.wheels]
    missing = [str(wheel) for wheel in wheels if not wheel.is_file()]
    if missing:
        print(f"[FAIL] wheels missing from disk: {missing}")
        return 1

    # --- 2. credentials + twine check (before anything irreversible) ----
    # Dry runs are previews, so missing credentials warn there and stop only when
    # an upload is actually about to happen.
    hint = credentials_hint()
    if hint:
        print(f"[{'FAIL' if options.apply else 'warn'}] {hint}")
        if options.apply:
            return 1
    else:
        print("[ok  ] PyPI credentials are configured (twine will read them)")

    if not options.skip_check:
        argv = twine_check_argv(wheels)
        print(f"[run ] {' '.join(str(part) for part in argv)}")
        completed = subprocess.run(argv, check=False)
        if completed.returncode != 0:
            print("[FAIL] twine check reported problems; fix the wheel metadata first")
            return 1
        print("[ok  ] twine check passed")

    # --- 3. phase plan -------------------------------------------------
    phases = ["testpypi"]
    if options.confirm_pypi:
        phases.append("pypi")

    print()
    for index in phases:
        state = index_state(index)
        if not state.known:
            print(f"[FAIL] {state.error}; cannot safely check for duplicate uploads")
            return 1
        if version in state.versions:
            print(
                f"[FAIL] orisys {version} already exists on {index} "
                f"(published: {', '.join(state.versions) or 'none'}) - "
                "an index version cannot be reused"
            )
            return 1
        print(f"[ok  ] {index}: {version} not present (published: {', '.join(state.versions) or 'none'})")

    print()
    for index in phases:
        print(f"[plan] upload {len(wheels)} wheel(s) to {index}")
    if "pypi" not in phases:
        print(
            "[plan] stop after TestPyPI. Re-run with "
            "`--confirm-pypi --apply` to publish to the real PyPI."
        )
    print()

    if not options.apply:
        print("dry run: nothing was uploaded. Re-run with --apply to execute.")
        return 0

    # --- 4. upload ----------------------------------------------------
    for index in phases:
        if _upload(wheels, index) != 0:
            print(f"[FAIL] twine upload to {index} failed")
            return 1
        print(f"[ok  ] uploaded to {index}")

        published = index_files(index, version)
        if published is None:
            print(
                f"[FAIL] {index} does not serve orisys {version} yet; "
                "the upload may still be indexing, verify manually"
            )
            return 1
        expected = {wheel.name for wheel in wheels}
        if set(published) != expected:
            print(
                f"[FAIL] {index} file set mismatch:\n  expected {sorted(expected)}\n"
                f"  got      {sorted(published)}"
            )
            return 1
        print(f"[ok  ] {index} serves all {len(published)} file(s) for {version}")

        if options.verify_install:
            if _fetch_back(index, version, wheels[0]) != 0:
                print(f"[FAIL] {index} artifact is not fetchable")
                return 1

    if "pypi" not in phases:
        print()
        print("=== release pypi: TestPyPI stage PASSED; PyPI untouched ===")
        print("next: python build/run.py release pypi --confirm-pypi --apply")
        return 0

    print()
    print("=== release pypi PASSED (TestPyPI + PyPI) ===")
    return 0