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
  12
  13 NUM_PKTS = 67
  14
  15
  16 class TestPolicerInput(VppTestCase):
  17     """ Policer on an input interface """
  18     vpp_worker_count = 2
  19
  20     def setUp(self):
  21         super(TestPolicerInput, self).setUp()
  22
  23         self.create_pg_interfaces(range(2))
  24         for i in self.pg_interfaces:
  25             i.admin_up()
  26             i.config_ip4()
  27             i.resolve_arp()
  28
  29         self.pkt = (Ether(src=self.pg0.remote_mac,
  30                           dst=self.pg0.local_mac) /
  31                     IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
  32                     UDP(sport=1234, dport=1234) /
  33                     Raw(b'\xa5' * 100))
  34
  35     def tearDown(self):
  36         for i in self.pg_interfaces:
  37             i.unconfig_ip4()
  38             i.admin_down()
  39         super(TestPolicerInput, self).tearDown()
  40
  41     def test_policer_input(self):
  42         """ Input Policing """
  43         pkts = self.pkt * NUM_PKTS
  44
  45         action_tx = PolicerAction(
  46             VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT,
  47             0)
  48         policer = VppPolicer(self, "pol1", 80, 0, 1000, 0,
  49                              conform_action=action_tx,
  50                              exceed_action=action_tx,
  51                              violate_action=action_tx)
  52         policer.add_vpp_config()
  53
  54         # Start policing on pg0
  55         policer.apply_vpp_config(self.pg0.sw_if_index, True)
  56
  57         rx = self.send_and_expect(self.pg0, pkts, self.pg1, worker=0)
  58         stats = policer.get_stats()
  59
  60         # Single rate, 2 colour policer - expect conform, violate but no exceed
  61         self.assertGreater(stats['conform_packets'], 0)
  62         self.assertEqual(stats['exceed_packets'], 0)
  63         self.assertGreater(stats['violate_packets'], 0)
  64
  65         # Stop policing on pg0
  66         policer.apply_vpp_config(self.pg0.sw_if_index, False)
  67
  68         rx = self.send_and_expect(self.pg0, pkts, self.pg1, worker=0)
  69
  70         statsnew = policer.get_stats()
  71
  72         # No new packets counted
  73         self.assertEqual(stats, statsnew)
  74
  75         policer.remove_vpp_config()
  76
  77     def test_policer_handoff(self):
  78         """ Worker thread handoff """
  79         pkts = self.pkt * NUM_PKTS
  80
  81         action_tx = PolicerAction(
  82             VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT,
  83             0)
  84         policer = VppPolicer(self, "pol2", 80, 0, 1000, 0,
  85                              conform_action=action_tx,
  86                              exceed_action=action_tx,
  87                              violate_action=action_tx)
  88         policer.add_vpp_config()
  89
  90         # Bind the policer to worker 1
  91         policer.bind_vpp_config(1, True)
  92
  93         # Start policing on pg0
  94         policer.apply_vpp_config(self.pg0.sw_if_index, True)
  95
  96         for worker in [0, 1]:
  97             self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
  98             self.logger.debug(self.vapi.cli("show trace max 100"))
  99
 100         stats = policer.get_stats()
 101         stats0 = policer.get_stats(worker=0)
 102         stats1 = policer.get_stats(worker=1)
 103
 104         # Worker 1, should have done all the policing
 105         self.assertEqual(stats, stats1)
 106
 107         # Worker 0, should have handed everything off
 108         self.assertEqual(stats0['conform_packets'], 0)
 109         self.assertEqual(stats0['exceed_packets'], 0)
 110         self.assertEqual(stats0['violate_packets'], 0)
 111
 112         # Unbind the policer from worker 1 and repeat
 113         policer.bind_vpp_config(1, False)
 114         for worker in [0, 1]:
 115             self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
 116             self.logger.debug(self.vapi.cli("show trace max 100"))
 117
 118         # The policer should auto-bind to worker 0 when packets arrive
 119         stats = policer.get_stats()
 120
 121         # The 2 workers should now have policed the same amount
 122         stats = policer.get_stats()
 123         stats0 = policer.get_stats(worker=0)
 124         stats1 = policer.get_stats(worker=1)
 125
 126         self.assertGreater(stats0['conform_packets'], 0)
 127         self.assertEqual(stats0['exceed_packets'], 0)
 128         self.assertGreater(stats0['violate_packets'], 0)
 129
 130         self.assertGreater(stats1['conform_packets'], 0)
 131         self.assertEqual(stats1['exceed_packets'], 0)
 132         self.assertGreater(stats1['violate_packets'], 0)
 133
 134         self.assertEqual(stats0['conform_packets'] + stats1['conform_packets'],
 135                          stats['conform_packets'])
 136
 137         self.assertEqual(stats0['violate_packets'] + stats1['violate_packets'],
 138                          stats['violate_packets'])
 139
 140         # Stop policing on pg0
 141         policer.apply_vpp_config(self.pg0.sw_if_index, False)
 142
 143         policer.remove_vpp_config()
 144
 145 if __name__ == '__main__':
 146     unittest.main(testRunner=VppTestRunner)