def add_vpp_config(self):
self.enable_exporter()
- self._test.vapi.ppcli("flowprobe params record %s active %s "
- "passive %s" % (self._collect, self._active,
- self._passive))
+ self._test.vapi.flowprobe_params(
+ record_l2=1 if 'l2' in self._collect.lower() else 0,
+ record_l3=1 if 'l3' in self._collect.lower() else 0,
+ record_l4=1 if 'l4' in self._collect.lower() else 0,
+ active_timer=self._active, passive_timer=self._passive)
self.enable_flowprobe_feature()
self._test.vapi.cli("ipfix flush")
self._configured = True
(self._intf, self._datapath))
def object_id(self):
- return "ipfix-collector-%s" % (self._src, self.dst)
+ return "ipfix-collector-%s-%s" % (self._src, self.dst)
def query_vpp_config(self):
return self._configured
# Create BD with MAC learning and unknown unicast flooding disabled
# and put interfaces to this BD
cls.vapi.bridge_domain_add_del(bd_id=1, uu_flood=1, learn=1)
- cls.vapi.sw_interface_set_l2_bridge(cls.pg1._sw_if_index, bd_id=1)
- cls.vapi.sw_interface_set_l2_bridge(cls.pg2._sw_if_index, bd_id=1)
+ cls.vapi.sw_interface_set_l2_bridge(
+ rx_sw_if_index=cls.pg1._sw_if_index, bd_id=1)
+ cls.vapi.sw_interface_set_l2_bridge(
+ rx_sw_if_index=cls.pg2._sw_if_index, bd_id=1)
# Set up all interfaces
for i in cls.pg_interfaces:
super(MethodHolder, cls).tearDownClass()
raise
+ @classmethod
+ def tearDownClass(cls):
+ super(MethodHolder, cls).tearDownClass()
+
def create_stream(self, src_if=None, dst_if=None, packets=None,
size=None, ip_ver='v4'):
"""Create a packet stream to tickle the plugin
ip_layer = capture[0][IPv6]
if data_set is not None:
for record in data:
- # skip flow if in/out gress interface is 0
+ # skip flow if ingress/egress interface is 0
if int(binascii.hexlify(record[10]), 16) == 0:
continue
if int(binascii.hexlify(record[14]), 16) == 0:
class Flowprobe(MethodHolder):
"""Template verification, timer tests"""
+ @classmethod
+ def setUpClass(cls):
+ super(Flowprobe, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(Flowprobe, cls).tearDownClass()
+
def test_0001(self):
""" timer less than template timeout"""
self.logger.info("FFP_TEST_START_0001")
class Datapath(MethodHolder):
"""collect information on Ethernet, IP4 and IP6 datapath (no timers)"""
+ @classmethod
+ def setUpClass(cls):
+ super(Datapath, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(Datapath, cls).tearDownClass()
+
def test_templatesL2(self):
""" verify template on L2 datapath"""
self.logger.info("FFP_TEST_START_0000")
ipfix.add_vpp_config()
# template packet should arrive immediately
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
ipfix.verify_templates(timeout=3, count=1)
self.collector.get_capture(1)
capture = self.send_packets()
# make sure the one packet we expect actually showed up
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
cflow = self.wait_for_cflow_packet(self.collector, templates[0])
self.verify_cflow_data_detail(ipfix_decoder, capture, cflow,
{2: 'packets', 256: 8})
capture = self.send_packets()
# make sure the one packet we expect actually showed up
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
cflow = self.wait_for_cflow_packet(self.collector, templates[0])
self.verify_cflow_data_detail(ipfix_decoder, capture, cflow,
{2: 'packets', 4: 17,
capture = self.send_packets()
# make sure the one packet we expect actually showed up
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
cflow = self.wait_for_cflow_packet(self.collector, templates[0])
self.verify_cflow_data_detail(ipfix_decoder, capture, cflow,
{2: 'packets', 7: 'sport', 11: 'dport'})
ipfix.add_vpp_config()
# template packet should arrive immediately
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
ipfix.verify_templates(timeout=3, count=1)
self.collector.get_capture(1)
capture = self.send_packets(src_if=self.pg3, dst_if=self.pg4)
# make sure the one packet we expect actually showed up
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
cflow = self.wait_for_cflow_packet(self.collector, templates[0])
self.verify_cflow_data_detail(ipfix_decoder, capture, cflow,
{2: 'packets', 256: 8})
capture = self.send_packets(src_if=self.pg3, dst_if=self.pg4)
# make sure the one packet we expect actually showed up
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
cflow = self.wait_for_cflow_packet(self.collector, templates[0])
self.verify_cflow_data_detail(ipfix_decoder, capture, cflow,
{1: 'octets', 2: 'packets',
capture = self.send_packets(src_if=self.pg3, dst_if=self.pg4)
# make sure the one packet we expect actually showed up
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
cflow = self.wait_for_cflow_packet(self.collector, templates[0])
self.verify_cflow_data_detail(ipfix_decoder, capture, cflow,
{2: 'packets', 7: 'sport', 11: 'dport'})
capture = self.send_packets(src_if=self.pg5, dst_if=self.pg6)
# make sure the one packet we expect actually showed up
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
cflow = self.wait_for_cflow_packet(self.collector, templates[0])
self.verify_cflow_data_detail(ipfix_decoder, capture, cflow,
{2: 'packets', 256: 56710},
capture = self.send_packets(src_if=self.pg5, dst_if=self.pg6)
# make sure the one packet we expect actually showed up
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
cflow = self.wait_for_cflow_packet(self.collector, templates[0])
self.verify_cflow_data_detail(ipfix_decoder, capture, cflow,
{2: 'packets',
capture = self.send_packets(src_if=self.pg5, dst_if=self.pg6)
# make sure the one packet we expect actually showed up
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
cflow = self.wait_for_cflow_packet(self.collector, templates[0])
self.verify_cflow_data_detail(ipfix_decoder, capture, cflow,
{2: 'packets', 7: 'sport', 11: 'dport'},
capture = self.send_packets()
# make sure the one packet we expect actually showed up
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
cflow = self.wait_for_cflow_packet(self.collector, templates[1])
self.verify_cflow_data_notimer(ipfix_decoder, capture, [cflow])
self.collector.get_capture(4)
ipfix_decoder = IPFIXDecoder()
# template packet should arrive immediately
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
templates = ipfix.verify_templates(ipfix_decoder)
self.create_stream(packets=6)
# make sure the one packet we expect actually showed up
cflows = []
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
cflows.append(self.wait_for_cflow_packet(self.collector,
templates[1]))
cflows.append(self.wait_for_cflow_packet(self.collector,
class DisableIPFIX(MethodHolder):
"""Disable IPFIX"""
+ @classmethod
+ def setUpClass(cls):
+ super(DisableIPFIX, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(DisableIPFIX, cls).tearDownClass()
+
def test_0001(self):
""" disable IPFIX after first packets"""
self.logger.info("FFP_TEST_START_0001")
self.send_packets()
# make sure the one packet we expect actually showed up
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
self.wait_for_cflow_packet(self.collector, templates[1])
self.collector.get_capture(4)
- # disble IPFIX
+ # disable IPFIX
ipfix.disable_exporter()
self.pg_enable_capture([self.collector])
self.send_packets()
# make sure no one packet arrived in 1 minute
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
self.wait_for_cflow_packet(self.collector, templates[1],
expected=False)
self.collector.get_capture(0)
class ReenableIPFIX(MethodHolder):
"""Re-enable IPFIX"""
+ @classmethod
+ def setUpClass(cls):
+ super(ReenableIPFIX, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(ReenableIPFIX, cls).tearDownClass()
+
def test_0011(self):
""" disable IPFIX after first packets and re-enable after few packets
"""
self.send_packets()
# make sure the one packet we expect actually showed up
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
self.wait_for_cflow_packet(self.collector, templates[1])
self.collector.get_capture(4)
- # disble IPFIX
+ # disable IPFIX
ipfix.disable_exporter()
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
self.pg_enable_capture([self.collector])
self.send_packets()
# make sure no one packet arrived in active timer span
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
self.wait_for_cflow_packet(self.collector, templates[1],
expected=False)
self.collector.get_capture(0)
class DisableFP(MethodHolder):
"""Disable Flowprobe feature"""
+ @classmethod
+ def setUpClass(cls):
+ super(DisableFP, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(DisableFP, cls).tearDownClass()
+
def test_0001(self):
""" disable flowprobe feature after first packets"""
self.logger.info("FFP_TEST_START_0001")
self.send_packets()
# make sure the one packet we expect actually showed up
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
self.wait_for_cflow_packet(self.collector, templates[1])
self.collector.get_capture(4)
- # disble IPFIX
+ # disable IPFIX
ipfix.disable_flowprobe_feature()
self.pg_enable_capture([self.collector])
self.send_packets()
# make sure no one packet arrived in active timer span
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
self.wait_for_cflow_packet(self.collector, templates[1],
expected=False)
self.collector.get_capture(0)
class ReenableFP(MethodHolder):
"""Re-enable Flowprobe feature"""
+ @classmethod
+ def setUpClass(cls):
+ super(ReenableFP, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(ReenableFP, cls).tearDownClass()
+
def test_0001(self):
""" disable flowprobe feature after first packets and re-enable
after few packets """
ipfix_decoder = IPFIXDecoder()
# template packet should arrive immediately
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
templates = ipfix.verify_templates(ipfix_decoder, timeout=3)
self.create_stream()
self.send_packets()
# make sure the one packet we expect actually showed up
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
self.wait_for_cflow_packet(self.collector, templates[1], 5)
self.collector.get_capture(4)
- # disble FPP feature
+ # disable FPP feature
ipfix.disable_flowprobe_feature()
self.pg_enable_capture([self.collector])
self.send_packets()
# make sure no one packet arrived in active timer span
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
self.wait_for_cflow_packet(self.collector, templates[1], 5,
expected=False)
self.collector.get_capture(0)
# enable FPP feature
ipfix.enable_flowprobe_feature()
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
templates = ipfix.verify_templates(ipfix_decoder, timeout=3)
self.send_packets()
# make sure the next packets (templates and data) we expect actually
# showed up
- self.vapi.cli("ipfix flush")
+ self.vapi.ipfix_flush()
self.wait_for_cflow_packet(self.collector, templates[1], 5)
self.collector.get_capture(4)