Skip to content
Snippets Groups Projects
Commit b2815d98 authored by Stefano Di Frischia's avatar Stefano Di Frischia
Browse files

Merge branch 'L2SS-1417-fix-broken-unit-tests' into 'master' 

L2SS-1417: fix broken unittests

Closes L2SS-1417

See merge request !759
parents 4abbc33f b822b927
Branches
Tags
1 merge request!759L2SS-1417: fix broken unittests
Show whitespace changes
Inline Side-by-side
tangostationcontrol/test/beam/managers/test_tilebeam_manager.py
+ 26
− 17
View file @ b2815d98
# Copyright (C) 2023 ASTRON (Netherlands Institute for Radio Astronomy)
# SPDX-License-Identifier: Apache-2.0
import datetime
import unittest
from unittest.mock import MagicMock
from unittest.mock import MagicMock, patch
from test import base
import numpy.testing
from tangostationcontrol.beam.managers import TileBeamManager
from tangostationcontrol.beam.managers import TileBeamManager, _tilebeam
from tangostationcontrol.common.constants import (
DEFAULT_N_HBA_TILES,
N_elements,
N_point_prop,
N_pol,
)
from test import base
class TestTileBeamManager(base.TestCase):
"""Test class for TileBeam manager"""
def test_delays(self):
"""Verify delays are retrieved with correct dimensions"""
dt = datetime.datetime.now()
_dt = datetime.datetime.now()
device_mock = MagicMock()
sut = TileBeamManager(device_mock)
......@@ -36,7 +36,7 @@ class TestTileBeamManager(base.TestCase):
pointings = [i**2 for i in range(DEFAULT_N_HBA_TILES)]
# verify delays method returns the correct dimensions
delays = sut.delays(pointings, dt).flatten()
delays = sut.delays(pointings, _dt).flatten()
numpy.testing.assert_array_equal(
delays,
numpy.concatenate(
......@@ -47,18 +47,21 @@ class TestTileBeamManager(base.TestCase):
),
)
for i, mock in enumerate(sut.HBAT_delay_calculators):
mock.set_measure_time.assert_called_with(dt)
mock.set_measure_time.assert_called_with(_dt)
mock.delays.assert_called_with(pointings[i], DEFAULT_N_HBA_TILES - i)
@unittest.skip("To be resolved in L2SS-1466")
def test_compute_weights(self):
dt = datetime.datetime.now()
@patch.object(_tilebeam, "create_device_proxy")
def test_compute_weights(self, mock_create_device_proxy):
"""Verify computed weights are retrieved correctly"""
_dt = datetime.datetime.now()
pointing_direction = numpy.full(1, 1)
def read_parent_attribute(attr):
match attr:
case "Control_to_RECV_mapping_R":
return numpy.array([[1, x] for x in range(0, DEFAULT_N_HBA_TILES)])
case "RECV_Devices_R":
return [recv_mock.name]
device_mock = MagicMock()
......@@ -74,6 +77,8 @@ class TestTileBeamManager(base.TestCase):
recv_mock.calculate_HBAT_bf_delay_steps.side_effect = lambda a: [a] * 2
device_mock.control.branch_child.return_value = recv_mock
mock_create_device_proxy.return_value = recv_mock
sut = TileBeamManager(device_mock)
sut.nr_tiles = DEFAULT_N_HBA_TILES
sut.delays = MagicMock(
......@@ -82,7 +87,7 @@ class TestTileBeamManager(base.TestCase):
)
)
bf_delay_steps = sut.compute_weights(pointing_direction, dt)
bf_delay_steps = sut.compute_weights(pointing_direction, _dt)
numpy.testing.assert_array_equal(
bf_delay_steps,
......@@ -91,9 +96,10 @@ class TestTileBeamManager(base.TestCase):
),
)
@unittest.skip("To be resolved in L2SS-1466")
def test_apply_weights(self):
dt = datetime.datetime.now()
@patch.object(_tilebeam, "create_device_proxy")
def test_apply_weights(self, mock_create_device_proxy):
"""Verify beam weights are applied correctly"""
_dt = datetime.datetime.now()
pointing_direction = numpy.array(range(DEFAULT_N_HBA_TILES))
bf_delay_steps = numpy.concatenate(
[numpy.full(N_elements, i) for i in range(DEFAULT_N_HBA_TILES)] * N_pol
......@@ -105,6 +111,8 @@ class TestTileBeamManager(base.TestCase):
return numpy.array([[1, x] for x in range(0, DEFAULT_N_HBA_TILES)])
case "ANT_mask_RW":
return [True, False] * (DEFAULT_N_HBA_TILES // 2)
case "RECV_Devices_R":
return [recv_mock.name]
device_mock = MagicMock()
......@@ -119,6 +127,7 @@ class TestTileBeamManager(base.TestCase):
recv_mock.calculate_HBAT_bf_delay_steps = MagicMock()
recv_mock.calculate_HBAT_bf_delay_steps.side_effect = lambda a: [a] * 2
device_mock.control.branch_child.return_value = recv_mock
mock_create_device_proxy.return_value = recv_mock
sut = TileBeamManager(device_mock)
sut.nr_tiles = DEFAULT_N_HBA_TILES
......@@ -128,11 +137,11 @@ class TestTileBeamManager(base.TestCase):
sut.current_pointing_direction = numpy.zeros(
(DEFAULT_N_HBA_TILES, N_point_prop), dtype="<U32"
)
sut.apply_weights(pointing_direction, dt, bf_delay_steps)
sut.apply_weights(pointing_direction, _dt, bf_delay_steps)
numpy.testing.assert_array_equal(
sut.current_pointing_timestamp,
numpy.array([dt.timestamp(), 0] * (DEFAULT_N_HBA_TILES // 2)),
numpy.array([_dt.timestamp(), 0] * (DEFAULT_N_HBA_TILES // 2)),
)
numpy.testing.assert_array_equal(
sut.current_pointing_direction,
......
tangostationcontrol/test/common/test_lofar_logging.py
+ 28
− 62
View file @ b2815d98
......@@ -2,36 +2,53 @@
# SPDX-License-Identifier: Apache-2.0
import logging
import unittest
from unittest import mock
from tango import device_server, DevFailed
from tango.server import Device
from test import base
from tango import DevFailed
from tango.test_context import DeviceTestContext
from tangostationcontrol.common import lofar_logging
from test import base
from tangostationcontrol.devices import antennafield
class TestExceptionToStr(base.TestCase):
"""Test class for logging strings check"""
def test_normal_exception(self):
"""Test whether logging strings work with normal exceptions"""
ex = Exception()
self.assertNotEqual("", lofar_logging.exception_to_str(ex))
def test_tango_exception(self):
"""Test whether logging strings work with Tango exceptions"""
ex = DevFailed()
self.assertNotEqual("", lofar_logging.exception_to_str(ex))
# Simulate DevFailed exception
try:
with DeviceTestContext(
antennafield.AntennaField, properties={}, process=True
) as proxy:
proxy.boot()
except DevFailed as ex:
original_exc = str(ex)
logged_exc = lofar_logging.exception_to_str(ex)
# Verbose original exception vs readable logged exception
self.assertTrue("PyDs_PythonError" in original_exc, msg=original_exc)
self.assertFalse("PyDs_PythonError" in logged_exc, msg=logged_exc)
self.assertTrue(len(original_exc) > len(logged_exc))
class TestLofarLogging(base.TestCase):
"""Test class for logging methods"""
def setUp(self):
super(TestLofarLogging, self).setUp()
# reset logging system
rootLogger = logging.getLogger()
rootLogger.filters = []
rootLogger.handlers = []
rootLogger.manager.loggerDict = {}
root_logger = logging.getLogger()
root_logger.filters = []
root_logger.handlers = []
root_logger.manager.loggerDict = {}
# record everything we log in memory so we can inspect it
class MemoryHandler(logging.Handler):
......@@ -45,7 +62,7 @@ class TestLofarLogging(base.TestCase):
self.records.append(record)
self.memory_handler = MemoryHandler()
rootLogger.addHandler(self.memory_handler)
root_logger.addHandler(self.memory_handler)
def test_configure_logging_basic_usage(self):
"""Test whether configure_logger indeed runs smoothly."""
......@@ -67,57 +84,6 @@ class TestLofarLogging(base.TestCase):
self.assertIn("device", self.memory_handler.records[0].__dict__)
self.assertIn("software_version", self.memory_handler.records[0].__dict__)
@unittest.skip("Logs are not sent to Tango device currently, to reduce logspam")
def test_configure_logging_uses_tango_device(self):
"""Test whether log records get annotated with the active Tango device after using configure_logger(), and whether logs get forwarded to it."""
logger = lofar_logging.configure_logger(debug=True)
# alias to distinguish from other "self"s
test_object = self
# Since we spawn a new process for the Device ("process=True" in the DeviceTestContext),
# and the mock doesn't work across processes, we do all of our mocking and checks in
# the device itself. If any Exception is raised, it will be marshalled to the device
# server by Tango and reraised here in the test as a DevFailed exception.
# create a Tango Device that logs something
class MyDevice(Device):
def init_device(self):
with mock.patch.object(
device_server.DeviceImpl, "__info_stream"
) as m_info_stream:
self.log_deeper_in_stack()
# check if we actually routed the log to self.info_stream
test_object.assertEqual(
1,
m_info_stream.call_count,
msg="configure_logger did not send logs to active Tango device",
)
# Lookup our "test" logline among f.e. the debug messages output by Tango
test_record = [
record
for record in test_object.memory_handler.records
if record.msg == "test log_deeper_in_stack"
]
# Tango uses slightly different class representations of MyDevice, so
# we can't compare them direclty. Just verify we're talking about the same thing.
test_object.assertEqual(
str(self),
str(test_record[0].device),
msg="configure_logging did not detect active Tango device",
)
def log_deeper_in_stack(self):
logger.info("test log_deeper_in_stack")
# if init_device fails, an exception will be thrown
with DeviceTestContext(MyDevice, process=False) as mydevice:
pass
def test_log_exceptions(self):
"""Test whether log_exceptions actually logs and reraises exceptions."""
......
tangostationcontrol/test/devices/sdp/test_digitalbeam_device.py
+ 0
− 102
View file @ b2815d98
# Copyright (C) 2022 ASTRON (Netherlands Institute for Radio Astronomy)
# SPDX-License-Identifier: Apache-2.0
# Builtin regular libraries
import copy
import unittest
# Builtin test libraries
from unittest import mock
# External regular libraries
import numpy
# External test libraries
from tango.test_context import DeviceTestContext
from tangostationcontrol.common.constants import (
MAX_ANTENNA,
N_beamlets_ctrl,
N_xyz,
N_pn,
A_pn,
)
# Internal regular imports
from tangostationcontrol.devices.sdp import digitalbeam
# Internal test imports
from test.devices import device_base
class TestDigitalBeamDevice(device_base.DeviceTestCase):
def setUp(self):
# DeviceTestCase setUp patches lofar_device DeviceProxy
super(TestDigitalBeamDevice, self).setUp()
@unittest.skip("Test for manual use, enable at most one (process=false)")
@mock.patch.object(digitalbeam.DigitalBeam, "_wait_to_apply_weights")
@mock.patch.object(digitalbeam.DigitalBeam, "_compute_weights")
@mock.patch.object(digitalbeam, "DeviceProxy")
def test_apply_weights_disabled(self, m_proxy, m_compute, m_wait):
"""Verify won't overwrite digitalbeam data if no input_selected"""
input_data = numpy.array(
[["AZELGEO", "0rad", "1.570796rad"]] * N_beamlets_ctrl
).flatten()
current_data = numpy.array([[16384] * (A_pn * N_beamlets_ctrl)] * N_pn)
m_proxy.return_value = mock.Mock(
read_attribute=mock.Mock(
return_value=mock.Mock(value=copy.copy(current_data))
),
Antenna_Usage_Mask_R=numpy.array([0] * MAX_ANTENNA),
Antenna_Field_Reference_ITRF_R=mock.MagicMock(),
HBAT_reference_ITRF_R=numpy.array([[0] * N_xyz] * MAX_ANTENNA),
)
new_data = numpy.array([[16384] * 2928 + [0] * 2928] * N_pn)
m_compute.return_value = copy.copy(new_data)
with DeviceTestContext(
digitalbeam.DigitalBeam,
process=False,
) as proxy:
proxy.initialise()
proxy.Tracking_enabled_RW = False
proxy.input_select_RW = numpy.array(
[[False] * N_beamlets_ctrl] * MAX_ANTENNA
)
proxy.set_pointing(input_data)
numpy.testing.assert_equal(
m_proxy.return_value.write_attribute.call_args[0][0], current_data
)
@unittest.skip("Test for manual use, enable at most one (process=false)")
@mock.patch.object(digitalbeam.DigitalBeam, "_wait_to_apply_weights")
@mock.patch.object(digitalbeam.DigitalBeam, "_compute_weights")
@mock.patch.object(digitalbeam, "DeviceProxy")
def test_apply_weights_enabled(self, m_proxy, m_compute, m_wait):
"""Verify can overwrite digitalbeam data if input_selected"""
input_data = numpy.array(
[["AZELGEO", "0rad", "1.570796rad"]] * N_beamlets_ctrl
).flatten()
current_data = numpy.array([[16384] * (A_pn * N_beamlets_ctrl)] * N_pn)
m_proxy.return_value = mock.Mock(
read_attribute=mock.Mock(return_value=mock.Mock(value=current_data)),
Antenna_Usage_Mask_R=numpy.array([0] * MAX_ANTENNA),
Antenna_Field_Reference_ITRF_R=mock.MagicMock(),
HBAT_reference_ITRF_R=numpy.array([[0] * N_xyz] * MAX_ANTENNA),
)
new_data = numpy.array([[16384] * 2928 + [0] * 2928] * N_pn)
m_compute.return_value = copy.copy(new_data)
with DeviceTestContext(
digitalbeam.DigitalBeam,
process=False,
) as proxy:
proxy.initialise()
proxy.Tracking_enabled_RW = False
proxy.input_select_RW = numpy.array(
[[True] * N_beamlets_ctrl] * MAX_ANTENNA
)
proxy.set_pointing(input_data)
numpy.testing.assert_equal(
m_proxy.return_value.write_attribute.call_args[0][0], new_data
)
tangostationcontrol/test/devices/test_antennafield_device.py
+ 0
− 32
View file @ b2815d98
......@@ -5,8 +5,6 @@
# pylint: disable=C0103
import logging
import unittest
from unittest import mock
from test import base
from test.devices import device_base
......@@ -1270,36 +1268,6 @@ class TestAntennafieldDevice(device_base.DeviceTestCase):
for i in range(len(antenna_names)):
self.assertTrue(proxy.Antenna_Names_R[i] == f"C{i}")
@unittest.skip("Test for manual use, enable at most one (process=false)")
@mock.patch.object(antennafield, "DeviceProxy")
def test_set_mapped_attribute(self, m_proxy):
"""Verify set_mapped_attribute only modifies controlled inputs"""
antenna_properties = {
"Control_Children": ["stat/RECVH/H0"],
}
data = numpy.array([[False] * N_rcu] * MAX_ANTENNA)
m_proxy.return_value = mock.Mock(
read_attribute=mock.Mock(return_value=mock.Mock(value=data))
)
with DeviceTestContext(
antennafield.AntennaField,
process=False,
properties={**self.AT_PROPERTIES, **antenna_properties},
) as proxy:
proxy.boot()
proxy.write_attribute(
"HBAT_PWR_on_RW", numpy.array([[False] * N_rcu] * DEFAULT_N_HBA_TILES)
)
numpy.testing.assert_equal(
m_proxy.return_value.write_attribute.call_args[0][1], data
)
def test_Antenna_Mask_all_antennas(self):
"""Verify if Antenna_Mask_R is correctly retrieved"""
with DeviceTestContext(
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment