kemal
8cd092460f
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.
161 lines
6.5 KiB
Python
161 lines
6.5 KiB
Python
"""Sample-sheet matrix — spec EDGE-007 (IN-001 … IN-014)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from simurg.core.errors import SimurgError
|
||
from simurg.formats.samplesheet import output_stem, parse_samplesheet
|
||
|
||
HEAD = "[Header]\nInvestigator Name,X\n[Manifests]\nA,foo.bpm\n[Data]\n"
|
||
|
||
|
||
def sheet(tmp_path: Path, body: str, name: str = "s.csv", raw: bytes | None = None) -> Path:
|
||
p = tmp_path / name
|
||
if raw is not None:
|
||
p.write_bytes(raw)
|
||
else:
|
||
p.write_text(body, encoding="utf-8")
|
||
return p
|
||
|
||
|
||
def code_of(tmp_path: Path, body: str) -> str:
|
||
with pytest.raises(SimurgError) as e:
|
||
parse_samplesheet(sheet(tmp_path, body))
|
||
return e.value.code
|
||
|
||
|
||
def test_basic_sections_and_passthrough(tmp_path) -> None: # type: ignore[no-untyped-def]
|
||
s = parse_samplesheet(
|
||
sheet(
|
||
tmp_path,
|
||
HEAD + "Sample_ID,SentrixBarcode_A,SentrixPosition_A,Gender,Tissue\na,203790110013,r04c01,Female,blood\n",
|
||
)
|
||
)
|
||
assert s.rows[0].sample_id == "203790110013_R04C01"
|
||
assert s.rows[0].reported_sex == "F"
|
||
assert s.passthrough_columns() == ["Sample_ID", "Gender", "Tissue"]
|
||
assert s.header == {"Investigator Name": "X"}
|
||
|
||
|
||
def test_sectionless(tmp_path) -> None: # type: ignore[no-untyped-def]
|
||
s = parse_samplesheet(sheet(tmp_path, "SentrixBarcode_A,SentrixPosition_A\n203790110013,R01C01\n"))
|
||
assert len(s.rows) == 1
|
||
|
||
|
||
def test_bom_crlf_quotes(tmp_path) -> None: # type: ignore[no-untyped-def]
|
||
raw = ('[Data]\r\nSentrixBarcode_A,SentrixPosition_A,Note\r\n203790110013,R01C01,"a, b"\r\n').encode()
|
||
s = parse_samplesheet(sheet(tmp_path, "", raw=raw))
|
||
assert s.rows[0].values["Note"] == "a, b"
|
||
|
||
|
||
def test_trailing_empty_columns(tmp_path) -> None: # type: ignore[no-untyped-def]
|
||
s = parse_samplesheet(sheet(tmp_path, "[Data]\nSentrixBarcode_A,SentrixPosition_A,,\n203790110013,R01C01,,\n"))
|
||
assert s.columns == ["SentrixBarcode_A", "SentrixPosition_A"]
|
||
assert [c for c, _ in s.warnings] == ["W_SHEET_EMPTY_COLUMN"]
|
||
assert "" not in s.rows[0].values
|
||
|
||
|
||
def test_manifests_section_with_values_ignored(tmp_path) -> None: # type: ignore[no-untyped-def]
|
||
body = (
|
||
"[Header],,\nDate,1,\n[Manifests],GSA.bpm,\nA,,\n[Data],,\nSentrixBarcode_A,SentrixPosition_A,x\n"
|
||
"203790110013,R01C01,1\n"
|
||
)
|
||
s = parse_samplesheet(sheet(tmp_path, body))
|
||
assert s.header == {"Date": "1"} and len(s.rows) == 1
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"body,code",
|
||
[
|
||
("[Data]\nSentrixBarcode_A,SentrixPosition_A,Sample_Name,sample_name\n1,2,3,4\n", "E_SHEET_DUP_COLUMN"),
|
||
("[Data]\nSentrixBarcode_A,SentrixPosition_A,Sex Estimate\n203790110013,R01C01,M\n", "E_SHEET_RESERVED"),
|
||
("[Data]\nSentrixBarcode_A,SentrixPosition_A,call_rate\n203790110013,R01C01,1\n", "E_SHEET_RESERVED"),
|
||
("[Data]\nSentrixBarcode_A,SentrixPosition_A,x\n203790110013,R01C01," + "a" * 501 + "\n", "E_SHEET_TOO_LONG"),
|
||
("[Data]\nSentrixBarcode_A,SentrixPosition_A\n2.04753E+11,R01C01\n", "E_SHEET_BARCODE_FORMAT"),
|
||
("[Data]\nSentrixBarcode_A,SentrixPosition_A\n203790110013,R1C1\n", "E_SHEET_BARCODE_FORMAT"),
|
||
("[Data]\nSentrixBarcode_A;SentrixPosition_A\n203790110013;R01C01\n", "E_SHEET_DELIMITER"),
|
||
('{"samples": []}', "E_SHEET_DELIMITER"),
|
||
("[Data]\nSample_ID,Foo\n1,2\n", "E_SHEET_NO_KEY_COLUMNS"),
|
||
("[Data]\nSentrixBarcode_A,SentrixPosition_A\n", "E_SHEET_EMPTY"),
|
||
("[Header]\nA,B\n", "E_SHEET_EMPTY"),
|
||
(
|
||
"[Data]\nSentrixBarcode_A,SentrixPosition_A,Sample_Name\n203790110013,R01C01,a\n203790110013,R02C01,\n",
|
||
"E_SHEET_NAME_EMPTY",
|
||
),
|
||
(
|
||
"[Data]\nSentrixBarcode_A,SentrixPosition_A,Sample_Name\n203790110013,R01C01,203790110013_R02C01\n"
|
||
"203790110013,R02C01,b\n",
|
||
"E_SHEET_NAME_COLLISION",
|
||
),
|
||
(
|
||
"[Data]\nSentrixBarcode_A,SentrixPosition_A,Sample_Name\n203790110013,R01C01,a\n203790110013,R02C01,a\n",
|
||
"E_SHEET_NAME_COLLISION",
|
||
),
|
||
(
|
||
"[Data]\nSentrixBarcode_A,SentrixPosition_A\n203790110013,R01C01\n203790110013,r01c01\n",
|
||
"E_SHEET_DUP_SAMPLE",
|
||
),
|
||
],
|
||
)
|
||
def test_error_matrix(tmp_path, body, code) -> None: # type: ignore[no-untyped-def]
|
||
assert code_of(tmp_path, body) == code
|
||
|
||
|
||
def test_barcode_decimal_repaired(tmp_path) -> None: # type: ignore[no-untyped-def]
|
||
s = parse_samplesheet(sheet(tmp_path, "[Data]\nSentrixBarcode_A,SentrixPosition_A\n204753010023.0,R01C01\n"))
|
||
assert s.rows[0].barcode == "204753010023"
|
||
assert s.warnings[0][0] == "W_SHEET_BARCODE_REPAIRED"
|
||
|
||
|
||
def test_scientific_notation_hint(tmp_path) -> None: # type: ignore[no-untyped-def]
|
||
with pytest.raises(SimurgError) as e:
|
||
parse_samplesheet(sheet(tmp_path, '[Data]\nSentrixBarcode_A,SentrixPosition_A\n"2,04753E+11",R01C01\n'))
|
||
assert e.value.hint and "0 decimals" in e.value.hint
|
||
|
||
|
||
def test_alias_keys_and_rootfolder(tmp_path) -> None: # type: ignore[no-untyped-def]
|
||
s = parse_samplesheet(
|
||
sheet(tmp_path, "[Header]\nRootFolder,idats\n[Data]\nbeadChipName,sampleSectionName\n203790110013,R01C01\n")
|
||
)
|
||
assert s.barcode_column == "beadChipName"
|
||
assert s.root_folder == tmp_path / "idats"
|
||
|
||
|
||
def test_sex_columns(tmp_path) -> None: # type: ignore[no-untyped-def]
|
||
s = parse_samplesheet(
|
||
sheet(
|
||
tmp_path,
|
||
"[Data]\nSentrixBarcode_A,SentrixPosition_A,Gender,Sex\n"
|
||
"203790110013,R01C01,Male,F\n203790110013,R02C01,,\n",
|
||
)
|
||
)
|
||
assert [r.reported_sex for r in s.rows] == ["F", "U"]
|
||
assert any(c == "W_SHEET_SEX_CONFLICT" for c, _ in s.warnings)
|
||
|
||
|
||
def test_turkish_case_folding_is_locale_independent(tmp_path) -> None: # type: ignore[no-untyped-def] # EDGE-001
|
||
s = parse_samplesheet(sheet(tmp_path, "[Data]\nSENTRIXBARCODE_A,SentrixPosition_A,İl\n203790110013,R01C01,x\n"))
|
||
assert s.barcode_column == "SENTRIXBARCODE_A"
|
||
|
||
|
||
def test_ten_thousand_rows_fast(tmp_path) -> None: # type: ignore[no-untyped-def] # PERF-007 (parse part)
|
||
import time
|
||
|
||
lines = ["[Data]", "SentrixBarcode_A,SentrixPosition_A,Sample_Name"]
|
||
for i in range(10_000):
|
||
lines.append(f"{200000000000 + i // 24},R{(i % 24) + 1:02d}C01,s{i}")
|
||
p = sheet(tmp_path, "\n".join(lines) + "\n")
|
||
t = time.perf_counter()
|
||
s = parse_samplesheet(p)
|
||
assert len(s.rows) == 10_000
|
||
assert time.perf_counter() - t < 2.0
|
||
|
||
|
||
def test_output_stem() -> None:
|
||
assert output_stem("NA 12878/ü") == "NA_12878__"
|
||
assert output_stem("..hidden") == "hidden"
|
||
assert len(output_stem("x" * 500)) == 200
|