Source code for pyicat_plus.tests.test_icat_datasets
import os
import re
import sys
from pathlib import Path
import pytest
from .. import errors
from ..concurrency import Empty
from .utils import compare
from .utils import generate
[docs]
def test_start_investigation(icat_metadata_client):
client, messages = icat_metadata_client
client.check_health()
client.start_investigation(proposal="hg123", beamline="id00")
message = messages.get(timeout=10)
assert messages.empty()
expected = {
"instrument": "id00",
"experiment": "hg123",
}
compare.assert_equal_investigation_message(message, expected)
[docs]
def test_start_bad_investigation(icat_metadata_client):
client, messages = icat_metadata_client
client.check_health()
client.start_investigation(proposal="hg666", beamline="id00")
with pytest.raises(Empty):
messages.get(timeout=2)
[docs]
def test_send_metadata(icat_metadata_client):
client, messages = icat_metadata_client
metadata = {
"Sample_name": "samplename",
"InstrumentBeam_distance": 0.01,
"InstrumentBeam_incident_energy": [10, 10.1, 10.2],
}
original_metadata = metadata.copy()
client.send_metadata(
proposal="hg123",
beamline="id00",
dataset="datasetname",
path=_dummy_path("dataset"),
metadata=metadata,
)
message = messages.get(timeout=10)
assert messages.empty()
assert metadata == original_metadata
expected = {
"complete": True,
"instrument": "id00",
"investigation": "hg123",
"location": str(_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_send_metadata_via_file(icat_metadata_client, tmpdir):
store_filename = tmpdir / "test.xml"
metadata = {
"Sample_name": "samplename",
"InstrumentBeam_distance": 0.01,
"InstrumentBeam_incident_energy": [10, 10.1, 10.2],
}
original_metadata = metadata.copy()
client, messages = icat_metadata_client
client.store_metadata(
str(store_filename),
proposal="hg123",
beamline="id00",
dataset="datasetname",
path=_dummy_path("dataset"),
metadata=metadata,
)
with pytest.raises(Empty):
message = messages.get(timeout=1)
assert metadata == original_metadata
assert store_filename.exists()
client.send_metadata_from_file(str(store_filename))
message = messages.get(timeout=10)
assert messages.empty()
expected = {
"complete": True,
"instrument": "id00",
"investigation": "hg123",
"location": str(_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_send_missing_data(icat_metadata_client):
client, messages = icat_metadata_client
with pytest.raises(AssertionError, match="ICAT requires the beamline name"):
client.send_metadata(
proposal=None,
beamline=None,
dataset=None,
path=None,
metadata=None,
)
[docs]
def test_send_missing_metadata(icat_metadata_client):
client, messages = icat_metadata_client
with pytest.raises(
AssertionError, match="ICAT metadata field 'Sample_name' is missing"
):
client.send_metadata(
proposal="hg123",
beamline="id00",
dataset="datasetname",
path=_dummy_path("dataset"),
metadata={},
)
[docs]
def test_send_metadata_with_machine_software(icat_metadata_client):
client, messages = icat_metadata_client
client.send_metadata(
proposal="hg123",
beamline="id00",
dataset="datasetname",
path=_dummy_path("dataset"),
metadata={
"Sample_name": "samplename",
"InstrumentBeam_distance": 0.01,
"InstrumentBeam_incident_energy": [10, 10.1, 10.2],
"machine": "mymachine",
"software": "mysoftware_version",
},
)
message = messages.get(timeout=10)
assert messages.empty()
expected = {
"complete": True,
"instrument": "id00",
"investigation": "hg123",
"location": str(_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"},
{"name": "machine", "value": "mymachine"},
{"name": "software", "value": "mysoftware_version"},
],
"sample": {"name": "samplename"},
}
compare.assert_equal_dataset_message(message, expected)
[docs]
def test_reschedule_investigation(icat_metadata_client):
client, messages = icat_metadata_client
client.check_health()
investigation_id = generate.icat_id()
client.reschedule_investigation(investigation_id=investigation_id)
message = messages.get(timeout=10)
assert messages.empty()
expected = {
"experiment": "",
"instrument": "",
"investigationId": investigation_id,
}
compare.assert_equal_investigation_message(message, expected)
[docs]
@pytest.mark.parametrize("path_defect", ["none", "trailing_slash", "up"])
def test_send_metadata_path_defects(icat_metadata_client, path_defect):
expected_path = _dummy_path("dataset")
expected_input_datasets = [_dummy_path("dataset1"), _dummy_path("dataset2")]
if path_defect == "trailing_slash":
path = _add_trailing_slash(expected_path)
assert path.endswith(os.path.sep)
input_datasets = [_add_trailing_slash(s) for s in expected_input_datasets]
elif path_defect == "up":
path = _add_up(expected_path)
input_datasets = [_add_up(s) for s in expected_input_datasets]
else:
path = expected_path
input_datasets = expected_input_datasets
client, messages = icat_metadata_client
client.send_metadata(
proposal="hg123",
beamline="id00",
dataset="datasetname",
path=path,
metadata={
"Sample_name": "samplename",
"InstrumentBeam_distance": 0.01,
"InstrumentBeam_incident_energy": [10, 10.1, 10.2],
"input_datasets": input_datasets,
},
)
message = messages.get(timeout=10)
assert messages.empty()
expected_path = str(expected_path)
input_datasets = ",".join([str(s) for s in expected_input_datasets])
expected = {
"complete": True,
"instrument": "id00",
"investigation": "hg123",
"location": str(expected_path),
"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"},
{"name": "input_datasets", "value": input_datasets},
],
"sample": {"name": "samplename"},
}
compare.assert_equal_dataset_message(message, expected)
[docs]
def test_send_metadata_normpath(icat_metadata_client):
client, messages = icat_metadata_client
if sys.platform == "win32":
path = "X:\\visitor\\es1755\\id27\\20251106\\RAW_DATA\\dummy\\..\\collection\\collection_0001\\"
normpath = (
"X:\\visitor\\es1755\\id27\\20251106\\RAW_DATA\\collection\\collection_0001"
)
else:
path = "/mnt/multipath-shares/data/visitor/es1755/id27/20251106/RAW_DATA/dummy/../collection/collection_0001/"
normpath = (
"/data/visitor/es1755/id27/20251106/RAW_DATA/collection/collection_0001"
)
metadata = {
"Sample_name": "sample",
}
client.send_metadata(
proposal="hg123",
beamline="id00",
dataset="0001",
path=path,
metadata=metadata,
)
message = messages.get(timeout=10)
assert messages.empty()
expected = {
"complete": True,
"instrument": "id00",
"investigation": "hg123",
"location": normpath,
"name": "0001",
"parameter": [
{"name": "complete", "value": "true"},
{"name": "beamlineID", "value": "id00"},
{"name": "proposal", "value": "hg123"},
{"name": "location", "value": normpath},
{"name": "Sample_name", "value": "sample"},
{"name": "datasetName", "value": "0001"},
],
"sample": {
"name": "sample",
},
}
compare.assert_equal_dataset_message(message, expected)
[docs]
def test_send_metadata_relative_path_warning(icat_metadata_client):
client, messages = icat_metadata_client
if sys.platform == "win32":
path = "relative\\path\\dataset"
escaped_path = re.escape(path)
else:
path = "relative/path/dataset"
escaped_path = path
kwargs = dict(
proposal="hg123",
beamline="id00",
dataset="0001",
path=path,
metadata={
"Sample_name": "samplename",
},
)
with pytest.raises(
errors.IcatMetadataValidationError,
match=f"Path must be absolute: {escaped_path}",
):
client.send_metadata(strict=True, **kwargs)
with pytest.warns(
errors.IcatMetadataValidationWarning,
match=f"Path must be absolute: {escaped_path}",
):
client.send_metadata(**kwargs)
message = messages.get(timeout=10)
assert messages.empty()
expected = {
"complete": True,
"instrument": "id00",
"investigation": "hg123",
"location": path,
"name": "0001",
"parameter": [
{"name": "complete", "value": "true"},
{"name": "beamlineID", "value": "id00"},
{"name": "proposal", "value": "hg123"},
{"name": "location", "value": path},
{"name": "Sample_name", "value": "samplename"},
{"name": "datasetName", "value": "0001"},
],
"sample": {
"name": "samplename",
},
}
compare.assert_equal_dataset_message(message, expected)
[docs]
def test_send_metadata_input_datasets_relative_path_warning(icat_metadata_client):
client, messages = icat_metadata_client
if sys.platform == "win32":
path = "relative\\input\\dataset"
escaped_path = re.escape(path)
else:
path = "relative/input/dataset"
escaped_path = path
kwargs = dict(
proposal="hg123",
beamline="id00",
dataset="0001",
path=_dummy_path("dataset"),
metadata={
"Sample_name": "samplename",
"input_datasets": [path],
},
)
with pytest.raises(
errors.IcatMetadataValidationError,
match=f"Path must be absolute: {escaped_path}",
):
client.send_metadata(strict=True, **kwargs)
with pytest.warns(
errors.IcatMetadataValidationWarning,
match=f"Path must be absolute: {escaped_path}",
):
client.send_metadata(**kwargs)
message = messages.get(timeout=10)
assert messages.empty()
expected = {
"complete": True,
"instrument": "id00",
"investigation": "hg123",
"location": str(_dummy_path("dataset")),
"name": "0001",
"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": "0001"},
{"name": "input_datasets", "value": path},
],
"sample": {
"name": "samplename",
},
}
compare.assert_equal_dataset_message(message, expected)
[docs]
def test_send_metadata_definition_warning(icat_metadata_client, esrfet_version):
client, messages = icat_metadata_client
kwargs = dict(
proposal="hg123",
beamline="id00",
dataset="datasetname",
path=_dummy_path("dataset"),
metadata={
"Sample_name": "samplename",
"InstrumentBeam_distance": 0.01,
"InstrumentBeam_incident_energy": [10, 10.1, 10.2],
"machine": "mymachine",
"software": "mysoftware_version",
"definition": "UNKNOWN XRF XAS",
},
)
with pytest.raises(
errors.IcatMetadataValidationError, match="Unknown technique names: UNKNOWN"
):
client.send_metadata(strict=True, **kwargs)
with pytest.warns(
errors.IcatMetadataValidationWarning, match="Unknown technique names: UNKNOWN"
):
client.send_metadata(**kwargs)
message = messages.get(timeout=10)
assert messages.empty()
expected = {
"complete": True,
"instrument": "id00",
"investigation": "hg123",
"location": str(_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": "definition", "value": "XAS XRF UNKNOWN"},
{
"name": "technique_pid",
"value": "https://w3id.org/PaN/ESRFET#XAS https://w3id.org/PaN/ESRFET#XRF",
},
{
"name": "technique_pid_esrfet_version",
"value": esrfet_version,
},
{"name": "InstrumentBeam_distance", "value": "0.01"},
{"name": "InstrumentBeam_incident_energy", "value": "10.0 10.1 10.2"},
{"name": "machine", "value": "mymachine"},
{"name": "software", "value": "mysoftware_version"},
],
"sample": {"name": "samplename"},
}
compare.assert_equal_dataset_message(message, expected)
def _add_trailing_slash(path: Path) -> str:
return str(path / "_")[:-1]
def _add_up(path: Path) -> Path:
return path / "_" / ".."
def _dummy_path(dirname: str) -> Path:
return Path.home() / dirname