aboutsummaryrefslogtreecommitdiffstats
path: root/src/vnet/policer
diff options
context:
space:
mode:
Diffstat (limited to 'src/vnet/policer')
-rw-r--r--src/vnet/policer/test/test_policer.py117
-rw-r--r--src/vnet/policer/test/test_policer_input.py146
2 files changed, 0 insertions, 263 deletions
diff --git a/src/vnet/policer/test/test_policer.py b/src/vnet/policer/test/test_policer.py
deleted file mode 100644
index 6b15a0234a3..00000000000
--- a/src/vnet/policer/test/test_policer.py
+++ /dev/null
@@ -1,117 +0,0 @@
-#!/usr/bin/env python3
-# Copyright (c) 2021 Graphiant, Inc.
-
-import unittest
-
-from framework import VppTestCase, VppTestRunner
-from vpp_policer import VppPolicer, PolicerAction
-
-# Default for the tests is 10s of "Green" packets at 8Mbps, ie. 10M bytes.
-# The policer helper CLI "sends" 500 byte packets, so default is 20000.
-
-TEST_RATE = 8000 # kbps
-TEST_BURST = 10000 # ms
-
-CIR_OK = 8500 # CIR in kbps, above test rate
-CIR_LOW = 7000 # CIR in kbps, below test rate
-EIR_OK = 9000 # EIR in kbps, above test rate
-EIR_LOW = 7500 # EIR in kbps, below test rate
-
-NUM_PKTS = 20000
-
-CBURST = 100000 # Committed burst in bytes
-EBURST = 200000 # Excess burst in bytes
-
-
-class TestPolicer(VppTestCase):
- """ Policer Test Case """
-
- def run_policer_test(self, type, cir, cb, eir, eb, rate=8000, burst=10000,
- colour=0):
- """
- Configure a Policer and push traffic through it.
- """
- types = {
- '1R2C': 0,
- '1R3C': 1,
- '2R3C': 3,
- }
-
- pol_type = types.get(type)
- policer = VppPolicer(self, "pol1", cir, eir, cb, eb, rate_type=0,
- type=pol_type, color_aware=colour)
- policer.add_vpp_config()
-
- error = self.vapi.cli(
- f"test policing index {policer.policer_index} rate {rate} "
- f"burst {burst} colour {colour}")
-
- stats = policer.get_stats()
- policer.remove_vpp_config()
-
- return stats
-
- def test_policer_1r2c(self):
- """ Single rate, 2 colour policer """
- stats = self.run_policer_test("1R2C", CIR_OK, CBURST, 0, 0)
- self.assertEqual(stats['conform_packets'], NUM_PKTS)
-
- stats = self.run_policer_test("1R2C", CIR_LOW, CBURST, 0, 0)
- self.assertLess(stats['conform_packets'], NUM_PKTS)
- self.assertEqual(stats['exceed_packets'], 0)
- self.assertGreater(stats['violate_packets'], 0)
-
- stats = self.run_policer_test("1R2C", CIR_LOW, CBURST, 0, 0, colour=2)
- self.assertEqual(stats['violate_packets'], NUM_PKTS)
-
- def test_policer_1r3c(self):
- """ Single rate, 3 colour policer """
- stats = self.run_policer_test("1R3C", CIR_OK, CBURST, 0, 0)
- self.assertEqual(stats['conform_packets'], NUM_PKTS)
-
- stats = self.run_policer_test("1R3C", CIR_LOW, CBURST, 0, EBURST)
- self.assertLess(stats['conform_packets'], NUM_PKTS)
- self.assertGreater(stats['exceed_packets'], 0)
- self.assertGreater(stats['violate_packets'], 0)
-
- stats = self.run_policer_test("1R3C", CIR_LOW, CBURST, 0, EBURST,
- colour=1)
- self.assertEqual(stats['conform_packets'], 0)
- self.assertGreater(stats['exceed_packets'], 0)
- self.assertGreater(stats['violate_packets'], 0)
-
- stats = self.run_policer_test("1R3C", CIR_LOW, CBURST, 0, EBURST,
- colour=2)
- self.assertEqual(stats['violate_packets'], NUM_PKTS)
-
- def test_policer_2r3c(self):
- """ Dual rate, 3 colour policer """
- stats = self.run_policer_test("2R3C", CIR_OK, CBURST, EIR_OK, EBURST)
- self.assertEqual(stats['conform_packets'], NUM_PKTS)
-
- stats = self.run_policer_test("2R3C", CIR_LOW, CBURST, EIR_OK, EBURST)
- self.assertLess(stats['conform_packets'], NUM_PKTS)
- self.assertGreater(stats['exceed_packets'], 0)
- self.assertEqual(stats['violate_packets'], 0)
-
- stats = self.run_policer_test("2R3C", CIR_LOW, CBURST, EIR_LOW, EBURST)
- self.assertLess(stats['conform_packets'], NUM_PKTS)
- self.assertGreater(stats['exceed_packets'], 0)
- self.assertGreater(stats['violate_packets'], 0)
-
- stats = self.run_policer_test("2R3C", CIR_LOW, CBURST, EIR_OK, EBURST,
- colour=1)
- self.assertEqual(stats['exceed_packets'], NUM_PKTS)
-
- stats = self.run_policer_test("2R3C", CIR_LOW, CBURST, EIR_LOW, EBURST,
- colour=1)
- self.assertEqual(stats['conform_packets'], 0)
- self.assertGreater(stats['exceed_packets'], 0)
- self.assertGreater(stats['violate_packets'], 0)
-
- stats = self.run_policer_test("2R3C", CIR_LOW, CBURST, EIR_OK, EBURST,
- colour=2)
- self.assertEqual(stats['violate_packets'], NUM_PKTS)
-
-if __name__ == '__main__':
- unittest.main(testRunner=VppTestRunner)
diff --git a/src/vnet/policer/test/test_policer_input.py b/src/vnet/policer/test/test_policer_input.py
deleted file mode 100644
index c95f6643ff2..00000000000
--- a/src/vnet/policer/test/test_policer_input.py
+++ /dev/null
@@ -1,146 +0,0 @@
-#!/usr/bin/env python3
-# Copyright (c) 2021 Graphiant, Inc.
-
-import unittest
-import scapy.compat
-from scapy.layers.inet import IP, UDP
-from scapy.layers.l2 import Ether
-from scapy.packet import Raw
-from framework import VppTestCase, VppTestRunner
-from vpp_papi import VppEnum
-from vpp_policer import VppPolicer, PolicerAction
-
-NUM_PKTS = 67
-
-
-class TestPolicerInput(VppTestCase):
- """ Policer on an input interface """
- vpp_worker_count = 2
-
- def setUp(self):
- super(TestPolicerInput, self).setUp()
-
- self.create_pg_interfaces(range(2))
- for i in self.pg_interfaces:
- i.admin_up()
- 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))
-
- def tearDown(self):
- for i in self.pg_interfaces:
- i.unconfig_ip4()
- i.admin_down()
- super(TestPolicerInput, self).tearDown()
-
- def test_policer_input(self):
- """ Input Policing """
- 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)
- policer.add_vpp_config()
-
- # Start policing on pg0
- policer.apply_vpp_config(self.pg0.sw_if_index, True)
-
- rx = self.send_and_expect(self.pg0, pkts, self.pg1, worker=0)
- 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)
-
- # Stop policing on pg0
- policer.apply_vpp_config(self.pg0.sw_if_index, False)
-
- rx = self.send_and_expect(self.pg0, pkts, self.pg1, worker=0)
-
- statsnew = policer.get_stats()
-
- # No new packets counted
- self.assertEqual(stats, statsnew)
-
- policer.remove_vpp_config()
-
- def test_policer_handoff(self):
- """ Worker thread handoff """
- 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)
- policer.add_vpp_config()
-
- # Bind the policer to worker 1
- policer.bind_vpp_config(1, True)
-
- # Start policing on pg0
- policer.apply_vpp_config(self.pg0.sw_if_index, True)
-
- for worker in [0, 1]:
- self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
- self.logger.debug(self.vapi.cli("show trace max 100"))
-
- stats = policer.get_stats()
- stats0 = policer.get_stats(worker=0)
- stats1 = policer.get_stats(worker=1)
-
- # Worker 1, should have done all the policing
- 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)
-
- # Unbind the policer from worker 1 and repeat
- policer.bind_vpp_config(1, False)
- for worker in [0, 1]:
- self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
- self.logger.debug(self.vapi.cli("show trace max 100"))
-
- # The policer should auto-bind to worker 0 when packets arrive
- stats = policer.get_stats()
-
- # The 2 workers should now have policed the same amount
- stats = policer.get_stats()
- 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(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['violate_packets'] + stats1['violate_packets'],
- stats['violate_packets'])
-
- # Stop policing on pg0
- policer.apply_vpp_config(self.pg0.sw_if_index, False)
-
- policer.remove_vpp_config()
-
-if __name__ == '__main__':
- unittest.main(testRunner=VppTestRunner)