-#!/usr/bin/env python
+#!/usr/bin/env python3
+from __future__ import print_function
+import binascii
import random
import socket
import unittest
import time
+import re
from scapy.packet import Raw
from scapy.layers.l2 import Ether
-from scapy.layers.inet import IP, UDP
+from scapy.layers.inet import IP, TCP, UDP
from scapy.layers.inet6 import IPv6
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from config import config
+from framework import tag_fixme_vpp_workers
+from framework import VppTestCase, VppTestRunner
+from framework import tag_run_solo
from vpp_object import VppObject
from vpp_pg_interface import CaptureTimeoutError
from util import ppp
from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
+from vpp_ip_route import VppIpRoute, VppRoutePath
+from vpp_papi.macaddress import mac_ntop
+from socket import inet_ntop
+from vpp_papi import VppEnum
class VppCFLOW(VppObject):
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))
+ l2_flag = 0
+ l3_flag = 0
+ l4_flag = 0
+ if 'l2' in self._collect.lower():
+ l2_flag = (VppEnum.vl_api_flowprobe_record_flags_t.
+ FLOWPROBE_RECORD_FLAG_L2)
+ if 'l3' in self._collect.lower():
+ l3_flag = (VppEnum.vl_api_flowprobe_record_flags_t.
+ FLOWPROBE_RECORD_FLAG_L3)
+ if 'l4' in self._collect.lower():
+ l4_flag = (VppEnum.vl_api_flowprobe_record_flags_t.
+ FLOWPROBE_RECORD_FLAG_L4)
+ self._test.vapi.flowprobe_params(
+ record_flags=(l2_flag | l3_flag | l4_flag),
+ active_timer=self._active, passive_timer=self._passive)
self.enable_flowprobe_feature()
self._test.vapi.cli("ipfix flush")
self._configured = True
def enable_exporter(self):
self._test.vapi.set_ipfix_exporter(
- collector_address=self._test.pg0.remote_ip4n,
- src_address=self._test.pg0.local_ip4n,
+ collector_address=self._test.pg0.remote_ip4,
+ src_address=self._test.pg0.local_ip4,
path_mtu=self._mtu,
template_interval=self._timeout)
(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
def verify_templates(self, decoder=None, timeout=1, count=3):
templates = []
- p = self._test.wait_for_cflow_packet(self._test.collector, 2, timeout)
- self._test.assertTrue(p.haslayer(IPFIX))
- if decoder is not None and p.haslayer(Template):
- templates.append(p[Template].templateID)
- decoder.add_template(p.getlayer(Template))
- if count > 1:
- p = self._test.wait_for_cflow_packet(self._test.collector, 2)
- self._test.assertTrue(p.haslayer(IPFIX))
- if decoder is not None and p.haslayer(Template):
- templates.append(p[Template].templateID)
- decoder.add_template(p.getlayer(Template))
- if count > 2:
- p = self._test.wait_for_cflow_packet(self._test.collector, 2)
+ self._test.assertIn(count, (1, 2, 3))
+ for _ in range(count):
+ p = self._test.wait_for_cflow_packet(self._test.collector, 2,
+ timeout)
self._test.assertTrue(p.haslayer(IPFIX))
if decoder is not None and p.haslayer(Template):
templates.append(p[Template].templateID)
super(MethodHolder, cls).setUpClass()
try:
# Create pg interfaces
- cls.create_pg_interfaces(range(7))
+ cls.create_pg_interfaces(range(9))
# Packet sizes
cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
# 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:
cls.pg3.resolve_arp()
cls.pg4.config_ip4()
cls.pg4.resolve_arp()
+ cls.pg7.config_ip4()
+ cls.pg8.config_ip4()
+ cls.pg8.configure_ipv4_neighbors()
cls.pg5.config_ip6()
cls.pg5.resolve_ndp()
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
p /= IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4)
else:
p /= IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6)
- p /= (UDP(sport=1234, dport=4321) / Raw(payload))
+ p /= UDP(sport=1234, dport=4321)
+ p /= Raw(payload)
info.data = p.copy()
self.extend_packet(p, pkt_size)
self.pkts.append(p)
if cflow.haslayer(Data):
data = decoder.decode_data_set(cflow.getlayer(Set))
for record in data:
- self.assertEqual(int(record[1].encode('hex'), 16), octets)
- self.assertEqual(int(record[2].encode('hex'), 16), packets)
+ self.assertEqual(int(binascii.hexlify(record[1]), 16), octets)
+ self.assertEqual(int(binascii.hexlify(record[2]), 16), packets)
def send_packets(self, src_if=None, dst_if=None):
if src_if is None:
data_set={1: 'octets', 2: 'packets'},
ip_ver='v4'):
if self.debug_print:
- print capture[0].show()
+ print(capture[0].show())
if cflow.haslayer(Data):
data = decoder.decode_data_set(cflow.getlayer(Set))
if self.debug_print:
- print data
+ print(data)
if ip_ver == 'v4':
ip_layer = capture[0][IP]
else:
ip_layer = capture[0][IPv6]
if data_set is not None:
for record in data:
- # skip flow if in/out gress interface is 0
- if int(record[10].encode('hex'), 16) == 0:
+ # skip flow if ingress/egress interface is 0
+ if int(binascii.hexlify(record[10]), 16) == 0:
continue
- if int(record[14].encode('hex'), 16) == 0:
+ if int(binascii.hexlify(record[14]), 16) == 0:
continue
for field in data_set:
else:
ip = socket.inet_pton(socket.AF_INET6,
ip_layer.src)
- value = int(ip.encode('hex'), 16)
+ value = int(binascii.hexlify(ip), 16)
elif value == 'dst_ip':
if ip_ver == 'v4':
ip = socket.inet_pton(socket.AF_INET,
else:
ip = socket.inet_pton(socket.AF_INET6,
ip_layer.dst)
- value = int(ip.encode('hex'), 16)
+ value = int(binascii.hexlify(ip), 16)
elif value == 'sport':
value = int(capture[0][UDP].sport)
elif value == 'dport':
value = int(capture[0][UDP].dport)
- self.assertEqual(int(record[field].encode('hex'), 16),
- value)
+ self.assertEqual(int(binascii.hexlify(
+ record[field]), 16),
+ value)
def verify_cflow_data_notimer(self, decoder, capture, cflows):
idx = 0
for rec in data:
p = capture[idx]
idx += 1
- self.assertEqual(p[IP].len, int(rec[1].encode('hex'), 16))
- self.assertEqual(1, int(rec[2].encode('hex'), 16))
+ self.assertEqual(p[IP].len, int(
+ binascii.hexlify(rec[1]), 16))
+ self.assertEqual(1, int(
+ binascii.hexlify(rec[2]), 16))
self.assertEqual(len(capture), idx)
- def wait_for_cflow_packet(self, collector_intf, set_id=2, timeout=1,
- expected=True):
+ def wait_for_cflow_packet(self, collector_intf, set_id=2, timeout=1):
""" wait for CFLOW packet and verify its correctness
:param timeout: how long to wait
- :returns: tuple (packet, time spent waiting for packet)
"""
self.logger.info("IPFIX: Waiting for CFLOW packet")
- deadline = time.time() + timeout
- counter = 0
# self.logger.debug(self.vapi.ppcli("show flow table"))
- while True:
- counter += 1
- # sanity check
- self.assert_in_range(counter, 0, 100, "number of packets ignored")
- time_left = deadline - time.time()
- try:
- if time_left < 0 and expected:
- # self.logger.debug(self.vapi.ppcli("show flow table"))
- raise CaptureTimeoutError(
- "Packet did not arrive within timeout")
- p = collector_intf.wait_for_packet(timeout=time_left)
- except CaptureTimeoutError:
- if expected:
- # self.logger.debug(self.vapi.ppcli("show flow table"))
- raise CaptureTimeoutError(
- "Packet did not arrive within timeout")
- else:
- return
- if not expected:
- raise CaptureTimeoutError("Packet arrived even not expected")
- self.assertEqual(p[Set].setID, set_id)
- # self.logger.debug(self.vapi.ppcli("show flow table"))
- self.logger.debug(ppp("IPFIX: Got packet:", p))
- break
+ p = collector_intf.wait_for_packet(timeout=timeout)
+ self.assertEqual(p[Set].setID, set_id)
+ # self.logger.debug(self.vapi.ppcli("show flow table"))
+ self.logger.debug(ppp("IPFIX: Got packet:", p))
return p
-class Timers(MethodHolder):
+@tag_run_solo
+@tag_fixme_vpp_workers
+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_0002")
+ self.logger.info("FFP_TEST_START_0001")
self.pg_enable_capture(self.pg_interfaces)
self.pkts = []
# template packet should arrive immediately
templates = ipfix.verify_templates(ipfix_decoder)
- self.create_stream(packets=2)
+ self.create_stream(packets=1)
self.send_packets()
- capture = self.pg2.get_capture(2)
+ capture = self.pg2.get_capture(1)
# make sure the one packet we expect actually showed up
cflow = self.wait_for_cflow_packet(self.collector, templates[1], 15)
self.verify_cflow_data(ipfix_decoder, capture, cflow)
ipfix.remove_vpp_config()
- self.logger.info("FFP_TEST_FINISH_0002")
+ self.logger.info("FFP_TEST_FINISH_0001")
def test_0002(self):
""" timer greater than template timeout"""
- self.logger.info("FFP_TEST_START_0003")
+ self.logger.info("FFP_TEST_START_0002")
self.pg_enable_capture(self.pg_interfaces)
self.pkts = []
self.verify_cflow_data(ipfix_decoder, capture, cflow)
ipfix.remove_vpp_config()
- self.logger.info("FFP_TEST_FINISH_0003")
+ self.logger.info("FFP_TEST_FINISH_0002")
+
+ def test_cflow_packet(self):
+ """verify cflow packet fields"""
+ self.logger.info("FFP_TEST_START_0000")
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pkts = []
+
+ ipfix = VppCFLOW(test=self, intf='pg8', datapath="ip4",
+ layer='l2 l3 l4', active=2)
+ ipfix.add_vpp_config()
+
+ route_9001 = VppIpRoute(self, "9.0.0.0", 24,
+ [VppRoutePath(self.pg8._remote_hosts[0].ip4,
+ self.pg8.sw_if_index)])
+ route_9001.add_vpp_config()
+
+ ipfix_decoder = IPFIXDecoder()
+ templates = ipfix.verify_templates(ipfix_decoder, count=1)
+
+ self.pkts = [(Ether(dst=self.pg7.local_mac,
+ src=self.pg7.remote_mac) /
+ IP(src=self.pg7.remote_ip4, dst="9.0.0.100") /
+ TCP(sport=1234, dport=4321, flags=80) /
+ Raw(b'\xa5' * 100))]
+ nowUTC = int(time.time())
+ nowUNIX = nowUTC+2208988800
+ self.send_packets(src_if=self.pg7, dst_if=self.pg8)
+ cflow = self.wait_for_cflow_packet(self.collector, templates[0], 10)
+ self.collector.get_capture(2)
+
+ if cflow[0].haslayer(IPFIX):
+ self.assertEqual(cflow[IPFIX].version, 10)
+ self.assertEqual(cflow[IPFIX].observationDomainID, 1)
+ self.assertEqual(cflow[IPFIX].sequenceNumber, 0)
+ self.assertAlmostEqual(cflow[IPFIX].exportTime, nowUTC, delta=5)
+ if cflow.haslayer(Data):
+ record = ipfix_decoder.decode_data_set(cflow[0].getlayer(Set))[0]
+ # ingress interface
+ self.assertEqual(int(binascii.hexlify(record[10]), 16), 8)
+ # egress interface
+ self.assertEqual(int(binascii.hexlify(record[14]), 16), 9)
+ # packets
+ self.assertEqual(int(binascii.hexlify(record[2]), 16), 1)
+ # src mac
+ self.assertEqual(mac_ntop(record[56]), self.pg8.local_mac)
+ # dst mac
+ self.assertEqual(mac_ntop(record[80]), self.pg8.remote_mac)
+ flowTimestamp = int(binascii.hexlify(record[156]), 16) >> 32
+ # flow start timestamp
+ self.assertAlmostEqual(flowTimestamp, nowUNIX, delta=1)
+ flowTimestamp = int(binascii.hexlify(record[157]), 16) >> 32
+ # flow end timestamp
+ self.assertAlmostEqual(flowTimestamp, nowUNIX, delta=1)
+ # ethernet type
+ self.assertEqual(int(binascii.hexlify(record[256]), 16), 8)
+ # src ip
+ self.assertEqual(inet_ntop(socket.AF_INET, record[8]),
+ self.pg7.remote_ip4)
+ # dst ip
+ self.assertEqual(inet_ntop(socket.AF_INET, record[12]),
+ "9.0.0.100")
+ # protocol (TCP)
+ self.assertEqual(int(binascii.hexlify(record[4]), 16), 6)
+ # src port
+ self.assertEqual(int(binascii.hexlify(record[7]), 16), 1234)
+ # dst port
+ self.assertEqual(int(binascii.hexlify(record[11]), 16), 4321)
+ # tcp flags
+ self.assertEqual(int(binascii.hexlify(record[6]), 16), 80)
+
+ ipfix.remove_vpp_config()
+ self.logger.info("FFP_TEST_FINISH_0000")
+
+
+@tag_fixme_vpp_workers
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,
self.logger.info("FFP_TEST_FINISH_0002")
-@unittest.skipUnless(running_extended_tests(), "part of extended tests")
+@unittest.skipUnless(config.extended, "part of extended tests")
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.wait_for_cflow_packet(self.collector, templates[1],
- expected=False)
- self.collector.get_capture(0)
+ self.vapi.ipfix_flush()
+ self.sleep(1, "wait before verifying no packets sent")
+ self.collector.assert_nothing_captured()
ipfix.remove_vpp_config()
self.logger.info("FFP_TEST_FINISH_0001")
-@unittest.skipUnless(running_extended_tests(), "part of extended tests")
+@unittest.skipUnless(config.extended, "part of extended tests")
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.wait_for_cflow_packet(self.collector, templates[1],
- expected=False)
- self.collector.get_capture(0)
+ self.vapi.ipfix_flush()
+ self.sleep(1, "wait before verifying no packets sent")
+ self.collector.assert_nothing_captured()
self.pg2.get_capture(5)
# enable IPFIX
self.logger.info("FFP_TEST_FINISH_0001")
-@unittest.skipUnless(running_extended_tests(), "part of extended tests")
+@unittest.skipUnless(config.extended, "part of extended tests")
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.wait_for_cflow_packet(self.collector, templates[1],
- expected=False)
- self.collector.get_capture(0)
+ self.vapi.ipfix_flush()
+ self.sleep(1, "wait before verifying no packets sent")
+ self.collector.assert_nothing_captured()
ipfix.remove_vpp_config()
self.logger.info("FFP_TEST_FINISH_0001")
-@unittest.skipUnless(running_extended_tests(), "part of extended tests")
+@unittest.skipUnless(config.extended, "part of extended tests")
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.wait_for_cflow_packet(self.collector, templates[1], 5,
- expected=False)
- self.collector.get_capture(0)
+ self.vapi.ipfix_flush()
+ self.sleep(5, "wait before verifying no packets sent")
+ self.collector.assert_nothing_captured()
# 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)
Code Review / vpp.git / blobdiff