Unit test pipeline (simonsobs#33)

* pipeline

* cleanup

* gha

* typo

* unit test

* unit test

* unit test

* misnomer

.github/workflows/ci.yml

@@ -0,0 +1,50 @@
+name: continuous-integration
+on:
+  push:
+    branches:
+      - master
+  pull_request:
+    branches:
+      - master
+jobs:
+  build:
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        os: [ ubuntu-latest ]
+        py: [ 3.8 ]
+
+    steps:
+      - uses: actions/checkout@v2
+
+      - name: Set up Python ${{ matrix.py }}
+        uses: actions/setup-python@v2
+        with:
+          python-version: ${{ matrix.py }}
+
+      - name: Install python dependencies
+        run: |
+          python -m pip install -U pip
+          pip install -U numpy
+          pip install -U healpy
+          pip install -U pytest
+
+      - name: Install BBPipe
+        run: |
+          python setup.py install --user
+
+      - name: Unit tests
+        run: |
+          export PATH=$PATH:$HOME/.local/bin
+          bbpipe test/test_ini.yml --python-cmd=python
+          pytest -vv test/
+
+      - name: Cache pip, brew
+        uses: actions/cache@v2
+        with:
+          path: |
+            ~/.cache/pip
+          key: ${{ runner.os }}-${{ matrix.py }}-cache
+          restore-keys: |
+            ${{ runner.os }}-${{ matrix.py }}-cache
+            ${{ runner.os }}-

bbpipe/main.py

@@ -13,8 +13,9 @@
 parser.add_argument('pipeline_config', help='Pipeline configuration file in YAML format.')
 parser.add_argument('--export-cwl', type=str, help='Exports pipeline in CWL format to provided path and exits')
 parser.add_argument('--dry-run', action='store_true', help='Just print out the commands the pipeline would run without running them')
+parser.add_argument('--python-cmd', type=str, default='python3', help='Command that calls the python interpreter')
 
-def run(pipeline_config_filename, dry_run=False):
+def run(pipeline_config_filename, dry_run=False, pycmd='python3'):
     """
     Runs the pipeline
     """
@@ -59,7 +60,7 @@ def run(pipeline_config_filename, dry_run=False):
         __import__(module)
 
     # Create and run pipeline
-    pipeline = Pipeline(launcher_config, stages)
+    pipeline = Pipeline(launcher_config, stages, pycmd=pycmd)
 
     if dry_run:
         pipeline.dry_run(inputs, output_dir, stages_config)
@@ -107,7 +108,7 @@ def main():
     if args.export_cwl is not None:
         export_cwl(args)
     else:
-        run(args.pipeline_config, dry_run=args.dry_run)
+        run(args.pipeline_config, dry_run=args.dry_run, pycmd=args.python_cmd)
 
 if __name__ == '__main__':
     main()

bbpipe/pipeline.py

@@ -11,10 +11,11 @@ def __init__(self, info):
         self.nprocess = info.get('nprocess', 1)
 
 class Pipeline:
-    def __init__(self, launcher_config, stages):
+    def __init__(self, launcher_config, stages, pycmd='python3'):
         self.stage_execution_config = {}
         self.stage_names = []
         self.mpi_command = launcher_config['mpi_command']
+        self.python_command = pycmd
         self.dfk = parsl.DataFlowKernel(launcher_config)
         for info in stages:
             self.add_stage(info)
@@ -91,7 +92,10 @@ def dry_run(self, overall_inputs, output_dir, stages_config):
 
         for stage in stages:
             sec = self.stage_execution_config[stage.name]
-            cmd = stage.generate_command(overall_inputs, stages_config, output_dir, sec.nprocess, self.mpi_command)
+            cmd = stage.generate_command(overall_inputs, stages_config, output_dir,
+                                         nprocess=sec.nprocess,
+                                         mpi_command=self.mpi_command,
+                                         python_command=self.python_command)
             print(cmd)
             print()
 
@@ -108,7 +112,9 @@ def run(self, overall_inputs, output_dir, log_dir, resume, stages_config):
 
         for stage in stages:
             sec = self.stage_execution_config[stage.name]
-            app = stage.generate(self.dfk, sec.nprocess, sec.site, log_dir, mpi_command=self.mpi_command)
+            app = stage.generate(self.dfk, sec.nprocess, sec.site, log_dir,
+                                 mpi_command=self.mpi_command,
+                                 python_command=self.python_command)
             inputs = self.find_inputs(stage, data_elements)
             outputs = self.find_outputs(stage, output_dir)
             # All pipeline stages implicitly get the overall configuration file

bbpipe/stage.py

@@ -525,7 +525,7 @@ def get_module(cls):
     def usage(cls):
         stage_names = "\n- ".join(cls.pipeline_stages.keys())
         sys.stderr.write(f"""
-Usage: python -m txpipe <stage_name> <stage_arguments>
+Usage: python -m <module_name> <stage_name> <stage_arguments>
 
 If no stage_arguments are given then usage information
 for the chosen stage will be given.
@@ -639,7 +639,8 @@ def _generate(cls, template, dfk):
         return function
 
     @classmethod
-    def generate_command(cls, external_inputs, config, outdir, nprocess=1, mpi_command='mpirun -n'):
+    def generate_command(cls, external_inputs, config, outdir, nprocess=1,
+                         mpi_command='mpirun -n', python_command='python3'):
         """
         Generate a command line that will run the stage
         """
@@ -672,11 +673,12 @@ def generate_command(cls, external_inputs, config, outdir, nprocess=1, mpi_comma
 
         # We just return this, instead of wrapping it in a
         # parsl job
-        cmd = f'{launcher} python3 -m {module} {flags} {mpi_flag}'
+        cmd = f'{launcher} {python_command} -m {module} {flags} {mpi_flag}'
         return cmd
 
     @classmethod
-    def generate(cls, dfk, nprocess, site_name, log_dir, mpi_command='mpirun -n'):
+    def generate(cls, dfk, nprocess, site_name, log_dir,
+                 mpi_command='mpirun -n', python_command='python3'):
         """
         Build a parsl bash app that executes this pipeline stage
         """
@@ -712,7 +714,7 @@ def generate(cls, dfk, nprocess, site_name, log_dir, mpi_command='mpirun -n'):
         template = f"""
 @parsl.App('bash', dfk, sites=['{site_name}'])
 def {cls.name}(inputs, outputs, stdout='{log_dir}/{cls.name}.out', stderr='{log_dir}/{cls.name}.err'):
-    cmd = '{launcher} python3 -m {module} {flags} {mpi_flag}'.format(inputs=inputs,outputs=outputs)
+    cmd = '{launcher} {python_command} -m {module} {flags} {mpi_flag}'.format(inputs=inputs,outputs=outputs)
     print("Compiling command:")
     print(cmd)
     return cmd