DMSC Integration Testing

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

Test: nexusjsontemplate-beamlime|bifrost|nexus_json_template|

nexus_json_template = PosixPath('bifrost-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='tdct', topic='bifrost_choppers', source='BIFRO-ChpSy1:Chop-PSC-101:00-TS-I'), StreamMod...d_R', 'topic': 'bifrost_choppers', 'dtype': 'double', 'value_units': 'Hz'}}]}, dtype='double', value_units='Hz')), ...)

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='bifrost_motion', source='BIFRO-SpRot:MC-RotZ-01:Mtr.RBV'), StreamModuleKey(module_type='f144', topic='bifrost_motion', source='BIFRO-SpGon:MC-RotX-01:Mtr.RBV'), StreamModuleKey(module_type='f144', topic='bifrost_motion', source='BIFRO-SpGon:MC-RotY-01:Mtr.RBV'), StreamModuleKey(module_type='f144', topic='bifrost_motion', source='BIFRO-DtCar:MC-RotZ-01:Mtr.RBV')])

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

nexus_json_template = PosixPath('bifrost-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='tdct', topic='bifrost_choppers', source='BIFRO-ChpSy1:Chop-PSC-101:00-TS-I'), StreamMod...d_R', 'topic': 'bifrost_choppers', 'dtype': 'double', 'value_units': 'Hz'}}]}, dtype='double', value_units='Hz')), ...)

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='bifrost_motion', source='BIFRO-SpRot:MC-RotZ-01:Mtr.RBV'), StreamModuleKey(module_type='f144', topic='bifrost_motion', source='BIFRO-SpGon:MC-RotX-01:Mtr.RBV'), StreamModuleKey(module_type='f144', topic='bifrost_motion', source='BIFRO-SpGon:MC-RotY-01:Mtr.RBV'), StreamModuleKey(module_type='f144', topic='bifrost_motion', source='BIFRO-DtCar:MC-RotZ-01:Mtr.RBV')])

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

nexus_json_template = PosixPath('bifrost-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='tdct', topic='bifrost_choppers', source='BIFRO-ChpSy1:Chop-PSC-101:00-TS-I'), StreamMod...d_R', 'topic': 'bifrost_choppers', 'dtype': 'double', 'value_units': 'Hz'}}]}, dtype='double', value_units='Hz')), ...)

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='bifrost_motion', source='BIFRO-SpRot:MC-RotZ-01:Mtr.RBV'), StreamModuleKey(module_type='f144', topic='bifrost_motion', source='BIFRO-SpGon:MC-RotX-01:Mtr.RBV'), StreamModuleKey(module_type='f144', topic='bifrost_motion', source='BIFRO-SpGon:MC-RotY-01:Mtr.RBV'), StreamModuleKey(module_type='f144', topic='bifrost_motion', source='BIFRO-DtCar:MC-RotZ-01:Mtr.RBV')])

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

nexus_json_template = PosixPath('bifrost-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='tdct', topic='bifrost_choppers', source='BIFRO-ChpSy1:Chop-PSC-101:00-TS-I'), StreamMod...d_R', 'topic': 'bifrost_choppers', 'dtype': 'double', 'value_units': 'Hz'}}]}, dtype='double', value_units='Hz')), ...)

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='bifrost_motion', source='BIFRO-SpRot:MC-RotZ-01:Mtr.RBV'), StreamModuleKey(module_type='f144', topic='bifrost_motion', source='BIFRO-SpGon:MC-RotX-01:Mtr.RBV'), StreamModuleKey(module_type='f144', topic='bifrost_motion', source='BIFRO-SpGon:MC-RotY-01:Mtr.RBV'), StreamModuleKey(module_type='f144', topic='bifrost_motion', source='BIFRO-DtCar:MC-RotZ-01:Mtr.RBV')])

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

nexus_json_template = PosixPath('bifrost-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='tdct', topic='bifrost_choppers', source='BIFRO-ChpSy1:Chop-PSC-101:00-TS-I'), StreamMod...d_R', 'topic': 'bifrost_choppers', 'dtype': 'double', 'value_units': 'Hz'}}]}, dtype='double', value_units='Hz')), ...)

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='bifrost_motion', source='BIFRO-SpRot:MC-RotZ-01:Mtr.RBV'), StreamModuleKey(module_type='f144', topic='bifrost_motion', source='BIFRO-SpGon:MC-RotX-01:Mtr.RBV'), StreamModuleKey(module_type='f144', topic='bifrost_motion', source='BIFRO-SpGon:MC-RotY-01:Mtr.RBV'), StreamModuleKey(module_type='f144', topic='bifrost_motion', source='BIFRO-DtCar:MC-RotZ-01:Mtr.RBV')])

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