Source code for ewoksmx.tests.tasks.test_slurm_pipeline

import pathlib

from ewoksmx.tasks.base_tasks.slurm_pipeline import PrepareSlurmPipeline




def test_prepare_slurm_pipeline(tmp_path):
    metadata = {"reprocess_path": str(tmp_path), "no_pipelines": 1}
    pipeline = _Dummy(inputs={"metadata": metadata, "Dummy_pipeline": True})
    pipeline.execute()

    assert pipeline.script_file_path.exists()

    expected = {
        "slurm_params": {
            "script_file_path": str(
                tmp_path / "Dummy_pipeline" / "nobackup" / "dummy_script.sh"
            ),
            "queue": "mx",
            "mem": 16000,
            "nodes": 1,
            "core": 20,
            "time": "2:00:00",
            "icat_dir": str(tmp_path / "Dummy_pipeline"),
            "icat_callback_url": None,
            "no_pipelines": 1,
            "pipeline_name": "Dummy_pipeline",
            "error_message": None,
        },
        "pipeline_name": "Dummy_pipeline",
    }
    assert pipeline.get_output_values() == expected



class _Dummy(PrepareSlurmPipeline):
    PIPELINE_NAME: str = "Dummy_pipeline"

    def prepare_pipeline(self):
        self.processing_base_dir.mkdir(parents=True, exist_ok=False, mode=0o755)
        with open(self.script_file_path, "w"):
            pass

    @property
    def script_file_path(self) -> pathlib.Path:
        return self.processing_base_dir / "dummy_script.sh"