-
Notifications
You must be signed in to change notification settings - Fork 9
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Fix devIocStats by including posix headers
This relies on the config_var "POSIX" being defined as True/False for the current platform. The test is pretty simple, just the CPU count and IOC CPU load as they're fairly easy to calculate
- Loading branch information
1 parent
289897e
commit ec2c1df
Showing
2 changed files
with
87 additions
and
4 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,70 @@ | ||
# File for tests related to devIocStats support module, which at time of writing | ||
# is built alongside PythonSoftIOC and optionally turned on at runtime | ||
|
||
import multiprocessing | ||
import pytest | ||
|
||
from conftest import ( | ||
create_random_prefix, | ||
TIMEOUT, | ||
select_and_recv, | ||
get_multiprocessing_context | ||
) | ||
|
||
from softioc import asyncio_dispatcher, builder, softioc | ||
|
||
def deviocstats_test_func( | ||
device_name, | ||
child_conn): | ||
"""Start the IOC with the specified validate method""" | ||
|
||
builder.SetDeviceName(device_name) | ||
|
||
dispatcher = asyncio_dispatcher.AsyncioDispatcher() | ||
builder.LoadDatabase() | ||
softioc.devIocStats(device_name) | ||
softioc.iocInit(dispatcher) | ||
|
||
child_conn.send("R") | ||
|
||
# Keep process alive while main thread runs CAGET | ||
if child_conn.poll(TIMEOUT): | ||
val = child_conn.recv() | ||
assert val == "D", "Did not receive expected Done character" | ||
|
||
def test_deviocstats(): | ||
|
||
ctx = get_multiprocessing_context() | ||
|
||
parent_conn, child_conn = ctx.Pipe() | ||
|
||
device_name = create_random_prefix() | ||
|
||
process = ctx.Process( | ||
target=deviocstats_test_func, | ||
args=(device_name, child_conn), | ||
) | ||
|
||
process.start() | ||
|
||
from cothread.catools import caget, _channel_cache | ||
|
||
try: | ||
# Wait for message that IOC has started | ||
select_and_recv(parent_conn, "R") | ||
|
||
# Suppress potential spurious warnings | ||
_channel_cache.purge() | ||
|
||
cpu_cnt = caget(device_name + ":CPU_CNT") | ||
assert cpu_cnt == multiprocessing.cpu_count() | ||
|
||
ioc_cpu_load = caget(device_name + ":IOC_CPU_LOAD") | ||
assert ioc_cpu_load == pytest.approx(0, abs=1e-2) | ||
|
||
|
||
finally: | ||
# Suppress potential spurious warnings | ||
_channel_cache.purge() | ||
parent_conn.send("D") # "Done" | ||
process.join(timeout=TIMEOUT) |