115 lines
3.1 KiB
Python
115 lines
3.1 KiB
Python
import pytest
|
|
import asyncio
|
|
import threading
|
|
import time
|
|
from contextlib import contextmanager
|
|
from io import StringIO
|
|
|
|
import colorama
|
|
from slic.utils.pv import PV, use_callback # Import both PV and use_callback
|
|
|
|
# Caproto is used to simulate a real EPICS IOC for testing
|
|
from caproto.server import pvproperty, PVGroup, ioc_arg_parser, run
|
|
|
|
|
|
# IOC definition: a single PV TEST:VAL with units
|
|
class TestIOC(PVGroup):
|
|
val = pvproperty(value=0.0, units='units')
|
|
|
|
def __init__(self, prefix, **kwargs):
|
|
super().__init__(prefix=prefix, **kwargs)
|
|
|
|
|
|
# Fixture to launch IOC in a background thread before tests
|
|
@pytest.fixture(scope="module", autouse=True)
|
|
def run_test_ioc():
|
|
prefix = "TEST:"
|
|
ioc = TestIOC(prefix=prefix)
|
|
|
|
_, run_options = ioc_arg_parser(default_prefix=prefix, desc="Test IOC", args=[])
|
|
|
|
def start_ioc():
|
|
run(ioc.pvdb, **run_options)
|
|
|
|
thread = threading.Thread(target=start_ioc, daemon=True)
|
|
thread.start()
|
|
|
|
time.sleep(2) # Let IOC start
|
|
yield
|
|
|
|
|
|
# Fixture to capture stdout (for tqdm / progress bar)
|
|
@pytest.fixture
|
|
def capture_stdout(monkeypatch):
|
|
buf = StringIO()
|
|
monkeypatch.setattr("sys.stdout", buf)
|
|
return buf
|
|
|
|
|
|
@pytest.mark.parametrize("value, expected_bar, expected_color", [
|
|
(25.0, "██▌ ", colorama.Fore.GREEN),
|
|
(50.0, "█████ ", colorama.Fore.GREEN),
|
|
(75.0, "███████▌ ", colorama.Fore.GREEN),
|
|
(100.0, "██████████", colorama.Fore.GREEN),
|
|
(150.0, ">>>>>>>>>>", colorama.Fore.RED),
|
|
(-50.0, "<<<<<<<<<<", colorama.Fore.RED)
|
|
])
|
|
def test_put_with_progress_and_repr(capture_stdout, value, expected_bar, expected_color):
|
|
# Use the custom PV class
|
|
pv = PV("TEST:VAL")
|
|
|
|
pv.wait_for_connection(timeout=2.0)
|
|
assert pv.connected
|
|
|
|
# Set to initial value
|
|
pv.put(0.0, wait=True)
|
|
|
|
# Wait for update to propagate
|
|
for _ in range(20):
|
|
val = pv.get()
|
|
if val is not None:
|
|
break
|
|
time.sleep(0.1)
|
|
|
|
assert val is not None
|
|
assert val == 0.0
|
|
|
|
# Put new value with progress bar
|
|
pv.put(value, show_progress=True)
|
|
|
|
output = capture_stdout.getvalue()
|
|
|
|
# Verify visual bar and color
|
|
assert f"|{expected_bar}|" in output
|
|
assert expected_color in output
|
|
assert str(value) in output
|
|
|
|
# Verify value updated
|
|
assert pv.get() == pytest.approx(value)
|
|
|
|
# Check custom and original repr
|
|
expected_repr = f'PV "TEST:VAL" at {value} units'
|
|
assert repr(pv) == expected_repr
|
|
assert pv.orig_repr().startswith('<epics.pv.PV')
|
|
|
|
|
|
def test_use_callback_context_manager():
|
|
pv = PV("TEST:VAL")
|
|
pv.put(0.0, wait=True)
|
|
|
|
seen_values = []
|
|
|
|
def callback(value=None, **kwargs):
|
|
seen_values.append(value)
|
|
|
|
initial_count = len(pv._callbacks)
|
|
|
|
# This is the function under test
|
|
with use_callback(pv, callback):
|
|
assert len(pv._callbacks) == initial_count + 1
|
|
pv.put(42.0, wait=True)
|
|
time.sleep(0.2) # Ensure callback has time to trigger
|
|
assert 42.0 in seen_values
|
|
|
|
# After context manager exit, callback should be removed
|
|
assert len(pv._callbacks) == initial_count |