"""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