P1: file parsers, sample sheet, discovery, CLI inspect
Readers for IDAT v3, BPM (locus entries v4-v8), CSV manifest, EGT (data blocks 5-9) and GTC v3-v5 (plus a layout-preserving GTC writer); sample-sheet validation (IN-001..IN-014), IDAT discovery and pairing (IN-020..IN-022), message catalogue and exit codes (CLI-014/016), JSON Lines logging (CLI-007/008), atomic writes (CLI-011), run manifest, and `simurg inspect`. Validation (docs/validation/P1.md): GTC, BPM and EGT readers agree exactly with Illumina BeadArrayFiles on GSA v4 / GDA v1 files; GTCs round-trip byte-identically; BPM and CSV manifests are consistent.
This commit is contained in:
1 parent
0a421dffc9
commit
5a20e7131c
40 files changed
+4742
No files matched your search
@@ -0,0 +1,95 @@
|
||||
"""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
|
||||
Reference in new issue
Block a user