test/vpp_policer.py

   1 from vpp_object import VppObject
   2 from vpp_ip import INVALID_INDEX
   3 from enum import Enum
   4
   5
   6 class Dir(Enum):
   7     RX = 0
   8     TX = 1
   9
  10
  11 class PolicerAction:
  12     """sse2 qos action"""
  13
  14     def __init__(self, type, dscp):
  15         self.type = type
  16         self.dscp = dscp
  17
  18     def encode(self):
  19         return {"type": self.type, "dscp": self.dscp}
  20
  21
  22 class VppPolicer(VppObject):
  23     """Policer"""
  24
  25     def __init__(
  26         self,
  27         test,
  28         name,
  29         cir,
  30         eir,
  31         commited_burst,
  32         excess_burst,
  33         rate_type=0,
  34         round_type=0,
  35         type=0,
  36         color_aware=False,
  37         conform_action=PolicerAction(1, 0),
  38         exceed_action=PolicerAction(0, 0),
  39         violate_action=PolicerAction(0, 0),
  40     ):
  41         self._test = test
  42         self.name = name
  43         self.cir = cir
  44         self.eir = eir
  45         self.commited_burst = commited_burst
  46         self.excess_burst = excess_burst
  47         self.rate_type = rate_type
  48         self.round_type = round_type
  49         self.type = 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
  55
  56     @property
  57     def policer_index(self):
  58         return self._policer_index
  59
  60     def add_vpp_config(self):
  61         r = self._test.vapi.policer_add_del(
  62             name=self.name,
  63             cir=self.cir,
  64             eir=self.eir,
  65             cb=self.commited_burst,
  66             eb=self.excess_burst,
  67             rate_type=self.rate_type,
  68             round_type=self.round_type,
  69             type=self.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(),
  74         )
  75         self._test.registry.register(self, self._test.logger)
  76         self._policer_index = r.policer_index
  77         return self
  78
  79     def remove_vpp_config(self):
  80         self._test.vapi.policer_add_del(is_add=False, name=self.name)
  81         self._policer_index = INVALID_INDEX
  82
  83     def bind_vpp_config(self, worker, bind):
  84         self._test.vapi.policer_bind(
  85             name=self.name, worker_index=worker, bind_enable=bind
  86         )
  87
  88     def apply_vpp_config(self, if_index, dir: Dir, apply):
  89         if dir == Dir.RX:
  90             self._test.vapi.policer_input(
  91                 name=self.name, sw_if_index=if_index, apply=apply
  92             )
  93         else:
  94             self._test.vapi.policer_output(
  95                 name=self.name, sw_if_index=if_index, apply=apply
  96             )
  97
  98     def query_vpp_config(self):
  99         dump = self._test.vapi.policer_dump(match_name_valid=True, match_name=self.name)
 100         for policer in dump:
 101             if policer.name == self.name:
 102                 return True
 103         return False
 104
 105     def object_id(self):
 106         return "policer-%s" % (self.name)
 107
 108     def get_stats(self, worker=None):
 109         conform = self._test.statistics.get_counter("/net/policer/conform")
 110         exceed = self._test.statistics.get_counter("/net/policer/exceed")
 111         violate = self._test.statistics.get_counter("/net/policer/violate")
 112
 113         counters = {"conform": conform, "exceed": exceed, "violate": violate}
 114
 115         total = {}
 116         for name, c in counters.items():
 117             total[f"{name}_packets"] = 0
 118             total[f"{name}_bytes"] = 0
 119             for i in range(len(c)):
 120                 t = c[i]
 121                 if worker is not None and i != worker + 1:
 122                     continue
 123                 stat_index = self._policer_index
 124                 total[f"{name}_packets"] += t[stat_index]["packets"]
 125                 total[f"{name}_bytes"] += t[stat_index]["bytes"]
 126
 127         return total