From 4e98368075b98979413c29f6c8b0785610831fbf Mon Sep 17 00:00:00 2001 From: Ivan Finch Date: Sun, 12 Jul 2026 20:40:52 +0100 Subject: [PATCH] Compensate for having to subclass SharedPV insead of just patching in the desired behaviour; include regression tests --- p4pillon/server/raw.py | 9 +++++++-- tests/unit/server/test_sharedpv_asyncio.py | 22 +++++++++++++++++++++- tests/unit/server/test_sharedpv_thread.py | 22 +++++++++++++++++++++- 3 files changed, 49 insertions(+), 4 deletions(-) diff --git a/p4pillon/server/raw.py b/p4pillon/server/raw.py index 57a6ec6..921ec91 100644 --- a/p4pillon/server/raw.py +++ b/p4pillon/server/raw.py @@ -5,6 +5,7 @@ import logging from abc import ABC +from p4p._p4p import SharedPV as _RawSharedPV from p4p.server.raw import SharedPV as _SharedPV _log = logging.getLogger(__name__) @@ -159,7 +160,9 @@ def open(self, value, nt=None, wrap=None, unwrap=None, **kwargs): else: open_fn(V) - _SharedPV.open(self, V) + # Call the C-extension base directly, bypassing p4p.server.raw.SharedPV.open(), + # which would wrap() V a second time (V is already wrapped above). + _RawSharedPV.open(self, V) def post(self, value, **kwargs): """Provide an update to the Value of this PV. @@ -185,7 +188,9 @@ def post(self, value, **kwargs): else: post_fn(self, V) - _SharedPV.post(self, V) + # Call the C-extension base directly, bypassing p4p.server.raw.SharedPV.post(), + # which would wrap() V a second time (V is already wrapped above). + _RawSharedPV.post(self, V) def close(self, destroy=False): """Close PV, disconnecting any clients. diff --git a/tests/unit/server/test_sharedpv_asyncio.py b/tests/unit/server/test_sharedpv_asyncio.py index 19954c0..619f903 100644 --- a/tests/unit/server/test_sharedpv_asyncio.py +++ b/tests/unit/server/test_sharedpv_asyncio.py @@ -1,4 +1,5 @@ -from p4p.nt import NTScalar +import numpy +from p4p.nt import NTNDArray, NTScalar from p4pillon.server.asyncio import Handler, SharedPV @@ -51,3 +52,22 @@ def teardown_method(self, _method): self.pv.close() del self.handler del self.pv + + +class TestNoDoubleWrapOfInitialValue: + """Regression tests for `SharedPV.open()`/`.post()` each wrapping `value` + via `nt.wrap()` themselves before delegating to p4p's own (already + wrapping) `SharedPV.open()`/`.post()`, which wraps a second time. Harmless + for `NTScalar`, whose `wrap()` tolerates being fed an already-wrapped + `Value` -- but `NTNDArray.wrap()` assumes a raw `numpy.ndarray` and raises + when handed a `Value` on the second pass. + """ + + def test_open_with_ntndarray_does_not_double_wrap(self): + pv = SharedPV(nt=NTNDArray(), initial=numpy.zeros((4, 4))) + assert numpy.array_equal(numpy.asarray(pv.current()).flatten(), numpy.zeros(16)) + + def test_post_with_ntndarray_does_not_double_wrap(self): + pv = SharedPV(nt=NTNDArray(), initial=numpy.zeros((4, 4))) + pv.post(numpy.ones((4, 4))) + assert numpy.array_equal(numpy.asarray(pv.current()).flatten(), numpy.ones(16)) diff --git a/tests/unit/server/test_sharedpv_thread.py b/tests/unit/server/test_sharedpv_thread.py index f15b8ae..25f2636 100644 --- a/tests/unit/server/test_sharedpv_thread.py +++ b/tests/unit/server/test_sharedpv_thread.py @@ -1,4 +1,5 @@ -from p4p.nt import NTScalar +import numpy +from p4p.nt import NTNDArray, NTScalar from p4pillon.server.thread import Handler, SharedPV @@ -51,3 +52,22 @@ def teardown_method(self, _method): self.pv.close() del self.handler del self.pv + + +class TestNoDoubleWrapOfInitialValue: + """Regression tests for `SharedPV.open()`/`.post()` each wrapping `value` + via `nt.wrap()` themselves before delegating to p4p's own (already + wrapping) `SharedPV.open()`/`.post()`, which wraps a second time. Harmless + for `NTScalar`, whose `wrap()` tolerates being fed an already-wrapped + `Value` -- but `NTNDArray.wrap()` assumes a raw `numpy.ndarray` and raises + when handed a `Value` on the second pass. + """ + + def test_open_with_ntndarray_does_not_double_wrap(self): + pv = SharedPV(nt=NTNDArray(), initial=numpy.zeros((4, 4))) + assert numpy.array_equal(numpy.asarray(pv.current()).flatten(), numpy.zeros(16)) + + def test_post_with_ntndarray_does_not_double_wrap(self): + pv = SharedPV(nt=NTNDArray(), initial=numpy.zeros((4, 4))) + pv.post(numpy.ones((4, 4))) + assert numpy.array_equal(numpy.asarray(pv.current()).flatten(), numpy.ones(16))