tests: replace pycodestyle with black
[vpp.git] / test / test_policer_input.py
index 9a4266c..9d44fc1 100644 (file)
@@ -14,7 +14,8 @@ NUM_PKTS = 67
 
 
 class TestPolicerInput(VppTestCase):
-    """ Policer on an interface """
+    """Policer on an interface"""
+
     vpp_worker_count = 2
 
     def setUp(self):
@@ -26,11 +27,12 @@ class TestPolicerInput(VppTestCase):
             i.config_ip4()
             i.resolve_arp()
 
-        self.pkt = (Ether(src=self.pg0.remote_mac,
-                          dst=self.pg0.local_mac) /
-                    IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
-                    UDP(sport=1234, dport=1234) /
-                    Raw(b'\xa5' * 100))
+        self.pkt = (
+            Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
+            / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
+            / UDP(sport=1234, dport=1234)
+            / Raw(b"\xa5" * 100)
+        )
 
     def tearDown(self):
         for i in self.pg_interfaces:
@@ -42,17 +44,22 @@ class TestPolicerInput(VppTestCase):
         pkts = self.pkt * NUM_PKTS
 
         action_tx = PolicerAction(
-            VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT,
-            0)
-        policer = VppPolicer(self, "pol1", 80, 0, 1000, 0,
-                             conform_action=action_tx,
-                             exceed_action=action_tx,
-                             violate_action=action_tx)
+            VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT, 0
+        )
+        policer = VppPolicer(
+            self,
+            "pol1",
+            80,
+            0,
+            1000,
+            0,
+            conform_action=action_tx,
+            exceed_action=action_tx,
+            violate_action=action_tx,
+        )
         policer.add_vpp_config()
 
-        sw_if_index = (self.pg0.sw_if_index
-                       if dir == Dir.RX
-                       else self.pg1.sw_if_index)
+        sw_if_index = self.pg0.sw_if_index if dir == Dir.RX else self.pg1.sw_if_index
 
         # Start policing on pg0
         policer.apply_vpp_config(sw_if_index, dir, True)
@@ -61,9 +68,9 @@ class TestPolicerInput(VppTestCase):
         stats = policer.get_stats()
 
         # Single rate, 2 colour policer - expect conform, violate but no exceed
-        self.assertGreater(stats['conform_packets'], 0)
-        self.assertEqual(stats['exceed_packets'], 0)
-        self.assertGreater(stats['violate_packets'], 0)
+        self.assertGreater(stats["conform_packets"], 0)
+        self.assertEqual(stats["exceed_packets"], 0)
+        self.assertGreater(stats["violate_packets"], 0)
 
         # Stop policing on pg0
         policer.apply_vpp_config(sw_if_index, dir, False)
@@ -78,28 +85,33 @@ class TestPolicerInput(VppTestCase):
         policer.remove_vpp_config()
 
     def test_policer_input(self):
-        """ Input Policing """
+        """Input Policing"""
         self.policer_interface_test(Dir.RX)
 
     def test_policer_output(self):
-        """ Output Policing """
+        """Output Policing"""
         self.policer_interface_test(Dir.TX)
 
     def policer_handoff_test(self, dir: Dir):
         pkts = self.pkt * NUM_PKTS
 
         action_tx = PolicerAction(
-            VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT,
-            0)
-        policer = VppPolicer(self, "pol2", 80, 0, 1000, 0,
-                             conform_action=action_tx,
-                             exceed_action=action_tx,
-                             violate_action=action_tx)
+            VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT, 0
+        )
+        policer = VppPolicer(
+            self,
+            "pol2",
+            80,
+            0,
+            1000,
+            0,
+            conform_action=action_tx,
+            exceed_action=action_tx,
+            violate_action=action_tx,
+        )
         policer.add_vpp_config()
 
-        sw_if_index = (self.pg0.sw_if_index
-                       if dir == Dir.RX
-                       else self.pg1.sw_if_index)
+        sw_if_index = self.pg0.sw_if_index if dir == Dir.RX else self.pg1.sw_if_index
 
         # Bind the policer to worker 1
         policer.bind_vpp_config(1, True)
@@ -119,9 +131,9 @@ class TestPolicerInput(VppTestCase):
         self.assertEqual(stats, stats1)
 
         # Worker 0, should have handed everything off
-        self.assertEqual(stats0['conform_packets'], 0)
-        self.assertEqual(stats0['exceed_packets'], 0)
-        self.assertEqual(stats0['violate_packets'], 0)
+        self.assertEqual(stats0["conform_packets"], 0)
+        self.assertEqual(stats0["exceed_packets"], 0)
+        self.assertEqual(stats0["violate_packets"], 0)
 
         # Unbind the policer from worker 1 and repeat
         policer.bind_vpp_config(1, False)
@@ -137,19 +149,23 @@ class TestPolicerInput(VppTestCase):
         stats0 = policer.get_stats(worker=0)
         stats1 = policer.get_stats(worker=1)
 
-        self.assertGreater(stats0['conform_packets'], 0)
-        self.assertEqual(stats0['exceed_packets'], 0)
-        self.assertGreater(stats0['violate_packets'], 0)
+        self.assertGreater(stats0["conform_packets"], 0)
+        self.assertEqual(stats0["exceed_packets"], 0)
+        self.assertGreater(stats0["violate_packets"], 0)
 
-        self.assertGreater(stats1['conform_packets'], 0)
-        self.assertEqual(stats1['exceed_packets'], 0)
-        self.assertGreater(stats1['violate_packets'], 0)
+        self.assertGreater(stats1["conform_packets"], 0)
+        self.assertEqual(stats1["exceed_packets"], 0)
+        self.assertGreater(stats1["violate_packets"], 0)
 
-        self.assertEqual(stats0['conform_packets'] + stats1['conform_packets'],
-                         stats['conform_packets'])
+        self.assertEqual(
+            stats0["conform_packets"] + stats1["conform_packets"],
+            stats["conform_packets"],
+        )
 
-        self.assertEqual(stats0['violate_packets'] + stats1['violate_packets'],
-                         stats['violate_packets'])
+        self.assertEqual(
+            stats0["violate_packets"] + stats1["violate_packets"],
+            stats["violate_packets"],
+        )
 
         # Stop policing on pg0
         policer.apply_vpp_config(sw_if_index, dir, False)
@@ -157,13 +173,13 @@ class TestPolicerInput(VppTestCase):
         policer.remove_vpp_config()
 
     def test_policer_handoff_input(self):
-        """ Worker thread handoff policer input"""
+        """Worker thread handoff policer input"""
         self.policer_handoff_test(Dir.RX)
 
     def test_policer_handoff_output(self):
-        """ Worker thread handoff policer output"""
+        """Worker thread handoff policer output"""
         self.policer_handoff_test(Dir.TX)
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     unittest.main(testRunner=VppTestRunner)