1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
|
import epics
import epics.ca
import epics.pv
import time
import subprocess
import pvnames
enabledpv = pvnames.clear_cache_enabled
beaconspv = pvnames.clear_cache_beacons
change_count = 0
def onChanges(pvname=None, value=None, **kw):
global change_count
change_count += 1
def run():
for beaconpv in beaconspv:
value = epics.caget(beaconpv)
assert value is not None, \
"PV {} is offline".format(beaconpv)
time.sleep(0.2)
epics.ca.clear_cache()
def test_use_monitor():
# This test verifies that clear_cache clears subscriptions created by
# use_monitor=True prior clearing channels.
# Otherwise, EPICS will crash on an attempt to clear subscription
# on a cleared channel.
assert epics.caget(enabledpv, use_monitor=True) is not None, \
"PV {} is offline".format(enabledpv)
pv = epics.pv.get_pv(enabledpv)
epics.ca.clear_cache()
pv.disconnect()
def test_subscribe():
# This test verifies that clear_cache does not
# leave registered callbacks behind which may cause
# a crash on shutdown on Linux randomly.
# Therefore, the test runs itself
# in a new process multiple times.
# This test takes quite long time to complete.
assert epics.caget(enabledpv) is not None, \
"PV {} is offline".format(enabledpv)
epics.caput(enabledpv, 1)
for run in range(20):
try:
subprocess.run(['python', __file__], check=True)
except subprocess.CalledProcessError as err:
epics.caput(enabledpv, 0)
if err.returncode > 0:
error = 'the return code {}'.format(err.returncode)
else:
error = 'the signal {}'.format(-err.returncode)
assert False, "Run {} failed due to {}".format(
run,
error)
epics.caput(enabledpv, 0)
if (__name__ == '__main__'):
run()
|
