51 lines
2.1 KiB
Python
51 lines
2.1 KiB
Python
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'python'))
|
|
from epics_source import IocSource
|
|
|
|
|
|
class FakePV:
|
|
def __init__(self, name, auto_monitor, callback, connection_callback):
|
|
self.name, self.callback, self.connection = name, callback, connection_callback
|
|
assert auto_monitor
|
|
def send(self, value): self.callback(value=value)
|
|
def connect(self): self.connection(conn=True)
|
|
def disconnect(self): self.connection(conn=False)
|
|
|
|
|
|
class SourceTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.source = IocSource('HEX:', FakePV)
|
|
def ready(self):
|
|
for key, pv in self.source.pvs.items():
|
|
pv.connect()
|
|
pv.send([1, 2, 3, 4, 5, 6] if key in IocSource.ARRAYS else 0)
|
|
def test_mapping_and_updates(self):
|
|
self.ready()
|
|
self.assertEqual(self.source.pvs['pose_actual'].name, 'HEX:hex-pose-actual')
|
|
self.assertEqual(self.source.pvs['leg_set'].name, 'HEX:hex-leg-set')
|
|
old = self.source.snapshot()
|
|
self.source.pvs['pose_actual'].send([6]*6)
|
|
self.assertEqual(self.source.snapshot()['pose_actual'], (6,)*6)
|
|
self.assertEqual(old['pose_actual'], (1,2,3,4,5,6))
|
|
def test_disconnect_and_reconnect_waits_for_data(self):
|
|
self.ready()
|
|
pv = self.source.pvs['leg_actual']
|
|
pv.disconnect()
|
|
with self.assertRaisesRegex(RuntimeError, 'disconnected'): self.source.snapshot()
|
|
pv.connect()
|
|
with self.assertRaisesRegex(RuntimeError, 'Waiting'): self.source.snapshot()
|
|
pv.send([0]*6)
|
|
self.assertEqual(self.source.snapshot()['leg_actual'], (0,)*6)
|
|
def test_bad_array_and_recovery(self):
|
|
self.ready()
|
|
for bad in ([1]*5, [float('nan')]*6, None):
|
|
self.source.pvs['pose_set'].send(bad)
|
|
with self.assertRaisesRegex(RuntimeError, 'hex-pose-set'): self.source.snapshot()
|
|
self.source.pvs['pose_set'].send([0]*6)
|
|
self.source.pvs['fault'].send(2)
|
|
self.assertEqual(self.source.snapshot()['fault'], 2)
|
|
|
|
if __name__ == '__main__': unittest.main()
|