Merge pull request #86 from LSSTDESC/mpi-error-message

Improve error message when --mpi used incorrectly

.github/workflows/ci.yml

@@ -35,13 +35,7 @@ jobs:
 
     - name: Tests
       run: |
-        ceci tests/test.yml
-        ceci --dry-run tests/test.yml
-        ceci --flow-chart test-flow.png tests/test.yml
-        test -f test-flow.png
         pytest --cov=ceci
-        # add a test with the memory monitor and profiling switched on
-        python3 -m ceci_example PZEstimationPipe   --DM=./tests/inputs/dm.txt   --fiducial_cosmology=./tests/inputs/fiducial_cosmology.txt   --config=./tests/config.yml   --photoz_pdfs=./tests/outputs/photoz_pdfs.txt --memmon=1 --cprofile=profile.stats
 
 
     - name: Upload coverage to Codecov

ceci/stage.py

@@ -559,9 +559,20 @@ def parse_command_line(cls, cmd=None):
             help="Report memory use. Argument gives interval in seconds between reports",
         )
 
+        # Error message we will return if --mpi used on a non-supported
+        # stage.
+        mpi_err = (
+            "Error: you used the --mpi flag (or set MPI parallelism options) "
+            f"for the stage {cls.name}, but that stage cannot be run in parallel."
+        )
+
         if cmd is None:
+            if ("--mpi" in sys.argv) and not cls.parallel:
+                raise ValueError(mpi_err)
             ret_args = parser.parse_args()
         else:
+            if ("--mpi" in cmd) and not cls.parallel:
+                raise ValueError(mpi_err)
             ret_args = parser.parse_args(cmd)
 
         return ret_args

ceci_example/example_stages.py

@@ -32,6 +32,7 @@ class PZEstimationPipe(PipelineStage):
     name = "PZEstimationPipe"
     inputs = [("DM", TextFile), ("fiducial_cosmology", TextFile)]
     outputs = [("photoz_pdfs", TextFile)]
+    parallel = False
 
     def run(self):
         for inp, _ in self.inputs:

tests/test_executables.py

@@ -0,0 +1,22 @@
+import pytest
+import subprocess
+import os
+
+def test_executable():
+    subprocess.check_call(["ceci", "tests/test.yml"])
+
+def test_dry_run():
+    subprocess.check_call(["ceci", "--dry-run", "tests/test.yml",])
+
+def test_flow_chart_run():
+    subprocess.check_call(["ceci", "--flow-chart", "test-flow.png", "tests/test.yml",])
+    assert os.path.exists("test-flow.png")
+
+def test_profiling_and_memmon_flags():
+    cmd = "python3 -m ceci_example PZEstimationPipe   --DM=./tests/inputs/dm.txt   --fiducial_cosmology=./tests/inputs/fiducial_cosmology.txt   --config=./tests/config.yml   --photoz_pdfs=./tests/outputs/photoz_pdfs.txt --memmon=1 --cprofile=profile.stats"
+    subprocess.check_call(cmd.split())
+
+def test_misuse_mpi():
+    cmd = "python3 -m ceci_example PZEstimationPipe --mpi"
+    with pytest.raises(subprocess.CalledProcessError):
+        subprocess.check_call(cmd.split())

tests/test_stage.py

@@ -511,8 +511,31 @@ def test_unknown_stage():
         PipelineStage.get_stage("ThisStageIsDeliberatelyLeftBlank")
 
 
+def test_wrong_mpi_flag():
+
+    class LimaParallel(PipelineStage):
+        name = f"LimaParallel"
+        inputs = []
+        outputs = []
+        config_options = {}
+
+        def run(self):
+            pass
+
+    class LimaSerial(LimaParallel):
+        name = f"LimaSerial"
+        parallel = False
+
+    assert LimaParallel.parse_command_line(["LimaParallel", "--mpi"]).mpi
+    assert not LimaParallel.parse_command_line(["LimaParallel"]).mpi
+
+    with pytest.raises(ValueError):
+        assert LimaSerial.parse_command_line(["LimaSerial", "--mpi"]).mpi
+
+
 # could add more tests here for constructor, but the regression tests here and in TXPipe are
 # pretty thorough.
 
 if __name__ == "__main__":
     test_construct()
+    test_wrong_mpi_flag()