Repository: widdowquinn/pyani

Issue:
Find genomes causing Blast to fail
Thanks for pyani.
With WARNING: At least one Blast run failed. ANIb may fail. 
Sometimes it does and sometimes it doesn't, but in either case it would be useful to identify the culprit genomes.
One can imagine they may be too distant or too fragmented.
But is there a way to identify which one's have failed?
Thanks


Discussion:

Patch:
diff --git a/pyani/run_multiprocessing.py b/pyani/run_multiprocessing.py
index 9d048b5..301042d 100644
--- a/pyani/run_multiprocessing.py
+++ b/pyani/run_multiprocessing.py
@@ -110,7 +110,9 @@ def populate_cmdsets(job: Job, cmdsets: List, depth: int) -> List:
 
 
 # Run a set of command lines using multiprocessing
-def multiprocessing_run(cmdlines: List, workers: Optional[int] = None) -> int:
+def multiprocessing_run(
+    cmdlines: List, workers: Optional[int] = None, logger: Optional[Logger] = None
+) -> int:
     """Distributes passed command-line jobs using multiprocessing.
 
     :param cmdlines:  iterable, command line strings
@@ -119,6 +121,8 @@ def multiprocessing_run(cmdlines: List, workers: Optional[int] = None) -> int:
     Returns the sum of exit codes from each job that was run. If
     all goes well, this should be 0. Anything else and the calling
     function should act accordingly.
+
+    Sends a warning to the logger if a comparison fails; the warning consists of the specific command line that has failed.
     """
     # Run jobs
     # If workers is None or greater than the number of cores available,
@@ -138,4 +142,11 @@ def multiprocessing_run(cmdlines: List, workers: Optional[int] = None) -> int:
     ]
     pool.close()
     pool.join()
+
+    # Print a warning so individual runs that fail can be identified
+    for r in results:
+        if r.get().returncode != 0:
+            if logger:  # Try to be informative, if the logger module is being used
+                logger.warning(f"Comparison failed: {' '.join(r.get().args)}")
+
     return sum([r.get().returncode for r in results])


Test patch:
diff --git a/tests/test_multiprocessing.py b/tests/test_multiprocessing.py
index 3757cd3..0c38778 100644
--- a/tests/test_multiprocessing.py
+++ b/tests/test_multiprocessing.py
@@ -46,10 +46,15 @@ pytest -v
 """
 
 import unittest
+import logging
 
 from pathlib import Path
 
 import pytest
+from typing import Dict, List, NamedTuple, Tuple
+import subprocess
+import pickle
+import multiprocessing
 
 from pyani.anib import fragment_fasta_files, make_blastcmd_builder, make_job_graph
 from pyani.pyani_jobs import Job
@@ -59,6 +64,9 @@ from pyani.run_multiprocessing import (
     run_dependency_graph,
 )
 
+assertions = unittest.TestCase("__init__")
+logger = logging.getLogger(__name__)
+
 
 @pytest.fixture
 def mp_cmdlist():
@@ -100,3 +108,50 @@ def test_dependency_graph_run(path_fna_two, fragment_length, tmp_path):
     jobgraph = make_job_graph(path_fna_two, fragresult[0], blastcmds)
     result = run_dependency_graph(jobgraph)
     assert 0 == result
+
+
+# Convenience struct wich mocks a multiprocessing.pool worker's task
+# pool.apply_async returns a multiprocessing.pool.ApplyResult object,
+# which is essentially a NamedTuple, but has a get method, hence
+# the need to define one here that doesn't appear to do much
+class MockProcess(NamedTuple):
+    returncode: int
+    args: List[str]
+    stdout: str
+    stderr: str
+
+    def get(self):
+        return self
+
+
+# This must be a top-level function so that the pool workers can pickle it
+def mock_run(*args, **kwargs):
+    """Mock a call to `subprocess.run()`."""
+    return MockProcess(1, ["ls", "-ltrh"], b"mock bytes", b"mock bytes")
+
+
+@pytest.fixture
+def bad_comparison(monkeypatch):
+    """
+    Mocks a call to `run_multiprocessing.multiprocessing_run()` that fails.
+    """
+    pool = multiprocessing.Pool(processes=8)
+
+    def mock_apply_async(*args, **kwargs):
+        """Mock a call to `multiprocessing.Pool().apply_async()`."""
+        return multiprocessing.reduction.ForkingPickler.dumps(
+            MockProcess(1, ["ls", "-ltrh"], b"mock bytes", b"mock bytes")
+        )
+
+    monkeypatch.setattr(pool, "apply_async", mock_apply_async)
+    monkeypatch.setattr(subprocess, "run", mock_run)
+
+
+def test_failed_comparison(mp_cmdlist, bad_comparison, monkeypatch):
+    """Test that a warning is logged when an individual comparison fails."""
+
+    with assertions.assertLogs(logger, level="WARNING") as scribe:
+        multiprocessing_run(mp_cmdlist, logger=logger)
+        assertions.assertEqual(
+            scribe.output[0], "WARNING:test_multiprocessing:Comparison failed: ls -ltrh"
+        )