diff --git a/tests/test_utils_pv.py b/tests/test_utils_pv.py new file mode 100644 index 000000000..4506fff42 --- /dev/null +++ b/tests/test_utils_pv.py @@ -0,0 +1,90 @@ +import pytest +from io import StringIO +import colorama +from slic.utils.pv import PV + +@pytest.fixture +def fake_epics_pv(monkeypatch): + # Simule a minimal PV EPICS for testing + class FakeEPICS: + def __init__(self): + self.value = 0.0 + self.units = "units" + self._callbacks = {} + + def get(self): + return self.value + + def put(self, value, **kwargs): + old = self.value + self.value = value + for cb in self._callbacks.values(): + cb(value, old=old, **kwargs) + + def add_callback(self, callback): + cb_id = id(callback) + self._callbacks[cb_id] = callback + return cb_id + + def remove_callback(self, cb_id): + self._callbacks.pop(cb_id, None) + + # Injecte our fake pv in the PV class + monkeypatch.setattr('epics.PV', FakeEPICS) + return PV("TEST:PV") # Retourn a PV instance + +@pytest.fixture +def capture_stdout(monkeypatch): + buf = StringIO() + monkeypatch.setattr("sys.stdout", buf) + return buf + +def test_progress_sequence(fake_epics_pv, capture_stdout): + pv = fake_epics_pv + cases = [ + (25.0, "██▌ ", colorama.Fore.GREEN), # 25% + (50.0, "█████ ", colorama.Fore.GREEN), # 50% + (75.0, "███████▌ ", colorama.Fore.GREEN), # 75% + (100.0, "██████████", colorama.Fore.GREEN), # 100% + (150.0, ">>>>>>>>>>", colorama.Fore.RED), # Overflow + (-50.0, "<<<<<<<<<<", colorama.Fore.RED) # Underflow + ] + + for value, expected_bar, expected_color in cases: + pv.put(value, show_progress=True) + output = capture_stdout.getvalue() + + # Check the bar + assert f"|{expected_bar}|" in output + assert expected_color in output + + # Check the actuel value + assert str(value) in output + assert pv.get() == value + + # Test origin_repr + assert f"PV TEST:PV at {value} units" == pv.orig_repr() + + capture_stdout.truncate(0) + +def test_use_callback_context_manager(fake_epics_pv): + # Test the use_callback context manager functionality + pv = fake_epics_pv + callback_log = [] + def test_callback(value, **kwargs): + callback_log.append(value) + + # Verify callback is added and removed properly + assert len(pv._pv._callbacks) == 0 # No callbacks initially + + with pv.use_callback(test_callback) as callback_id: + # Callback should be added when entering context + assert len(pv._pv._callbacks) == 1 + assert callback_id in pv._pv._callbacks + + # Trigger callback + pv.put(10.0) + assert callback_log == [10.0] + + # Callback should be removed after context + assert len(pv._pv._callbacks) == 0 \ No newline at end of file