1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
|
#!/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 setUp(self):
super(TestPolicer, self).setUp()
def tearDown(self):
super(TestPolicer, self).tearDown()
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)
|
