ae0cbb7e9b
Shell out to m2pack-secure with --json, parse its envelope, and translate into the standard metin-release Result envelope under data.m2pack. Non-zero exit and non-JSON output map to SubprocessError with m2pack_failed / m2pack_invalid_json / m2pack_empty_output codes.
79 lines
2.7 KiB
Python
79 lines
2.7 KiB
Python
"""Shared helper to invoke the real m2pack CLI and translate its JSON.
|
|
|
|
m2pack-secure emits its own JSON envelopes with ``--json``. Their shape is
|
|
not identical to our :class:`~metin_release.result.Result` envelope, so we
|
|
wrap the raw dict under ``data["m2pack"]`` and promote a few well-known
|
|
fields (artifact paths, counts) where it makes sense per command.
|
|
|
|
If m2pack exits non-zero or prints non-JSON, we raise
|
|
:class:`~metin_release.errors.SubprocessError` so the CLI exits with code
|
|
1 and a readable error envelope.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from ..errors import SubprocessError
|
|
from ..log import get_logger
|
|
from ..m2pack_binary import resolve_m2pack_binary
|
|
|
|
|
|
def run_m2pack(subcommand: str, argv: list[str]) -> dict[str, Any]:
|
|
"""Invoke ``m2pack <subcommand> [argv...] --json`` and return parsed JSON.
|
|
|
|
Raises :class:`SubprocessError` (exit code 1) on any failure: binary
|
|
missing, non-zero exit, empty stdout, or non-JSON stdout. Missing
|
|
binary is handled inside :func:`resolve_m2pack_binary` by raising a
|
|
:class:`~metin_release.errors.ValidationError` which the dispatcher
|
|
already converts into the standard error envelope.
|
|
"""
|
|
log = get_logger()
|
|
binary: Path = resolve_m2pack_binary()
|
|
cmd = [str(binary), subcommand, *argv, "--json"]
|
|
log.debug("running m2pack: %s", " ".join(cmd))
|
|
try:
|
|
proc = subprocess.run(
|
|
cmd, capture_output=True, text=True, check=False
|
|
)
|
|
except OSError as exc:
|
|
raise SubprocessError(
|
|
f"failed to spawn m2pack: {exc}",
|
|
error_code="m2pack_spawn_failed",
|
|
) from exc
|
|
|
|
stdout = (proc.stdout or "").strip()
|
|
stderr = (proc.stderr or "").strip()
|
|
|
|
if proc.returncode != 0:
|
|
detail = stderr or stdout or f"exit code {proc.returncode}"
|
|
raise SubprocessError(
|
|
f"m2pack {subcommand} failed (rc={proc.returncode}): {detail}",
|
|
error_code="m2pack_failed",
|
|
)
|
|
|
|
if not stdout:
|
|
raise SubprocessError(
|
|
f"m2pack {subcommand} produced no stdout; stderr={stderr!r}",
|
|
error_code="m2pack_empty_output",
|
|
)
|
|
|
|
try:
|
|
parsed = json.loads(stdout)
|
|
except json.JSONDecodeError as exc:
|
|
raise SubprocessError(
|
|
f"m2pack {subcommand} returned non-JSON output: {exc}; "
|
|
f"first 200 chars={stdout[:200]!r}",
|
|
error_code="m2pack_invalid_json",
|
|
) from exc
|
|
|
|
if not isinstance(parsed, dict):
|
|
raise SubprocessError(
|
|
f"m2pack {subcommand} JSON was not an object: {type(parsed).__name__}",
|
|
error_code="m2pack_invalid_json",
|
|
)
|
|
return parsed
|