summaryrefslogtreecommitdiffstats
path: root/test/test_policer_input.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/test_policer_input.py')
-rw-r--r--test/test_policer_input.py106
1 files changed, 61 insertions, 45 deletions
diff --git a/test/test_policer_input.py b/test/test_policer_input.py
index 9a4266ceb12..9d44fc1a21c 100644
--- a/test/test_policer_input.py
+++ b/test/test_policer_input.py
@@ -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)