added preflight check
This commit is contained in:
@@ -0,0 +1,141 @@
|
||||
"""Version consistency and duplicate-release guards for the release gate."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
PYPI_JSON_URL = "https://pypi.org/pypi/orisys/json"
|
||||
|
||||
# orisys-<version>-<pytag>-<abitag>-<platformtag>.whl
|
||||
_WHEEL_NAME_RE = re.compile(r"^orisys-(?P<version>[^-]+)-[^-]+-[^-]+-.+\.whl$")
|
||||
|
||||
|
||||
def wheel_version(wheel_path: Path) -> str | None:
|
||||
match = _WHEEL_NAME_RE.match(Path(wheel_path).name)
|
||||
return match.group("version") if match else None
|
||||
|
||||
|
||||
def sdk_versions(repo_root: Path) -> list[str]:
|
||||
"""Version directory names present under ``SDK/``, in ascending order.
|
||||
|
||||
Only digit-leading names count as versions, so stray directories (``tmp``,
|
||||
``latest``) never appear as releases.
|
||||
"""
|
||||
sdk_root = Path(repo_root) / "SDK"
|
||||
if not sdk_root.is_dir():
|
||||
return []
|
||||
|
||||
return sorted(
|
||||
(
|
||||
entry.name
|
||||
for entry in sdk_root.iterdir()
|
||||
if entry.is_dir() and entry.name[:1].isdigit()
|
||||
),
|
||||
key=_version_sort_key,
|
||||
)
|
||||
|
||||
|
||||
def sdk_wheels(repo_root: Path, version: str) -> list[Path]:
|
||||
"""Wheels already published into ``SDK/<version>/``."""
|
||||
version_dir = Path(repo_root) / "SDK" / version
|
||||
if not version_dir.is_dir():
|
||||
return []
|
||||
return sorted(version_dir.glob("*.whl"))
|
||||
|
||||
|
||||
def dynamic_version_attr(repo_root: Path) -> str | None:
|
||||
"""Return pyproject's configured dynamic version attr (e.g. orisys.__version__)."""
|
||||
from common.pyproject_utils import load_pyproject
|
||||
|
||||
config = (
|
||||
load_pyproject(Path(repo_root))
|
||||
.get("tool", {})
|
||||
.get("setuptools", {})
|
||||
.get("dynamic", {})
|
||||
.get("version")
|
||||
)
|
||||
if isinstance(config, dict):
|
||||
attr = config.get("attr")
|
||||
if isinstance(attr, str):
|
||||
return attr
|
||||
return None
|
||||
|
||||
|
||||
def pypi_released_versions(timeout: float = 20.0) -> tuple[list[str] | None, str | None]:
|
||||
"""Read the published version list from PyPI.
|
||||
|
||||
Returns ``(versions, error)``; ``versions`` is ``None`` when the lookup
|
||||
failed. This is the only network call in the release gate and it is
|
||||
read-only, so it is safe to run before an irreversible upload.
|
||||
"""
|
||||
request = urllib.request.Request(
|
||||
PYPI_JSON_URL,
|
||||
headers={"User-Agent": "orisys-release-preflight", "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 [], None
|
||||
return None, f"PyPI returned HTTP {exc.code}"
|
||||
except Exception as exc: # network, DNS, TLS, JSON
|
||||
return None, f"{type(exc).__name__}: {exc}"
|
||||
|
||||
releases = payload.get("releases")
|
||||
if not isinstance(releases, dict):
|
||||
return None, "unexpected PyPI response shape (no 'releases' mapping)"
|
||||
listed = [
|
||||
str(version)
|
||||
for version, files in releases.items()
|
||||
if isinstance(files, list) and files
|
||||
]
|
||||
return sorted(listed, key=_version_sort_key), None
|
||||
|
||||
|
||||
def _version_sort_key(name: str):
|
||||
parts = re.findall(r"\d+", name)
|
||||
return ([int(part) for part in parts] or [0], name)
|
||||
|
||||
|
||||
def conda_envs_root() -> Path | None:
|
||||
"""Locate the conda envs directory the same way build/wheel/win.bat does."""
|
||||
override = os.getenv("CONDA_ENVS_ROOT")
|
||||
if override:
|
||||
path = Path(override)
|
||||
return path if path.is_dir() else None
|
||||
|
||||
home = Path.home()
|
||||
candidates = (
|
||||
home / "anaconda3" / "envs",
|
||||
home / "miniconda3" / "envs",
|
||||
Path(os.getenv("LOCALAPPDATA", "")) / "anaconda3" / "envs",
|
||||
Path(os.getenv("LOCALAPPDATA", "")) / "miniconda3" / "envs",
|
||||
)
|
||||
for candidate in candidates:
|
||||
if candidate.is_dir():
|
||||
return candidate
|
||||
return None
|
||||
|
||||
|
||||
def python_for_tag(tag: str, envs_root: Path | None = None) -> Path | None:
|
||||
"""Resolve an interpreter for a wheel tag like ``cp310`` (env ``py310``)."""
|
||||
digits = tag[2:] if tag.startswith("cp") else tag
|
||||
env_name = f"py{digits}"
|
||||
root = envs_root or conda_envs_root()
|
||||
if root is None:
|
||||
return None
|
||||
|
||||
for relative in (
|
||||
Path(env_name) / "python.exe",
|
||||
Path(env_name) / "bin" / "python",
|
||||
):
|
||||
candidate = root / relative
|
||||
if candidate.is_file():
|
||||
return candidate
|
||||
return None
|
||||
Reference in New Issue
Block a user