From 47ffa16aa67c352539885e91247fa8b897bb6c76 Mon Sep 17 00:00:00 2001 From: Peter Heesterman Date: Mon, 21 Sep 2026 14:22:28 +0100 Subject: [PATCH] Removed use of the _rec_cache variable in db.py as cached and then re-used record instances do not work. This usage causes segmentation faults and/or unpredictable behaviour as highlighted in https://github.com/epics-modules/pyDevSup/issues/60 Added tests for use of array record types, although the noted issue has turned out to be generic to all types. --- devsupApp/src/devsup/db.py | 10 +- devsupApp/src/devsup/test/test_db.py | 203 ++++++++++++++++++++++----- 2 files changed, 171 insertions(+), 42 deletions(-) diff --git a/devsupApp/src/devsup/db.py b/devsupApp/src/devsup/db.py index 49ea01f..bf66110 100644 --- a/devsupApp/src/devsup/db.py +++ b/devsupApp/src/devsup/db.py @@ -6,7 +6,6 @@ from . import _dbapi -_rec_cache = {} _no_such_field = object() __all__ = [ @@ -26,12 +25,7 @@ def getRecord(name): >>> R = getRecord("my:record:name") Record("my:record:name") """ - try: - return _rec_cache[name] - except KeyError: - rec = Record(name) - _rec_cache[name] = rec - return rec + return Record(name) class IOScanListBlock(object): """A list of records which will be processed together. @@ -335,7 +329,7 @@ def __getattr__(self, name): try: F = self.field(name) except ValueError: - raise AttributeError('No such field') + raise AttributeError('No such field ' + name) else: return F.getval() diff --git a/devsupApp/src/devsup/test/test_db.py b/devsupApp/src/devsup/test/test_db.py index 07d0fbd..8d89e44 100644 --- a/devsupApp/src/devsup/test/test_db.py +++ b/devsupApp/src/devsup/test/test_db.py @@ -53,10 +53,32 @@ class TestField(IOCHelper): field(FTVL, "DOUBLE") field(NELM, "10") } + record(waveform, "rec:wf:l") { + field(FTVL, "LONG") + field(NELM, "10") + } record(waveform, "rec:wf:s") { field(FTVL, "STRING") field(NELM, "10") } + record(aai, "rec:aai:f") { + field(FTVL, "DOUBLE") + field(NELM, "10") + } + record(aai, "rec:aai:l") { + field(FTVL, "LONG") + field(NELM, "10") + } + record(aao, "src:aao:f") { + field(FTVL, "DOUBLE") + field(OUT, "rec:aai:f PP") + field(NELM, "10") + } + record(aao, "src:aao:l") { + field(FTVL, "LONG") + field(OUT, "rec:aai:l PP") + field(NELM, "10") + } """ def test_ai(self): @@ -91,34 +113,102 @@ def test_si(self): # TODO: test unicode - def test_wf_float(self): - rec = getRecord("rec:wf:a") + def array_float_test(self, src, tgt=None): + with src: + assert_array_almost_equal(src.VAL, []) + if tgt is not None: + assert_array_almost_equal(tgt.VAL, []) + + src.VAL = numpy.arange(5) + assert_array_almost_equal(src.VAL, numpy.arange(5)) + if tgt is not None: + src.scan(sync=True) + assert_array_almost_equal(tgt.VAL, src.VAL) + + src.VAL = numpy.arange(10) + assert_array_almost_equal(src.VAL, numpy.arange(10)) + if tgt is not None: + src.scan(sync=True) + assert_array_almost_equal(tgt.VAL, src.VAL) - with rec: - assert_array_almost_equal(rec.VAL, []) + with self.assertRaises(ValueError): + src.VAL = numpy.arange(15) - rec.VAL = numpy.arange(5) - assert_array_almost_equal(rec.VAL, numpy.arange(5)) + src.VAL = [] + assert_array_almost_equal(src.VAL, []) + if tgt is not None: + src.scan(sync=True) + assert_array_almost_equal(tgt.VAL, src.VAL) - rec.VAL = numpy.arange(10) - assert_array_almost_equal(rec.VAL, numpy.arange(10)) + # in-place modification + fld = src.field('VAL') + fld.putarraylen(5) + arr = fld.getarray() + self.assertEqual(arr.shape, (10,)) # size of NELM + arr[:5] = numpy.arange(5) # we only fill in the part in use + arr[2] = 42.0 + + assert_array_almost_equal(src.VAL, [0.0, 1.0, 42.0, 3.0, 4.0]) + if tgt is not None: + src.scan(sync=True) + assert_array_almost_equal(tgt.VAL, src.VAL) + + def array_long_test(self, src, tgt=None): + with src: + assert_array_equal(src.VAL, []) + if tgt is not None: + assert_array_equal(tgt.VAL, []) + + src.VAL = range(5) + assert_array_equal(src.VAL, range(5)) + if tgt is not None: + src.scan(sync=True) + assert_array_equal(tgt.VAL, src.VAL) + + src.VAL = range(10) + assert_array_equal(src.VAL, range(10)) + if tgt is not None: + src.scan(sync=True) + assert_array_equal(tgt.VAL, src.VAL) with self.assertRaises(ValueError): - rec.VAL = numpy.arange(15) + src.VAL = range(15) - rec.VAL = [] - assert_array_almost_equal(rec.VAL, []) + src.VAL = [] + assert_array_equal(src.VAL, []) + if tgt is not None: + src.scan(sync=True) + assert_array_equal(tgt.VAL, src.VAL) # in-place modification - fld = rec.field('VAL') + fld = src.field('VAL') fld.putarraylen(5) arr = fld.getarray() self.assertEqual(arr.shape, (10,)) # size of NELM - arr[:5] = numpy.arange(5) # we only fill in the part in use + arr[:5] = range(5) # we only fill in the part in use arr[2] = 42 - assert_array_almost_equal(rec.VAL, [0, 1, 42, 3, 4]) + assert_array_equal(src.VAL, [0, 1, 42, 3, 4]) + if tgt is not None: + src.scan(sync=True) + assert_array_equal(tgt.VAL, src.VAL) + + def test_wf_float(self): + rec = getRecord("rec:wf:a") + self.array_float_test(rec) + def test_wf_long(self): + rec = getRecord("rec:wf:l") + self.array_long_test(rec) + + def test_aai_float(self): + rec = getRecord("rec:aai:f") + self.array_float_test(rec) + + def test_aai_long(self): + rec = getRecord("rec:aai:l") + self.array_long_test(rec) + def test_wf_string(self): rec = getRecord("rec:wf:s") @@ -129,6 +219,14 @@ def test_wf_string(self): assert_array_equal(rec.VAL, numpy.asarray(["zero", "", "one", "This is a really long string which shoul", "", "last"], dtype='S40')) + + def test_aao_long(self): + tgt, src = getRecord('rec:aai:l'), getRecord('src:aao:l') + self.array_long_test(src, tgt) + + def test_aao_float(self): + tgt, src = getRecord('rec:aai:f'), getRecord('src:aao:f') + self.array_float_test(src, tgt) class TestDset(IOCHelper): db = """ @@ -262,29 +360,66 @@ def test_calcoutrecord(self): rec.B = 2 self.assertEqual(rec.scan(sync=True), 0) self.assertEqual(rec.VAL, 42) - - - + class TestAlarm(IOCHelper): db = """ - record(longin, "rec:alarm:msg") {} - record(longin, "rec:alarm:plain") {} + record(longin, "rec:inalarm:amsg") { + field(PINI, "YES") + } + record(longin, "rec:inalarm:plain") { + field(PINI, "YES") + } + record(longout, "rec:outalarm:amsg") { + field(PINI, "YES") + field(VAL, "0") + } + record(longout, "rec:outalarm:plain") { + field(PINI, "YES") + field(VAL, "0") + } """ - - def test_set_severity_message(self): - rec = getRecord("rec:alarm:msg") - + def test_setin_severity_message(self): + rec = getRecord("rec:inalarm:amsg") with rec: - rec.setSevr(_dbapi.MAJOR_ALARM, _dbapi.HIHI_ALARM, amsg="Meaningful alarm message") - self.assertEqual(rec.NSEV, _dbapi.MAJOR_ALARM) - self.assertEqual(rec.NSTA, _dbapi.HIHI_ALARM) + self.assertEqual(rec.SEVR, _dbapi.NO_ALARM) + self.assertEqual(rec.STAT, _dbapi.NO_ALARM) if _dbapi.epicsver[:4] >= (7, 0, 6, 0): - self.assertEqual(rec.NAMSG, "Meaningful alarm message") - - def test_set_severity_without_message(self): - rec = getRecord("rec:alarm:plain") - + self.assertEqual(rec.AMSG, "") + self.assertEqual(rec.setSevr(_dbapi.MAJOR_ALARM, _dbapi.HIHI_ALARM, amsg="Meaningful input alarm message"), None) + self.assertEqual(rec.scan(sync=True), 0) + self.assertEqual(rec.STAT, _dbapi.HIHI_ALARM) + if _dbapi.epicsver[:4] >= (7, 0, 6, 0): + self.assertEqual(rec.AMSG, "Meaningful input alarm message") + + def test_setin_severity_without_message(self): + rec = getRecord("rec:inalarm:plain") + with rec: + self.assertEqual(rec.SEVR, _dbapi.NO_ALARM) + self.assertEqual(rec.STAT, _dbapi.NO_ALARM) + self.assertEqual(rec.setSevr(_dbapi.MAJOR_ALARM, _dbapi.COMM_ALARM), None) + self.assertEqual(rec.scan(sync=True), 0) + self.assertEqual(rec.SEVR, _dbapi.MAJOR_ALARM) + self.assertEqual(rec.STAT, _dbapi.COMM_ALARM) + + def test_setout_severity_message(self): + rec = getRecord("rec:outalarm:amsg") + with rec: + self.assertEqual(rec.SEVR, _dbapi.NO_ALARM) + self.assertEqual(rec.STAT, _dbapi.NO_ALARM) + if _dbapi.epicsver[:4] >= (7, 0, 6, 0): + self.assertEqual(rec.AMSG, "") + self.assertEqual(rec.setSevr(_dbapi.MAJOR_ALARM, _dbapi.HIHI_ALARM, amsg="Meaningful output alarm message"), None) + self.assertEqual(rec.scan(sync=True), 0) + self.assertEqual(rec.STAT, _dbapi.HIHI_ALARM) + if _dbapi.epicsver[:4] >= (7, 0, 6, 0): + self.assertEqual(rec.AMSG, "Meaningful output alarm message") + + def test_setout_severity_without_message(self): + rec = getRecord("rec:outalarm:plain") with rec: - rec.setSevr(_dbapi.MAJOR_ALARM, _dbapi.COMM_ALARM) - self.assertEqual(rec.NSEV, _dbapi.MAJOR_ALARM) - self.assertEqual(rec.NSTA, _dbapi.COMM_ALARM) + self.assertEqual(rec.SEVR, _dbapi.NO_ALARM) + self.assertEqual(rec.STAT, _dbapi.NO_ALARM) + self.assertEqual(rec.setSevr(_dbapi.MAJOR_ALARM, _dbapi.COMM_ALARM), None) + self.assertEqual(rec.scan(sync=True), 0) + self.assertEqual(rec.SEVR, _dbapi.MAJOR_ALARM) + self.assertEqual(rec.STAT, _dbapi.COMM_ALARM)