Add unit tests for consensus_markers script

.github/workflows/tests.yml

@@ -31,6 +31,5 @@ jobs:
       - name: Install Dependencies
         run: python -m pip install pytest
 
-      # TODO: Add demo unit tests
-      #- name: Run Tests
-      #  run: pytest .tests/unit/
+      - name: Run Tests
+        run: pytest .tests/unit/

.tests/e2e/test_full_run.py

@@ -100,4 +100,4 @@ def test_full_run(run_sunbeam):
             lines[-1].split("\t")[0]
             == "k__Bacteria|p__Proteobacteria|c__Gammaproteobacteria|o__Enterobacterales|f__Enterobacteriaceae|g__Escherichia|s__Escherichia_coli|t__SGB10068"
         )
-        assert int(lines[-1].split("\t")[2]) >= 95
+        assert float(lines[-1].split("\t")[2]) >= 95

.tests/unit/test_consensus_markers.py

@@ -0,0 +1,72 @@
+import bz2
+import os
+import pickle
+import pytest
+import sys
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.append(Path(__file__).parents[2].joinpath("scripts").as_posix())
+sys.path.append(os.path.join(os.path.dirname(__file__), "..", "scripts"))
+print(Path(__file__).parents[2].joinpath("scripts").as_posix())
+from consensus_markers_f import is_bz2_empty, run_sample2markers
+
+
+@pytest.fixture
+def empty_bz2_file(tmpdir):
+    fp = tmpdir.join("empty.bz2")
+    with bz2.open(fp, "wt") as f:
+        f.write("")
+
+    yield Path(fp)
+
+
+@pytest.fixture
+def non_empty_bz2_file(tmpdir):
+    fp = tmpdir.join("nonempty.bz2")
+    with bz2.open(fp, "wt") as f:
+        f.write("CONTENT")
+
+    yield Path(fp)
+
+
+@pytest.fixture
+def output_fp(tmpdir):
+    # Create a temporary output file
+    output_file = tmpdir.join("output.txt")
+    return Path(output_file)
+
+
+@pytest.fixture
+def pickle_fp(tmpdir):
+    # Create a temporary pickle file
+    pickle_file = tmpdir.join("data.pickle")
+    return Path(pickle_file)
+
+
+@pytest.fixture
+def output_dir(tmpdir):
+    # Create a temporary output directory
+    return Path(tmpdir)
+
+
+@pytest.fixture
+def threads():
+    return 4
+
+
+def test_is_bz2_empty_with_empty_file(empty_bz2_file):
+    assert is_bz2_empty(empty_bz2_file) is True
+
+
+def test_is_bz2_empty_with_non_empty_file(non_empty_bz2_file):
+    assert is_bz2_empty(non_empty_bz2_file) is False
+
+
+def test_run_sample2markers_with_empty_input(
+    empty_bz2_file, output_fp, pickle_fp, output_dir, threads
+):
+    with patch("consensus_markers_f.is_bz2_empty", return_value=True):
+        run_sample2markers(empty_bz2_file, output_fp, pickle_fp, output_dir, threads)
+
+    assert pickle.load(open(output_fp, "rb")) == []

scripts/consensus_markers.py

@@ -1,14 +1,9 @@
-import bz2
-import pickle
-import subprocess as sp
+from consensus_markers import run_sample2markers
 
-with bz2.open(snakemake.input[0], "r") as f_in:
-    if f_in.read():
-        empty = False
-
-if empty:
-    args = ["sample2markers.py", "-i", snakemake.input[0], "-d", snakemake.params.pickle, "-o", snakemake.params.outdir, "-n", str(snakemake.threads)]
-    sp.run(args, check=True)
-else:
-    with open(snakemake.output[0], "wb") as f_out:
-        pickle.dump([], f_out)
+run_sample2markers(
+    snakemake.input[0],
+    snakemake.output[0],
+    snakemake.params.pickle,
+    snakemake.params.outdir,
+    snakemake.threads,
+)

scripts/consensus_markers_f.py

@@ -0,0 +1,52 @@
+import bz2
+import pickle
+import subprocess as sp
+
+
+def is_bz2_empty(fp):
+    """
+    Check if a bz2 file is empty.
+
+    Args:
+        fp (str): The file path of the bz2 file.
+
+    Returns:
+        bool: True if the file is empty, False otherwise.
+    """
+    with bz2.open(fp, "r") as f_in:
+        if f_in.read():
+            return False
+        else:
+            return True
+
+
+def run_sample2markers(input_fp, output_fp, pickle_fp, output_dir, threads):
+    """
+    Run the sample2markers.py script with the given input and output parameters.
+
+    Args:
+        input_fp (str): File path of the input file.
+        output_fp (str): File path of the output file.
+        pickle_fp (str): File path of the pickle file.
+        output_dir (str): Directory path for the output files.
+        threads (int): Number of threads to use.
+
+    Returns:
+        None
+    """
+    if not is_bz2_empty(input_fp):
+        args = [
+            "sample2markers.py",
+            "-i",
+            input_fp,
+            "-d",
+            pickle_fp,
+            "-o",
+            output_dir,
+            "-n",
+            str(threads),
+        ]
+        sp.run(args, check=True)
+    else:
+        with open(output_fp, "wb") as f_out:
+            pickle.dump([], f_out)