65 lines
2.2 KiB
Python
65 lines
2.2 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from gfa_core.wgs_samples import WgsSample, load_wgs_samples
|
||
|
|
|
||
|
|
|
||
|
|
def test_load_wgs_samples_resolves_paths_from_yaml_parent(tmp_path):
|
||
|
|
reads = tmp_path / "reads"
|
||
|
|
reads.mkdir()
|
||
|
|
for name in ("S1_R1.fastq.gz", "S1_R2.fastq.gz"):
|
||
|
|
(reads / name).write_bytes(b"reads")
|
||
|
|
config = tmp_path / "wgs_samples.yaml"
|
||
|
|
config.write_text(
|
||
|
|
"samples:\n S1:\n r1: reads/S1_R1.fastq.gz\n r2: reads/S1_R2.fastq.gz\n",
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
assert load_wgs_samples(config) == (
|
||
|
|
WgsSample("S1", reads / "S1_R1.fastq.gz", reads / "S1_R2.fastq.gz"),
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_load_wgs_samples_rejects_include(tmp_path):
|
||
|
|
r1 = tmp_path / "r1.fastq.gz"
|
||
|
|
r2 = tmp_path / "r2.fastq.gz"
|
||
|
|
r1.write_bytes(b"r1")
|
||
|
|
r2.write_bytes(b"r2")
|
||
|
|
path = tmp_path / "wgs_samples.yaml"
|
||
|
|
path.write_text(
|
||
|
|
f"samples:\n S1: {{r1: {r1}, r2: {r2}, include: false}}\n",
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
with pytest.raises(ValueError, match="include"):
|
||
|
|
load_wgs_samples(path)
|
||
|
|
|
||
|
|
|
||
|
|
def test_load_wgs_samples_sorts_ids_and_rejects_identical_mates(tmp_path):
|
||
|
|
for name in ("a_R1.fastq", "a_R2.fastq", "z_R1.fastq", "z_R2.fastq"):
|
||
|
|
(tmp_path / name).write_bytes(b"reads")
|
||
|
|
config = tmp_path / "wgs_samples.yaml"
|
||
|
|
config.write_text(
|
||
|
|
"samples:\n"
|
||
|
|
" Z: {r1: z_R1.fastq, r2: z_R2.fastq}\n"
|
||
|
|
" A: {r1: a_R1.fastq, r2: a_R2.fastq}\n",
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
assert [sample.sample_id for sample in load_wgs_samples(config)] == ["A", "Z"]
|
||
|
|
config.write_text("samples:\n A: {r1: a_R1.fastq, r2: a_R1.fastq}\n", encoding="utf-8")
|
||
|
|
with pytest.raises(ValueError, match="different"):
|
||
|
|
load_wgs_samples(config)
|
||
|
|
|
||
|
|
|
||
|
|
def test_load_wgs_samples_rejects_duplicate_sample_id(tmp_path):
|
||
|
|
for name in ("S1_R1.fastq", "S1_R2.fastq"):
|
||
|
|
(tmp_path / name).write_bytes(b"reads")
|
||
|
|
config = tmp_path / "wgs_samples.yaml"
|
||
|
|
config.write_text(
|
||
|
|
"samples:\n"
|
||
|
|
" S1: {r1: S1_R1.fastq, r2: S1_R2.fastq}\n"
|
||
|
|
" S1: {r1: S1_R1.fastq, r2: S1_R2.fastq}\n",
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
with pytest.raises(ValueError, match="duplicate key"):
|
||
|
|
load_wgs_samples(config)
|