Source code for pyicat_plus.tests.utils.message
import socket
from typing import Optional
from ... import __version__
[docs]
def check_parameter_exists(dataset, param_name):
return any(param["name"] == param_name for param in dataset["parameter"])
[docs]
def assert_dataset_message(message: dict, expected: dict):
expected["dataset"]["startDate"] = message["dataset"]["startDate"]
expected["dataset"]["endDate"] = message["dataset"]["endDate"]
expected["dataset"]["parameter"].append(_get_parameter(message, "startDate"))
expected["dataset"]["parameter"].append(_get_parameter(message, "endDate"))
if not check_parameter_exists(expected["dataset"], "machine"):
expected["dataset"]["parameter"].append(
{"name": "machine", "value": socket.getfqdn()}
)
if not check_parameter_exists(expected["dataset"], "software"):
expected["dataset"]["parameter"].append(
{"name": "software", "value": "pyicat-plus_v" + __version__}
)
expected["dataset"]["parameter"] = sorted(
expected["dataset"]["parameter"], key=lambda adict: adict["name"]
)
message["dataset"]["parameter"] = sorted(
message["dataset"]["parameter"], key=lambda adict: adict["name"]
)
assert message == expected
[docs]
def assert_investigation_message(message: dict, expected: dict):
expected["investigation"]["startDate"] = message["investigation"]["startDate"]
assert message == expected
def _get_parameter(root: dict, parameter_name: str) -> Optional[dict]:
for parameter in root["dataset"]["parameter"]:
if parameter["name"] == parameter_name:
return parameter