test/vpp_qos.py

   1 """
   2   QoS
   3
   4   object abstractions for representing QoS config VPP
   5 """
   6
   7 from vpp_object import VppObject
   8
   9
  10 class VppQosRecord(VppObject):
  11     """ QoS Record(ing) configuration """
  12
  13     def __init__(self, test, intf, source):
  14         self._test = test
  15         self.intf = intf
  16         self.source = source
  17
  18     def add_vpp_config(self):
  19         self._test.vapi.qos_record_enable_disable(
  20             enable=1,
  21             record={'sw_if_index': self.intf.sw_if_index,
  22                     'input_source': self.source})
  23         self._test.registry.register(self, self._test.logger)
  24         return self
  25
  26     def remove_vpp_config(self):
  27         self._test.vapi.qos_record_enable_disable(
  28             enable=0,
  29             record={'sw_if_index': self.intf.sw_if_index,
  30                     'input_source': self.source})
  31
  32     def query_vpp_config(self):
  33         rs = self._test.vapi.qos_record_dump()
  34
  35         for r in rs:
  36             if self.intf.sw_if_index == r.record.sw_if_index and \
  37                self.source == r.record.input_source:
  38                 return True
  39         return False
  40
  41     def object_id(self):
  42         return ("qos-record-%s-%d" % (self.intf, self.source))
  43
  44
  45 class VppQosEgressMap(VppObject):
  46     """ QoS Egress Map(ping) configuration """
  47
  48     def __init__(self, test, id, rows):
  49         self._test = test
  50         self.id = id
  51         self.rows = rows
  52
  53     def add_vpp_config(self):
  54         self._test.vapi.qos_egress_map_update(
  55             map={'id': self.id,
  56                  'rows': self.rows})
  57         self._test.registry.register(self, self._test.logger)
  58         return self
  59
  60     def remove_vpp_config(self):
  61         self._test.vapi.qos_egress_map_delete(id=self.id)
  62
  63     def query_vpp_config(self):
  64         rs = self._test.vapi.qos_egress_map_dump()
  65
  66         for r in rs:
  67             if self.id == r.map.id:
  68                 return True
  69         return False
  70
  71     def object_id(self):
  72         return ("qos-map-%d" % (self.id))
  73
  74
  75 class VppQosMark(VppObject):
  76     """ QoS Mark(ing) configuration """
  77
  78     def __init__(self, test, intf, map, source):
  79         self._test = test
  80         self.intf = intf
  81         self.source = source
  82         self.map = map
  83
  84     def add_vpp_config(self):
  85         self._test.vapi.qos_mark_enable_disable(
  86             enable=1,
  87             mark={'sw_if_index': self.intf.sw_if_index,
  88                   'map_id': self.map.id,
  89                   'output_source': self.source})
  90         self._test.registry.register(self, self._test.logger)
  91         return self
  92
  93     def remove_vpp_config(self):
  94         self._test.vapi.qos_mark_enable_disable(
  95             enable=0,
  96             mark={'sw_if_index': self.intf.sw_if_index,
  97                   'output_source': self.source})
  98
  99     def query_vpp_config(self):
 100         ms = self._test.vapi.qos_mark_dump()
 101
 102         for m in ms:
 103             if self.intf.sw_if_index == m.mark.sw_if_index and \
 104                self.source == m.mark.output_source and \
 105                self.map.id == m.mark.map_id:
 106                 return True
 107         return False
 108
 109     def object_id(self):
 110         return ("qos-mark-%s-%d" % (self.intf, self.source))