summaryrefslogtreecommitdiffstats
path: root/test/vpp_qos.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/vpp_qos.py')
-rw-r--r--test/vpp_qos.py76
1 files changed, 43 insertions, 33 deletions
diff --git a/test/vpp_qos.py b/test/vpp_qos.py
index a7fa9e748aa..a7374d1b1d8 100644
--- a/test/vpp_qos.py
+++ b/test/vpp_qos.py
@@ -8,7 +8,7 @@ from vpp_object import VppObject
class VppQosRecord(VppObject):
- """ QoS Record(ing) configuration """
+ """QoS Record(ing) configuration"""
def __init__(self, test, intf, source):
self._test = test
@@ -18,32 +18,34 @@ class VppQosRecord(VppObject):
def add_vpp_config(self):
self._test.vapi.qos_record_enable_disable(
enable=1,
- record={'sw_if_index': self.intf.sw_if_index,
- 'input_source': self.source})
+ record={"sw_if_index": self.intf.sw_if_index, "input_source": self.source},
+ )
self._test.registry.register(self, self._test.logger)
return self
def remove_vpp_config(self):
self._test.vapi.qos_record_enable_disable(
enable=0,
- record={'sw_if_index': self.intf.sw_if_index,
- 'input_source': self.source})
+ record={"sw_if_index": self.intf.sw_if_index, "input_source": self.source},
+ )
def query_vpp_config(self):
rs = self._test.vapi.qos_record_dump()
for r in rs:
- if self.intf.sw_if_index == r.record.sw_if_index and \
- self.source == r.record.input_source:
+ if (
+ self.intf.sw_if_index == r.record.sw_if_index
+ and self.source == r.record.input_source
+ ):
return True
return False
def object_id(self):
- return ("qos-record-%s-%d" % (self.intf, self.source))
+ return "qos-record-%s-%d" % (self.intf, self.source)
class VppQosStore(VppObject):
- """ QoS Store(ing) configuration """
+ """QoS Store(ing) configuration"""
def __init__(self, test, intf, source, value):
self._test = test
@@ -54,34 +56,39 @@ class VppQosStore(VppObject):
def add_vpp_config(self):
self._test.vapi.qos_store_enable_disable(
enable=1,
- store={'sw_if_index': self.intf.sw_if_index,
- 'input_source': self.source,
- 'value': self.value})
+ store={
+ "sw_if_index": self.intf.sw_if_index,
+ "input_source": self.source,
+ "value": self.value,
+ },
+ )
self._test.registry.register(self, self._test.logger)
return self
def remove_vpp_config(self):
self._test.vapi.qos_store_enable_disable(
enable=0,
- store={'sw_if_index': self.intf.sw_if_index,
- 'input_source': self.source})
+ store={"sw_if_index": self.intf.sw_if_index, "input_source": self.source},
+ )
def query_vpp_config(self):
rs = self._test.vapi.qos_store_dump()
for r in rs:
- if self.intf.sw_if_index == r.store.sw_if_index and \
- self.source == r.store.input_source and \
- self.value == r.store.value:
+ if (
+ self.intf.sw_if_index == r.store.sw_if_index
+ and self.source == r.store.input_source
+ and self.value == r.store.value
+ ):
return True
return False
def object_id(self):
- return ("qos-store-%s-%d" % (self.intf, self.source))
+ return "qos-store-%s-%d" % (self.intf, self.source)
class VppQosEgressMap(VppObject):
- """ QoS Egress Map(ping) configuration """
+ """QoS Egress Map(ping) configuration"""
def __init__(self, test, id, rows):
self._test = test
@@ -89,9 +96,7 @@ class VppQosEgressMap(VppObject):
self.rows = rows
def add_vpp_config(self):
- self._test.vapi.qos_egress_map_update(
- map={'id': self.id,
- 'rows': self.rows})
+ self._test.vapi.qos_egress_map_update(map={"id": self.id, "rows": self.rows})
self._test.registry.register(self, self._test.logger)
return self
@@ -107,11 +112,11 @@ class VppQosEgressMap(VppObject):
return False
def object_id(self):
- return ("qos-map-%d" % (self.id))
+ return "qos-map-%d" % (self.id)
class VppQosMark(VppObject):
- """ QoS Mark(ing) configuration """
+ """QoS Mark(ing) configuration"""
def __init__(self, test, intf, map, source):
self._test = test
@@ -122,27 +127,32 @@ class VppQosMark(VppObject):
def add_vpp_config(self):
self._test.vapi.qos_mark_enable_disable(
enable=1,
- mark={'sw_if_index': self.intf.sw_if_index,
- 'map_id': self.map.id,
- 'output_source': self.source})
+ mark={
+ "sw_if_index": self.intf.sw_if_index,
+ "map_id": self.map.id,
+ "output_source": self.source,
+ },
+ )
self._test.registry.register(self, self._test.logger)
return self
def remove_vpp_config(self):
self._test.vapi.qos_mark_enable_disable(
enable=0,
- mark={'sw_if_index': self.intf.sw_if_index,
- 'output_source': self.source})
+ mark={"sw_if_index": self.intf.sw_if_index, "output_source": self.source},
+ )
def query_vpp_config(self):
ms = self._test.vapi.qos_mark_dump()
for m in ms:
- if self.intf.sw_if_index == m.mark.sw_if_index and \
- self.source == m.mark.output_source and \
- self.map.id == m.mark.map_id:
+ if (
+ self.intf.sw_if_index == m.mark.sw_if_index
+ and self.source == m.mark.output_source
+ and self.map.id == m.mark.map_id
+ ):
return True
return False
def object_id(self):
- return ("qos-mark-%s-%d" % (self.intf, self.source))
+ return "qos-mark-%s-%d" % (self.intf, self.source)