mirror of
https://github.com/bec-project/ophyd_devices.git
synced 2026-02-08 08:08:41 +01:00
refactor(static-decvice-test): add test for run_with_list_output
This commit is contained in:
@@ -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
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user