Skip to content
Snippets Groups Projects
Commit f948e2fa authored by Corné Lukken's avatar Corné Lukken
Browse files

Add tests for read_modify_write of lofar_device_proxy

parent 2448b39a
No related branches found
No related tags found
1 merge request!437Add tests for read_modify_write of lofar_device_proxy
......@@ -9,13 +9,13 @@ RUN sudo apt-get install -y python3-dev libboost-python-dev pkg-config && sudo a
RUN sudo apt-get install -y rsync && sudo apt-get clean
COPY lofar-device-base/lofar-requirements.txt /lofar-requirements.txt
RUN sudo pip3 install -r /lofar-requirements.txt
# Manually install all requirements from the .txt as part of the base image
# This reduces runtime overhead as well as preventing issues around dependency
# installation for development builds (pip install ./ ignores requirements.txt)
COPY tmp/requirements.txt /tangostationcontrol-requirements.txt
RUN sudo pip3 install -r /tangostationcontrol-requirements.txt
RUN sudo pip3 install -r /tangostationcontrol-requirements.txt -r /lofar-requirements.txt
# install and use ephimerides and geodetic ("measures") tables for casacore.
# we install a _stub_ since the tables need to be deployed explicitly from within the software.
......
......@@ -26,7 +26,7 @@ package_dir=
packages=find:
python_requires => 3.7
install_requires =
importlib-metadata>=0.12, <5.0;python_version<"3.8"
importlib-metadata<2.0.0,>=0.12;python_version<"3.8"
pip>=1.5
[options.packages.find]
......
......@@ -3,23 +3,14 @@
# Distributed under the terms of the APACHE license.
# See LICENSE.txt for more info.
from collections.abc import Sequence
import numpy
def is_sequence(obj):
"""True for sequences, positionally ordered collections
See https://www.pythontutorial.net/advanced-python/python-sequences/
"""
return isinstance(obj, Sequence) or isinstance(obj, numpy.ndarray)
from tango.utils import is_seq
def sequence_not_str(obj):
"""True for sequences that are not str, bytes or bytearray"""
return is_sequence(obj) and not isinstance(obj, (str, bytes, bytearray))
return is_seq(obj) and not isinstance(obj, (str, bytes, bytearray))
def type_not_sequence(obj):
"""True for types that are not sequences"""
return not is_sequence(obj) and isinstance(obj, type)
return not is_seq(obj) and isinstance(obj, type)
......@@ -6,6 +6,8 @@
#
# Distributed under the terms of the APACHE license.
# See LICENSE.txt for more info.
from tango.utils import is_seq
import numpy
from tangostationcontrol.common import type_checking
......@@ -32,7 +34,10 @@ class TestTypeChecking(base.TestCase):
return False
def sequence_test(self, obj):
"""Test object is sequence based on properties and verify is_sequence"""
"""Test object is sequence based on properties and verify is_sequence
Recover alternative is_sequence method from commit 73177e9 if tests fail
"""
result = (
self.subscriptable(obj) & self.iterable(obj)
......@@ -40,12 +45,15 @@ class TestTypeChecking(base.TestCase):
)
self.assertEqual(
result, type_checking.is_sequence(obj),
result, is_seq(obj),
F"Test failed for type {type(obj)}"
)
def test_is_sequence_for_types(self):
"""Types to be tested by is_sequence"""
"""Types to be tested by is_sequence
Recover alternative is_seq method from commit 73177e9 if tests fail
"""
test_types = [
(False,),
......
......@@ -367,44 +367,6 @@ class TestAntennaToRecvMapper(base.TestCase):
actual = mapper.map_write("HBAT_PWR_on_RW", set_values)
numpy.testing.assert_equal(expected, actual)
# def test_merge_write(self):
# """Verify all None fields are replaced by merge_write if no control"""
#
# mapper = AntennaToRecvMapper(
# self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 1
# )
#
# merge_values = [[None] * 32] * 96
# current_values = [[False] * 32] * 96
#
# mapper.merge_write(merge_values, current_values)
# numpy.testing.assert_equal(merge_values, current_values)
#
# results = []
# for _i in range(25):
# start_time = time.monotonic_ns()
# mapper.merge_write(merge_values, current_values)
# stop_time = time.monotonic_ns()
# results.append(stop_time - start_time)
#
# logging.error(
# f"Merge write performance: Median {statistics.median(results) / 1.e9} "
# f"Stdev {statistics.stdev(results) / 1.e9}"
# )
#
# def test_merge_write_values(self):
# """Verify all fields with values are retained by merge_write"""
#
# mapper = AntennaToRecvMapper(
# self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 1
# )
#
# merge_values = [[True] * 32] * 2 + [[None] * 32] * 94
# current_values = [[True] * 32] * 2 + [[False] * 32] * 94
#
# mapper.merge_write(merge_values, current_values)
# numpy.testing.assert_equal(merge_values, current_values)
class TestAntennafieldDevice(device_base.DeviceTestCase):
......
......@@ -7,12 +7,20 @@
# Distributed under the terms of the APACHE license.
# See LICENSE.txt for more info.
import numpy
from tango.test_context import DeviceTestContext
from tango.server import attribute
from tango import DevState, DevFailed
from tango.server import command
from tango import AttrWriteType
from tango import DevFailed
from tango import DevState
from tango import DevVarBooleanArray
from tangostationcontrol.devices import lofar_device
from unittest import mock
from tangostationcontrol.test.devices import device_base
......@@ -75,8 +83,54 @@ class TestLofarDevice(device_base.DeviceTestCase):
# Just for demo, do not use class variables to store attribute state
_bool_array = [False] * BOOL_ARRAY_DIM
bool_array = attribute(
dtype=(bool,), max_dim_x=BOOL_ARRAY_DIM,
access=AttrWriteType.READ_WRITE, fget="get_bool_array",
fset="set_bool_array"
)
@attribute(dtype=(bool,), max_dim_x=BOOL_ARRAY_DIM)
def bool_array(self):
def get_bool_array(self):
return self._bool_array
def set_bool_array(self, bool_array):
self._bool_array = bool_array
@command(dtype_in=DevVarBooleanArray)
def do_read_modify_write(self, values: numpy.array):
bool_array_half = int(self.BOOL_ARRAY_DIM / 2)
# We have to mock the proxy because lofar_device.proxy will be
# patched
t_write = mock.Mock()
t_proxy = mock.Mock(
read_attribute=mock.Mock(
return_value=mock.Mock(
value=numpy.array(self._bool_array)
)
),
write_attribute=t_write
)
self.atomic_read_modify_write_attribute(
values, t_proxy, "bool_array",
numpy.array([True, False] * bool_array_half)
)
# Fake the write, extract the call argument from t_write mock
self._bool_array = t_write.call_args[0][1]
with DeviceTestContext(
AttributeLofarDevice, process=True
) as proxy:
bool_array_half = int(AttributeLofarDevice.BOOL_ARRAY_DIM / 2)
excepted_result = [True, False] * bool_array_half
proxy.initialise()
proxy.do_read_modify_write(
[True] * AttributeLofarDevice.BOOL_ARRAY_DIM
)
numpy.testing.assert_array_equal(
excepted_result, proxy.bool_array
)
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment