diff --git a/scripts/publish_zenodo_release.py b/scripts/publish_zenodo_release.py index cd4cee6..83db4fc 100644 --- a/scripts/publish_zenodo_release.py +++ b/scripts/publish_zenodo_release.py @@ -13,6 +13,7 @@ import argparse import json import os +import re import sys import tempfile import urllib.error @@ -22,13 +23,67 @@ ZENODO_API = "https://zenodo.org/api" GITHUB_REPO = "brunomartinsmv/eto-methods-comparison" -CHANGELOG_ANCHOR = "200---2026-07-12" +DEFAULT_TIMEOUT = 120 +DEFAULT_ZENODO_JSON = Path(".zenodo.json") +DEFAULT_CHANGELOG = Path("CHANGELOG.md") class ZenodoError(RuntimeError): pass +def _github_heading_anchor(heading: str) -> str: + slug = re.sub(r"[^\w\s-]", "", heading.lower()) + return re.sub(r"\s+", "-", slug.strip()) + + +def _changelog_entry_for_tag(tag: str, changelog_path: Path = DEFAULT_CHANGELOG) -> tuple[str, str]: + version = tag.removeprefix("v") + pattern = re.compile( + rf"^## \[{re.escape(version)}\] - (\d{{4}}-\d{{2}}-\d{{2}})\s*$", + re.MULTILINE, + ) + match = pattern.search(changelog_path.read_text(encoding="utf-8")) + if match is None: + raise ZenodoError( + f"No CHANGELOG entry found for {tag!r} in {changelog_path}. " + f"Expected a heading like '## [{version}] - YYYY-MM-DD'." + ) + heading = match.group(0).lstrip("#").strip() + return match.group(1), _github_heading_anchor(heading) + + +def _build_metadata( + tag: str, + *, + zenodo_path: Path = DEFAULT_ZENODO_JSON, + changelog_path: Path = DEFAULT_CHANGELOG, +) -> dict: + base = json.loads(zenodo_path.read_text(encoding="utf-8")) + publication_date, anchor = _changelog_entry_for_tag(tag, changelog_path) + changelog_url = ( + f"https://github.com/{GITHUB_REPO}/blob/{tag}/CHANGELOG.md#{anchor}" + ) + description = base["description"] + if not description.startswith("

"): + description = f"

{description}

" + base["description"] = ( + f"{description}" + f'

Changelog: {changelog_url}

' + ) + base["publication_date"] = publication_date + base["version"] = tag + base["related_identifiers"] = [ + { + "relation": "isSupplementTo", + "identifier": f"https://github.com/{GITHUB_REPO}/tree/{tag}", + "resource_type": "software", + "scheme": "url", + } + ] + return {"metadata": base} + + def _request( method: str, url: str, @@ -36,92 +91,67 @@ def _request( token: str, data: bytes | None = None, headers: dict[str, str] | None = None, + timeout: float = DEFAULT_TIMEOUT, ) -> dict: request_headers = {"Authorization": f"Bearer {token}"} if headers: request_headers.update(headers) request = urllib.request.Request(url, data=data, headers=request_headers, method=method) try: - with urllib.request.urlopen(request) as response: + with urllib.request.urlopen(request, timeout=timeout) as response: body = response.read().decode("utf-8") return json.loads(body) if body else {} except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace") raise ZenodoError(f"{method} {url} failed ({exc.code}): {detail}") from exc + except urllib.error.URLError as exc: + raise ZenodoError(f"{method} {url} failed: {exc.reason}") from exc -def _download(url: str, destination: Path) -> None: - with urllib.request.urlopen(url) as response, destination.open("wb") as handle: - handle.write(response.read()) +def _download(url: str, destination: Path, *, timeout: float = DEFAULT_TIMEOUT) -> None: + try: + with urllib.request.urlopen(url, timeout=timeout) as response, destination.open("wb") as handle: + handle.write(response.read()) + except urllib.error.HTTPError as exc: + detail = exc.read().decode("utf-8", errors="replace") + raise ZenodoError(f"GET {url} failed ({exc.code}): {detail}") from exc + except urllib.error.URLError as exc: + raise ZenodoError(f"GET {url} failed: {exc.reason}") from exc -def _delete_inherited_files(deposition: dict, token: str) -> None: +def _delete_inherited_files(deposition: dict, token: str, *, timeout: float) -> None: for file_info in deposition.get("files", []): - _request("DELETE", file_info["links"]["self"], token=token) + _request("DELETE", file_info["links"]["self"], token=token, timeout=timeout) -def _build_metadata(tag: str) -> dict: - changelog_url = ( - f"https://github.com/{GITHUB_REPO}/blob/{tag}/CHANGELOG.md#{CHANGELOG_ANCHOR}" - ) - return { - "metadata": { - "title": f"{GITHUB_REPO}: {tag}", - "upload_type": "software", - "description": ( - "

Major release with full ET0 computation pipeline, UX commands, " - "uncertainty and sensitivity diagnostics, safety-net CI, and LaTeX " - f"equation documentation. Full changelog: " - f'{changelog_url}

' - ), - "publication_date": "2026-07-12", - "version": tag, - "access_right": "open", - "license": "mit", - "creators": [ - { - "name": "Vieira, Bruno Martins M.", - "affiliation": "Universidade Federal do Mato Grosso", - } - ], - "keywords": [ - "evapotranspiration", - "reference evapotranspiration", - "Penman-Monteith", - "FAO-56", - "reproducible research", - "scientific computing", - ], - "related_identifiers": [ - { - "relation": "isSupplementTo", - "identifier": f"https://github.com/{GITHUB_REPO}/tree/{tag}", - "resource_type": "software", - "scheme": "url", - } - ], - } - } - - -def publish_release(*, tag: str, deposition_id: str, token: str) -> dict: +def publish_release( + *, + tag: str, + deposition_id: str, + token: str, + dry_run: bool = False, + timeout: float = DEFAULT_TIMEOUT, + zenodo_path: Path = DEFAULT_ZENODO_JSON, + changelog_path: Path = DEFAULT_CHANGELOG, +) -> dict: new_version = _request( "POST", f"{ZENODO_API}/deposit/depositions/{deposition_id}/actions/newversion", token=token, + timeout=timeout, ) draft_url = new_version["links"]["latest_draft"] - draft = _request("GET", draft_url, token=token) + draft = _request("GET", draft_url, token=token, timeout=timeout) draft_id = str(draft["id"]) bucket_url = draft["links"]["bucket"] - _delete_inherited_files(draft, token) + _delete_inherited_files(draft, token, timeout=timeout) archive_name = f"{GITHUB_REPO.replace('/', '-')}-{tag}.zip" archive_url = f"https://github.com/{GITHUB_REPO}/archive/refs/tags/{tag}.zip" with tempfile.TemporaryDirectory() as tmp_dir: archive_path = Path(tmp_dir) / archive_name - _download(archive_url, archive_path) + _download(archive_url, archive_path, timeout=timeout) with archive_path.open("rb") as handle: _request( "PUT", @@ -129,22 +159,29 @@ def publish_release(*, tag: str, deposition_id: str, token: str) -> dict: token=token, data=handle.read(), headers={"Content-Type": "application/octet-stream"}, + timeout=timeout, ) - _request( + updated = _request( "PUT", f"{ZENODO_API}/deposit/depositions/{draft_id}", token=token, - data=json.dumps(_build_metadata(tag)).encode("utf-8"), + data=json.dumps(_build_metadata(tag, zenodo_path=zenodo_path, changelog_path=changelog_path)).encode( + "utf-8" + ), headers={"Content-Type": "application/json"}, + timeout=timeout, ) - published = _request( + if dry_run: + return updated + + return _request( "POST", f"{ZENODO_API}/deposit/depositions/{draft_id}/actions/publish", token=token, + timeout=timeout, ) - return published def main(argv: list[str] | None = None) -> int: @@ -160,25 +197,39 @@ def main(argv: list[str] | None = None) -> int: default=os.environ.get("ZENODO_ACCESS_TOKEN", ""), help="Zenodo personal access token (or set ZENODO_ACCESS_TOKEN)", ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Create and update a draft deposition without publishing", + ) + parser.add_argument( + "--timeout", + type=float, + default=DEFAULT_TIMEOUT, + help=f"HTTP timeout in seconds (default: {DEFAULT_TIMEOUT})", + ) args = parser.parse_args(argv) if not args.token: print("Missing Zenodo token. Set ZENODO_ACCESS_TOKEN or pass --token.", file=sys.stderr) return 1 - published = publish_release(tag=args.tag, deposition_id=args.deposition_id, token=args.token) - doi = published.get("doi", "") - record_id = published.get("id", "") - print( - json.dumps( - { - "doi": doi, - "record_id": record_id, - "record_url": published.get("links", {}).get("record_html"), - }, - indent=2, - ) + result = publish_release( + tag=args.tag, + deposition_id=args.deposition_id, + token=args.token, + dry_run=args.dry_run, + timeout=args.timeout, ) + output = { + "doi": result.get("doi", ""), + "record_id": result.get("id", ""), + "record_url": result.get("links", {}).get("record_html"), + "state": result.get("state", ""), + } + if args.dry_run: + output["draft_url"] = result.get("links", {}).get("self") + print(json.dumps(output, indent=2)) return 0 diff --git a/tests/test_publish_zenodo_release.py b/tests/test_publish_zenodo_release.py new file mode 100644 index 0000000..a714548 --- /dev/null +++ b/tests/test_publish_zenodo_release.py @@ -0,0 +1,112 @@ +"""Tests for Zenodo release publishing helpers.""" + +from __future__ import annotations + +import json +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +from scripts import publish_zenodo_release as zenodo + + +def test_build_metadata_v200_uses_zenodo_json_and_changelog() -> None: + metadata = zenodo._build_metadata("v2.0.0")["metadata"] + + assert metadata["version"] == "v2.0.0" + assert metadata["publication_date"] == "2026-07-12" + assert "Reproducible Python pipeline" in metadata["description"] + assert "CHANGELOG.md#200---2026-07-12" in metadata["description"] + assert metadata["related_identifiers"][0]["identifier"].endswith("/tree/v2.0.0") + assert metadata["license"] == "mit" + + +def test_changelog_entry_for_tag_unknown_version_raises(tmp_path: Path) -> None: + changelog = tmp_path / "CHANGELOG.md" + changelog.write_text("## [Unreleased]\n\n## [2.0.0] - 2026-07-12\n", encoding="utf-8") + + with pytest.raises(zenodo.ZenodoError, match="No CHANGELOG entry found"): + zenodo._changelog_entry_for_tag("v9.9.9", changelog) + + +def test_github_heading_anchor_matches_github_slug() -> None: + assert zenodo._github_heading_anchor("[2.0.0] - 2026-07-12") == "200---2026-07-12" + + +def test_download_http_error_raises_zenodo_error(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + import urllib.error + + def fake_urlopen(*_args, **_kwargs): + raise urllib.error.HTTPError( + "https://example.com/archive.zip", + 404, + "Not Found", + hdrs=None, + fp=MagicMock(read=lambda: b"release not found"), + ) + + monkeypatch.setattr(zenodo.urllib.request, "urlopen", fake_urlopen) + + with pytest.raises(zenodo.ZenodoError, match="GET .* failed \\(404\\)"): + zenodo._download("https://example.com/archive.zip", tmp_path / "archive.zip") + + +def test_publish_release_dry_run_skips_publish(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + calls: list[tuple[str, str]] = [] + + def fake_request(method: str, url: str, **_kwargs) -> dict: + calls.append((method, url)) + if method == "POST" and url.endswith("/actions/newversion"): + return {"links": {"latest_draft": "https://zenodo.org/api/deposit/depositions/99"}} + if method == "GET": + return { + "id": 99, + "links": {"bucket": "https://zenodo.org/api/files/bucket/abc"}, + "files": [{"links": {"self": "https://zenodo.org/api/files/1"}}], + } + if method == "DELETE": + return {} + if method == "PUT" and "/depositions/" in url: + return { + "id": 99, + "state": "draft", + "links": { + "self": "https://zenodo.org/api/deposit/depositions/99", + "record_html": "https://zenodo.org/deposit/99", + }, + } + if method == "PUT": + return {} + raise AssertionError(f"Unexpected request: {method} {url}") + + def fake_download(url: str, destination: Path, **_kwargs) -> None: + destination.write_bytes(b"archive") + + monkeypatch.setattr(zenodo, "_request", fake_request) + monkeypatch.setattr(zenodo, "_download", fake_download) + + zenodo_json = tmp_path / ".zenodo.json" + changelog = tmp_path / "CHANGELOG.md" + zenodo_json.write_text( + json.dumps({"description": "Test description", "license": "mit"}), + encoding="utf-8", + ) + changelog.write_text("## [1.0.0] - 2026-02-11\n", encoding="utf-8") + + result = zenodo.publish_release( + tag="v1.0.0", + deposition_id="18615164", + token="token", + dry_run=True, + zenodo_path=zenodo_json, + changelog_path=changelog, + ) + + assert result["state"] == "draft" + assert not any(method == "POST" and url.endswith("/actions/publish") for method, url in calls) + + +def test_main_missing_token_returns_error(capsys: pytest.CaptureFixture[str]) -> None: + assert zenodo.main(["--tag", "v2.0.0", "--token", ""]) == 1 + assert "Missing Zenodo token" in capsys.readouterr().err