Merge branch 'L2SS-1614-fix-test-calibration' into 'master'

L2SS-1614: fix test calibration

Closes L2SS-1614

See merge request !779

tangostationcontrol/test/common/test_calibration.py

@@ -6,7 +6,7 @@ from unittest.mock import patch, Mock, call, PropertyMock
 from test import base
 
 import numpy
-from numpy.testing import assert_array_equal, assert_equal
+from numpy.testing import assert_array_equal
 
 from tangostationcontrol.common import consul
 from tangostationcontrol.common.calibration import (
@@ -17,7 +17,7 @@ from tangostationcontrol.common.calibration import (
     CalibrationTable,
 )
 from tangostationcontrol.common.constants import S_pn, N_subbands, N_pn, SDP_UNIT_WEIGHT
-from tangostationcontrol.common.sdp import complex_to_weights, are_subbands_decreasing
+from tangostationcontrol.common.sdp import complex_to_weights
 
 
 class MockMinio:
@@ -75,11 +75,8 @@ class TestCalibrationManager(base.TestCase):
             ]
         )
 
-    @patch("tangostationcontrol.common.calibration.read_hdf5")
-    def test_calibrate_subband_weights(self, hdf_reader, _):
-        """Test whether calibration values are correctly applied to weights"""
-
-        antenna_field_mock = Mock(
+    def _setup_mock_antennafield(self):
+        return Mock(
             Antenna_to_SDP_Mapping_R=numpy.array(
                 [[1, 1], [1, 2]], dtype=numpy.int32
             ).reshape(-1, 2),
@@ -88,6 +85,19 @@ class TestCalibrationManager(base.TestCase):
             antenna_type_R="HBA",
             **{"name.return_value": "Stat/AFH/HBA0"},
         )
+
+    def _setup_mock_caltable(self):
+        return Mock(
+            observation_station="unittest-station",
+            antennas={
+                "T1": Mock(x=numpy.arange(0, 512), y=numpy.arange(512, 1024)),
+                "T2": Mock(x=numpy.arange(1024, 1536), y=numpy.arange(1536, 2048)),
+            },
+        )
+
+    def _test_calibration(self, nyquist_zone, hdf_reader):
+        """Common logic of calibration tests"""
+        antenna_field_mock = self._setup_mock_antennafield()
         subband_weights = numpy.array([[SDP_UNIT_WEIGHT] * S_pn * N_subbands] * N_pn)
 
         def subband_weights_side_effect(new_value=None):
@@ -97,23 +107,16 @@ class TestCalibrationManager(base.TestCase):
             return subband_weights
 
         sdp_mock = Mock(
-            nyquist_zone_RW=numpy.array([[0] * N_pn] * S_pn),
+            nyquist_zone_RW=nyquist_zone,
             FPGA_spectral_inversion_R=numpy.array([[0] * N_pn] * S_pn),
         )
         subband_property_mock = PropertyMock(side_effect=subband_weights_side_effect)
         type(sdp_mock).FPGA_subband_weights_RW = subband_property_mock
-        caltable_mock = Mock(
-            observation_station="unittest-station",
-            antennas={
-                "T1": Mock(x=numpy.arange(0, 512), y=numpy.arange(512, 1024)),
-                "T2": Mock(x=numpy.arange(1024, 1536), y=numpy.arange(1536, 2048)),
-            },
-        )
+        caltable_mock = self._setup_mock_caltable()
         hdf_reader.return_value.__enter__.return_value = caltable_mock
 
         sut = CalibrationManager("http://server:1234", "unittest-station")
         sut.calibrate_subband_weights(antenna_field_mock, sdp_mock)
-
         hdf_reader.assert_has_calls(
             [
                 call(
@@ -130,6 +133,14 @@ class TestCalibrationManager(base.TestCase):
                 call().__exit__(None, None, None),
             ]
         )
+        return subband_weights
+
+    @patch("tangostationcontrol.common.calibration.read_hdf5")
+    def test_calibrate_subband_weights(self, hdf_reader, _):
+        """Test whether calibration values are correctly applied to weights"""
+        nyquist_zone = numpy.array([[0] * N_pn] * S_pn)
+        subband_weights = self._test_calibration(nyquist_zone, hdf_reader)
+
         assert_array_equal(
             subband_weights[1, 1024:1536],
             complex_to_weights(numpy.arange(0, 512)),
@@ -151,73 +162,10 @@ class TestCalibrationManager(base.TestCase):
     def test_calibrate_reverse_order(self, hdf_reader, _):
         """Test whether calibration values are applied in decreasing order
         when subband frequencies are decreasing"""
-        antenna_field_mock = Mock(
-            Antenna_to_SDP_Mapping_R=numpy.array(
-                [[1, 1], [1, 2]], dtype=numpy.int32
-            ).reshape(-1, 2),
-            Antenna_Names_R=[f"T{n + 1}" for n in range(2)],
-            RCU_band_select_RW=numpy.array([[1, 1], [2, 2]]),
-            **{"name.return_value": "Stat/AFH/HBA0"},
-        )
-        subband_weights = numpy.array([[SDP_UNIT_WEIGHT] * S_pn * N_subbands] * N_pn)
-
-        def subband_weights_side_effect(new_value=None):
-            nonlocal subband_weights
-            if new_value is not None:
-                subband_weights = new_value
-            return subband_weights
-
-        sdp_mock = Mock(
-            nyquist_zone_RW=numpy.array(
+        nyquist_zone = numpy.array(
             [[0] * N_pn] + [[1] * N_pn] + [[0] * N_pn] * (S_pn - 2)
-            ),
-            FPGA_spectral_inversion_R=numpy.array([[0] * N_pn] * S_pn),
-        )
-        subband_property_mock = PropertyMock(side_effect=subband_weights_side_effect)
-        type(sdp_mock).FPGA_subband_weights_RW = subband_property_mock
-        caltable_mock = Mock(
-            observation_station="unittest-station",
-            antennas={
-                "T1": Mock(x=numpy.arange(0, 512), y=numpy.arange(512, 1024)),
-                "T2": Mock(x=numpy.arange(1024, 1536), y=numpy.arange(1536, 2048)),
-            },
-        )
-        hdf_reader.return_value.__enter__.return_value = caltable_mock
-
-        sut = CalibrationManager("http://server:1234", "unittest-station")
-        sut.calibrate_subband_weights(antenna_field_mock, sdp_mock)
-
-        hdf_reader.assert_has_calls(
-            [
-                call(
-                    f"{sut._tmp_dir.name}/CalTable-unittest-station-HBA-200MHz.h5",
-                    CalibrationTable,
-                ),
-                call().__enter__(),
-                call().__exit__(None, None, None),
-                call(
-                    f"{sut._tmp_dir.name}/CalTable-unittest-station-HBA-150MHz.h5",
-                    CalibrationTable,
-                ),
-                call().__enter__(),
-                call().__exit__(None, None, None),
-            ]
-        )
-
-        assert_equal(
-            are_subbands_decreasing(
-                sdp_mock.nyquist_zone_RW[0, 0],
-                sdp_mock.FPGA_spectral_inversion_R[0, 0],
-            ),
-            False,
-        )
-        assert_equal(
-            are_subbands_decreasing(
-                sdp_mock.nyquist_zone_RW[1, 0],
-                sdp_mock.FPGA_spectral_inversion_R[0, 0],
-            ),
-            True,
         )
+        subband_weights = self._test_calibration(nyquist_zone, hdf_reader)
 
         assert_array_equal(
             subband_weights[1, 1024:1536],