src/vnet/policer/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
  19     def setUp(self):
  20         super(TestPolicerInput, self).setUp()
  21
  22         self.create_pg_interfaces(range(2))
  23         for i in self.pg_interfaces:
  24             i.admin_up()
  25             i.config_ip4()
  26             i.resolve_arp()
  27
  28         self.pkt = (Ether(src=self.pg0.remote_mac,
  29                           dst=self.pg0.local_mac) /
  30                     IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
  31                     UDP(sport=1234, dport=1234) /
  32                     Raw(b'\xa5' * 100))
  33
  34     def tearDown(self):
  35         for i in self.pg_interfaces:
  36             i.unconfig_ip4()
  37             i.admin_down()
  38         super(TestPolicerInput, self).tearDown()
  39
  40     def test_policer_input(self):
  41         action_tx = PolicerAction(
  42             VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT,
  43             0)
  44         policer = VppPolicer(self, "pol1", 80, 0, 1000, 0,
  45                              conform_action=action_tx,
  46                              exceed_action=action_tx,
  47                              violate_action=action_tx)
  48         policer.add_vpp_config()
  49
  50         # Start policing on pg0
  51         policer.apply_vpp_config(self.pg0.sw_if_index, True)
  52
  53         rx = self.send_and_expect(self.pg0, self.pkt * NUM_PKTS, self.pg1)
  54         stats = policer.get_stats()
  55
  56         # Single rate, 2 colour policer - expect conform, violate but no exceed
  57         self.assertGreater(stats['conform_packets'], 0)
  58         self.assertEqual(stats['exceed_packets'], 0)
  59         self.assertGreater(stats['violate_packets'], 0)
  60
  61         # Stop policing on pg0
  62         policer.apply_vpp_config(self.pg0.sw_if_index, False)
  63
  64         rx = self.send_and_expect(self.pg0, self.pkt * NUM_PKTS, self.pg1)
  65         statsnew = policer.get_stats()
  66
  67         # No new packets counted
  68         self.assertEqual(stats, statsnew)
  69
  70         policer.remove_vpp_config()
  71
  72
  73 if __name__ == '__main__':
  74     unittest.main(testRunner=VppTestRunner)