Benjamin Labrecque
106 lines
2.4 KiB
Python
106 lines
2.4 KiB
Python
from unittest.mock import patch
|
|
|
|
from agebd.service.pvs import BasePVs
|
|
from agebd.runner import CallbackRunner, PeriodicRunner
|
|
from agebd.service.base import BaseService
|
|
|
|
|
|
class DummyPV:
|
|
def __init__(self, name):
|
|
self.pvname = name
|
|
self.value = None
|
|
|
|
def put(self, value):
|
|
self.value = value
|
|
|
|
def get(self):
|
|
return self.value
|
|
|
|
|
|
class DummyAbortPV:
|
|
def __init__(self, name):
|
|
self.pvname = name
|
|
self.i = 0
|
|
|
|
def put(self, value):
|
|
self.value = value
|
|
|
|
def get(self):
|
|
"""
|
|
Return False on first call to run code once, then return False
|
|
"""
|
|
if self.i == 0:
|
|
self.i += 1
|
|
return False
|
|
return True
|
|
|
|
|
|
class DummyCallbackPV:
|
|
def __init__(self, name):
|
|
self.pvname = name
|
|
self.value = None
|
|
|
|
self._added_callback = False
|
|
self._removed_callback = False
|
|
|
|
def put(self, value):
|
|
self.value = value
|
|
|
|
def get(self):
|
|
return self.value
|
|
|
|
def add_callback(self, *args, **kwargs):
|
|
self._added_callback = True
|
|
|
|
def remove_callback(self, *args, **kwargs):
|
|
self._removed_callback = True
|
|
|
|
|
|
class FakePVs(BasePVs):
|
|
CallbackPV = DummyCallbackPV("DummyCallbackPV")
|
|
|
|
def __init__(self, service_name: str = "test-service", pv_factory=DummyPV):
|
|
super().__init__(service_name, pv_factory)
|
|
self.abort = DummyAbortPV("DummyAbortPV")
|
|
self.onoff.value = 1
|
|
|
|
|
|
class FakeService(BaseService):
|
|
def __init__(self, name: str, pvs: FakePVs, version: str = "test-version"):
|
|
super().__init__(name, pvs, version)
|
|
|
|
def update(self):
|
|
pass
|
|
|
|
|
|
def test_callbackrunner():
|
|
version = "version-0"
|
|
pvs = FakePVs()
|
|
svc_name = "test-service"
|
|
svc = FakeService(svc_name, pvs, version)
|
|
|
|
runner = CallbackRunner(svc)
|
|
|
|
# Mock the pyepics poll function so it doesn't try to look at network sockets
|
|
with patch("agebd.runner.poll"):
|
|
runner.start()
|
|
|
|
assert pvs.CallbackPV._added_callback
|
|
assert pvs.CallbackPV._removed_callback
|
|
assert pvs.status.get() == "killed"
|
|
|
|
|
|
def test_periodicrunner():
|
|
version = "version-0"
|
|
pvs = FakePVs()
|
|
svc_name = "test-service"
|
|
svc = FakeService(svc_name, pvs, version)
|
|
|
|
runner = PeriodicRunner(svc)
|
|
|
|
# Mock the pyepics sleep function so it doesn't try to sleep during tests
|
|
with patch("agebd.runner.sleep"):
|
|
runner.start()
|
|
|
|
assert pvs.status.get() == "killed"
|