Source code for ewoksmx.tests.tasks.test_ednaproc_pipeline

from ewoksmx.tasks.EDNA_proc_pipeline import EDNA_proc_pipeline




def test_edna_proc_prepare(tmp_path):
    metadata = {
        "beamline": "id30a2",
        "proposal": "mx2112",
        "MX_dataCollectionId": 123456,
        "reprocess_path": str(tmp_path),
        "xds_inp_path": "/tmp/XDS.INP",
        "exclude_range": [[10, 20], [30, 40]],
        "no_pipelines": 1,
    }
    pipeline = EDNA_proc_pipeline(inputs={"metadata": metadata, "EDNA_proc": True})
    pipeline.execute()

    assert pipeline.input_file_path.exists()
    with open(pipeline.input_file_path) as f:
        input_xml = f.read()
    print(input_xml)
    assert pipeline.script_file_path.exists()

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