from datetime import datetime
from datetime import timedelta
from pathlib import Path
from ..apps import store_from_file
from ..apps import store_processed
from ..apps import store_raw
from .utils import compare
[docs]
def test_store_raw(icat_metadata_client, icat_backend):
_, messages = icat_metadata_client
argv = [
"",
"--beamline",
"id00",
"--proposal",
"hg123",
"--dataset",
"datasetname",
"--path",
_dummy_path("dataset"),
"--sample",
"samplename",
"-p",
"InstrumentBeam_distance=0.01",
"-p",
"InstrumentBeam_incident_energy=[10, 10.1, 10.2]",
"--queue",
icat_backend["metadata_urls"][0],
]
store_raw.main(argv)
message = messages.get(timeout=10)
assert messages.empty()
expected = {
"complete": True,
"instrument": "id00",
"investigation": "hg123",
"location": _dummy_path("dataset"),
"name": "datasetname",
"parameter": [
{"name": "complete", "value": "true"},
{"name": "beamlineID", "value": "id00"},
{"name": "proposal", "value": "hg123"},
{"name": "location", "value": str(_dummy_path("dataset"))},
{"name": "Sample_name", "value": "samplename"},
{"name": "datasetName", "value": "datasetname"},
{"name": "InstrumentBeam_distance", "value": "0.01"},
{"name": "InstrumentBeam_incident_energy", "value": "10.0 10.1 10.2"},
],
"sample": {"name": "samplename"},
}
compare.assert_equal_dataset_message(message, expected)
[docs]
def test_store_processed(icat_metadata_client, icat_backend):
_, messages = icat_metadata_client
argv = [
"",
"--beamline",
"id00",
"--proposal",
"hg123",
"--dataset",
"datasetname",
"--path",
_dummy_path("processed"),
"--sample",
"samplename",
"-p",
"InstrumentBeam_distance=0.01",
"-p",
"InstrumentBeam_incident_energy=[10, 10.1, 10.2]",
"--queue",
icat_backend["metadata_urls"][0],
"--raw",
_dummy_path("dataset1"),
"--raw",
_dummy_path("dataset2"),
]
store_processed.main(argv)
expected = {
"complete": True,
"instrument": "id00",
"investigation": "hg123",
"location": _dummy_path("processed"),
"name": "datasetname",
"parameter": [
{"name": "complete", "value": "true"},
{"name": "beamlineID", "value": "id00"},
{"name": "proposal", "value": "hg123"},
{"name": "location", "value": str(_dummy_path("processed"))},
{"name": "Sample_name", "value": "samplename"},
{"name": "datasetName", "value": "datasetname"},
{"name": "InstrumentBeam_distance", "value": "0.01"},
{"name": "InstrumentBeam_incident_energy", "value": "10.0 10.1 10.2"},
{
"name": "input_datasets",
"value": f"{_dummy_path('dataset1')},{_dummy_path('dataset2')}",
},
],
"sample": {"name": "samplename"},
}
message = messages.get(timeout=10)
assert messages.empty()
compare.assert_equal_dataset_message(message, expected)
[docs]
def test_store_from_file(icat_main_client, icat_backend, tmp_path):
client, messages = icat_main_client
icat_dir = tmp_path / "__icat__"
icat_dir.mkdir()
start_date = datetime.now()
client.start_investigation(
beamline="id00",
proposal="hg123",
start_datetime=start_date,
)
messages.get(timeout=10)
dataset_start = start_date + timedelta(hours=1)
dataset_end = start_date + timedelta(hours=2)
file1 = icat_dir / "dataset1.xml"
file1.write_text(
_build_dataset_xml(
"0001", str(icat_dir / "sample_0001"), dataset_start, dataset_end
)
)
argv = [
"store_from_file_script",
str(file1),
"--queue",
icat_backend["metadata_urls"][0],
]
store_from_file.main(argv)
message = messages.get(timeout=10)
expected = {
"complete": True,
"name": "0001",
"instrument": "id00",
"investigation": "hg123",
"location": str(icat_dir / "sample_0001"),
"parameter": [
{"name": "startDate", "value": dataset_start.isoformat()},
{"name": "endDate", "value": dataset_end.isoformat()},
{"name": "Sample_name", "value": "sample"},
{"name": "machine", "value": "machine"},
{"name": "software", "value": "test"},
],
"sample": {"name": "sample", "parameter": [{}]},
}
compare.assert_equal_dataset_message(message, expected)
assert messages.empty()
def _build_dataset_xml(name, location, start_date, end_date):
return (
f'<tns:dataset xmlns:tns="http://www.esrf.fr/icat" complete="true">'
f"<tns:investigation>hg123</tns:investigation>"
f"<tns:instrument>id00</tns:instrument>"
f"<tns:name>{name}</tns:name>"
f"<tns:location>{location}</tns:location>"
f"<tns:parameter><tns:name>Sample_name</tns:name><tns:value>sample</tns:value></tns:parameter>"
f"<tns:parameter><tns:name>startDate</tns:name><tns:value>{start_date.isoformat()}</tns:value></tns:parameter>"
f"<tns:parameter><tns:name>endDate</tns:name><tns:value>{end_date.isoformat()}</tns:value></tns:parameter>"
f"<tns:parameter><tns:name>machine</tns:name><tns:value>machine</tns:value></tns:parameter>"
f"<tns:parameter><tns:name>software</tns:name><tns:value>test</tns:value></tns:parameter>"
f"<tns:startDate>{start_date.isoformat()}</tns:startDate>"
f"<tns:endDate>{end_date.isoformat()}</tns:endDate>"
f"<tns:sample><tns:name>sample</tns:name><tns:parameter></tns:parameter></tns:sample>"
f"</tns:dataset>"
)
def _dummy_path(dirname: str) -> str:
return str(Path.home() / dirname)