kemal
522429b731
Found by the P10 fresh-VM run: `simurg inspect … | head` raised BrokenPipeError, reported as an internal error, and stopped examples/gsa_demo.sh (set -o pipefail). run_guarded now flushes stdout inside the guard and maps BrokenPipeError to ExitCode.SIGPIPE (141) without a message. p10.sh: SKIP_DOCKER/SKIP_CONDA, clearer digest report; p10t.sh runs build+pip+tutorial only.
111 lines
4.0 KiB
Python
111 lines
4.0 KiB
Python
"""CLI behaviour (CLI-007, CLI-014, CLI-020) and the message catalogue (CLI-016)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
from typer.testing import CliRunner
|
|
|
|
from simurg.cli.main import app
|
|
from simurg.core.errors import CATALOGUE, render_markdown
|
|
from simurg.synth.files import SynthIdat, SynthLocus, encode_bpm, encode_egt, encode_idat
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
runner = CliRunner()
|
|
|
|
|
|
def _files(tmp_path: Path) -> dict[str, Path]:
|
|
idat = tmp_path / "200000000001_R01C01_Grn.idat"
|
|
idat.write_bytes(
|
|
encode_idat(SynthIdat(address_ids=np.arange(1, 6, dtype=np.int32), mean=np.arange(5, dtype=np.uint16)))
|
|
)
|
|
bpm = tmp_path / "m.bpm"
|
|
bpm.write_bytes(encode_bpm([SynthLocus("rs1"), SynthLocus("rs2", address_a=2)]))
|
|
egt = tmp_path / "c.egt"
|
|
egt.write_bytes(encode_egt(["rs1"], theta_mean=[[0.1, 0.5, 0.9]], r_mean=[[1, 1, 1]]))
|
|
return {"idat": idat, "bpm": bpm, "egt": egt}
|
|
|
|
|
|
def test_version() -> None:
|
|
r = runner.invoke(app, ["--version"])
|
|
assert r.exit_code == 0 and r.stdout.startswith("SimurgArray ")
|
|
|
|
|
|
def test_inspect_json(tmp_path) -> None: # type: ignore[no-untyped-def]
|
|
f = _files(tmp_path)
|
|
r = runner.invoke(app, ["inspect", "--json", str(f["idat"]), str(f["bpm"]), str(f["egt"])])
|
|
assert r.exit_code == 0, r.output
|
|
out = json.loads(r.stdout)
|
|
assert [o["kind"] for o in out] == ["idat", "bpm", "egt"]
|
|
assert out[1]["num_loci"] == 2 and out[2]["num_records"] == 1
|
|
|
|
|
|
def test_inspect_missing_file_exit2() -> None:
|
|
r = runner.invoke(app, ["inspect", "/nonexistent/x.idat"])
|
|
assert r.exit_code == 2
|
|
assert "E_FILE_NOT_FOUND" in r.output
|
|
|
|
|
|
def test_inspect_partial_failure_exit3(tmp_path) -> None: # type: ignore[no-untyped-def]
|
|
f = _files(tmp_path)
|
|
bad = tmp_path / "bad.gtc"
|
|
bad.write_bytes(b"gtc\x05" + b"\x00" * 3)
|
|
r = runner.invoke(app, ["inspect", "--log-json", str(f["idat"]), str(bad)])
|
|
assert r.exit_code == 3
|
|
line = next(ln for ln in r.output.splitlines() if ln.startswith('{"schema"'))
|
|
rec = json.loads(line)
|
|
assert rec["schema"] == "simurg.log/1" and rec["code"] == "E_GTC_CORRUPT" and rec["level"] == "error"
|
|
|
|
|
|
def test_unknown_option_is_usage_error(tmp_path) -> None: # type: ignore[no-untyped-def]
|
|
r = runner.invoke(app, ["inspect", "--no-such-option", "x"])
|
|
assert r.exit_code == 2
|
|
|
|
|
|
def test_catalogue_integrity() -> None:
|
|
for code, d in CATALOGUE.items():
|
|
assert code == d.code
|
|
assert code[:2] in ("E_", "W_", "I_")
|
|
assert d.level in ("error", "warning", "info")
|
|
|
|
|
|
def test_messages_doc_is_current() -> None:
|
|
doc = ROOT / "docs" / "messages.md"
|
|
assert doc.read_text(encoding="utf-8") == render_markdown(), "regenerate: python -m simurg.core.errors"
|
|
|
|
|
|
def test_locale_independent_output(tmp_path) -> None: # type: ignore[no-untyped-def] # EDGE-001
|
|
f = _files(tmp_path)
|
|
outs = set()
|
|
for loc in ("C.UTF-8", "tr_TR.UTF-8", "de_DE.UTF-8"):
|
|
env = {"LC_ALL": loc, "PATH": "/usr/bin:/bin"}
|
|
r = subprocess.run(
|
|
[sys.executable, "-m", "simurg.cli.main", "inspect", "--json", str(f["bpm"])],
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
cwd=ROOT,
|
|
)
|
|
assert r.returncode == 0, r.stderr
|
|
outs.add(r.stdout)
|
|
assert len(outs) == 1
|
|
|
|
|
|
def test_closed_stdout_is_not_an_internal_error(tmp_path) -> None: # type: ignore[no-untyped-def]
|
|
# `simurg inspect … | head`: the reader goes away; the CLI must exit like a SIGPIPE'd tool (141), silently
|
|
f = _files(tmp_path)
|
|
p = subprocess.Popen(
|
|
[sys.executable, "-m", "simurg.cli.main", "inspect", "--json", str(f["bpm"]), str(f["egt"]), str(f["idat"])],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
)
|
|
assert p.stdout is not None and p.stderr is not None
|
|
p.stdout.close() # nobody reads stdout
|
|
err = p.stderr.read().decode()
|
|
assert p.wait() == 141, err
|
|
assert "E_INTERNAL" not in err and "Traceback" not in err and "Exception ignored" not in err
|