diff --git a/tests/test_cli/test_cli.py b/tests/test_cli/test_cli.py index 07805497..d5e3247a 100644 --- a/tests/test_cli/test_cli.py +++ b/tests/test_cli/test_cli.py @@ -9,6 +9,21 @@ from neat.cli.cli import Cli, main +def _write_min_cfg(tmp_path: Path) -> Path: + """ + Write a minimal config file for read-simulator. + + These CLI tests validate argument handling, logging, and return codes, + not the full simulation behavior, but read-simulator still requires -c. + """ + ref = tmp_path / "ref.fa" + ref.write_text(">chr1\nACGT\n", encoding="utf-8") + + cfg = tmp_path / "conf.yml" + cfg.write_text(f"reference: {ref}\nproduce_fastq: true\n", encoding="utf-8") + return cfg + + def test_cli_registers_read_simulator_subcommand(): cli = Cli() # Argparse stores subparsers in a private map; ensure our command is registered @@ -50,10 +65,15 @@ def test_logging_creates_named_log_file_and_announces(monkeypatch, tmp_path: Pat lambda *args, **kwargs: None, ) + cfg = _write_min_cfg(tmp_path) + rc = main(cli.parser, [ "--log-name", str(logname), # Supply a benign subcommand with minimal required args - "read-simulator", "-o", str(tmp_path), "-p", "pref" + "read-simulator", + "-c", str(cfg), + "-o", str(tmp_path), + "-p", "pref", ]) out = capsys.readouterr().out # main should create/log the file path and return 0 (success) @@ -68,13 +88,12 @@ def test_read_simulator_success_invokes_runner(monkeypatch, tmp_path: Path): called = {} def fake_runner(cfg, outdir, prefix): - called['args'] = (cfg, outdir, prefix) + called["args"] = (cfg, outdir, prefix) # Patch runner used by command monkeypatch.setattr("neat.cli.commands.read_simulator.read_simulator_runner", fake_runner) - cfg = tmp_path / "conf.yml" - cfg.write_text("reference: ''\n", encoding="utf-8") # minimal content; not validated here + cfg = _write_min_cfg(tmp_path) rc = main(cli.parser, [ "--no-log", @@ -85,7 +104,7 @@ def fake_runner(cfg, outdir, prefix): ]) assert rc == 0 - assert called['args'] == (str(cfg), str(tmp_path), "myprefix") + assert called["args"] == (str(cfg), str(tmp_path), "myprefix") def test_read_simulator_failure_returns_1_and_prints_error(monkeypatch, tmp_path: Path, capsys): @@ -94,11 +113,15 @@ def test_read_simulator_failure_returns_1_and_prints_error(monkeypatch, tmp_path def boom(*args, **kwargs): raise RuntimeError("kaboom") - monkeypatch.setattr("neat.read_simulator.read_simulator_runner", boom) + # Patch the runner symbol used by the read-simulator command + monkeypatch.setattr("neat.cli.commands.read_simulator.read_simulator_runner", boom) + + cfg = _write_min_cfg(tmp_path) rc = main(cli.parser, [ "--no-log", "read-simulator", + "-c", str(cfg), "-o", str(tmp_path), "-p", "x", ]) @@ -106,4 +129,4 @@ def boom(*args, **kwargs): out = capsys.readouterr().out assert rc == 1 # Error path prints a line starting with 'ERROR:' - assert "ERROR:" in out + assert "ERROR:" in out \ No newline at end of file diff --git a/tests/test_models/test_error_and_mut_models.py b/tests/test_models/test_error_and_mut_models.py index cc8caec3..a255ac96 100644 --- a/tests/test_models/test_error_and_mut_models.py +++ b/tests/test_models/test_error_and_mut_models.py @@ -43,8 +43,7 @@ def test_mutation_model_generate_snv_trinuc(): def test_sequencing_error_model_zero_error_returns_none_or_empty(): """ - avg_seq_error == 0 should yield no errors. Some versions return just the list, - others return a (list, padding) tuple — accept both. + avg_seq_error == 0 should yield no errors. """ rng = default_rng(4) sem = SequencingErrorModel(avg_seq_error=0.0) @@ -89,3 +88,105 @@ def test_sequencing_error_model_basic_snvs_only(): assert hasattr(e, "location") assert hasattr(e, "ref") assert hasattr(e, "alt") + +def test_mutation_model_insertion_reproducible_with_seed(): + """Same seed and inputs should give the same insertion (length and alt).""" + rng1 = default_rng(123) + rng2 = default_rng(123) + m = MutationModel() + ref = Seq("ACGT") + + ins1 = m.generate_insertion(location=10, ref=ref, rng=rng1) + ins2 = m.generate_insertion(location=10, ref=ref, rng=rng2) + + assert isinstance(ins1, Insertion) + assert isinstance(ins2, Insertion) + assert ins1.length == ins2.length + assert str(ins1.alt) == str(ins2.alt) + + +def test_mutation_model_deletion_reproducible_with_seed(): + """Same seed and inputs should give the same deletion object shape.""" + rng1 = default_rng(456) + rng2 = default_rng(456) + m = MutationModel() + + del1 = m.generate_deletion(location=25, rng=rng1) + del2 = m.generate_deletion(location=25, rng=rng2) + + assert isinstance(del1, Deletion) + assert isinstance(del2, Deletion) + assert del1.length == del2.length + assert del1.position1 == del2.position1 + + +def test_mutation_model_snv_does_not_keep_reference_base(): + """ + For a given trinucleotide, the generated SNV should change the central base. + """ + rng = default_rng(7) + m = MutationModel() + trinuc = Seq("ACA") + central = str(trinuc[1]) + + snv = m.generate_snv(trinucleotide=trinuc, reference_location=100, rng=rng) + assert isinstance(snv, SingleNucleotideVariant) + assert snv.alt in ["A", "C", "G", "T"] + assert snv.alt != central + + +def test_traditional_quality_model_reproducible_with_seed(): + """Quality model should be deterministic given the same RNG state.""" + rng1 = default_rng(8) + rng2 = default_rng(8) + qm = TraditionalQualityModel(average_error=0.01) + + qs1 = qm.get_quality_scores(model_read_length=151, length=100, rng=rng1) + qs2 = qm.get_quality_scores(model_read_length=151, length=100, rng=rng2) + + assert np.array_equal(qs1, qs2) + + +def test_sequencing_error_model_reproducible_with_seed(): + """Error placement should be deterministic given the same RNG state.""" + sem = SequencingErrorModel(avg_seq_error=0.05) + ref = SeqRecord(Seq("ACGT" * 30), id="chr1") + quals = np.array([30] * len(ref), dtype=int) + + rng1 = default_rng(9) + rng2 = default_rng(9) + + introduced1, pad1 = sem.get_sequencing_errors( + padding=20, reference_segment=ref, quality_scores=quals, rng=rng1 + ) + introduced2, pad2 = sem.get_sequencing_errors( + padding=20, reference_segment=ref, quality_scores=quals, rng=rng2 + ) + + assert pad1 == pad2 + proj1 = [(e.error_type, e.location, e.ref, e.alt) for e in introduced1] + proj2 = [(e.error_type, e.location, e.ref, e.alt) for e in introduced2] + assert proj1 == proj2 + + +def test_sequencing_error_model_nonzero_error_introduces_in_bounds_errors(): + """ + With a non-zero average error rate, we expect at least some errors and their + locations must be within the reference segment. + """ + rng = default_rng(10) + sem = SequencingErrorModel(avg_seq_error=0.2) + ref = SeqRecord(Seq("ACGT" * 50), id="chr1") + quals = np.array([10] * len(ref), dtype=int) + + introduced, pad = sem.get_sequencing_errors( + padding=20, reference_segment=ref, quality_scores=quals, rng=rng + ) + + # At least one error is expected for these settings. + assert len(introduced) > 0 + # All error locations should be within the sequence. + for e in introduced: + assert 0 <= e.location < len(ref) + assert e.ref in ["A", "C", "G", "T"] + assert e.alt in ["A", "C", "G", "T"] \ No newline at end of file diff --git a/tests/test_read_simulator/test_options.py b/tests/test_read_simulator/test_options.py index a5528ab4..dad1cbfe 100644 --- a/tests/test_read_simulator/test_options.py +++ b/tests/test_read_simulator/test_options.py @@ -1,6 +1,7 @@ from neat.read_simulator.utils.options import Options from pathlib import Path as _PathAlias +import logging as _logging import numpy as _np import textwrap as _textwrap import pytest as _pytest @@ -10,8 +11,42 @@ def _project_root() -> _PathAlias: return _PathAlias(__file__).resolve().parents[2] -# Redefine the function name used above to override the brittle test -# so pytest only sees this correct version. +@_pytest.fixture(autouse=True) +def _isolate_neat_logging(): + """ + Prevent flaky 'ValueError: I/O operation on closed file' logging errors under pytest. + """ + # Clear handlers on NEAT and all child loggers + for name, logger in list(_logging.Logger.manager.loggerDict.items()): + if name == "neat" or name.startswith("neat."): + if isinstance(logger, _logging.Logger): + for h in list(logger.handlers): + logger.removeHandler(h) + try: + h.close() + except Exception: + pass + logger.handlers.clear() + logger.propagate = True # child loggers will propagate to 'neat' + + neat_logger = _logging.getLogger("neat") + neat_logger.handlers.clear() + neat_logger.addHandler(_logging.NullHandler()) + neat_logger.propagate = False # stop at 'neat' (do not reach root) + + yield + + # Rremove NullHandler + for h in list(neat_logger.handlers): + neat_logger.removeHandler(h) + try: + h.close() + except Exception: + pass + neat_logger.handlers.clear() + neat_logger.propagate = True + + def test_basic_options(): reference = _project_root() / "data" / "H1N1.fa" base_options = Options(reference) @@ -57,7 +92,6 @@ def test_rng_seed_reproducible(): def test_from_cli_single_end_with_threads_and_splits(tmp_path: _PathAlias): - # Build a minimal YAML config using repository-relative paths cfg = _textwrap.dedent( f""" reference: {(_project_root() / 'data' / 'H1N1.fa').as_posix()} @@ -76,8 +110,8 @@ def test_from_cli_single_end_with_threads_and_splits(tmp_path: _PathAlias): rng_seed: 42 overwrite_output: true - mode: contig - size: 500000 + parallel_mode: size + parallel_block_size: 500000 threads: 2 cleanup_splits: false reuse_splits: false @@ -92,14 +126,12 @@ def test_from_cli_single_end_with_threads_and_splits(tmp_path: _PathAlias): opts = Options.from_cli(outdir, "fromcli", yml_path) - # Basics propagated assert opts.reference == _project_root() / "data" / "H1N1.fa" assert opts.read_len == 75 assert opts.coverage == 5 assert opts.ploidy == 2 assert opts.rng_seed == 42 - # Output construction via log_configuration() inside from_cli assert opts.output_dir == outdir assert opts.output_prefix == "fromcli" assert opts.fq1 == outdir / "fromcli.fastq.gz" @@ -107,9 +139,7 @@ def test_from_cli_single_end_with_threads_and_splits(tmp_path: _PathAlias): assert opts.bam is None assert opts.vcf is None - # Parallel-related settings assert opts.threads == 2 - # cleanup_splits: false -> splits dir under output_dir assert opts.splits_dir == outdir / "splits" assert opts.splits_dir.is_dir() @@ -132,7 +162,7 @@ def test_from_cli_paired_end_fragments(tmp_path: _PathAlias): rng_seed: 7 overwrite_output: true - mode: contig + parallel_mode: contig threads: 1 cleanup_splits: true reuse_splits: false @@ -162,6 +192,8 @@ def test_from_cli_reuse_splits_missing_dir_raises(tmp_path: _PathAlias): produce_bam: false produce_vcf: false threads: 4 + parallel_mode: size + parallel_block_size: 500000 cleanup_splits: true reuse_splits: true overwrite_output: true @@ -174,6 +206,5 @@ def test_from_cli_reuse_splits_missing_dir_raises(tmp_path: _PathAlias): outdir = tmp_path / "out" outdir.mkdir(parents=True, exist_ok=True) - options = Options.from_cli(outdir, "reuse", yml_path) - # should issue a warning but continue in this case - assert options.reuse_splits == True + with _pytest.raises(FileNotFoundError, match=r"reuse_splits=True"): + Options.from_cli(outdir, "reuse", yml_path) \ No newline at end of file