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 VppQosStore(VppObject):
  46     """ QoS Store(ing) configuration """
  47
  48     def __init__(self, test, intf, source, value):
  49         self._test = test
  50         self.intf = intf
  51         self.source = source
  52         self.value = value
  53
  54     def add_vpp_config(self):
  55         self._test.vapi.qos_store_enable_disable(
  56             enable=1,
  57             store={'sw_if_index': self.intf.sw_if_index,
  58                    'input_source': self.source,
  59                    'value': self.value})
  60         self._test.registry.register(self, self._test.logger)
  61         return self
  62
  63     def remove_vpp_config(self):
  64         self._test.vapi.qos_store_enable_disable(
  65             enable=0,
  66             store={'sw_if_index': self.intf.sw_if_index,
  67                    'input_source': self.source})
  68
  69     def query_vpp_config(self):
  70         rs = self._test.vapi.qos_store_dump()
  71
  72         for r in rs:
  73             if self.intf.sw_if_index == r.store.sw_if_index and \
  74                self.source == r.store.input_source and \
  75                self.value == r.store.value:
  76                 return True
  77         return False
  78
  79     def object_id(self):
  80         return ("qos-store-%s-%d" % (self.intf, self.source))
  81
  82
  83 class VppQosEgressMap(VppObject):
  84     """ QoS Egress Map(ping) configuration """
  85
  86     def __init__(self, test, id, rows):
  87         self._test = test
  88         self.id = id
  89         self.rows = rows
  90
  91     def add_vpp_config(self):
  92         self._test.vapi.qos_egress_map_update(
  93             map={'id': self.id,
  94                  'rows': self.rows})
  95         self._test.registry.register(self, self._test.logger)
  96         return self
  97
  98     def remove_vpp_config(self):
  99         self._test.vapi.qos_egress_map_delete(id=self.id)
 100
 101     def query_vpp_config(self):
 102         rs = self._test.vapi.qos_egress_map_dump()
 103
 104         for r in rs:
 105             if self.id == r.map.id:
 106                 return True
 107         return False
 108
 109     def object_id(self):
 110         return ("qos-map-%d" % (self.id))
 111
 112
 113 class VppQosMark(VppObject):
 114     """ QoS Mark(ing) configuration """
 115
 116     def __init__(self, test, intf, map, source):
 117         self._test = test
 118         self.intf = intf
 119         self.source = source
 120         self.map = map
 121
 122     def add_vpp_config(self):
 123         self._test.vapi.qos_mark_enable_disable(
 124             enable=1,
 125             mark={'sw_if_index': self.intf.sw_if_index,
 126                   'map_id': self.map.id,
 127                   'output_source': self.source})
 128         self._test.registry.register(self, self._test.logger)
 129         return self
 130
 131     def remove_vpp_config(self):
 132         self._test.vapi.qos_mark_enable_disable(
 133             enable=0,
 134             mark={'sw_if_index': self.intf.sw_if_index,
 135                   'output_source': self.source})
 136
 137     def query_vpp_config(self):
 138         ms = self._test.vapi.qos_mark_dump()
 139
 140         for m in ms:
 141             if self.intf.sw_if_index == m.mark.sw_if_index and \
 142                self.source == m.mark.output_source and \
 143                self.map.id == m.mark.map_id:
 144                 return True
 145         return False
 146
 147     def object_id(self):
 148         return ("qos-mark-%s-%d" % (self.intf, self.source))