diff --git a/.github/actions/upload-release-assets/action.yml b/.github/actions/upload-release-assets/action.yml new file mode 100644 index 00000000..39c42aad --- /dev/null +++ b/.github/actions/upload-release-assets/action.yml @@ -0,0 +1,55 @@ +name: Upload release artefacts +description: Upload staged artefacts to a GitHub release or validate them in dry-run mode. +inputs: + release-tag: + description: Git tag identifying the release to publish to. + required: true + bin-name: + description: Binary name used to derive artefact names. + required: true + dist-dir: + description: Directory containing staged artefacts. + required: false + default: dist + dry-run: + description: When true, only validate artefacts and print the upload plan. + required: false + default: "false" +outputs: + upload-error: + description: Whether the invocation detected an error. + value: ${{ steps.invoke.outputs.upload-error }} + error-message: + description: Summary of the error when ``upload-error`` is ``true``. + value: ${{ steps.invoke.outputs.error-message }} +runs: + using: composite + steps: + - id: invoke + name: Upload release artefacts + shell: bash + env: + INPUT_RELEASE_TAG: ${{ inputs.release-tag }} + INPUT_BIN_NAME: ${{ inputs.bin-name }} + INPUT_DIST_DIR: ${{ inputs.dist-dir }} + INPUT_DRY_RUN: ${{ inputs.dry-run }} + run: | + set -uo pipefail + log_file="$(mktemp)" + if "$GITHUB_WORKSPACE/scripts/upload_release_assets.py" >"$log_file" 2>&1; then + cat "$log_file" + { + echo "upload-error=false" + echo "error-message=" + } >>"$GITHUB_OUTPUT" + else + status=$? + cat "$log_file" + { + echo "upload-error=true" + echo "error-message<<'EOF'" + cat "$log_file" + printf 'Exit code: %s\n' "$status" + echo 'EOF' + } >>"$GITHUB_OUTPUT" + fi diff --git a/.github/workflows/release-dry-run.yml b/.github/workflows/release-dry-run.yml index 17fa02b3..4dfe3353 100644 --- a/.github/workflows/release-dry-run.yml +++ b/.github/workflows/release-dry-run.yml @@ -27,14 +27,17 @@ jobs: with: path: dist pattern: ${{ env.REPO_NAME }}-* - - name: Ensure artefacts exist - shell: bash + - id: validate + name: Validate artefacts + uses: ./.github/actions/upload-release-assets + with: + release-tag: ${{ github.ref_name }} + bin-name: ${{ needs.release.outputs.bin_name }} + dist-dir: dist + dry-run: "true" + - name: Check validation errors + if: steps.validate.outputs.upload-error == 'true' run: | - set -euo pipefail - shopt -s nullglob - mapfile -t files < <(find dist -type f) - if [ "${#files[@]}" -eq 0 ]; then - echo '::error title=Missing artefacts::No artefacts downloaded' - exit 1 - fi - printf 'Found %s artefact(s) in dist/\n' "${#files[@]}" + echo "Error validating release assets:" + printf '%s\n' "${{ steps.validate.outputs.error-message }}" + exit 1 diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 6ab5dc3f..877a2033 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -1,7 +1,7 @@ --- name: Release Binary -'on': +on: push: tags: - 'v*.*.*' @@ -15,6 +15,16 @@ name: Release Binary required: false type: boolean default: false + outputs: + bin_name: + description: Binary name extracted from Cargo.toml + value: ${{ jobs.metadata.outputs.bin_name }} + version: + description: Crate version resolved from Cargo.toml + value: ${{ jobs.metadata.outputs.version }} + should_publish: + description: Whether artefacts should be published to the release + value: ${{ jobs.metadata.outputs.should_publish }} env: REPO_NAME: ${{ github.event.repository.name }} @@ -298,52 +308,19 @@ jobs: with: path: dist pattern: ${{ env.REPO_NAME }}-* - - name: Upload artefacts to release - shell: bash - run: | - set -euo pipefail - shopt -s nullglob - bin_name="${BIN_NAME:?BIN_NAME must be provided}" - mapfile -t files < <( - find dist -type f \ - \( -name "*.deb" -o -name "*.rpm" -o -name "*.pkg" -o -name "*.msi" \ - -o -name "${bin_name}" -o -name "${bin_name}.exe" \ - -o -name "${bin_name}.1" -o -name "*.sha256" \) - -print - ) - if [ "${#files[@]}" -eq 0 ]; then - message='::error title=No artefacts uploaded::No files found in ' - message+='dist/' - echo "$message" - exit 1 - fi - declare -A seen=() - uploaded=0 - for file in "${files[@]}"; do - dir_name="$(basename "$(dirname "$file")")" - base_name="$(basename "$file")" - if [[ "$base_name" =~ \.(deb|rpm|pkg)$ ]]; then - asset_name="$base_name" - else - asset_name="${dir_name}-${base_name}" - fi - if [[ -n "${seen[$asset_name]:-}" ]]; then - printf '::error title=Duplicate release asset::Asset name '\''%s'\'' would be uploaded more than once\n' \ - "${asset_name}" - exit 1 - fi - seen[$asset_name]=1 - gh release upload "${{ github.ref_name }}" \ - "$file#${asset_name}" --clobber \ - && uploaded=$((uploaded + 1)) - done - if [ "$uploaded" -eq 0 ]; then - message='::error title=No artefacts uploaded::No files were ' - message+='published to the release' - echo "$message" - exit 1 - fi + - id: upload_assets + name: Upload artefacts to release + uses: ./.github/actions/upload-release-assets + with: + release-tag: ${{ github.ref_name }} + bin-name: ${{ needs.metadata.outputs.bin_name }} + dist-dir: dist env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} - BIN_NAME: ${{ needs.metadata.outputs.bin_name }} + - name: Check asset upload errors + if: steps.upload_assets.outputs.upload-error == 'true' + run: | + echo "Error uploading release assets:" + printf '%s\n' "${{ steps.upload_assets.outputs.error-message }}" + exit 1 diff --git a/.github/workflows/scripts/read_manifest.py b/.github/workflows/scripts/read_manifest.py index 33336785..477909c1 100755 --- a/.github/workflows/scripts/read_manifest.py +++ b/.github/workflows/scripts/read_manifest.py @@ -1,5 +1,43 @@ #!/usr/bin/env python3 -"""Utility helpers for extracting fields from Cargo.toml.""" +""" +Utility helpers for extracting fields from Cargo.toml. + +Summary +------- +Parse and extract package metadata fields from Cargo manifest files. + +Purpose +------- +Provide both a CLI tool and programmatic API for reading ``name`` and +``version`` fields from Cargo.toml manifests, with robust error handling +for missing files, invalid TOML, and unexpected structure. + +Usage +----- +CLI invocation:: + + python read_manifest.py name --manifest-path /path/to/Cargo.toml + python read_manifest.py version + +Programmatic usage:: + + from pathlib import Path + manifest = read_manifest(Path("Cargo.toml")) + name = get_field(manifest, "name") + +Examples +-------- +Extract the package name from a manifest:: + + $ python read_manifest.py name --manifest-path Cargo.toml + netsuke + +Use the CARGO_TOML_PATH environment variable:: + + $ export CARGO_TOML_PATH=/path/to/Cargo.toml + $ python read_manifest.py version + 1.2.3 +""" from __future__ import annotations @@ -10,18 +48,33 @@ import tomllib +PARSER_DESCRIPTION = " ".join( + [ + "Read selected fields from a Cargo.toml manifest and print them to", + "stdout.", + ] +) + def parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser( - description=( - "Read selected fields from a Cargo.toml manifest and print them to " - "stdout." - ) - ) + """ + Return the parsed CLI arguments for manifest field extraction. + + Returns + ------- + argparse.Namespace + Parsed arguments containing ``field`` (str) and optional + ``manifest_path`` (str or None). + + Examples + -------- + >>> args = parse_args() # With sys.argv = ["script.py", "name"] + >>> args.field + 'name' + """ + parser = argparse.ArgumentParser(description=PARSER_DESCRIPTION) parser.add_argument( - "field", - choices=("name", "version"), - help="The manifest field to print." + "field", choices=("name", "version"), help="The manifest field to print." ) parser.add_argument( "--manifest-path", @@ -36,23 +89,96 @@ def parse_args() -> argparse.Namespace: def read_manifest(path: Path) -> dict[str, object]: + """ + Load and return the parsed Cargo manifest as a dictionary. + + Parameters + ---------- + path : Path + Path to the ``Cargo.toml`` file. + + Returns + ------- + dict[str, object] + Parsed manifest fields keyed by section. + + Raises + ------ + FileNotFoundError + If the manifest file does not exist. + tomllib.TOMLDecodeError + If the manifest contains invalid TOML syntax. + + Examples + -------- + >>> from pathlib import Path + >>> manifest_path = Path("Cargo.toml") + >>> data = read_manifest(manifest_path) + >>> "package" in data + True + """ if not path.is_file(): - raise FileNotFoundError(f"Manifest {path} does not exist") + message = f"Manifest {path} does not exist" + raise FileNotFoundError(message) with path.open("rb") as handle: return tomllib.load(handle) def get_field(manifest: dict[str, object], field: str) -> str: + """ + Extract a package field from the manifest, raising if it is missing. + + Parameters + ---------- + manifest : dict[str, object] + The parsed Cargo manifest dictionary. + field : str + The package field to extract, such as ``"name"`` or ``"version"``. + + Returns + ------- + str + The non-empty field value from the package section. + + Raises + ------ + KeyError + If the package table is missing or the field is absent or blank. + + Examples + -------- + >>> manifest = {"package": {"name": "netsuke", "version": "1.2.3"}} + >>> get_field(manifest, "name") + 'netsuke' + """ package = manifest.get("package") or {} if not isinstance(package, dict): - raise KeyError("package table missing from manifest") + message = "package table missing from manifest" + raise KeyError(message) value = package.get(field, "") if not isinstance(value, str) or not value: - raise KeyError(f"package.{field} is missing") + message = f"package.{field} is missing" + raise KeyError(message) return value def main() -> int: + """ + Entry point for the manifest reader CLI. + + Returns + ------- + int + Exit code: 0 for success, 1 for errors (missing file, invalid + TOML, or missing fields). + + Examples + -------- + Typical CLI invocation:: + + $ python read_manifest.py name --manifest-path Cargo.toml + netsuke + """ args = parse_args() manifest_path = args.manifest_path or os.environ.get( "CARGO_TOML_PATH", "Cargo.toml" @@ -60,7 +186,7 @@ def main() -> int: try: manifest = read_manifest(Path(manifest_path)) value = get_field(manifest, args.field) - except (KeyError, FileNotFoundError) as exc: + except (KeyError, FileNotFoundError, tomllib.TOMLDecodeError) as exc: print(exc, file=sys.stderr) return 1 print(value, end="") diff --git a/ruff.toml b/ruff.toml index d1db8276..50c5da8c 100644 --- a/ruff.toml +++ b/ruff.toml @@ -38,6 +38,7 @@ select = [ ] per-file-ignores = {"**/test_*.py" = ["S101"]} ignore = ["D205"] +task-tags = ["TODO", "FIXME"] [lint.flake8-import-conventions] # Declare the banned `from` imports. @@ -61,5 +62,5 @@ msgspec = "ms" typing = "typ" [lint.pydocstyle] -# Enforce NumPy docstring +# Enforce NumPy docstring convention = "numpy" diff --git a/scripts/tests/test_upload_release_assets.py b/scripts/tests/test_upload_release_assets.py new file mode 100644 index 00000000..52c12d52 --- /dev/null +++ b/scripts/tests/test_upload_release_assets.py @@ -0,0 +1,162 @@ +"""Tests for the upload_release_assets helper script.""" + +from __future__ import annotations + +import importlib.util +import subprocess +import sys +import typing as typ +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[2] +SCRIPT_PATH = REPO_ROOT / "scripts" / "upload_release_assets.py" + + +if typ.TYPE_CHECKING: + from types import ModuleType +else: + ModuleType = type(sys) + + +@pytest.fixture(scope="session") +def module() -> ModuleType: + """Load the upload_release_assets script once for reuse across tests.""" + spec = importlib.util.spec_from_file_location("upload_release_assets", SCRIPT_PATH) + if spec is None: + message = "Failed to create module spec for upload_release_assets" + raise RuntimeError(message) + if spec.loader is None: + message = "Module spec missing loader for upload_release_assets" + raise RuntimeError(message) + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def create_file(path: Path, content: bytes = b"data") -> None: + """Create a file with the given content, ensuring parent directories exist.""" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(content) + + +def test_discover_assets_collects_expected_files( + module: ModuleType, tmp_path: Path +) -> None: + """Verify that asset discovery preserves expected ordering and names.""" + dist = tmp_path / "dist" + create_file(dist / "linux" / "netsuke", b"binary") + create_file(dist / "linux" / "netsuke.sha256", b"checksum") + create_file(dist / "linux" / "netsuke.deb", b"deb") + create_file(dist / "windows" / "netsuke.exe", b"exe") + create_file(dist / "windows" / "netsuke.msi", b"msi") + create_file(dist / "man" / "netsuke.1", b"man") + + assets = module.discover_assets(dist, bin_name="netsuke") + + assert [asset.asset_name for asset in assets] == [ + "linux-netsuke", + "linux-netsuke.deb", + "linux-netsuke.sha256", + "man-netsuke.1", + "windows-netsuke.exe", + "windows-netsuke.msi", + ], "Asset discovery order does not match expected sequence" + + +def test_discover_assets_disambiguates_duplicate_filenames( + module: ModuleType, tmp_path: Path +) -> None: + """It prefixes nested directories to avoid collisions for package files.""" + dist = tmp_path / "dist" + create_file(dist / "linux" / "pkg" / "netsuke.pkg", b"pkg-linux") + create_file(dist / "windows" / "pkg" / "netsuke.pkg", b"pkg-windows") + + assets = module.discover_assets(dist, bin_name="netsuke") + + assert [ + asset.asset_name for asset in assets if asset.asset_name.endswith(".pkg") + ] == [ + "linux__pkg-netsuke.pkg", + "windows__pkg-netsuke.pkg", + ] + + +def test_discover_assets_rejects_empty_files( + module: ModuleType, tmp_path: Path +) -> None: + """It refuses to publish zero-byte artefacts.""" + dist = tmp_path / "dist" + create_file(dist / "linux" / "netsuke", b"") + + with pytest.raises(module.AssetError) as exc: + module.discover_assets(dist, bin_name="netsuke") + + assert "is empty" in str(exc.value) + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + (True, True), + (False, False), + ("true", True), + ("false", False), + ("YES", True), + ("no", False), + ("1", True), + ("0", False), + (" on ", True), + (" off ", False), + ("", False), + ], +) +def test_coerce_bool_handles_common_inputs( + module: ModuleType, value: object, expected: object +) -> None: + """The coercion helper accepts boolean strings in various casings.""" + assert module._coerce_bool(value) is expected + + +def test_coerce_bool_rejects_unknown_input(module: ModuleType) -> None: + """Unexpected values surface a descriptive error.""" + with pytest.raises(ValueError, match="Cannot interpret 'maybe'"): + module._coerce_bool("maybe") + + +def test_cli_dry_run_outputs_summary(module: ModuleType, tmp_path: Path) -> None: + """It prints the planned gh commands instead of uploading during dry runs.""" + dist = tmp_path / "dist" + create_file(dist / "linux" / "netsuke", b"binary") + create_file(dist / "linux" / "netsuke.sha256", b"checksum") + + result = subprocess.run( # noqa: S603 # Security: CLI invocation uses trusted arguments in tests. + [ + sys.executable, + str(SCRIPT_PATH), + "--release-tag", + "v1.2.3", + "--bin-name", + "netsuke", + "--dist-dir", + str(dist), + "--dry-run", + ], + capture_output=True, + text=True, + check=False, + ) + + assert result.returncode == 0, "CLI dry-run should exit successfully" + assert "Planned uploads:" in result.stdout, ( + "Dry-run output should include upload summary" + ) + assert "linux-netsuke" in result.stdout, "Dry-run output should list linux binary" + assert "linux-netsuke.sha256" in result.stdout, ( + "Dry-run output should list checksum file" + ) + assert "[dry-run] gh release upload v1.2.3" in result.stdout, ( + "Dry-run output should show gh command" + ) diff --git a/scripts/upload_release_assets.py b/scripts/upload_release_assets.py new file mode 100755 index 00000000..d69330b0 --- /dev/null +++ b/scripts/upload_release_assets.py @@ -0,0 +1,292 @@ +#!/usr/bin/env -S uv run python +# /// script +# requires-python = ">=3.13" +# dependencies = ["cyclopts>=2.9", "plumbum"] +# /// + +"""Upload packaged release artefacts to a GitHub release. + +The script discovers artefacts in a staging directory, validates their +filenames and sizes, and optionally uploads them using the GitHub CLI. It is +idempotent and supports a dry-run mode used by the release dry-run workflow to +assert expected asset names without mutating state. + +Examples +-------- +Upload artefacts to the ``v1.2.3`` release:: + + upload_release_assets --release-tag v1.2.3 --bin-name netsuke + +Inspect the planned uploads without publishing anything:: + + upload_release_assets --release-tag v1.2.3 --bin-name netsuke --dry-run +""" + +from __future__ import annotations + +import dataclasses as dc +import sys +import typing as typ +from pathlib import Path + +import cyclopts +from cyclopts import App, Parameter +from plumbum import local +from plumbum.commands import CommandNotFound, ProcessExecutionError + +if typ.TYPE_CHECKING: + import collections.abc as cabc + + from plumbum.commands.base import BoundCommand + + +class AssetError(RuntimeError): + """Raised when the staged artefacts are invalid.""" + + +@dc.dataclass(frozen=True) +class ReleaseAsset: + """Artefact staged for upload to a GitHub release.""" + + path: Path + asset_name: str + size: int + + +app = App(config=cyclopts.config.Env("INPUT_", command=False)) + + +def _is_candidate(path: Path, bin_name: str) -> bool: + name = path.name + if name in {bin_name, f"{bin_name}.exe", f"{bin_name}.1"}: + return True + if name.endswith(".sha256"): + return True + return path.suffix in {".deb", ".rpm", ".pkg", ".msi"} + + +def _coerce_bool(value: object) -> bool: + """Return ``value`` as a strict boolean.""" + if isinstance(value, bool): + return value + if not isinstance(value, str): + message = f"Cannot interpret {value!r} as a boolean" + raise TypeError(message) + normalised = value.strip().lower() + if normalised in {"", "false", "0", "no", "off"}: + return False + if normalised in {"true", "1", "yes", "on"}: + return True + message = f"Cannot interpret {value!r} as a boolean" + raise ValueError(message) + + +def _resolve_asset_name(path: Path, *, dist_dir: Path) -> str: + """Return a unique asset name derived from ``path`` within ``dist_dir``.""" + relative_path = path.relative_to(dist_dir) + if relative_path.parent == Path(): + return relative_path.name + prefix = relative_path.parent.as_posix().replace("/", "__") + return f"{prefix}-{relative_path.name}" + + +def _iter_candidate_paths(dist_dir: Path, bin_name: str) -> cabc.Iterator[Path]: + for path in sorted(dist_dir.rglob("*")): + if path.is_file() and _is_candidate(path, bin_name): + yield path + + +def _require_non_empty(path: Path) -> int: + size = path.stat().st_size + if size <= 0: + message = f"Artefact {path} is empty" + raise AssetError(message) + return size + + +def _register_asset(asset_name: str, path: Path, seen: dict[str, Path]) -> None: + if previous := seen.get(asset_name): + message = ( + "Asset name collision: " + f"{asset_name} would upload both {previous} and {path}" + ) + raise AssetError(message) + seen[asset_name] = path + + +def discover_assets(dist_dir: Path, *, bin_name: str) -> list[ReleaseAsset]: + """Return the artefacts that should be published. + + Parameters + ---------- + dist_dir : Path + Root directory that contains the staged artefacts. + bin_name : str + Binary name used to match platform-specific artefacts. + + Returns + ------- + list[ReleaseAsset] + Ordered collection of artefacts ready to upload. + + Raises + ------ + AssetError + If no artefacts are found, an artefact is empty, or multiple files would + upload with the same asset name. + + Examples + -------- + >>> discover_assets(Path("dist"), bin_name="netsuke") # doctest: +SKIP + [ReleaseAsset(path=PosixPath('dist/netsuke'), ...)] + """ + if not dist_dir.exists(): + message = f"Artefact directory {dist_dir} does not exist" + raise AssetError(message) + + assets: list[ReleaseAsset] = [] + seen: dict[str, Path] = {} + + for path in _iter_candidate_paths(dist_dir, bin_name): + size = _require_non_empty(path) + asset_name = _resolve_asset_name(path, dist_dir=dist_dir) + _register_asset(asset_name, path, seen) + assets.append(ReleaseAsset(path=path, asset_name=asset_name, size=size)) + + if not assets: + message = f"No artefacts discovered in {dist_dir}" + raise AssetError(message) + + return assets + + +def _render_summary(assets: cabc.Iterable[ReleaseAsset]) -> str: + lines = ["Planned uploads:"] + lines.extend( + f" - {asset.asset_name} ({asset.size} bytes) -> {asset.path}" + for asset in assets + ) + return "\n".join(lines) + + +def upload_assets( + *, release_tag: str, assets: cabc.Iterable[ReleaseAsset], dry_run: bool = False +) -> None: + """Upload artefacts to GitHub using the ``gh`` CLI. + + Parameters + ---------- + release_tag : str + Git tag identifying the release that should receive the artefacts. + assets : Iterable[ReleaseAsset] + Iterable of artefacts to publish. + dry_run : bool + When ``True``, print the planned ``gh`` invocations without executing + them. + + Raises + ------ + ProcessExecutionError + If ``gh`` returns a non-zero status while uploading. + CommandNotFound + If the ``gh`` executable is not available in ``PATH``. + + Examples + -------- + >>> upload_assets( # doctest: +SKIP + ... release_tag="v1.2.3", + ... assets=[ReleaseAsset(Path("dist/netsuke"), "netsuke", 1024)], + ... dry_run=True, + ... ) + """ + gh_cmd: BoundCommand | None = None + for asset in assets: + descriptor = f"{asset.path}#{asset.asset_name}" + if dry_run: + print(f"[dry-run] gh release upload {release_tag} {descriptor} --clobber") + continue + if gh_cmd is None: + gh_cmd = local["gh"] + gh_cmd[ + "release", + "upload", + release_tag, + descriptor, + "--clobber", + ]() + + +def main( + *, + release_tag: str, + bin_name: str, + dist_dir: Path = Path("dist"), + dry_run: bool = False, +) -> int: + """Entry point shared by the CLI and tests. + + Parameters + ---------- + release_tag : str + Git tag identifying the release to publish to. + bin_name : str + Binary name used to derive artefact names during discovery. + dist_dir : Path + Directory containing staged artefacts. + dry_run : bool + When ``True``, validate artefacts and print the upload plan without + uploading. + + Returns + ------- + int + Exit code: ``0`` on success, ``1`` when artefact discovery or upload + fails. + + Examples + -------- + >>> main( # doctest: +SKIP + ... release_tag="v1.2.3", + ... bin_name="netsuke", + ... dist_dir=Path("dist"), + ... dry_run=True, + ... ) + 0 + """ + try: + assets = discover_assets(dist_dir, bin_name=bin_name) + except AssetError as exc: + print(exc, file=sys.stderr) + return 1 + + if dry_run: + print(_render_summary(assets)) + + try: + upload_assets(release_tag=release_tag, assets=assets, dry_run=dry_run) + except (ProcessExecutionError, CommandNotFound) as exc: # pragma: no cover + print(exc, file=sys.stderr) + return 1 + + return 0 + + +@app.default +def cli( + *, + release_tag: typ.Annotated[str, Parameter(required=True)], + bin_name: typ.Annotated[str, Parameter(required=True)], + dist_dir: Path = Path("dist"), + dry_run: bool | str = False, +) -> int: + """Cyclopts-bound CLI entry point.""" + return main( + release_tag=release_tag, + bin_name=bin_name, + dist_dir=dist_dir, + dry_run=_coerce_bool(dry_run), + ) + + +if __name__ == "__main__": # pragma: no cover - exercised via CLI + raise SystemExit(app()) diff --git a/tests_python/test_read_manifest.py b/tests_python/test_read_manifest.py new file mode 100644 index 00000000..94033974 --- /dev/null +++ b/tests_python/test_read_manifest.py @@ -0,0 +1,396 @@ +"""Comprehensive tests for the read_manifest helper script. + +Summary +------- +Tests for the ``read_manifest`` helper script that interrogates Cargo +metadata. + +Purpose +------- +Exercise ``.github/workflows/scripts/read_manifest.py`` to confirm it +extracts package fields, honours CLI arguments and environment overrides, +and surfaces descriptive errors for missing manifests, invalid TOML, and +unexpected structure. + +Usage +----- +Run the full suite:: + + python -m pytest tests_python/test_read_manifest.py + +Execute a targeted test:: + + python -m pytest tests_python/test_read_manifest.py::test_get_field_returns_value + +Examples +-------- +Create a minimal manifest and query the package name:: + + from pathlib import Path + from textwrap import dedent + import subprocess + import sys + + path = Path("/tmp/Cargo.toml") + path.write_text(dedent( + ''' + [package] + name = "netsuke" + version = "1.2.3" + ''' + ), encoding="utf-8") + + completed = subprocess.run( + [ + sys.executable, + ".github/workflows/scripts/read_manifest.py", + "name", + "--manifest-path", + str(path), + ], + check=False, + capture_output=True, + text=True, + ) + assert completed.stdout == "netsuke" +""" + +from __future__ import annotations + +import dataclasses +import importlib.util +import os +import subprocess +import sys +import types +import typing as typ +from contextlib import ExitStack, contextmanager, redirect_stderr, redirect_stdout +from io import StringIO +from pathlib import Path +from textwrap import dedent +from unittest import mock + +import pytest + +REPO_ROOT = Path(__file__).resolve().parent.parent +SCRIPT_PATH = REPO_ROOT / ".github" / "workflows" / "scripts" / "read_manifest.py" + + +@dataclasses.dataclass(slots=True) +class CLIResult: + """ + Result container returned by ReadManifestTests._invoke_cli. + + Attributes + ---------- + exit_code : int + Exit status returned by the CLI process. + stdout : str + Captured standard output emitted by the CLI. + stderr : str + Captured standard error emitted by the CLI. + """ + + exit_code: int + stdout: str + stderr: str + + +@dataclasses.dataclass(slots=True) +class CLIInvocationConfig: + """ + Configuration for invoking the CLI when reading manifest fields. + + Attributes + ---------- + cli_args : tuple[str, ...] | None + Optional CLI arguments to pass; defaults to field and manifest path. + env : dict[str, str] | None + Optional environment variables to set during invocation. + cwd : Path | None + Optional working directory for the invocation. + """ + + cli_args: tuple[str, ...] | None = None + env: dict[str, str] | None = None + cwd: Path | None = None + + +@contextmanager +def change_directory(path: Path) -> typ.Iterator[None]: + """ + Temporarily change the working directory for the current process. + + Parameters + ---------- + path : Path + Target directory that should become the working directory. + + Yields + ------ + None + Control to the caller while the working directory is set to ``path``. + + Examples + -------- + >>> from pathlib import Path + >>> with change_directory(Path("/tmp")): + ... Path.cwd().as_posix() + '/tmp' + """ + original = Path.cwd() + os.chdir(path) + try: + yield + finally: + os.chdir(original) + + +def load_script_module() -> types.ModuleType: + """ + Import the read_manifest script as a module for reuse in tests. + + Returns + ------- + types.ModuleType + The loaded read_manifest module with all its functions and constants. + + Examples + -------- + >>> module = load_script_module() + >>> manifest = {"package": {"name": "foo"}} + >>> module.get_field(manifest, "name") + 'foo' + """ + spec = importlib.util.spec_from_file_location("read_manifest", SCRIPT_PATH) + if spec is None: + message = "Failed to create module spec for read_manifest" + raise RuntimeError(message) + if spec.loader is None: + message = "Module spec missing loader for read_manifest" + raise RuntimeError(message) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + assert isinstance(module, types.ModuleType) + return module + + +@dataclasses.dataclass(slots=True) +class ReadManifestTests: + """ + Helpers that exercise the manifest-reading CLI in different scenarios. + + Methods + ------- + _write_manifest(content: str) -> Path + Write TOML content to a temporary ``Cargo.toml``. + _invoke_cli(*args, env=None, cwd=None) -> CLIResult + Execute the CLI in-process and capture output. + _assert_manifest_error(manifest_path, expected_stderr_fragment=None) -> None + Assert the CLI fails for the given manifest path. + _assert_successful_field_read( + manifest_content, field, expected_value, *, config=None + ) -> None + Assert the CLI prints the expected value for the field. + """ + + module: types.ModuleType + temp_path: Path + + def _write_manifest(self, content: str) -> Path: + """Write ``content`` to ``Cargo.toml`` in the temporary directory.""" + manifest = self.temp_path / "Cargo.toml" + manifest.write_text(dedent(content), encoding="utf-8") + return manifest + + def _invoke_cli( + self, + *args: str, + env: dict[str, str] | None = None, + cwd: Path | None = None, + ) -> CLIResult: + """Execute the CLI and capture its exit code and output streams.""" + stdout = StringIO() + stderr = StringIO() + with ExitStack() as stack: + stack.enter_context( + mock.patch.object(sys, "argv", [str(SCRIPT_PATH), *args]) + ) + if env: + stack.enter_context(mock.patch.dict(os.environ, env, clear=False)) + if cwd: + stack.enter_context(change_directory(cwd)) + stack.enter_context(redirect_stdout(stdout)) + stack.enter_context(redirect_stderr(stderr)) + exit_code = self.module.main() + return CLIResult( + exit_code=exit_code, + stdout=stdout.getvalue(), + stderr=stderr.getvalue(), + ) + + def _assert_manifest_error( + self, + manifest_path: Path, + expected_stderr_fragment: str | None = None, + ) -> None: + """Assert that invoking the CLI fails for ``manifest_path``.""" + # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-audit.dangerous-subprocess-use-audit -- test-only; argv fully controlled # noqa: E501 + result = subprocess.run( # noqa: S603 # TODO(release-ci): FIXME: Security false positive; executed with trusted inputs in tests. https://github.com/leynos/netsuke/pull/179#discussion_r2404108802 + [ + sys.executable, + str(SCRIPT_PATH), + "name", + "--manifest-path", + str(manifest_path), + ], + check=False, + capture_output=True, + text=True, + ) + assert result.returncode != 0 + if expected_stderr_fragment is not None: + assert expected_stderr_fragment in result.stderr + else: + assert result.stderr + assert result.stdout == "" + + def _assert_successful_field_read( + self, + manifest_content: str, + field: str, + expected_value: str, + *, + config: CLIInvocationConfig | None = None, + ) -> None: + """Assert that the CLI prints ``expected_value`` for ``field``.""" + config = config or CLIInvocationConfig() + manifest = self._write_manifest(manifest_content) + args = config.cli_args or (field, "--manifest-path", str(manifest)) + result = self._invoke_cli(*args, env=config.env, cwd=config.cwd) + assert result.exit_code == 0 + assert result.stdout == expected_value + assert result.stderr == "" + + +@pytest.fixture(scope="module") +def read_manifest_module() -> types.ModuleType: + """Load the read_manifest script once for all tests.""" + return load_script_module() + + +@pytest.fixture +def read_manifest_tests( + read_manifest_module: types.ModuleType, + tmp_path: Path, +) -> ReadManifestTests: + """Provide helpers that operate within a temporary working directory.""" + return ReadManifestTests(module=read_manifest_module, temp_path=tmp_path) + + +@pytest.mark.parametrize( + ("field", "expected"), + [ + ("name", "netsuke"), + ("version", "1.2.3"), + ], +) +def test_get_field_returns_value( + read_manifest_module: types.ModuleType, + field: str, + expected: str, +) -> None: + """It returns the requested package metadata from the manifest.""" + manifest = {"package": {"name": "netsuke", "version": "1.2.3"}} + assert read_manifest_module.get_field(manifest, field) == expected + + +def test_get_field_raises_when_missing(read_manifest_module: types.ModuleType) -> None: + """It raises when the requested field is absent.""" + manifest = {"package": {"name": "netsuke"}} + with pytest.raises(KeyError): + read_manifest_module.get_field(manifest, "version") + + +def test_get_field_rejects_non_string_values( + read_manifest_module: types.ModuleType, +) -> None: + """It rejects non-string manifest entries.""" + manifest = { + "package": { + "name": "netsuke", + "version": 123, + "authors": ["alice", "bob"], + "metadata": {"license": "MIT"}, + } + } + with pytest.raises(KeyError): + read_manifest_module.get_field(manifest, "version") + with pytest.raises(KeyError): + read_manifest_module.get_field(manifest, "authors") + with pytest.raises(KeyError): + read_manifest_module.get_field(manifest, "metadata") + + +def test_main_reads_manifest_path_argument( + read_manifest_tests: ReadManifestTests, +) -> None: + """It reads manifests from the path provided via CLI arguments.""" + read_manifest_tests._assert_successful_field_read( + """ + [package] + name = "netsuke" + version = "1.2.3" + """, + field="name", + expected_value="netsuke", + ) + + +def test_main_prefers_environment_manifest_path( + read_manifest_tests: ReadManifestTests, +) -> None: + """It prefers the manifest path supplied via environment variable.""" + manifest = read_manifest_tests._write_manifest( + """ + [package] + name = "netsuke" + version = "1.2.3" + """ + ) + env = {"CARGO_TOML_PATH": str(manifest)} + result = read_manifest_tests._invoke_cli( + "version", + env=env, + cwd=read_manifest_tests.temp_path, + ) + assert result.exit_code == 0 + assert result.stdout == "1.2.3" + assert result.stderr == "" + + +def test_main_reports_missing_manifest( + read_manifest_tests: ReadManifestTests, +) -> None: + """It surfaces errors when the manifest file does not exist.""" + missing = read_manifest_tests.temp_path / "missing.toml" + read_manifest_tests._assert_manifest_error(missing, "does not exist") + + +def test_main_reports_invalid_toml(read_manifest_tests: ReadManifestTests) -> None: + """It surfaces errors for invalid TOML content.""" + manifest = read_manifest_tests._write_manifest("not = [valid") + read_manifest_tests._assert_manifest_error(manifest) + + +def test_main_reports_valid_toml_with_unexpected_structure( + read_manifest_tests: ReadManifestTests, +) -> None: + """It reports a descriptive error when required sections are missing.""" + manifest = read_manifest_tests._write_manifest( + """ + [unexpected_section] + foo = "bar" + """ + ) + read_manifest_tests._assert_manifest_error(manifest, "missing")