105 lines
2.9 KiB
Python
105 lines
2.9 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 # The custom PV class to test
|
|
|
|
# -------- caproto imports for the test IOC --------
|
|
from caproto.server import pvproperty, PVGroup, run
|
|
|
|
# Create a simple IOC with one PV (TEST:VAL)
|
|
class TestIOC(PVGroup):
|
|
val = pvproperty(value=0.0, units='units')
|
|
|
|
def __init__(self, prefix, **kwargs):
|
|
super().__init__(prefix=prefix, **kwargs)
|
|
|
|
|
|
# Run the IOC in a background thread (shared across module)
|
|
@pytest.fixture(scope="module", autouse=True)
|
|
def run_test_ioc():
|
|
prefix = "TEST:"
|
|
ioc = TestIOC(prefix=prefix)
|
|
|
|
# Define async loop to serve IOC
|
|
def start_ioc():
|
|
asyncio.run(run(ioc.pvdb, startup_hook=None))
|
|
|
|
thread = threading.Thread(target=start_ioc, daemon=True)
|
|
thread.start()
|
|
|
|
# Wait a moment to ensure IOC is ready before test starts
|
|
time.sleep(1.5)
|
|
|
|
# Nothing to yield; test can now run
|
|
yield
|
|
|
|
|
|
# Helper to capture stdout output
|
|
@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):
|
|
# Connect to the test PV using the custom slic PV class
|
|
pv = PV("TEST:VAL")
|
|
|
|
# Set initial value to 0.0
|
|
pv.put(0.0, wait=True)
|
|
assert pv.get() == 0.0
|
|
|
|
# Call put() with progress enabled
|
|
pv.put(value, show_progress=True)
|
|
|
|
# Get printed output from RangeBar
|
|
output = capture_stdout.getvalue()
|
|
|
|
# Check progress bar content and color
|
|
assert f"|{expected_bar}|" in output
|
|
assert expected_color in output
|
|
assert str(value) in output
|
|
|
|
# Verify PV value was correctly updated
|
|
assert pv.get() == pytest.approx(value)
|
|
|
|
# Check 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():
|
|
callback_values = []
|
|
|
|
pv = PV("TEST:VAL")
|
|
pv.put(0.0, wait=True)
|
|
|
|
def cb(value=None, **kwargs):
|
|
callback_values.append(value)
|
|
|
|
# Ensure initially no callback is set
|
|
initial_callbacks = len(pv._callbacks)
|
|
|
|
with pv.use_callback(cb):
|
|
assert len(pv._callbacks) == initial_callbacks + 1
|
|
pv.put(10.0, wait=True)
|
|
assert callback_values[-1] == 10.0
|
|
|
|
# Callback should be removed after context exit
|
|
assert len(pv._callbacks) == initial_callbacks
|