L2SS-998: Auto format job for tox

tangostationcontrol/tangostationcontrol/integration_test/default/toolkit/test_archiver.py

@@ -24,21 +24,21 @@ class TestArchiver(BaseIntegrationTestCase):
         super().setUp()
         self.archiver = Archiver()
         self.assertIsNotNone(self.archiver)
-    
+
     def test_archiver_initialisation(self):
         """Test archiver main attributes"""            
         self.assertEqual(self.archiver.cm_name,"archiving/hdbppts/confmanager01")
         self.assertTrue(len(self.archiver.es_list))  # subscribers list not empty
-    
+
     def test_hdbpp_library(self):
         """Test if the correct hdbpp library is retrieved (TimescaleDB)"""
         cm_name = self.archiver.cm_name
         self.assertEqual('libhdb++timescale.so', self.archiver.get_hdbpp_libname(cm_name))
-    
+
     def test_next_subscriber(self):
         """Test if there is an available subscriber"""
         self.assertIsNotNone(self.archiver.get_next_subscriber())
-    
+
     def test_archiver_configuration(self):
         """Test archiver configuration file"""
         dev_config_dict = self.archiver.get_configuration('lofar2_dev')
@@ -46,7 +46,7 @@ class TestArchiver(BaseIntegrationTestCase):
         self.assertIsNotNone(dev_config_dict)
         self.assertIsNotNone(prod_config_dict)
         self.assertNotEqual(dev_config_dict, prod_config_dict)
-    
+
     def test_archive_scalar_attribute(self):
         """Test if a scalar attribute is correctly archived"""
         # Start RECV Device
@@ -67,7 +67,7 @@ class TestArchiver(BaseIntegrationTestCase):
         time.sleep(3)
         # Test if the attribute has been correctly added to event subscriber
         self.assertTrue(self.archiver.is_attribute_archived(attribute_fqdn(attr_fullname)))
-        
+
         # Retrieve data from DB views
         self.retriever = RetrieverTimescale()
         self.assertIsNotNone(self.retriever)
@@ -79,15 +79,15 @@ class TestArchiver(BaseIntegrationTestCase):
         self.assertEqual(datetime,type(item.data_time))           # column datetime
         self.assertEqual(int,type(item.value))                    # column value
 
-        
+
         # Remove attribute at the end of the test
         self.archiver.remove_attribute_from_archiver(attr_fullname)
         time.sleep(3)
         # Test if the attribute has been correctly removed
         self.assertFalse(self.archiver.is_attribute_archived(attribute_fqdn(attr_fullname)))
-        
+
         recv_proxy.off()
-    
+
     def test_archive_array_attribute(self):
         """Test if an array attribute is correctly archived"""
         # Start SDP Device
@@ -100,7 +100,7 @@ class TestArchiver(BaseIntegrationTestCase):
         sdp_proxy.set_defaults()
         sdp_proxy.on()
         self.assertEqual(DevState.ON, sdp_proxy.state())
-       
+
         polling_period=1000
         archive_event_period=3000
         attr_fullname = 'stat/sdp/1/fpga_temp_r'  # double
@@ -121,15 +121,15 @@ class TestArchiver(BaseIntegrationTestCase):
         self.assertEqual(int,type(item.x))                                  # column index
         self.assertEqual(float,type(item.value))                            # column value
 
-        
+
         # Remove attribute at the end of the test
         self.archiver.remove_attribute_from_archiver(attr_fullname)
         time.sleep(3)
         # Test if the attribute has been correctly removed
         self.assertFalse(self.archiver.is_attribute_archived(attribute_fqdn(attr_fullname)))
-        
+
         sdp_proxy.off()
-    
+
     def test_archive_image_boolean_attribute(self):
         """Test if a boolean image attribute is correctly archived"""
         # Start RECV Device
@@ -142,7 +142,7 @@ class TestArchiver(BaseIntegrationTestCase):
         recv_proxy.set_defaults()
         recv_proxy.on()
         self.assertEqual(DevState.ON, recv_proxy.state())
-        
+
         polling_period=1000
         archive_event_period=5000
         attr_fullname = 'stat/recv/1/hbat_pwr_on_rw'  # boolean 96x32
@@ -165,15 +165,15 @@ class TestArchiver(BaseIntegrationTestCase):
         self.assertEqual(int,type(item.value))                              # column value (bool stored as int)
         self.assertLessEqual(item.value,1)                                  # column value (must be 0 or 1)
 
-        
+
         # Remove attribute at the end of the test
         self.archiver.remove_attribute_from_archiver(attr_fullname)
         time.sleep(3)
         # Test if the attribute has been correctly removed
         self.assertFalse(self.archiver.is_attribute_archived(attribute_fqdn(attr_fullname)))
-        
+
         recv_proxy.off()
-    
+
     def test_get_maximum_device_load(self):
         """ Test if the maximum device load is correctly computed """
         # Start RECV Device
@@ -194,7 +194,7 @@ class TestArchiver(BaseIntegrationTestCase):
         time.sleep(3)
         max_load = self.archiver.get_maximum_device_load(device_name)
         self.assertGreater(max_load,0)
-    
+
     def test_archive_right_number_of_attributes(self):
         """ Test if the right number of attributes are archived, following the JSON configuration file"""
         # Start SDP Device
@@ -208,7 +208,7 @@ class TestArchiver(BaseIntegrationTestCase):
         sdp_proxy.set_defaults()
         sdp_proxy.on()
         self.assertEqual(DevState.ON, sdp_proxy.state())
-        
+
         config_dict = self.archiver.get_configuration('lofar2_dev')
         self.archiver.apply_configuration(config_dict)
         # 4 SDP_attributes contain the suffix '_error_R'
@@ -219,11 +219,11 @@ class TestArchiver(BaseIntegrationTestCase):
         archived_attrs = self.archiver.get_subscriber_attributes()
         sdp_archived_attrs = [a for a in archived_attrs if device_name.lower() in a]
         self.assertEqual(9, len(sdp_archived_attrs))
-    
+
     def _wait_for_archiving(self, attr_fullname: str, archive_event_period: int, max_wait: int = 10):
         wait = 0
         records = self.retriever.get_lofar_attribute(attr_fullname)
-        while (not(len(records)>0) and wait<max_wait):
+        while (not (len(records) > 0) and wait < max_wait):
             time.sleep(archive_event_period)
             records = self.retriever.get_lofar_attribute(attr_fullname)
             wait+=1

tangostationcontrol/test-requirements.txt

@@ -13,14 +13,13 @@ flake8>=3.8.0 # MIT
 flake8-bugbear>=22.1.11 # MIT
 flake8-breakpoint>=1.1.0 # MIT
 flake8-debugger>=4.0.0 #MIT
-flake8-mock>=0.3 #GPL
-hacking>=3.2.0,<3.3.0 # Apache-2.0
+autopep8>=1.7.0 # MIT
 python-subunit>=1.4.0 # Apache-2.0/BSD
 Pygments>=2.6.0 # BSD
 stestr>=3.0.0 # Apache-2.0
 testscenarios>=0.5.0 # Apache-2.0/BSD
 testtools>=2.4.0 # MIT
-timeout-decorator>=0.5  # MIT
+timeout-decorator>=0.5.0  # MIT
 xenon>=0.8.0 # MIT
 prometheus_client # Apache-2.0
 python-logstash-async # MIT

tangostationcontrol/tox.ini

@@ -83,6 +83,11 @@ commands =
     {envpython} -m doc8 docs/source/ --ignore D001
     {envpython} -m flake8
 
+[testenv:format]
+commands =
+    {envpython} -m autopep8 --version
+    {envpython} -m autopep8 -v -aa --in-place --recursive tangostationcontrol/
+
 [testenv:bandit];
 ; B104: hardcoded_bind_all_interfaces
 ;           - We disable this warning as Docker serves as our firewall.