DMSC Integration Testing

Last updated: September 15, 2025 23:36:07

Test: nexusjsontemplate-beamlime|loki|nexus_json_template|

nexus_json_template = PosixPath('loki-dynamic.json')

def test_nexus_json_template(nexus_json_template: pathlib.Path) -> None:
downloaded = json.loads(nexus_json_template.read_text())
assert "children" in downloaded # Check if the file was downloaded correctly
# collect_streaming_modules will validate the structure
# that is required by `beamlime`.
# It is currently removed from `beamlime` package.
# We should update the method that is used by `beamlime` once it is added back
# or updated in `beamlime`.
> collect_streaming_modules(downloaded)

tests/nexusjsontemplate-beamlime/nexus_json_template_test.py:201:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
tests/nexusjsontemplate-beamlime/nexus_json_template_test.py:174: in collect_streaming_modules
_validate_module_keys(key_value_pairs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key_value_pairs = ((StreamModuleKey(module_type='f144', topic='loki_motion', source='LOKI-DtCar1:MC-LinX-01:Mtr.RBV'), StreamModuleValue...urce': 'caen'}}], 'attributes': [{'name': 'NX_class', 'values': 'NXevent_data'}]}, dtype=None, value_units=None)), ...)

def _validate_module_keys(
key_value_pairs: tuple[tuple[StreamModuleKey, StreamModuleValue], ...],
) -> None:
"""Validate the module keys."""
from collections import Counter

key_counts = Counter([key for key, _ in key_value_pairs])
duplicated_keys = [key for key, count in key_counts.items() if count > 1]
if duplicated_keys:
> raise InvalidNexusStructureError(
"Duplicate module place holder(s) found in the nexus structure: ",
duplicated_keys,
)
E tests.nexusjsontemplate-beamlime.nexus_json_template_test.InvalidNexusStructureError: ('Duplicate module place holder(s) found in the nexus structure: ', [StreamModuleKey(module_type='f144', topic='loki_motion', source='LOKI-DtCar1:MC-LinX-01:Mtr.RBV')])

tests/nexusjsontemplate-beamlime/nexus_json_template_test.py:106: InvalidNexusStructureError

nexus_json_template = PosixPath('loki-dynamic.json')

def test_nexus_json_template(nexus_json_template: pathlib.Path) -> None:
downloaded = json.loads(nexus_json_template.read_text())
assert "children" in downloaded # Check if the file was downloaded correctly
# collect_streaming_modules will validate the structure
# that is required by `beamlime`.
# It is currently removed from `beamlime` package.
# We should update the method that is used by `beamlime` once it is added back
# or updated in `beamlime`.
> collect_streaming_modules(downloaded)

tests/nexusjsontemplate-beamlime/nexus_json_template_test.py:201:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
tests/nexusjsontemplate-beamlime/nexus_json_template_test.py:174: in collect_streaming_modules
_validate_module_keys(key_value_pairs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key_value_pairs = ((StreamModuleKey(module_type='f144', topic='loki_motion', source='LOKI-DtCar1:MC-LinX-01:Mtr.RBV'), StreamModuleValue...urce': 'caen'}}], 'attributes': [{'name': 'NX_class', 'values': 'NXevent_data'}]}, dtype=None, value_units=None)), ...)

def _validate_module_keys(
key_value_pairs: tuple[tuple[StreamModuleKey, StreamModuleValue], ...],
) -> None:
"""Validate the module keys."""
from collections import Counter

key_counts = Counter([key for key, _ in key_value_pairs])
duplicated_keys = [key for key, count in key_counts.items() if count > 1]
if duplicated_keys:
> raise InvalidNexusStructureError(
"Duplicate module place holder(s) found in the nexus structure: ",
duplicated_keys,
)
E tests.nexusjsontemplate-beamlime.nexus_json_template_test.InvalidNexusStructureError: ('Duplicate module place holder(s) found in the nexus structure: ', [StreamModuleKey(module_type='f144', topic='loki_motion', source='LOKI-DtCar1:MC-LinX-01:Mtr.RBV')])

tests/nexusjsontemplate-beamlime/nexus_json_template_test.py:106: InvalidNexusStructureError

nexus_json_template = PosixPath('loki-dynamic.json')

def test_nexus_json_template(nexus_json_template: pathlib.Path) -> None:
downloaded = json.loads(nexus_json_template.read_text())
assert "children" in downloaded # Check if the file was downloaded correctly
# collect_streaming_modules will validate the structure
# that is required by `beamlime`.
# It is currently removed from `beamlime` package.
# We should update the method that is used by `beamlime` once it is added back
# or updated in `beamlime`.
> collect_streaming_modules(downloaded)

tests/nexusjsontemplate-beamlime/nexus_json_template_test.py:201:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
tests/nexusjsontemplate-beamlime/nexus_json_template_test.py:174: in collect_streaming_modules
_validate_module_keys(key_value_pairs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key_value_pairs = ((StreamModuleKey(module_type='f144', topic='loki_motion', source='LOKI-DtCar1:MC-LinX-01:Mtr.RBV'), StreamModuleValue...urce': 'caen'}}], 'attributes': [{'name': 'NX_class', 'values': 'NXevent_data'}]}, dtype=None, value_units=None)), ...)

def _validate_module_keys(
key_value_pairs: tuple[tuple[StreamModuleKey, StreamModuleValue], ...],
) -> None:
"""Validate the module keys."""
from collections import Counter

key_counts = Counter([key for key, _ in key_value_pairs])
duplicated_keys = [key for key, count in key_counts.items() if count > 1]
if duplicated_keys:
> raise InvalidNexusStructureError(
"Duplicate module place holder(s) found in the nexus structure: ",
duplicated_keys,
)
E tests.nexusjsontemplate-beamlime.nexus_json_template_test.InvalidNexusStructureError: ('Duplicate module place holder(s) found in the nexus structure: ', [StreamModuleKey(module_type='f144', topic='loki_motion', source='LOKI-DtCar1:MC-LinX-01:Mtr.RBV')])

tests/nexusjsontemplate-beamlime/nexus_json_template_test.py:106: InvalidNexusStructureError

nexus_json_template = PosixPath('loki-dynamic.json')

def test_nexus_json_template(nexus_json_template: pathlib.Path) -> None:
downloaded = json.loads(nexus_json_template.read_text())
assert "children" in downloaded # Check if the file was downloaded correctly
# collect_streaming_modules will validate the structure
# that is required by `beamlime`.
# It is currently removed from `beamlime` package.
# We should update the method that is used by `beamlime` once it is added back
# or updated in `beamlime`.
> collect_streaming_modules(downloaded)

tests/nexusjsontemplate-beamlime/nexus_json_template_test.py:201:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
tests/nexusjsontemplate-beamlime/nexus_json_template_test.py:174: in collect_streaming_modules
_validate_module_keys(key_value_pairs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key_value_pairs = ((StreamModuleKey(module_type='f144', topic='loki_motion', source='LOKI-DtCar1:MC-LinX-01:Mtr.RBV'), StreamModuleValue...urce': 'caen'}}], 'attributes': [{'name': 'NX_class', 'values': 'NXevent_data'}]}, dtype=None, value_units=None)), ...)

def _validate_module_keys(
key_value_pairs: tuple[tuple[StreamModuleKey, StreamModuleValue], ...],
) -> None:
"""Validate the module keys."""
from collections import Counter

key_counts = Counter([key for key, _ in key_value_pairs])
duplicated_keys = [key for key, count in key_counts.items() if count > 1]
if duplicated_keys:
> raise InvalidNexusStructureError(
"Duplicate module place holder(s) found in the nexus structure: ",
duplicated_keys,
)
E tests.nexusjsontemplate-beamlime.nexus_json_template_test.InvalidNexusStructureError: ('Duplicate module place holder(s) found in the nexus structure: ', [StreamModuleKey(module_type='f144', topic='loki_motion', source='LOKI-DtCar1:MC-LinX-01:Mtr.RBV')])

tests/nexusjsontemplate-beamlime/nexus_json_template_test.py:106: InvalidNexusStructureError

nexus_json_template = PosixPath('loki-dynamic.json')

def test_nexus_json_template(nexus_json_template: pathlib.Path) -> None:
downloaded = json.loads(nexus_json_template.read_text())
assert "children" in downloaded # Check if the file was downloaded correctly
# collect_streaming_modules will validate the structure
# that is required by `beamlime`.
# It is currently removed from `beamlime` package.
# We should update the method that is used by `beamlime` once it is added back
# or updated in `beamlime`.
> collect_streaming_modules(downloaded)

tests/nexusjsontemplate-beamlime/nexus_json_template_test.py:201:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
tests/nexusjsontemplate-beamlime/nexus_json_template_test.py:174: in collect_streaming_modules
_validate_module_keys(key_value_pairs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

key_value_pairs = ((StreamModuleKey(module_type='f144', topic='loki_motion', source='LOKI-DtCar1:MC-LinX-01:Mtr.RBV'), StreamModuleValue...urce': 'caen'}}], 'attributes': [{'name': 'NX_class', 'values': 'NXevent_data'}]}, dtype=None, value_units=None)), ...)

def _validate_module_keys(
key_value_pairs: tuple[tuple[StreamModuleKey, StreamModuleValue], ...],
) -> None:
"""Validate the module keys."""
from collections import Counter

key_counts = Counter([key for key, _ in key_value_pairs])
duplicated_keys = [key for key, count in key_counts.items() if count > 1]
if duplicated_keys:
> raise InvalidNexusStructureError(
"Duplicate module place holder(s) found in the nexus structure: ",
duplicated_keys,
)
E tests.nexusjsontemplate-beamlime.nexus_json_template_test.InvalidNexusStructureError: ('Duplicate module place holder(s) found in the nexus structure: ', [StreamModuleKey(module_type='f144', topic='loki_motion', source='LOKI-DtCar1:MC-LinX-01:Mtr.RBV')])

tests/nexusjsontemplate-beamlime/nexus_json_template_test.py:106: InvalidNexusStructureError