chore: fix command execution when job_path is not cwd

Author: AcquaDiGiorgio

pixi.lock

@@ -6,6 +6,7 @@
 import json
 import logging
 import os
+from pathlib import Path
 
 from DIRAC.AccountingSystem.Client.DataStoreClient import DataStoreClient
 from DIRAC.Core.Utilities.ReturnValues import SErrorException, returnValueOrRaise
@@ -76,7 +77,7 @@ def _execute(self, job_path: os.PathLike[str], workflow_commons: WorkflowCommons
         if workflow_commons.step_status == StepStatus.Done:
             self.job_report.setApplicationStatus("Job Finished Successfully", True)
 
-        self.generate_failover_file(workflow_commons)
+        self.generate_failover_file(job_path, workflow_commons)
 
     def _resolve_clients(self, workflow_commons: WorkflowCommons):
         super()._resolve_clients(workflow_commons)
@@ -87,7 +88,7 @@ def _resolve_clients(self, workflow_commons: WorkflowCommons):
         if not self.dsc:
             self.dsc = DataStoreClient()
 
-    def generate_failover_file(self, workflow_commons: WorkflowCommons):
+    def generate_failover_file(self, job_path: os.PathLike[str], workflow_commons: WorkflowCommons):
         """Create a request.json file."""
         try:
             diset_op = returnValueOrRaise(self.job_report.generateForwardDISET())
@@ -114,7 +115,9 @@ def generate_failover_file(self, workflow_commons: WorkflowCommons):
             request_json_content = returnValueOrRaise(self.request.toJSON())
 
             # Write it
-            fname = f"{workflow_commons.production_id}_{workflow_commons.prod_job_id}_request.json"
+            fname = Path(job_path).joinpath(
+                f"{workflow_commons.production_id}_{workflow_commons.prod_job_id}_request.json"
+            )
             with open(fname, "w", encoding="utf-8") as f:
                 json.dump(request_json_content, f)
 

src/dirac_cwl/commands/create_failover_request.py

@@ -3,6 +3,7 @@
 import copy
 import logging
 import os
+from pathlib import Path
 from typing import Any, Dict
 
 from DIRAC.Core.Utilities.ReturnValues import SErrorException, returnValueOrRaise
@@ -102,6 +103,7 @@ def _execute_for_step(
                 workflow_commons.prod_job_id,
                 step_commons.number,
                 step_commons.id,
+                path=job_path,
             )
 
         info_dict = {
@@ -145,9 +147,10 @@ def _execute_for_step(
             step_commons.xf_o,
             step_commons.outputs,
             step_commons.inputs,
-            step_commons.size,
-            step_commons.md5,
-            step_commons.guid,
+            path=job_path,
+            size_map=step_commons.size,
+            md5_map=step_commons.md5,
+            guid_map=step_commons.guid,
         )
 
         # Generate SimulationConditions
@@ -158,7 +161,7 @@ def _execute_for_step(
         doc = job_info.to_xml()
 
         # Write to file
-        bfilename = f"bookkeeping_{step_commons.id}.xml"
+        bfilename = Path(job_path).joinpath(f"bookkeeping_{step_commons.id}.xml")
         with open(bfilename, "wb") as bfile:
             bfile.write(doc)
 

src/dirac_cwl/commands/report_bookkeeping.py

@@ -3,6 +3,7 @@
 import logging
 import os
 import shlex
+from pathlib import Path
 
 from DIRAC.ConfigurationSystem.Client.Helpers.Operations import Operations
 from DIRAC.Core.Utilities.ReturnValues import SErrorException, returnSingleResult, returnValueOrRaise
@@ -75,7 +76,7 @@ def _execute(self, job_path: os.PathLike[str], workflow_commons: WorkflowCommons
         _prepareRequest(self.request, workflow_commons.job_id)
 
         try:
-            file_list = returnValueOrRaise(systemCall(0, shlex.split("ls -al")))
+            file_list = returnValueOrRaise(systemCall(0, shlex.split(f"ls -al {str(job_path)}")))
             if file_list:
                 logger.info("The contents of the working directory...")
                 logger.info(str(file_list[1]))
@@ -85,7 +86,7 @@ def _execute(self, job_path: os.PathLike[str], workflow_commons: WorkflowCommons
             logger.error("Failed to list the log directory\n%s", e)
 
         workflow_commons.log_dir = os.path.realpath(
-            os.path.join(job_path, f"./job/log/{workflow_commons.production_id}/{workflow_commons.prod_job_id}")
+            Path(job_path).joinpath("job", "log", workflow_commons.production_id, workflow_commons.production_id)
         )
         logger.info("Selected log files will be temporarily stored in %s", workflow_commons.log_dir)
 
@@ -94,7 +95,7 @@ def _execute(self, job_path: os.PathLike[str], workflow_commons: WorkflowCommons
         logger.info("Determining the files to be saved in the logs.")
 
         try:
-            selected_files = returnValueOrRaise(_determineRelevantFiles(log_extensions))
+            selected_files = returnValueOrRaise(_determineRelevantFiles(log_extensions, path=job_path))
         except SErrorException as e:
             logger.error("Completely failed to select relevant log files.", exc_info=e)
             return  # Does not fail
@@ -123,7 +124,7 @@ def _execute(self, job_path: os.PathLike[str], workflow_commons: WorkflowCommons
 
         # zip all files
         try:
-            zip_file_name = returnValueOrRaise(_zip_files(workflow_commons.prod_job_id, selected_files))
+            zip_file_name = returnValueOrRaise(_zip_files(workflow_commons.prod_job_id, selected_files, path=job_path))
         except SErrorException as e:
             logger.error("Failed to create zip of log files %s", e)
             self.job_report.setApplicationStatus("Failed to create zip of log files")
@@ -133,7 +134,7 @@ def _execute(self, job_path: os.PathLike[str], workflow_commons: WorkflowCommons
 
         # logFilePath is something like /lhcb/MC/2016/LOG/00095376/0000/
         # the zipFileName should have the same name, e.g. 00000381.zip
-        zip_path = os.path.join(workflow_commons.log_file_path, zip_file_name)
+        zip_path = Path(job_path).joinpath(workflow_commons.log_file_path, zip_file_name)
         log_https_url = _get_log_url(log_se, zip_path)
 
         logger.info("putFile %s to %s", zip_file_name, log_se)
@@ -150,10 +151,7 @@ def _execute(self, job_path: os.PathLike[str], workflow_commons: WorkflowCommons
             try:
                 upload_result_dict = returnValueOrRaise(
                     _uploadLogToFailoverSE(
-                        self.failover_transfer,
-                        zip_file_name,
-                        log_lfn_path,
-                        workflow_commons.site_name,
+                        self.failover_transfer, zip_file_name, log_lfn_path, workflow_commons.site_name, path=job_path
                     )
                 )
 

src/dirac_cwl/commands/upload_log_file.py

@@ -4,6 +4,7 @@
 import os
 import random
 import time
+from pathlib import Path
 
 from DIRAC.Core.Utilities.ReturnValues import SErrorException, returnValueOrRaise
 from DIRAC.DataManagementSystem.Client.DataManager import DataManager
@@ -75,6 +76,7 @@ def _execute(self, job_path: os.PathLike[str], workflow_commons: WorkflowCommons
             workflow_commons.output_data_file_mask,
             workflow_commons.output_data_step,
             workflow_commons.output_SEs,
+            path=job_path,
         )
 
         if not file_metadata:
@@ -118,7 +120,7 @@ def _execute(self, job_path: os.PathLike[str], workflow_commons: WorkflowCommons
                 self.file_report.setFileStatus(int(workflow_commons.production_id), lfns_with_descendents, "Processed")
                 raise WorkflowProcessingException("Input Data Already Processed")
 
-        bk_files = _getBKFiles()
+        bk_files = _getBKFiles(path=job_path)
         logger.info("The following BK records will be sent\n%s", ", ".join(bk_files))
 
         for bk_file in bk_files:
@@ -134,7 +136,7 @@ def _execute(self, job_path: os.PathLike[str], workflow_commons: WorkflowCommons
                 logger.info("Preparing DISET request for %s", bk_file)
 
         logger.info("Creating DISABLE_WATCHDOG_CPU_WALLCLOCK_CHECK in order to disable the Watchdog")
-        with open("DISABLE_WATCHDOG_CPU_WALLCLOCK_CHECK", "w") as f:
+        with open(Path(job_path).joinpath("DISABLE_WATCHDOG_CPU_WALLCLOCK_CHECK"), "w") as f:
             f.write(f"{time.asctime()}")
 
         perform_bk_registration = []
@@ -143,7 +145,9 @@ def _execute(self, job_path: os.PathLike[str], workflow_commons: WorkflowCommons
         for file_name, metadata in final.items():
             target_se = metadata["resolvedSE"]
 
-            logger.info("Attempting to store file to SE %s to the following SE(s):\n%s", file_name, ", ".join(target_se))
+            logger.info(
+                "Attempting to store file to SE %s to the following SE(s):\n%s", file_name, ", ".join(target_se)
+            )
 
             file_meta_dict = _createMetaDict(metadata)
 

src/dirac_cwl/commands/upload_output_data.py

@@ -1,5 +1,6 @@
 """Common pytest fixture used by test modules."""
 
+import shutil
 from pathlib import Path
 
 import pytest
@@ -56,7 +57,16 @@ def sample_job(sample_command_line_tool):
 @pytest.fixture
 def job_path():
     """Job Path Fixture."""
-    return Path(".")
+    path = Path(".").joinpath("examplepath")
+
+    if path.exists():
+        raise Exception("NON EMPTY DIRECTORY !!!")
+
+    path.mkdir()
+
+    yield path
+
+    shutil.rmtree(path)
 
 
 @pytest.fixture

test/conftest.py

@@ -82,7 +82,7 @@ def test_report_bk_prod_mcsimulation_success(
         wf_commons["steps"][0]["outputs"] = [
             {"outputDataName": "00209455_00001537_1.sim", "outputDataType": "sim"},
         ]
-        Path(wf_commons["steps"][0]["outputs"][0]["outputDataName"]).touch()
+        Path(job_path).joinpath(wf_commons["steps"][0]["outputs"][0]["outputDataName"]).touch()
 
         # Mock the XMLSummary object
         xml_content = dedent("""\
@@ -327,8 +327,8 @@ def test_report_bk_prod_mcreconstruction_success(
             {"outputDataName": "00209455_00001537_1", "outputDataType": "digi"},
         ]
         wf_commons["steps"][0]["application_log"] = "application.log"
-        Path(wf_commons["steps"][0]["application_log"]).touch()
-        Path(wf_commons["steps"][0]["outputs"][0]["outputDataName"]).touch()
+        Path(job_path).joinpath(wf_commons["steps"][0]["application_log"]).touch()
+        Path(job_path).joinpath(wf_commons["steps"][0]["outputs"][0]["outputDataName"]).touch()
 
         # Mock the XMLSummary object
         xml_content = dedent("""\

test/test_command_bookkeeping_report.py

@@ -35,7 +35,7 @@ def uplogfile(self, mocker: MockerFixture, wf_commons, job_path):
         Path(job_path).joinpath(f"{wf_commons['prod_job_id']}.zip").unlink(missing_ok=True)
         Path(job_path).joinpath("workflow_commons.json").unlink(missing_ok=True)
 
-        shutil.rmtree(Path(job_path).joinpath("unzipped"), ignore_errors=True)
+        shutil.rmtree(Path("unzipped"), ignore_errors=True)
 
     @pytest.fixture
     def prodconf_json(self, job_path):
@@ -45,7 +45,7 @@ def prodconf_json(self, job_path):
         with open(file, "w") as f:
             f.write('{"foo": "bar"}')
 
-        yield file
+        yield file.name
 
         file.unlink(missing_ok=True)
 
@@ -57,7 +57,7 @@ def prodconf_py(self, job_path):
         with open(file, "w") as f:
             f.write('foo = "bar"')
 
-        yield file
+        yield file.name
 
         file.unlink(missing_ok=True)
 
@@ -96,7 +96,7 @@ def test_uploadLogFile_success(
         assert zipFile.exists()
 
         zipfile.ZipFile(zipFile, "r").extractall("unzipped")
-        unzipped = Path("unzipped").joinpath(updated_wf_commons.prod_job_id)
+        unzipped = Path("unzipped").joinpath(updated_wf_commons.prod_job_id, job_path)
         assert unzipped.joinpath(prodconf_json).exists()
         assert unzipped.joinpath(prodconf_py).exists()
         assert unzipped.joinpath(prodconf_json).read_text() == '{"foo": "bar"}'
@@ -288,7 +288,7 @@ def test_uploadLogFile_SEError(
         assert zipFile.exists()
 
         zipfile.ZipFile(zipFile, "r").extractall("unzipped")
-        unzipped = Path("unzipped").joinpath(updated_wf_commons.prod_job_id)
+        unzipped = Path("unzipped").joinpath(updated_wf_commons.prod_job_id, job_path)
         assert unzipped.joinpath(prodconf_json).exists()
         assert unzipped.joinpath(prodconf_py).exists()
         assert unzipped.joinpath(prodconf_json).read_text() == '{"foo": "bar"}'
@@ -355,7 +355,7 @@ def test_uploadLogFile_transferError(
         assert zipFile.exists()
 
         zipfile.ZipFile(zipFile, "r").extractall("unzipped")
-        unzipped = Path("unzipped").joinpath(updated_wf_commons.prod_job_id)
+        unzipped = Path("unzipped").joinpath(updated_wf_commons.prod_job_id, job_path)
         assert unzipped.joinpath(prodconf_json).exists()
         assert unzipped.joinpath(prodconf_py).exists()
         assert unzipped.joinpath(prodconf_json).read_text() == '{"foo": "bar"}'

test/test_command_upload_log_file.py

@@ -34,7 +34,7 @@ def sim_file(self, wf_commons, job_path):
         with open(path, "w") as f:
             f.write("Bookkeeping file content")
 
-        yield str(path)
+        yield str(path.name)
 
         path.unlink(missing_ok=True)
 
@@ -555,7 +555,7 @@ def test_uploadOutputData_noOutput(self, mocker: MockerFixture, upload_output, j
         wf_commons["output_data_step"] = self.OUTPUT_DATA_STEP
 
         # Remove the output
-        Path(sim_file).unlink(missing_ok=True)
+        Path(job_path).joinpath(sim_file).unlink(missing_ok=True)
 
         WorkflowCommons(**wf_commons).save(job_path)