lb: remove api boilerplate
[vpp.git] / 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))