diff --git a/ophyd_devices/utils/static_device_test.py b/ophyd_devices/utils/static_device_test.py index 06ed18d..ec3d2e7 100644 --- a/ophyd_devices/utils/static_device_test.py +++ b/ophyd_devices/utils/static_device_test.py @@ -2,6 +2,7 @@ import argparse import copy import os import traceback +from collections import namedtuple from io import TextIOWrapper from unittest import mock @@ -16,6 +17,8 @@ try: except ImportError: device_manager = None +TestResult = namedtuple("TestResult", ["name", "success", "message"]) + class StaticDeviceAnalysisError(Exception): """Error class for static device analysis""" @@ -26,11 +29,13 @@ class StaticDeviceTest: def __init__( self, - output_file: TextIOWrapper, + output_file: TextIOWrapper | None = None, config_file: str | None = None, config_dict: dict[str, dict] | None = None, ) -> None: """ + Initialize the StaticDeviceTest class. Either output_file or config_dict must be provided. + Config_file will have precedence over config_dict. Args: config(str): path to the config file config_dict(dict): device configuration dictionary. Same formatting as in device_manager @@ -279,9 +284,10 @@ class StaticDeviceTest: text(str): text to print and write """ print(text) - self.file.write(text + "\n") + if self.file is not None: # Write only if no output file is provided + self.file.write(text + "\n") - def run_with_list_output(self, connect: bool = False) -> list[tuple[str, bool, str]]: + def run_with_list_output(self, connect: bool = False) -> list[TestResult]: """ Run the tests and return a list of tuples with the device name, success status, and error message. @@ -315,10 +321,13 @@ class StaticDeviceTest: if return_val == 0: status = True + self.print_and_write(f"{name} is OK") except Exception as e: self.print_and_write(f"ERROR: {name} failed: {e}") finally: - results.append((name, status, "\n".join(print_and_write))) + results.append( + TestResult(name=name, success=status, message="\n".join(print_and_write)) + ) print_and_write.clear() return results diff --git a/tests/test_static_device_test.py b/tests/test_static_device_test.py index 967ab1f..6145f41 100644 --- a/tests/test_static_device_test.py +++ b/tests/test_static_device_test.py @@ -1,12 +1,44 @@ import os import sys +from unittest import mock import bec_lib -from ophyd_devices.utils.static_device_test import launch +from ophyd_devices.utils.static_device_test import StaticDeviceTest, launch def test_static_device_test(): config_path = os.path.join(os.path.dirname(bec_lib.__file__), "configs", "demo_config.yaml") sys.argv = ["", "--config", config_path, "--connect"] launch() + + +def test_static_device_test_with_config_dict(): + """First device is okay, second one is not""" + device_dict = { + "waveform": { + "readoutPriority": "async", + "deviceClass": "ophyd_devices.SimWaveform", + "deviceConfig": { + "waveform_shape": 1000, + "sim_init": { + "model": "GaussianModel", + "params": {"amplitude": 100, "center": 500, "sigma": 50}, + }, + }, + "deviceTags": ["detector"], + "enabled": True, + "readOnly": False, + "softwareTrigger": True, + }, + "wrong": {"this is not corect": 0}, + } + test = StaticDeviceTest(config_dict=device_dict) + ret = test.run_with_list_output(connect=False) + assert len(ret) == 2 + assert ret[0].name == "waveform" + assert ret[0].success is True + assert ret[0].message == "waveform is OK" + assert ret[1].name == "wrong" + assert ret[1].success is False + assert isinstance(ret[1].message, str)