Testing CLI and new tests for error and mutation models.

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,16 +113,20 @@ 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",
     ])
 
     out = capsys.readouterr().out
     assert rc == 1
     # Error path prints a line starting with 'ERROR:'
-    assert "ERROR:" in out
+    assert "ERROR:" in out

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

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,24 +126,20 @@ 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"
     assert opts.fq2 is None
     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)