olink_run_cmd_fix

igf_airflow/utils/dag50_olink_reveal_nextflow_utils.py

@@ -52,7 +52,7 @@ def prepare_olink_nextflow_script(
       plate_design_csv_key: str = "plate_design_csv",
       indexPlate_key: str = "indexPlate",
       plate_design_csv_file: str = "plate_design.csv"
-    ) -> str:
+    ) -> dict[str, str]:
   try:
     ## read design and get sample metadata
     design_file = \
@@ -138,7 +138,14 @@ def prepare_olink_nextflow_script(
         WORKDIR=work_dir
       )
     )
-    return {"run_script": nf_script_file.as_posix()}
+    run_script = nf_script_file.as_posix()
+    script_dir = os.path.dirname(run_script)
+    run_cmd = f"""set -eo pipefail;
+## Move to the script dir
+cd {script_dir} ;
+chmod u+x {run_script} ;
+bash {run_script} ;"""
+    return {"run_script": run_script, "run_cmd": run_cmd}
   except Exception as e:
     log.error(e)
     send_airflow_failed_logs_to_channels(
@@ -153,15 +160,9 @@ def prepare_olink_nextflow_script(
   queue='hpc_8G4t72hr',
   pool='batch_job',
   retries=4)
-def run_olink_nextflow_script(run_script: str) -> str:
+def run_olink_nextflow_script(run_cmd: str) -> str:
   try:
-    script_dir = os.path.dirname(run_script)
-    bash_cmd = f"""set -eo pipefail;
-## Move to the script dir
-cd {script_dir} ;
-chmod u+x {run_script} ;
-bash {run_script} ;"""
-    return bash_cmd
+    return run_cmd
   except Exception as e:
     log.error(e)
     send_airflow_failed_logs_to_channels(

test/igf_airflow/test_dag50_olink_reveal_nextflow_utils.py

@@ -77,19 +77,22 @@ def test_prepare_olink_nextflow_script(self, *args):
     assert len(csv_data.split("\n")) == 5
     assert "well_id;sample_id;sample_type\nA1;A200;SAMPLE" in csv_data
     assert "G11;Olink_external_control1;NEGATIVE_CONTROL" in csv_data
+    run_cmd = analysis_script_conf["run_cmd"]
+    script_path = os.path.dirname(analysis_script)
+    assert run_cmd == f"""set -eo pipefail;
+## Move to the script dir
+cd {script_path} ;
+chmod u+x {analysis_script} ;
+bash {analysis_script} ;"""
 
 
 
   def test_run_olink_nextflow_script(self):
     script_file = "/tmp/t.sh"
     analysis_cmd = run_olink_nextflow_script.function(
-      script_file
+      run_cmd="echo test"
     )
-    assert analysis_cmd == f"""set -eo pipefail;
-## Move to the script dir
-cd {os.path.dirname(script_file)} ;
-chmod u+x {script_file} ;
-bash {script_file} ;"""
+    assert analysis_cmd == "echo test"
 
 if __name__=='__main__':
   unittest.main()