Latest Threat Research:SANDWORM_MODE: Shai-Hulud-Style npm Worm Hijacks CI Workflows and Poisons AI Toolchains.Details
Socket
Book a DemoSign in
Socket

softioc

Package Overview
Dependencies
Maintainers
0
Versions
16
Alerts
File Explorer

Advanced tools

Socket logo

Install Socket

Detect and block malicious and high-risk dependencies

Install

softioc - pypi Package Compare versions

Comparing version
4.3.0
to
4.4.0
+71
tests/test_deviocstats.py
# 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"
@pytest.mark.asyncio
async 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 aioca import caget, purge_channel_caches
try:
# Wait for message that IOC has started
select_and_recv(parent_conn, "R")
# Suppress potential spurious warnings
purge_channel_caches()
cpu_cnt = await caget(device_name + ":CPU_CNT")
assert cpu_cnt == multiprocessing.cpu_count()
ioc_cpu_load = await caget(device_name + ":IOC_CPU_LOAD")
assert ioc_cpu_load == pytest.approx(0, abs=1e-2)
finally:
# Suppress potential spurious warnings
purge_channel_caches()
parent_conn.send("D") # "Done"
process.join(timeout=TIMEOUT)
+1
-1
Metadata-Version: 2.1
Name: softioc
Version: 4.3.0
Version: 4.4.0
Summary: Embed an EPICS IOC in a Python process

@@ -5,0 +5,0 @@ Home-page: https://github.com/dls-controls/pythonSoftIOC

[build-system]
requires = ["setuptools", "wheel", "setuptools_dso>=2.1", "epicscorelibs>=7.0.7.99.0.0"]
requires = ["setuptools", "wheel", "setuptools_dso>=2.1", "epicscorelibs>=7.0.7.99.0.2"]
build-backend = "setuptools.build_meta:__legacy__"

@@ -42,9 +42,11 @@ [metadata]

scipy
aioca >=1.3
aioca >=1.6
dev =
pytest-cov
pytest-flake8
flake8 <5.0.0
sphinx ==4.3.2
sphinx-rtd-theme-github-versions
pytest-asyncio
aioca >=1.3
aioca >=1.6
cothread; sys_platform != "win32"

@@ -51,0 +53,0 @@ p4p

@@ -44,2 +44,3 @@ import os

devIocStats_src = os.path.join("softioc", "iocStats", "devIocStats")
devIocStats_posix = os.path.join(devIocStats_src, "os", "posix")
devIocStats_os = os.path.join(devIocStats_src, "os", get_config_var('OS_CLASS'))

@@ -56,2 +57,17 @@ devIocStats_default = os.path.join(devIocStats_src, "os", "default")

include_dirs = [
epicscorelibs.path.include_path,
devIocStats_src,
devIocStats_os,
devIocStats_default
]
if get_config_var("POSIX"):
# If we're on a POSIX system, insert the POSIX folder into the list after
# the os-specific one so that os-specific header files are used first.
include_dirs.insert(
include_dirs.index(devIocStats_os) + 1,
devIocStats_posix
)
# Extension with all our C code

@@ -61,11 +77,4 @@ ext = Extension(

sources = sources,
include_dirs=[
epicscorelibs.path.include_path,
devIocStats_src, devIocStats_os, devIocStats_default
],
include_dirs = include_dirs,
dsos = [
'epicscorelibs.lib.qsrv',
'epicscorelibs.lib.pvAccessIOC',
'epicscorelibs.lib.pvAccess',
'epicscorelibs.lib.pvData',
'epicscorelibs.lib.dbRecStd',

@@ -101,2 +110,3 @@ 'epicscorelibs.lib.dbCore',

epicscorelibs.version.abi_requires(),
"pvxslibs>=1.2.2",
"numpy",

@@ -103,0 +113,0 @@ "epicsdbbuilder>=1.4"

Metadata-Version: 2.1
Name: softioc
Version: 4.3.0
Version: 4.4.0
Summary: Embed an EPICS IOC in a Python process

@@ -5,0 +5,0 @@ Home-page: https://github.com/dls-controls/pythonSoftIOC

@@ -1,2 +0,3 @@

epicscorelibs<7.0.7.99.1,>=7.0.7.99.0.0
epicscorelibs<7.0.7.99.1,>=7.0.7.99.0.2
pvxslibs>=1.2.2
numpy

@@ -8,5 +9,7 @@ epicsdbbuilder>=1.4

pytest-flake8
flake8<5.0.0
sphinx==4.3.2
sphinx-rtd-theme-github-versions
pytest-asyncio
aioca>=1.3
aioca>=1.6
p4p

@@ -20,2 +23,2 @@

scipy
aioca>=1.3
aioca>=1.6

@@ -84,3 +84,4 @@ LICENSE

tests/test_cothread.py
tests/test_deviocstats.py
tests/test_record_values.py
tests/test_records.py
'''Python soft IOC module.'''
import os
import ctypes
from epicscorelibs import path
from setuptools_dso.runtime import find_dso
import epicscorelibs.path
import pvxslibs.path
from epicscorelibs.ioc import \

@@ -20,7 +23,10 @@ iocshRegisterCommon, registerRecordDeviceDriver, pdbbase

iocshRegisterCommon()
for dbd in ('base.dbd', 'PVAServerRegister.dbd', 'qsrv.dbd'):
dbLoadDatabase(dbd, os.path.join(path.base_path, 'dbd'), None)
base_dbd_path = os.path.join(epicscorelibs.path.base_path, 'dbd')
dbLoadDatabase('base.dbd', base_dbd_path, None)
dbLoadDatabase('pvxsIoc.dbd', pvxslibs.path.dbd_path, None)
iocStats = os.path.join(os.path.dirname(__file__), "iocStats", "devIocStats")
dbLoadDatabase('devIocStats.dbd', iocStats, None)
ctypes.CDLL(find_dso('pvxslibs.lib.pvxsIoc'), ctypes.RTLD_GLOBAL)
os.environ.setdefault('PVXS_QSRV_ENABLE', 'YES')
if registerRecordDeviceDriver(pdbbase):

@@ -27,0 +33,0 @@ raise RuntimeError('Error registering')

@@ -11,4 +11,4 @@ # Compute a version number from a git repo or archive

# These will be filled in if git archive is run or by setup.py cmdclasses
GIT_REFS = 'tag: 4.3.0'
GIT_SHA1 = 'b22b497'
GIT_REFS = 'tag: 4.4.0'
GIT_SHA1 = '07b1168'

@@ -15,0 +15,0 @@ # Git describe gives us sha1, last version-like tag, and commits since then

@@ -264,7 +264,10 @@ import os

fields['_wf_nelm'] = length
fields['_wf_dtype'] = numpy.dtype('uint8')
fields['_wf_dtype'] = numpy.dtype('int8')
fields['NELM'] = length
fields['FTVL'] = 'UCHAR'
fields['FTVL'] = 'CHAR'
def qform_string(rec):
rec.add_info("Q:form", "String")
return rec

@@ -274,7 +277,7 @@ def longStringIn(name, **fields):

_set_in_defaults(fields)
return PythonDevice.long_stringin(name, **fields)
return qform_string(PythonDevice.long_stringin(name, **fields))
def longStringOut(name, **fields):
_long_string(fields)
return PythonDevice.long_stringout(name, **fields)
return qform_string(PythonDevice.long_stringout(name, **fields))

@@ -281,0 +284,0 @@

@@ -83,6 +83,5 @@ '''Device support files for pythonSoftIoc Python EPICS device support.

def __str__(self):
return str(self.__builder)
return self.__device._name
class PythonDevice(object):

@@ -89,0 +88,0 @@ DbdFileList = ['device']

@@ -55,2 +55,8 @@ import asyncio

assert await ctx.get(pre + ":AI") == 23.45
long_pv = pre + ":AVERYLONGRECORDSUFFIXTOMAKELONGPV"
long_str = "A long string that is more than 40 characters long"
assert await ctx.get(long_pv) == long_str
assert await ctx.get(long_pv + ".NAME") == long_pv
await ctx.put(pre + ":LONGSTRING", long_str)
assert await ctx.get(pre + ":LONGSTRING") == long_str

@@ -57,0 +63,0 @@ conn.send("D") # "Done"

@@ -292,3 +292,61 @@ import asyncio

def str_ioc_test_func(device_name, conn):
builder.SetDeviceName(device_name)
record_name = "MyAI"
full_name = device_name + ":" + record_name
ai = builder.aIn(record_name)
log("CHILD: Created ai record: ", ai.__str__())
assert (
ai.__str__() == full_name
), f"Record name {ai.__str__()} before LoadDatabase did not match " \
f"expected string {full_name}"
builder.LoadDatabase()
assert (
ai.__str__() == full_name
), f"Record name {ai.__str__()} after LoadDatabase did not match " \
f"expected string {full_name}"
dispatcher = asyncio_dispatcher.AsyncioDispatcher()
softioc.iocInit(dispatcher)
assert (
ai.__str__() == full_name
), f"Record name {ai.__str__()} after iocInit did not match expected " \
f"string {full_name}"
conn.send("R") # "Ready"
log("CHILD: Sent R over Connection to Parent")
def test_record_wrapper_str():
"""Test that the __str__ method on the RecordWrapper behaves as expected,
both before and after loading the record database"""
ctx = get_multiprocessing_context()
parent_conn, child_conn = ctx.Pipe()
device_name = create_random_prefix()
ioc_process = ctx.Process(
target=str_ioc_test_func,
args=(device_name, child_conn),
)
ioc_process.start()
# Wait for message that IOC has started
# If we never receive R it probably means an assert failed
select_and_recv(parent_conn, "R")
def validate_fixture_names(params):

@@ -295,0 +353,0 @@ """Provide nice names for the out_records fixture in TestValidate class"""