test/test_policer_input.py

   1 #!/usr/bin/env python3
   2 # Copyright (c) 2021 Graphiant, Inc.
   3
   4 import unittest
   5 import scapy.compat
   6 from scapy.layers.inet import IP, UDP
   7 from scapy.layers.l2 import Ether
   8 from scapy.packet import Raw
   9 from framework import VppTestCase, VppTestRunner
  10 from vpp_papi import VppEnum
  11 from vpp_policer import VppPolicer, PolicerAction, Dir
  12
  13 NUM_PKTS = 67
  14
  15
  16 class TestPolicerInput(VppTestCase):
  17     """Policer on an interface"""
  18
  19     vpp_worker_count = 2
  20
  21     def setUp(self):
  22         super(TestPolicerInput, self).setUp()
  23
  24         self.create_pg_interfaces(range(2))
  25         for i in self.pg_interfaces:
  26             i.admin_up()
  27             i.config_ip4()
  28             i.resolve_arp()
  29
  30         self.pkt = (
  31             Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
  32             / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
  33             / UDP(sport=1234, dport=1234)
  34             / Raw(b"\xa5" * 100)
  35         )
  36
  37     def tearDown(self):
  38         for i in self.pg_interfaces:
  39             i.unconfig_ip4()
  40             i.admin_down()
  41         super(TestPolicerInput, self).tearDown()
  42
  43     def policer_interface_test(self, dir: Dir):
  44         pkts = self.pkt * NUM_PKTS
  45
  46         action_tx = PolicerAction(
  47             VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT, 0
  48         )
  49         policer = VppPolicer(
  50             self,
  51             "pol1",
  52             80,
  53             0,
  54             1000,
  55             0,
  56             conform_action=action_tx,
  57             exceed_action=action_tx,
  58             violate_action=action_tx,
  59         )
  60         policer.add_vpp_config()
  61
  62         sw_if_index = self.pg0.sw_if_index if dir == Dir.RX else self.pg1.sw_if_index
  63
  64         # Start policing on pg0
  65         policer.apply_vpp_config(sw_if_index, dir, True)
  66
  67         rx = self.send_and_expect(self.pg0, pkts, self.pg1, worker=0)
  68         stats = policer.get_stats()
  69
  70         # Single rate, 2 colour policer - expect conform, violate but no exceed
  71         self.assertGreater(stats["conform_packets"], 0)
  72         self.assertEqual(stats["exceed_packets"], 0)
  73         self.assertGreater(stats["violate_packets"], 0)
  74
  75         # Stop policing on pg0
  76         policer.apply_vpp_config(sw_if_index, dir, False)
  77
  78         rx = self.send_and_expect(self.pg0, pkts, self.pg1, worker=0)
  79
  80         statsnew = policer.get_stats()
  81
  82         # No new packets counted
  83         self.assertEqual(stats, statsnew)
  84
  85         policer.remove_vpp_config()
  86
  87     def test_policer_input(self):
  88         """Input Policing"""
  89         self.policer_interface_test(Dir.RX)
  90
  91     def test_policer_output(self):
  92         """Output Policing"""
  93         self.policer_interface_test(Dir.TX)
  94
  95     def policer_handoff_test(self, dir: Dir):
  96         pkts = self.pkt * NUM_PKTS
  97
  98         action_tx = PolicerAction(
  99             VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT, 0
 100         )
 101         policer = VppPolicer(
 102             self,
 103             "pol2",
 104             80,
 105             0,
 106             1000,
 107             0,
 108             conform_action=action_tx,
 109             exceed_action=action_tx,
 110             violate_action=action_tx,
 111         )
 112         policer.add_vpp_config()
 113
 114         sw_if_index = self.pg0.sw_if_index if dir == Dir.RX else self.pg1.sw_if_index
 115
 116         # Bind the policer to worker 1
 117         policer.bind_vpp_config(1, True)
 118
 119         # Start policing on pg0
 120         policer.apply_vpp_config(sw_if_index, dir, True)
 121
 122         for worker in [0, 1]:
 123             self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
 124             self.logger.debug(self.vapi.cli("show trace max 100"))
 125
 126         stats = policer.get_stats()
 127         stats0 = policer.get_stats(worker=0)
 128         stats1 = policer.get_stats(worker=1)
 129
 130         # Worker 1, should have done all the policing
 131         self.assertEqual(stats, stats1)
 132
 133         # Worker 0, should have handed everything off
 134         self.assertEqual(stats0["conform_packets"], 0)
 135         self.assertEqual(stats0["exceed_packets"], 0)
 136         self.assertEqual(stats0["violate_packets"], 0)
 137
 138         # Unbind the policer from worker 1 and repeat
 139         policer.bind_vpp_config(1, False)
 140         for worker in [0, 1]:
 141             self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
 142             self.logger.debug(self.vapi.cli("show trace max 100"))
 143
 144         # The policer should auto-bind to worker 0 when packets arrive
 145         stats = policer.get_stats()
 146
 147         # The 2 workers should now have policed the same amount
 148         stats = policer.get_stats()
 149         stats0 = policer.get_stats(worker=0)
 150         stats1 = policer.get_stats(worker=1)
 151
 152         self.assertGreater(stats0["conform_packets"], 0)
 153         self.assertEqual(stats0["exceed_packets"], 0)
 154         self.assertGreater(stats0["violate_packets"], 0)
 155
 156         self.assertGreater(stats1["conform_packets"], 0)
 157         self.assertEqual(stats1["exceed_packets"], 0)
 158         self.assertGreater(stats1["violate_packets"], 0)
 159
 160         self.assertEqual(
 161             stats0["conform_packets"] + stats1["conform_packets"],
 162             stats["conform_packets"],
 163         )
 164
 165         self.assertEqual(
 166             stats0["violate_packets"] + stats1["violate_packets"],
 167             stats["violate_packets"],
 168         )
 169
 170         # Stop policing on pg0
 171         policer.apply_vpp_config(sw_if_index, dir, False)
 172
 173         policer.remove_vpp_config()
 174
 175     def test_policer_handoff_input(self):
 176         """Worker thread handoff policer input"""
 177         self.policer_handoff_test(Dir.RX)
 178
 179     def test_policer_handoff_output(self):
 180         """Worker thread handoff policer output"""
 181         self.policer_handoff_test(Dir.TX)
 182
 183
 184 if __name__ == "__main__":
 185     unittest.main(testRunner=VppTestRunner)