import json
import pytest
from ewoksmx.tasks.MXPipelineInput import MXPipelineInput
[docs]
@pytest.fixture()
def mxpipeline_artefacts(tmp_path):
raw_data_path = tmp_path / "RAW_DATA"
raw_data_path.mkdir()
metadata = {"MX_dataCollectionId": 123456, "MX_template": "template"}
metadata_path = raw_data_path / "metadata.json"
with open(metadata_path, "w") as f:
f.write(json.dumps(metadata, indent=4))
reprocess_root_path = tmp_path / "PROCESSED_DATA"
reprocess_root_path.mkdir()
test_autoproc_dir = reprocess_root_path / "autoproc"
test_autoproc_dir.mkdir()
xds_path = test_autoproc_dir / "XDS.INP"
with open(xds_path, "w") as f:
f.write("XDS INP\n")
return {
"raw_data_path": raw_data_path,
"reprocess_root_path": reprocess_root_path,
"xds_path": xds_path,
}
[docs]
def test_run_valid(mxpipeline_artefacts):
mxpipeline_input = MXPipelineInput(
inputs={
"raw_data_path": [str(mxpipeline_artefacts["raw_data_path"])],
"callback": None,
"mx_pipeline_name": [
"EDNA_proc",
"autoPROC",
"XIA2_DIALS",
"grenades_fastproc",
"grenades_parallelproc",
],
"beamline": "id30a-1",
"icat_metadata": None,
"forced_spacegroup": "P1",
"forced_cell": None,
"start_image": 10,
"end_image": 990,
"anomalous": False,
"low_res_limit": 4.0,
"high_res_limit": 1.0,
"exclude_range": "10-20, 30-40",
}
)
mxpipeline_input.run()
output_values = mxpipeline_input.get_output_values()
assert output_values["EDNA_proc"]
assert output_values["XIA2_DIALS"]
assert output_values["autoPROC"]
assert output_values["grenades_fastproc"]
assert output_values["grenades_parallelproc"]
metadata = output_values["metadata"]
assert metadata["MX_dataCollectionId"] == 123456
assert metadata["MX_template"] == "template"
assert not metadata["anomalous"]
assert metadata["beamline"] == "unknown"
assert metadata["proposal"] == "unknown"
assert metadata["start_image"] == 10
assert metadata["end_image"] == 990
assert metadata["forced_cell"] is None
assert metadata["forced_spacegroup"] == "P1"
assert metadata["high_res_limit"] == 1.0
assert metadata["icat_callback_url"] is None
assert metadata["job_id"] is None
assert metadata["low_res_limit"] == 4.0
assert metadata["no_pipelines"] == 5
assert metadata["reprocess_path"] == str(
mxpipeline_artefacts["reprocess_root_path"] / "reprocess_template_run_1"
)
assert metadata["xds_inp_path"] == str(mxpipeline_artefacts["xds_path"])
assert metadata["exclude_range"] == ((10, 20), (30, 40))
[docs]
def test_run_invalid_exclude_range(mxpipeline_artefacts):
mxpipeline_input = MXPipelineInput(
inputs={
"raw_data_path": [str(mxpipeline_artefacts["raw_data_path"])],
"callback": None,
}
)
mxpipeline_input.run()
output_values = mxpipeline_input.get_output_values()
assert "exclude_range" not in output_values
mxpipeline_input = MXPipelineInput(
inputs={
"raw_data_path": [str(mxpipeline_artefacts["raw_data_path"])],
"callback": None,
"exclude_range": None,
}
)
mxpipeline_input.run()
output_values = mxpipeline_input.get_output_values()
assert "exclude_range" not in output_values
mxpipeline_input = MXPipelineInput(
inputs={
"raw_data_path": [str(mxpipeline_artefacts["raw_data_path"])],
"callback": None,
"exclude_range": " ",
}
)
mxpipeline_input.run()
output_values = mxpipeline_input.get_output_values()
assert "exclude_range" not in output_values
mxpipeline_input = MXPipelineInput(
inputs={
"raw_data_path": [str(mxpipeline_artefacts["raw_data_path"])],
"callback": None,
"exclude_range": "10-20a, 30-40",
}
)
with pytest.raises(ValueError, match="Invalid exclude_range format '10-20a,30-40'"):
mxpipeline_input.run()
[docs]
def test_run_long_space_group_name(mxpipeline_artefacts):
mxpipeline_input = MXPipelineInput(
inputs={
"raw_data_path": [str(mxpipeline_artefacts["raw_data_path"])],
"callback": None,
"forced_spacegroup": " P 1 2 1",
}
)
mxpipeline_input.run()
metadata = mxpipeline_input.get_output_values()["metadata"]
assert metadata["forced_spacegroup"] == "P2"