1 from vpp_object import VppObject
2 from vpp_ip import INVALID_INDEX
14 def __init__(self, type, dscp):
19 return {"type": self.type, "dscp": self.dscp}
22 class VppPolicer(VppObject):
37 conform_action=PolicerAction(1, 0),
38 exceed_action=PolicerAction(0, 0),
39 violate_action=PolicerAction(0, 0),
45 self.commited_burst = commited_burst
46 self.excess_burst = excess_burst
47 self.rate_type = rate_type
48 self.round_type = round_type
50 self.color_aware = color_aware
51 self.conform_action = conform_action
52 self.exceed_action = exceed_action
53 self.violate_action = violate_action
54 self._policer_index = INVALID_INDEX
57 def policer_index(self):
58 return self._policer_index
65 "cb": self.commited_burst,
66 "eb": self.excess_burst,
67 "rate_type": self.rate_type,
68 "round_type": self.round_type,
70 "color_aware": self.color_aware,
71 "conform_action": self.conform_action.encode(),
72 "exceed_action": self.exceed_action.encode(),
73 "violate_action": self.violate_action.encode(),
76 def add_vpp_config(self):
77 r = self._test.vapi.policer_add(name=self.name, infos=self.config)
78 self._test.registry.register(self, self._test.logger)
79 self._policer_index = r.policer_index
83 self._test.vapi.policer_update(
84 policer_index=self._policer_index, infos=self.config
87 def remove_vpp_config(self):
88 self._test.vapi.policer_del(policer_index=self._policer_index)
89 self._policer_index = INVALID_INDEX
91 def bind_vpp_config(self, worker, bind):
92 self._test.vapi.policer_bind_v2(
93 policer_index=self._policer_index, worker_index=worker, bind_enable=bind
96 def apply_vpp_config(self, if_index, dir: Dir, apply):
98 self._test.vapi.policer_input_v2(
99 policer_index=self._policer_index, sw_if_index=if_index, apply=apply
102 self._test.vapi.policer_output_v2(
103 policer_index=self._policer_index, sw_if_index=if_index, apply=apply
106 def query_vpp_config(self):
107 dump = self._test.vapi.policer_dump_v2(policer_index=self._policer_index)
109 if policer.name == self.name:
114 return "policer-%s" % (self.name)
116 def get_details(self):
117 dump = self._test.vapi.policer_dump_v2(policer_index=self._policer_index)
119 if policer.name == self.name:
121 raise self._test.vapi.VPPValueError("Missing policer")
123 def get_stats(self, worker=None):
124 conform = self._test.statistics.get_counter("/net/policer/conform")
125 exceed = self._test.statistics.get_counter("/net/policer/exceed")
126 violate = self._test.statistics.get_counter("/net/policer/violate")
128 counters = {"conform": conform, "exceed": exceed, "violate": violate}
131 for name, c in counters.items():
132 total[f"{name}_packets"] = 0
133 total[f"{name}_bytes"] = 0
134 for i in range(len(c)):
136 if worker is not None and i != worker + 1:
138 stat_index = self._policer_index
139 total[f"{name}_packets"] += t[stat_index]["packets"]
140 total[f"{name}_bytes"] += t[stat_index]["bytes"]