Source code for ewoksmx.tests.tasks.mx_pipelines.test_edna_pipeline

from typing import Type

from ewoksmx.models.edml.common import XSData
from ewoksmx.tasks.base_tasks.edna_pipeline import PrepareEdnaPipeline




def test_prepare_slurm_pipeline(tmp_path):
    metadata = metadata = {
        "beamline": "id30a2",
        "proposal": "mx2112",
        "MX_dataCollectionId": 123456,
        "reprocess_path": str(tmp_path),
        "no_pipelines": 1,
    }
    pipeline = _Dummy_EDNA(inputs={"metadata": metadata, "Dummy_EDNA_pipeline": True})
    pipeline.execute()

    assert pipeline.script_file_path.exists()

    with open(pipeline.script_file_path, "r") as f:
        script = f.read()
    assert 'pluginName = "EDPluginDummy"' in script

    expected = {
        "slurm_params": {
            "script_file_path": str(
                tmp_path / "Dummy_EDNA_pipeline" / "nobackup" / "edna_script.py"
            ),
            "queue": "dummy_edna_mx",
            "mem": 8000,
            "nodes": 8,
            "core": 8,
            "time": 800,
            "icat_dir": str(tmp_path / "Dummy_EDNA_pipeline"),
            "icat_callback_url": None,
            "no_pipelines": 1,
            "pipeline_name": "Dummy_EDNA_pipeline",
            "error_message": None,
        },
        "pipeline_name": "Dummy_EDNA_pipeline",
    }
    assert pipeline.get_output_values() == expected



class _Dummy_EDNA(PrepareEdnaPipeline):
    PIPELINE_NAME: str = "Dummy_EDNA_pipeline"
    EDNA_PLUGIN_NAME: str = "EDPluginDummy"
    DATA_MODEL: Type[XSData] = XSData