diff --git a/tangostationcontrol/tangostationcontrol/test/devices/test_antennafield_device.py b/tangostationcontrol/tangostationcontrol/test/devices/test_antennafield_device.py index 48692d08b255bea2a232776737a7d17f18633606..94256ba5ce4bbba274f45d41ca6098acdd434381 100644 --- a/tangostationcontrol/tangostationcontrol/test/devices/test_antennafield_device.py +++ b/tangostationcontrol/tangostationcontrol/test/devices/test_antennafield_device.py @@ -20,15 +20,15 @@ from tangostationcontrol.test.devices import device_base class TestHBATToRecvMapper(base.TestCase): # A mapping where HBATs are all not mapped to power RCUs - power_not_connected = [[-1, -1]] * 48 + POWER_NOT_CONNECTED = [[-1, -1]] * 48 # A mapping where HBATs are all not mapped to control RCUs - control_not_connected = [[-1, -1]] * 48 + CONTROL_NOT_CONNECTED = [[-1, -1]] * 48 # A mapping where first two HBATs are mapped on the first Receiver. # The first HBAT control line on RCU 1 and the second HBAT control line on RCU 0. - control_hba_0_and_1_on_rcu_1_and_0_of_recv_1 = [[1, 1], [1, 0]] + [[-1, -1]] * 46 + CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1 = [[1, 1], [1, 0]] + [[-1, -1]] * 46 def test_ant_read_mask_r_no_mapping(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 3) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 3) receiver_values = [[[False] * 3] * 32, [[False] * 3] * 32, [[False] * 3] * 32] expected = [False] * 48 @@ -36,7 +36,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_ant_read_mask_r_hba_0_and_1_on_rcu_1_and_0_of_recv_1(self): - mapper = HBATToRecvMapper(self.control_hba_0_and_1_on_rcu_1_and_0_of_recv_1, self.power_not_connected, 3) + mapper = HBATToRecvMapper(self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1, self.POWER_NOT_CONNECTED, 3) receiver_values = [[[False, True, False]] + [[False, False, False]] * 31, [[False] * 3] * 32, [[False] * 3] * 32] expected = [True, False] + [False] * 46 @@ -45,14 +45,14 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_rcu_band_select_no_mapping(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 3) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 3) receiver_values = [[[0] * 3] * 32, [[0] * 3] * 32, [[0] * 3] * 32] expected = [0] * 48 actual = mapper.map_read("RCU_band_select_RW", receiver_values) numpy.testing.assert_equal(expected, actual) def test_bf_read_delay_steps_r_no_mapping(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 3) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 3) receiver_values = [[[0] * 32] * 96, [[0] * 32] * 96, [[0] * 32] * 96] expected = [[0] * 32] * 48 @@ -60,7 +60,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_bf_read_delay_steps_r_hba_0_and_1_on_rcu_1_and_0_of_recv_1(self): - mapper = HBATToRecvMapper(self.control_hba_0_and_1_on_rcu_1_and_0_of_recv_1, self.power_not_connected, 3) + mapper = HBATToRecvMapper(self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1, self.POWER_NOT_CONNECTED, 3) receiver_values = [[[2] * 32, [1] * 32] + [[0] * 32] * 94, [[0] * 32] * 96, [[0] * 32] * 96] expected = [[1] * 32, [2] * 32] + [[0] * 32] * 46 @@ -69,7 +69,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_bf_read_delay_steps_rw_no_mapping(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 3) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 3) receiver_values = [[[0] * 32] * 96, [[0] * 32] * 96, [[0] * 32] * 96] expected = [[0] * 32] * 48 @@ -77,7 +77,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_bf_read_delay_steps_rw_hba_0_and_1_on_rcu_1_and_0_of_recv_1(self): - mapper = HBATToRecvMapper(self.control_hba_0_and_1_on_rcu_1_and_0_of_recv_1, self.power_not_connected, 3) + mapper = HBATToRecvMapper(self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1, self.POWER_NOT_CONNECTED, 3) receiver_values = [[[2] * 32, [1] * 32] + [[0] * 32] * 94, [[0] * 32] * 96, [[0] * 32] * 96] expected = [[1] * 32, [2] * 32] + [[0] * 32] * 46 @@ -86,7 +86,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_read_led_on_r_unmapped(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 3) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 3) receiver_values = [[[False] * 32] * 96, [[False] * 32] * 96, [[False] * 32] * 96] expected = [[False] * 32] * 48 @@ -94,7 +94,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_read_led_on_r_hba_0_and_1_on_rcu_1_and_0_of_recv_1(self): - mapper = HBATToRecvMapper(self.control_hba_0_and_1_on_rcu_1_and_0_of_recv_1, self.power_not_connected, 3) + mapper = HBATToRecvMapper(self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1, self.POWER_NOT_CONNECTED, 3) receiver_values = [[[False, True] * 16, [True, False] * 16] + [[False] * 32] * 94, [[False] * 32] * 96, [[False] * 32] * 96] @@ -103,7 +103,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_read_led_on_rw_unmapped(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 3) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 3) receiver_values = [[[False] * 32] * 96, [[False] * 32] * 96, [[False] * 32] * 96] expected = [[False] * 32] * 48 @@ -111,7 +111,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_read_led_on_rw_hba_0_and_1_on_rcu_1_and_0_of_recv_1(self): - mapper = HBATToRecvMapper(self.control_hba_0_and_1_on_rcu_1_and_0_of_recv_1, self.power_not_connected, 3) + mapper = HBATToRecvMapper(self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1, self.POWER_NOT_CONNECTED, 3) receiver_values = [[[False, True] * 16, [True, False] * 16] + [[False] * 32] * 94, [[False] * 32] * 96, [[False] * 32] * 96] @@ -120,7 +120,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_read_pwr_lna_on_r_unmapped(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 3) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 3) receiver_values = [[[False] * 32] * 96, [[False] * 32] * 96, [[False] * 32] * 96] expected = [[False] * 32] * 48 @@ -128,7 +128,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_read_pwr_lna_on_r_hba_0_and_1_on_rcu_1_and_0_of_recv_1(self): - mapper = HBATToRecvMapper(self.control_hba_0_and_1_on_rcu_1_and_0_of_recv_1, self.power_not_connected, 3) + mapper = HBATToRecvMapper(self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1, self.POWER_NOT_CONNECTED, 3) receiver_values = [[[False, True] * 16, [True, False] * 16] + [[False] * 32] * 94, [[False] * 32] * 96, [[False] * 32] * 96] @@ -137,7 +137,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_read_pwr_lna_on_rw_unmapped(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 3) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 3) receiver_values = [[[False] * 32] * 96, [[False] * 32] * 96, [[False] * 32] * 96] expected = [[False] * 32] * 48 @@ -145,7 +145,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_read_pwr_lna_on_rw_hba_0_and_1_on_rcu_1_and_0_of_recv_1(self): - mapper = HBATToRecvMapper(self.control_hba_0_and_1_on_rcu_1_and_0_of_recv_1, self.power_not_connected, 3) + mapper = HBATToRecvMapper(self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1, self.POWER_NOT_CONNECTED, 3) receiver_values = [[[False, True] * 16, [True, False] * 16] + [[False] * 32] * 94, [[False] * 32] * 96, [[False] * 32] * 96] @@ -154,7 +154,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_read_pwr_on_r_unmapped(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 3) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 3) receiver_values = [[[False] * 32] * 96, [[False] * 32] * 96, [[False] * 32] * 96] expected = [[False] * 32] * 48 @@ -162,7 +162,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_read_pwr_on_r_hba_0_and_1_on_rcu_1_and_0_of_recv_1(self): - mapper = HBATToRecvMapper(self.control_hba_0_and_1_on_rcu_1_and_0_of_recv_1, self.power_not_connected, 3) + mapper = HBATToRecvMapper(self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1, self.POWER_NOT_CONNECTED, 3) receiver_values = [[[False, True] * 16, [True, False] * 16] + [[False] * 32] * 94, [[False] * 32] * 96, [[False] * 32] * 96] @@ -171,7 +171,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_read_pwr_on_rw_unmapped(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 3) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 3) receiver_values = [[[False] * 32] * 96, [[False] * 32] * 96, [[False] * 32] * 96] expected = [[False] * 32] * 48 @@ -179,7 +179,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_read_pwr_on_rw_hba_0_and_1_on_rcu_1_and_0_of_recv_1(self): - mapper = HBATToRecvMapper(self.control_hba_0_and_1_on_rcu_1_and_0_of_recv_1, self.power_not_connected, 3) + mapper = HBATToRecvMapper(self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1, self.POWER_NOT_CONNECTED, 3) receiver_values = [[[False, True] * 16, [True, False] * 16] + [[False] * 32] * 94, [[False] * 32] * 96, [[False] * 32] * 96] @@ -190,7 +190,7 @@ class TestHBATToRecvMapper(base.TestCase): # Rename to write def test_map_write_ant_mask_rw_no_mapping_and_one_receiver(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 1) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 1) set_values = [False] * 48 expected = [[[False] * 3] * 32] @@ -198,7 +198,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_write_ant_mask_rw_no_mapping_and_two_receivers(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 2) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 2) set_values = [False] * 48 expected = [[[False] * 3] * 32] * 2 @@ -206,7 +206,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_write_ant_mask_rw_hba_0_and_1_on_rcu_1_and_0_of_recv_1(self): - mapper = HBATToRecvMapper(self.control_hba_0_and_1_on_rcu_1_and_0_of_recv_1, self.power_not_connected, 1) + mapper = HBATToRecvMapper(self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1, self.POWER_NOT_CONNECTED, 1) set_values = [True, False] + [False] * 46 expected = [[[False, True, False]] + [[False] * 3] * 31] @@ -214,7 +214,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_write_rcu_band_select_no_mapping_and_one_receiver(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 1) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 1) set_values = [0] * 48 expected = [[[0] * 3] * 32] @@ -222,7 +222,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_write_rcu_band_select_no_mapping_and_two_receivers(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 2) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 2) set_values = [0] * 48 expected = [[[0] * 3] * 32] * 2 @@ -230,7 +230,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_write_rcu_band_select_hba_0_and_1_on_rcu_1_and_0_of_recv_1(self): - mapper = HBATToRecvMapper(self.control_hba_0_and_1_on_rcu_1_and_0_of_recv_1, self.power_not_connected, 1) + mapper = HBATToRecvMapper(self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1, self.POWER_NOT_CONNECTED, 1) set_values = [1, 0] + [0] * 46 expected = [[[0, 1, 0]] + [[0] * 3] * 31] @@ -238,7 +238,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_write_bf_delay_steps_rw_no_mapping_and_one_receiver(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 1) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 1) set_values = [[1] * 32] * 48 expected = [[[0] * 32] * 96] @@ -246,7 +246,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_write_bf_delay_steps_rw_no_mapping_and_two_receivers(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 2) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 2) set_values = [[1] * 32] * 48 expected = [[[0] * 32] * 96, [[0] * 32] * 96] @@ -254,7 +254,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_write_bf_delay_steps_rw_hba_0_and_1_on_rcu_1_and_0_of_recv_1(self): - mapper = HBATToRecvMapper(self.control_hba_0_and_1_on_rcu_1_and_0_of_recv_1, self.power_not_connected, 1) + mapper = HBATToRecvMapper(self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1, self.POWER_NOT_CONNECTED, 1) set_values = [[1] * 32, [2] * 32] + [[0] * 32] * 46 expected = [[[2] * 32, [1] * 32] + [[0] * 32] * 94] @@ -262,7 +262,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_write_led_on_rw_no_mapping_and_one_receiver(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 1) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 1) set_values = [[False] * 32] * 48 expected = [[[False] * 32] * 96] @@ -270,7 +270,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_write_led_on_rw_no_mapping_and_two_receivers(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 2) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 2) set_values = [[False] * 32] * 48 expected = [[[False] * 32] * 96, [[False] * 32] * 96] @@ -278,7 +278,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_write_led_on_rw_hba_0_and_1_on_rcu_1_and_0_of_recv_1(self): - mapper = HBATToRecvMapper(self.control_hba_0_and_1_on_rcu_1_and_0_of_recv_1, self.power_not_connected, 1) + mapper = HBATToRecvMapper(self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1, self.POWER_NOT_CONNECTED, 1) set_values = [[False, True] * 16, [True, False] * 16] + [[False] * 32] * 46 expected = [[[True, False] * 16, [False, True] * 16] + [[False] * 32] * 94] @@ -286,7 +286,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_write_pwr_lna_on_rw_no_mapping_and_one_receiver(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 1) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 1) set_values = [[False] * 32] * 48 expected = [[[False] * 32] * 96] @@ -294,7 +294,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_write_pwr_lna_on_rw_no_mapping_and_two_receivers(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 2) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 2) set_values = [[False] * 32] * 48 expected = [[[False] * 32] * 96, [[False] * 32] * 96] @@ -302,7 +302,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_write_pwr_lna_on_rw_hba_0_and_1_on_rcu_1_and_0_of_recv_1(self): - mapper = HBATToRecvMapper(self.control_hba_0_and_1_on_rcu_1_and_0_of_recv_1, self.power_not_connected, 1) + mapper = HBATToRecvMapper(self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1, self.POWER_NOT_CONNECTED, 1) set_values = [[False, True] * 16, [True, False] * 16] + [[False] * 32] * 46 expected = [[[True, False] * 16, [False, True] * 16] + [[False] * 32] * 94] @@ -310,7 +310,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_write_pwr_on_rw_no_mapping_and_one_receiver(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 1) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 1) set_values = [[False] * 32] * 48 expected = [[[False] * 32] * 96] @@ -318,7 +318,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_write_lna_on_rw_no_mapping_and_two_receivers(self): - mapper = HBATToRecvMapper(self.control_not_connected, self.power_not_connected, 2) + mapper = HBATToRecvMapper(self.CONTROL_NOT_CONNECTED, self.POWER_NOT_CONNECTED, 2) set_values = [[False] * 32] * 48 expected = [[[False] * 32] * 96, [[False] * 32] * 96] @@ -326,7 +326,7 @@ class TestHBATToRecvMapper(base.TestCase): numpy.testing.assert_equal(expected, actual) def test_map_write_pwr_on_rw_hba_0_and_1_on_rcu_1_and_0_of_recv_1(self): - mapper = HBATToRecvMapper(self.control_hba_0_and_1_on_rcu_1_and_0_of_recv_1, self.power_not_connected, 1) + mapper = HBATToRecvMapper(self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1, self.POWER_NOT_CONNECTED, 1) set_values = [[False, True] * 16, [True, False] * 16] + [[False] * 32] * 46 expected = [[[True, False] * 16, [False, True] * 16] + [[False] * 32] * 94] @@ -337,7 +337,7 @@ class TestHBATToRecvMapper(base.TestCase): class TestAntennafieldDevice(device_base.DeviceTestCase): # some dummy values for mandatory properties - at_properties = {'OPC_Server_Name': 'example.com', 'OPC_Server_Port': 4840, 'OPC_Time_Out': 5.0, + AT_PROPERTIES = {'OPC_Server_Name': 'example.com', 'OPC_Server_Port': 4840, 'OPC_Time_Out': 5.0, 'Antenna_Field_Reference_ITRF' : [3.0, 3.0, 3.0], 'Antenna_Field_Reference_ETRS' : [7.0, 7.0, 7.0]} def setUp(self): @@ -347,7 +347,7 @@ class TestAntennafieldDevice(device_base.DeviceTestCase): def test_read_Antenna_Field_Reference(self): """Verify if Antenna coordinates are correctly provided""" # Device uses ITRF coordinates by default - with DeviceTestContext(antennafield.AntennaField, properties=self.at_properties, process=True) as proxy: + with DeviceTestContext(antennafield.AntennaField, properties=self.AT_PROPERTIES, process=True) as proxy: self.assertEqual(3.0, proxy.Antenna_Field_Reference_ITRF_R[0]) # Device derives coordinates from ETRS if ITRF ones are not found at_properties_v2 = {'OPC_Server_Name': 'example.com', 'OPC_Server_Port': 4840, 'OPC_Time_Out': 5.0, 'Antenna_Field_Reference_ETRS' : [7.0, 7.0, 7.0]} diff --git a/tangostationcontrol/tangostationcontrol/test/devices/test_recv_device.py b/tangostationcontrol/tangostationcontrol/test/devices/test_recv_device.py index 1e1c08570f20c217aa5be54fdee43571aa886707..fe53bc69729aa809dcbddc9bec410a7bcde69330 100644 --- a/tangostationcontrol/tangostationcontrol/test/devices/test_recv_device.py +++ b/tangostationcontrol/tangostationcontrol/test/devices/test_recv_device.py @@ -19,7 +19,7 @@ from tangostationcontrol.test.devices import device_base class TestRecvDevice(device_base.DeviceTestCase): # some dummy values for mandatory properties - recv_properties = {'OPC_Server_Name': 'example.com', 'OPC_Server_Port': 4840, 'OPC_Time_Out': 5.0} + RECV_PROPERTIES = {'OPC_Server_Name': 'example.com', 'OPC_Server_Port': 4840, 'OPC_Time_Out': 5.0} def setUp(self): # DeviceTestCase setUp patches lofar_device DeviceProxy @@ -27,14 +27,14 @@ class TestRecvDevice(device_base.DeviceTestCase): def test_calculate_HBAT_bf_delay_steps(self): """Verify HBAT beamforming calculations are correctly executed""" - with DeviceTestContext(recv.RECV, properties=self.recv_properties, process=True) as proxy: + with DeviceTestContext(recv.RECV, properties=self.RECV_PROPERTIES, process=True) as proxy: delays = numpy.random.rand(96,16).flatten() HBAT_bf_delay_steps = proxy.calculate_HBAT_bf_delay_steps(delays) self.assertEqual(3072, len(HBAT_bf_delay_steps)) # 96x32=3072 def test_get_rcu_band_from_filter(self): """Verify filter lookup table values are correctly retrieved""" - with DeviceTestContext(recv.RECV, properties=self.recv_properties, process=True) as proxy: + with DeviceTestContext(recv.RECV, properties=self.RECV_PROPERTIES, process=True) as proxy: filter_name = "HBA_170_230" self.assertEqual(1, proxy.get_rcu_band_from_filter(filter_name)) self.assertEqual(-1, proxy.get_rcu_band_from_filter('MOCK_FILTER')) diff --git a/tangostationcontrol/tangostationcontrol/test/devices/test_snmp_device.py b/tangostationcontrol/tangostationcontrol/test/devices/test_snmp_device.py index 23d81f6d5634b6e4cb1e7ca9ee935484efbdbcc7..609b30ede7f95d3fcb88ef40b9dfbf86f75cbdd0 100644 --- a/tangostationcontrol/tangostationcontrol/test/devices/test_snmp_device.py +++ b/tangostationcontrol/tangostationcontrol/test/devices/test_snmp_device.py @@ -19,15 +19,15 @@ from tangostationcontrol.test.devices import device_base class TestSNMPDevice(device_base.DeviceTestCase): # some dummy values for mandatory properties - snmp_properties = {'SNMP_community': 'localhost', 'SNMP_host': 161, 'SNMP_rel_mib_dir': "SNMP_mib_loading", 'SNMP_timeout': 5.0} + SNMP_PROPERTIES = {'SNMP_community': 'localhost', 'SNMP_host': 161, 'SNMP_rel_mib_dir': "SNMP_mib_loading", 'SNMP_timeout': 5.0} def setUp(self): # DeviceTestCase setUp patches lofar_device DeviceProxy super(TestSNMPDevice, self).setUp() def test_get_mib_dir(self): - with DeviceTestContext(snmp_device.SNMP, properties=self.snmp_properties, process=True) as proxy: + with DeviceTestContext(snmp_device.SNMP, properties=self.SNMP_PROPERTIES, process=True) as proxy: mib_dir = proxy.get_mib_dir() - self.assertEqual(mib_dir, f"{path.dirname(snmp_device.__file__)}/{self.snmp_properties['SNMP_rel_mib_dir']}") + self.assertEqual(mib_dir, f"{path.dirname(snmp_device.__file__)}/{self.SNMP_PROPERTIES['SNMP_rel_mib_dir']}") diff --git a/tangostationcontrol/tangostationcontrol/test/toolkit/test_archiver_configurator.py b/tangostationcontrol/tangostationcontrol/test/toolkit/test_archiver_configurator.py index 28465a5b48b65417505fb6491d320e033abc6414..aaf8a622faf044362c2ff13b931ecad30b8bd327 100644 --- a/tangostationcontrol/tangostationcontrol/test/toolkit/test_archiver_configurator.py +++ b/tangostationcontrol/tangostationcontrol/test/toolkit/test_archiver_configurator.py @@ -14,37 +14,37 @@ import pkg_resources class TestArchiverConfigurator(base.TestCase): - device_name = 'STAT/RECV/1' - attribute_name = 'ant_mask_rw' - prod_config_dict = json.load(pkg_resources.resource_stream('tangostationcontrol.toolkit', f'archiver_config/lofar2_prod.json')) - dev_config_dict = json.load(pkg_resources.resource_stream('tangostationcontrol.toolkit', f'archiver_config/lofar2_dev.json')) - dev_suffixes = dev_config_dict['global']['suffixes'] - prod_suffixes = prod_config_dict['global']['suffixes'] - prod_infixes = prod_config_dict['global']['infixes'] + DEVICE_NAME = 'STAT/RECV/1' + ATTRIBUTE_NAME = 'ant_mask_rw' + PROD_CONFIG_DICT = json.load(pkg_resources.resource_stream('tangostationcontrol.toolkit', f'archiver_config/lofar2_prod.json')) + DEV_CONFIG_DICT = json.load(pkg_resources.resource_stream('tangostationcontrol.toolkit', f'archiver_config/lofar2_dev.json')) + DEV_SUFFIXES = DEV_CONFIG_DICT['global']['suffixes'] + PROD_SUFFIXES = PROD_CONFIG_DICT['global']['suffixes'] + PROD_INFIXES = PROD_CONFIG_DICT['global']['infixes'] def test_get_parameters_from_attribute(self): """Test if the attribute archiving parameters are correctly retrieved from the JSON config file""" - self.assertIsNotNone(self.dev_config_dict) - archive_period, event_period, abs_change, rel_change = get_parameters_from_attribute(self.device_name, self.attribute_name, - self.dev_config_dict) - self.assertEqual(archive_period,int(self.dev_suffixes[2]['archive_period'])) - self.assertEqual(event_period,int(self.dev_suffixes[2]['event_period'])) - self.assertEqual(abs_change,float(self.dev_suffixes[2]['abs_change'])) - self.assertEqual(rel_change, self.dev_suffixes[2]['rel_change'] and int(self.dev_suffixes[2]['rel_change'])) + self.assertIsNotNone(self.DEV_CONFIG_DICT) + archive_period, event_period, abs_change, rel_change = get_parameters_from_attribute(self.DEVICE_NAME, self.ATTRIBUTE_NAME, + self.DEV_CONFIG_DICT) + self.assertEqual(archive_period,int(self.DEV_SUFFIXES[2]['archive_period'])) + self.assertEqual(event_period,int(self.DEV_SUFFIXES[2]['event_period'])) + self.assertEqual(abs_change,float(self.DEV_SUFFIXES[2]['abs_change'])) + self.assertEqual(rel_change, self.DEV_SUFFIXES[2]['rel_change'] and int(self.DEV_SUFFIXES[2]['rel_change'])) """Test if the attribute archiving parameters are correctly retrieved from the infixes list (production environment)""" attribute_name = 'rcu_temp_r' # 'TEMP' is in the infixes list - archive_period, event_period, abs_change, rel_change = get_parameters_from_attribute(self.device_name, attribute_name, - self.prod_config_dict) - self.assertEqual(archive_period,int(self.prod_infixes[2]['archive_period'])) - self.assertEqual(event_period,int(self.prod_infixes[2]['event_period'])) - self.assertEqual(abs_change,float(self.prod_infixes[2]['abs_change'])) - self.assertEqual(rel_change,self.prod_infixes[2]['rel_change'] and int(self.prod_infixes[2]['rel_change'])) + archive_period, event_period, abs_change, rel_change = get_parameters_from_attribute(self.DEVICE_NAME, attribute_name, + self.PROD_CONFIG_DICT) + self.assertEqual(archive_period,int(self.PROD_INFIXES[2]['archive_period'])) + self.assertEqual(event_period,int(self.PROD_INFIXES[2]['event_period'])) + self.assertEqual(abs_change,float(self.PROD_INFIXES[2]['abs_change'])) + self.assertEqual(rel_change,self.PROD_INFIXES[2]['rel_change'] and int(self.PROD_INFIXES[2]['rel_change'])) def test_get_global_env_parameters(self): """Test if the include attribute list is correctly retrieved from the JSON config file""" - self.assertIsNotNone(self.prod_config_dict) - polling_time, archive_abs_change, archive_rel_change, archive_period, event_period, strategy = get_global_env_parameters(self.prod_config_dict) + self.assertIsNotNone(self.PROD_CONFIG_DICT) + polling_time, archive_abs_change, archive_rel_change, archive_period, event_period, strategy = get_global_env_parameters(self.PROD_CONFIG_DICT) self.assertEqual(type(polling_time),int) self.assertEqual(type(archive_abs_change), int) self.assertEqual(f"{type(archive_rel_change)}", f"<class 'NoneType'>") diff --git a/tangostationcontrol/tangostationcontrol/test/toolkit/test_archiver_util.py b/tangostationcontrol/tangostationcontrol/test/toolkit/test_archiver_util.py index 842aff8367ab78a30ba55eb31bd48fcb29193d8e..1d2f92b754620969a63669ea27a0e075381efb88 100644 --- a/tangostationcontrol/tangostationcontrol/test/toolkit/test_archiver_util.py +++ b/tangostationcontrol/tangostationcontrol/test/toolkit/test_archiver_util.py @@ -12,28 +12,28 @@ from tangostationcontrol.toolkit.archiver_util import get_attribute_from_fqdn, s class TestArchiverUtil(base.TestCase): - device_name = 'STAT/RECV/1' - attribute_name = 'ant_mask_rw' + DEVICE_NAME = 'STAT/RECV/1' + ATTRIBUTE_NAME = 'ant_mask_rw' def test_get_attribute_from_fqdn(self): """Test if a Tango attribute name is correctly retrieved from a Tango FQDN""" - fqdn = f"tango://databaseds:10000/{self.device_name}/{self.attribute_name}" + fqdn = f"tango://databaseds:10000/{self.DEVICE_NAME}/{self.ATTRIBUTE_NAME}" self.assertEqual('STAT/RECV/1/ant_mask_rw', get_attribute_from_fqdn(fqdn)) def test_device_fqdn(self): """Test if a device name is correctly converted in a Tango FQDN""" - self.assertEqual(f"tango://databaseds:10000/{self.device_name}".lower(), device_fqdn(self.device_name, "databaseds:10000")) + self.assertEqual(f"tango://databaseds:10000/{self.DEVICE_NAME}".lower(), device_fqdn(self.DEVICE_NAME, "databaseds:10000")) def test_attribute_fqdn(self): """Test if an attribute name is correctly converted in a Tango FQDN""" - self.assertEqual(f"tango://databaseds:10000/{self.device_name}/{self.attribute_name}".lower(), - attribute_fqdn(f"{self.device_name}/{self.attribute_name}", "databaseds:10000")) - self.assertRaises(ValueError, lambda: attribute_fqdn(self.attribute_name)) + self.assertEqual(f"tango://databaseds:10000/{self.DEVICE_NAME}/{self.ATTRIBUTE_NAME}".lower(), + attribute_fqdn(f"{self.DEVICE_NAME}/{self.ATTRIBUTE_NAME}", "databaseds:10000")) + self.assertRaises(ValueError, lambda: attribute_fqdn(self.ATTRIBUTE_NAME)) def test_split_tango_name(self): """Test if the Tango full qualified domain names are correctly splitted""" - self.assertEqual(('STAT','RECV','1'), split_tango_name(self.device_name, 'device')) - self.assertEqual(('STAT','RECV','1', 'ant_mask_rw'), split_tango_name(f"{self.device_name}/{self.attribute_name}", 'attribute')) + self.assertEqual(('STAT','RECV','1'), split_tango_name(self.DEVICE_NAME, 'device')) + self.assertEqual(('STAT','RECV','1', 'ant_mask_rw'), split_tango_name(f"{self.DEVICE_NAME}/{self.ATTRIBUTE_NAME}", 'attribute')) def test_get_size_from_datatype(self): """Test if the bytesize of a certain datatype is correctly retrieved""" diff --git a/tangostationcontrol/tangostationcontrol/toolkit/archiver.py b/tangostationcontrol/tangostationcontrol/toolkit/archiver.py index 610ae3c0554e9e5bfbc1bd0df6fe900a863296b1..4e40ebeab891fca63f20708348f456cc4dd6de5c 100644 --- a/tangostationcontrol/tangostationcontrol/toolkit/archiver.py +++ b/tangostationcontrol/tangostationcontrol/toolkit/archiver.py @@ -55,12 +55,12 @@ class Archiver(): """ # Global environment variables set by configuration file - global_polling_time = 1000 - global_archive_abs_change = 1 - global_archive_rel_change = None - global_archive_period = 10000 # 3600000 (prod) - global_event_period = 1000 # 60000 (prod) - global_strategy = 'RUN' + GLOBAL_POLLING_TIME = 1000 + GLOBAL_ARCHIVE_ABS_CHANGE = 1 + GLOBAL_ARCHIVE_REL_CHANGE = None + GLOBAL_ARCHIVE_PERIOD = 10000 # 3600000 (prod) + GLOBAL_EVENT_PERIOD = 1000 # 60000 (prod) + GLOBAL_STRATEGY = 'RUN' def __init__(self, cm_name: str = 'archiving/hdbppts/confmanager01', context: str = 'RUN'): self.cm_name = cm_name @@ -122,7 +122,7 @@ class Archiver(): Apply the customized strategy defined by the given archiver configuration. """ # Set global development env variables - self.global_polling_time, self.global_archive_abs_change, self.global_archive_rel_change, self.global_archive_period, self.global_event_period, self.global_strategy = get_global_env_parameters(config_dict) + self.GLOBAL_POLLING_TIME, self.GLOBAL_ARCHIVE_ABS_CHANGE, self.GLOBAL_ARCHIVE_REL_CHANGE, self.GLOBAL_ARCHIVE_PERIOD, self.GLOBAL_EVENT_PERIOD, self.GLOBAL_STRATEGY = get_global_env_parameters(config_dict) # Set devices archiving env_dict = config_dict['devices'] # Check if device has more than one member (domain/family/*) @@ -164,11 +164,11 @@ class Archiver(): archive_period, event_period, abs_change, rel_change = get_parameters_from_attribute(device,att,config_dict) att_fqname = attribute_fqdn(att) # Add the attribute to the archiver setting either specific or global parameters - self.add_attribute_to_archiver(att_fqname, self.global_polling_time, archive_period or self.global_archive_period, - self.global_strategy, abs_change or self.global_archive_abs_change, - rel_change or self.global_archive_rel_change) - self.add_attributes_by_device(device, self.global_archive_period, self.global_archive_abs_change, - self.global_archive_rel_change, exclude=exclude_att_list) + self.add_attribute_to_archiver(att_fqname, self.GLOBAL_POLLING_TIME, archive_period or self.GLOBAL_ARCHIVE_PERIOD, + self.GLOBAL_STRATEGY, abs_change or self.GLOBAL_ARCHIVE_ABS_CHANGE, + rel_change or self.GLOBAL_ARCHIVE_REL_CHANGE) + self.add_attributes_by_device(device, self.GLOBAL_ARCHIVE_PERIOD, self.GLOBAL_ARCHIVE_ABS_CHANGE, + self.GLOBAL_ARCHIVE_REL_CHANGE, exclude=exclude_att_list) # Remove attributes by custom configuration if already present # The following cycle is a security check in the special case that an attribute is in the # included list in DEV mode, and in the excluded list in PROD mode