Skip to content

Commit 7783c7f

Browse files
OCoppingOCopping
committed
[WIP] Added conftest.py to setup subprocess for IOC to run in to allow tests to run with same IOC (wrapper) object, along with IOC startup script to be called with this; Renamed test_pfc_wrapper to test_pfc_ioc; Added test autosave file
1 parent 84b611f commit 7783c7f

File tree

5 files changed

+368
-39
lines changed

5 files changed

+368
-39
lines changed

‎tests/conftest.py‎

Lines changed: 140 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,140 @@
1+
import atexit
2+
import os
3+
import random
4+
import string
5+
import subprocess
6+
import sys
7+
from datetime import datetime
8+
9+
import pytest
10+
from epicsdbbuilder import ResetRecords
11+
12+
# Must import softioc before epicsdbbuilder
13+
from softioc.device_core import RecordLookup
14+
15+
requires_cothread = pytest.mark.skipif(
16+
sys.platform.startswith("win"), reason="Cothread doesn't work on windows"
17+
)
18+
19+
# Default length used to initialise Waveform and longString records.
20+
# Length picked to match string record length, so we can re-use test strings.
21+
WAVEFORM_LENGTH = 40
22+
23+
# Default timeout for many operations across testing
24+
TIMEOUT = 10 # Seconds
25+
26+
# Address for multiprocessing Listener/Client pair
27+
ADDRESS = ("localhost", 2345)
28+
29+
30+
def create_random_prefix():
31+
"""Create 12-character random string, for generating unique Device Names"""
32+
return "".join(random.choice(string.ascii_uppercase) for _ in range(12))
33+
34+
35+
# Can't use logging as it's not multiprocess safe, and
36+
# alteratives are overkill
37+
def log(*args):
38+
print(datetime.now().strftime("%H:%M:%S"), *args)
39+
40+
41+
class SubprocessIOC:
42+
def __init__(self, ioc_py):
43+
self.pv_prefix = create_random_prefix()
44+
sim_ioc = os.path.join(os.path.dirname(__file__), ioc_py)
45+
cmd = [sys.executable, sim_ioc, self.pv_prefix]
46+
self.proc = subprocess.Popen(
47+
cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE
48+
)
49+
50+
def kill(self):
51+
if self.proc.returncode is None:
52+
# still running, kill it and print the output
53+
self.proc.kill()
54+
out, err = self.proc.communicate(timeout=TIMEOUT)
55+
print(out.decode())
56+
print(err.decode())
57+
58+
59+
def aioca_cleanup():
60+
from aioca import _catools, purge_channel_caches
61+
62+
# Unregister the aioca atexit handler as it conflicts with the one installed
63+
# by cothread. If we don't do this we get a seg fault. This is not a problem
64+
# in production as we won't mix aioca and cothread, but we do mix them in
65+
# the tests so need to do this.
66+
atexit.unregister(_catools._catools_atexit)
67+
# purge the channels before the event loop goes
68+
purge_channel_caches()
69+
70+
71+
@pytest.fixture
72+
def pfc_ioc():
73+
ioc = SubprocessIOC("test_pfc_wrapper.py")
74+
yield ioc
75+
ioc.kill()
76+
aioca_cleanup()
77+
78+
79+
# @pytest.fixture
80+
# def asyncio_ioc_override():
81+
# ioc = SubprocessIOC("sim_asyncio_ioc_override.py")
82+
# yield ioc
83+
# ioc.kill()
84+
# aioca_cleanup()
85+
86+
87+
def _clear_records():
88+
# Remove any records created at epicsdbbuilder layer
89+
ResetRecords()
90+
# And at pythonSoftIoc level
91+
# TODO: Remove this hack and use use whatever comes out of
92+
# https://github.com/dls-controls/pythonSoftIOC/issues/56
93+
RecordLookup._RecordDirectory.clear()
94+
95+
96+
@pytest.fixture(autouse=True)
97+
def clear_records():
98+
"""Deletes all records before and after every test"""
99+
_clear_records()
100+
yield
101+
_clear_records()
102+
103+
104+
@pytest.fixture(autouse=True)
105+
def enable_code_coverage():
106+
"""Ensure code coverage works as expected for `multiprocesses` tests.
107+
As its harmless for other types of test, we always run this fixture."""
108+
try:
109+
from pytest_cov.embed import cleanup_on_sigterm
110+
except ImportError:
111+
pass
112+
else:
113+
cleanup_on_sigterm()
114+
115+
116+
def select_and_recv(conn, expected_char=None):
117+
"""Wait for the given Connection to have data to receive, and return it.
118+
If a character is provided check its correct before returning it."""
119+
# Must use cothread's select if cothread is present, otherwise we'd block
120+
# processing on all cothread processing. But we don't want to use it
121+
# unless we have to, as importing cothread can cause issues with forking.
122+
if "cothread" in sys.modules:
123+
from cothread import select
124+
125+
rrdy, _, _ = select([conn], [], [], TIMEOUT)
126+
else:
127+
# Would use select.select(), but Windows doesn't accept Pipe handles
128+
# as selectable objects.
129+
if conn.poll(TIMEOUT):
130+
rrdy = True
131+
132+
if rrdy:
133+
val = conn.recv()
134+
else:
135+
pytest.fail("Did not receive expected char before TIMEOUT expired")
136+
137+
if expected_char:
138+
assert val == expected_char, "Expected character did not match"
139+
140+
return val

‎tests/test_asyncio_pfc_wrapper.py‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
import os
2+
from pathlib import Path
3+
4+
# require('pygelf')
5+
# Import the basic framework components.
6+
from softioc import asyncio_dispatcher, builder, softioc
7+
8+
from pmacfiltercontrol.pmacFilterControlWrapper import (
9+
Wrapper as pmacFilterControlWrapper,
10+
)
11+
12+
# A couple of identification PVs, assumes this file is the name of the IOC
13+
device_name = "pytest_pfcw"
14+
builder.SetDeviceName(device_name)
15+
builder.stringIn("WHOAMI", initial_value="Test Fast Attenutator Control")
16+
builder.stringIn("HOSTNAME", VAL=os.uname()[1])
17+
18+
filter_set_total = 2
19+
filters_per_set = 2
20+
21+
autosave_pos_file = f"{Path.cwd()}/tests/test_autosave.txt"
22+
23+
dispatcher = asyncio_dispatcher.AsyncioDispatcher()
24+
25+
wrapper = pmacFilterControlWrapper(
26+
"127.0.0.1",
27+
9000,
28+
9001,
29+
builder=builder,
30+
device_name=device_name,
31+
filter_set_total=filter_set_total,
32+
filters_per_set=filters_per_set,
33+
detector="BLXXI-EA-EXCBR-01",
34+
motors="BLXXI-OP-FILT-01",
35+
autosave_file_path=autosave_pos_file,
36+
hdf_file_path=f"{Path.cwd()}/tests/",
37+
)
38+
39+
# dispatcher(wrapper.run_forever)
40+
41+
# setup_logging(default_level=logging.DEBUG)
42+
43+
# Now get the IOC started
44+
builder.LoadDatabase()
45+
softioc.iocInit(dispatcher)
46+
47+
# wrapper.set_device_info()
48+
49+
# Leave the iocsh running
50+
softioc.interactive_ioc(globals())

‎tests/test_autosave.txt‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
pytest_pfcw:FILTER_SET:1:IN:1 100.0
2+
pytest_pfcw:FILTER_SET:1:IN:2 100.0
3+
pytest_pfcw:FILTER_SET:2:IN:1 100.0
4+
pytest_pfcw:FILTER_SET:2:IN:2 100.0
5+
pytest_pfcw:FILTER_SET:1:OUT:1 0.0
6+
pytest_pfcw:FILTER_SET:1:OUT:2 0.0
7+
pytest_pfcw:FILTER_SET:2:OUT:1 0.0
8+
pytest_pfcw:FILTER_SET:2:OUT:2 0.0
9+
pytest_pfcw:SHUTTER:OPEN 0.0
10+
pytest_pfcw:SHUTTER:CLOSED 500.0
11+
pytest_pfcw:HIGH:THRESHOLD:EXTREME 10.0
12+
pytest_pfcw:HIGH:THRESHOLD:UPPER 5.0
13+
pytest_pfcw:HIGH:THRESHOLD:LOWER 2.0
14+
pytest_pfcw:LOW:THRESHOLD:UPPER 2.0
15+
pytest_pfcw:LOW:THRESHOLD:LOWER 5.0
16+
High3 10000.0
17+
High2 500.0
18+
High1 300.0
19+
Low1 500.0
20+
Low2 10000.0
21+
pytest_pfcw:FILTER_SET 1

‎tests/test_pfc_ioc.py‎

Lines changed: 157 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,157 @@
1+
import asyncio
2+
3+
# from importlib import reload
4+
# from pathlib import Path
5+
from typing import Dict
6+
7+
import pytest
8+
from mock import Mock
9+
from mock.mock import create_autospec
10+
11+
from pmacfiltercontrol.hdfadapter import HDFAdapter
12+
13+
# from pmacfiltercontrol.pmacFilterControlWrapper import Wrapper
14+
from pmacfiltercontrol.zmqadapter import ZeroMQAdapter
15+
16+
# -------------------------------------------------
17+
18+
19+
# https://www.roguelynn.com/words/asyncio-testing/
20+
# used to patch/mock asyncio coroutines
21+
@pytest.fixture
22+
def create_mock_coro(mocker, monkeypatch):
23+
def _create_mock_patch_coro(to_patch=None):
24+
mock = mocker.Mock()
25+
26+
async def _coro(*args, **kwargs):
27+
return mock(*args, **kwargs)
28+
29+
if to_patch: # <-- may not need/want to patch anything
30+
monkeypatch.setattr(to_patch, _coro)
31+
return mock, _coro
32+
33+
return _create_mock_patch_coro
34+
35+
36+
# -------------------------------------------------
37+
# Mock Socket
38+
39+
40+
@pytest.fixture
41+
def mock_run_coroutine_threadsafe(mocker, monkeypatch):
42+
_run_coroutine_threadsafe = mocker.Mock()
43+
monkeypatch.setattr(asyncio, "run_coroutine_threadsafe", _run_coroutine_threadsafe)
44+
return _run_coroutine_threadsafe.return_value
45+
46+
47+
# -------------------------------------------------
48+
# Objects
49+
50+
51+
@pytest.fixture
52+
def mock_zmq_adapter() -> Mock:
53+
return create_autospec(ZeroMQAdapter)
54+
55+
56+
@pytest.fixture
57+
def mock_hdf_adapter() -> Mock:
58+
return create_autospec(HDFAdapter)
59+
60+
61+
# @pytest.fixture
62+
# def builder_fixture():
63+
# from softioc import builder, pythonSoftIoc
64+
65+
# reload(pythonSoftIoc)
66+
# # reload(device)
67+
# builder = reload(builder)
68+
# return builder
69+
70+
71+
# @pytest.fixture
72+
# def pfc_ioc(builder_fixture) -> Wrapper:
73+
# test_wrapper = Wrapper(
74+
# "127.0.0.1",
75+
# 9000,
76+
# 9001,
77+
# builder=builder_fixture,
78+
# device_name="pytest_pfcw",
79+
# filter_set_total=2,
80+
# filters_per_set=2,
81+
# detector="BLXXI-TEST-EXCBR-01",
82+
# motors="BLXXI-TEST-FILT-01",
83+
# autosave_file_path=f"{Path.cwd()}/tests/test_autosave.txt",
84+
# hdf_file_path=f"{Path.cwd()}/tests/",
85+
# )
86+
# return test_wrapper
87+
88+
89+
# -------------------------------------------------
90+
91+
92+
# def test_pfc_ioc_constructor(builder_=builder):
93+
# # Don't want these to be called so made into Mocks
94+
# Wrapper._generate_filter_pos_records = Mock()
95+
# Wrapper._generate_shutter_records = Mock()
96+
# Wrapper._generate_pixel_threshold_records = Mock()
97+
98+
# test_w = Wrapper(
99+
# "127.0.0.1",
100+
# 9998,
101+
# 9999,
102+
# builder=builder_,
103+
# device_name="test_test",
104+
# filter_set_total=2,
105+
# filters_per_set=2,
106+
# detector="BLXXI-TEST-EXCBR-01",
107+
# motors="BLXXI-TEST-FILT-01",
108+
# autosave_file_path=f"{Path.cwd()}/tests/test_autosave.txt",
109+
# hdf_file_path=f"{Path.cwd()}/tests/",
110+
# )
111+
112+
# assert test_w.ip == "127.0.0.1"
113+
# assert test_w.timeout.get() == 3
114+
115+
116+
@pytest.mark.asyncio
117+
async def test_pfc_ioc_send_initial_config(pfc_ioc, mock_run_coroutine_threadsafe):
118+
pfc_ioc.connected = True
119+
pfc_ioc._autosave_dict = {"pytest_pfcw:FILTER_SET": 1}
120+
121+
pfc_ioc._configure_param = Mock()
122+
pfc_ioc._setup_hist_thresholds = Mock()
123+
pfc_ioc.shutter_pos_closed.get = Mock()
124+
pfc_ioc._set_filter_set = Mock()
125+
126+
await pfc_ioc._send_initial_config()
127+
128+
assert pfc_ioc.attenuation.get() == 15
129+
130+
131+
def test_pfc_ioc_get_autosave(pfc_ioc):
132+
133+
autosave_dict: Dict[str, float] = pfc_ioc._get_autosave()
134+
135+
assert autosave_dict == {
136+
"pytest_pfcw:FILTER_SET:1:IN:1": 100.0,
137+
"pytest_pfcw:FILTER_SET:1:IN:2": 100.0,
138+
"pytest_pfcw:FILTER_SET:2:IN:1": 100.0,
139+
"pytest_pfcw:FILTER_SET:2:IN:2": 100.0,
140+
"pytest_pfcw:FILTER_SET:1:OUT:1": 0.0,
141+
"pytest_pfcw:FILTER_SET:1:OUT:2": 0.0,
142+
"pytest_pfcw:FILTER_SET:2:OUT:1": 0.0,
143+
"pytest_pfcw:FILTER_SET:2:OUT:2": 0.0,
144+
"pytest_pfcw:SHUTTER:OPEN": 0.0,
145+
"pytest_pfcw:SHUTTER:CLOSED": 500.0,
146+
"pytest_pfcw:HIGH:THRESHOLD:EXTREME": 10.0,
147+
"pytest_pfcw:HIGH:THRESHOLD:UPPER": 5.0,
148+
"pytest_pfcw:HIGH:THRESHOLD:LOWER": 2.0,
149+
"pytest_pfcw:LOW:THRESHOLD:UPPER": 2.0,
150+
"pytest_pfcw:LOW:THRESHOLD:LOWER": 5.0,
151+
"High3": 10000.0,
152+
"High2": 500.0,
153+
"High1": 300.0,
154+
"Low1": 500.0,
155+
"Low2": 10000.0,
156+
"pytest_pfcw:FILTER_SET": 1,
157+
}

0 commit comments

Comments
 (0)