test/test_flowprobe.py

   1 #!/usr/bin/env python3
   2 from __future__ import print_function
   3 import binascii
   4 import random
   5 import socket
   6 import unittest
   7 import time
   8 import re
   9
  10 from scapy.packet import Raw
  11 from scapy.layers.l2 import Ether
  12 from scapy.layers.inet import IP, TCP, UDP
  13 from scapy.layers.inet6 import IPv6
  14 from scapy.contrib.lacp import SlowProtocol, LACP
  15
  16 from config import config
  17 from framework import tag_fixme_vpp_workers, tag_fixme_ubuntu2204, tag_fixme_debian11
  18 from framework import is_distro_ubuntu2204, is_distro_debian11
  19 from framework import VppTestCase, VppTestRunner
  20 from framework import tag_run_solo
  21 from vpp_object import VppObject
  22 from vpp_pg_interface import CaptureTimeoutError
  23 from util import ppp
  24 from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
  25 from vpp_ip_route import VppIpRoute, VppRoutePath
  26 from vpp_papi.macaddress import mac_ntop
  27 from socket import inet_ntop
  28 from vpp_papi import VppEnum
  29
  30
  31 class VppCFLOW(VppObject):
  32     """CFLOW object for IPFIX exporter and Flowprobe feature"""
  33
  34     def __init__(
  35         self,
  36         test,
  37         intf="pg2",
  38         active=0,
  39         passive=0,
  40         timeout=100,
  41         mtu=1024,
  42         datapath="l2",
  43         layer="l2 l3 l4",
  44         direction="tx",
  45     ):
  46         self._test = test
  47         self._intf = intf
  48         self._intf_obj = getattr(self._test, intf)
  49         self._active = active
  50         if passive == 0 or passive < active:
  51             self._passive = active + 1
  52         else:
  53             self._passive = passive
  54         self._datapath = datapath  # l2 ip4 ip6
  55         self._collect = layer  # l2 l3 l4
  56         self._direction = direction  # rx tx both
  57         self._timeout = timeout
  58         self._mtu = mtu
  59         self._configured = False
  60
  61     def add_vpp_config(self):
  62         self.enable_exporter()
  63         l2_flag = 0
  64         l3_flag = 0
  65         l4_flag = 0
  66         if "l2" in self._collect.lower():
  67             l2_flag = VppEnum.vl_api_flowprobe_record_flags_t.FLOWPROBE_RECORD_FLAG_L2
  68         if "l3" in self._collect.lower():
  69             l3_flag = VppEnum.vl_api_flowprobe_record_flags_t.FLOWPROBE_RECORD_FLAG_L3
  70         if "l4" in self._collect.lower():
  71             l4_flag = VppEnum.vl_api_flowprobe_record_flags_t.FLOWPROBE_RECORD_FLAG_L4
  72         self._test.vapi.flowprobe_set_params(
  73             record_flags=(l2_flag | l3_flag | l4_flag),
  74             active_timer=self._active,
  75             passive_timer=self._passive,
  76         )
  77         self.enable_flowprobe_feature()
  78         self._test.vapi.cli("ipfix flush")
  79         self._configured = True
  80
  81     def remove_vpp_config(self):
  82         self.disable_exporter()
  83         self.disable_flowprobe_feature()
  84         self._test.vapi.cli("ipfix flush")
  85         self._configured = False
  86
  87     def enable_exporter(self):
  88         self._test.vapi.set_ipfix_exporter(
  89             collector_address=self._test.pg0.remote_ip4,
  90             src_address=self._test.pg0.local_ip4,
  91             path_mtu=self._mtu,
  92             template_interval=self._timeout,
  93         )
  94
  95     def _enable_disable_flowprobe_feature(self, is_add):
  96         which_map = {
  97             "l2": VppEnum.vl_api_flowprobe_which_t.FLOWPROBE_WHICH_L2,
  98             "ip4": VppEnum.vl_api_flowprobe_which_t.FLOWPROBE_WHICH_IP4,
  99             "ip6": VppEnum.vl_api_flowprobe_which_t.FLOWPROBE_WHICH_IP6,
 100         }
 101         direction_map = {
 102             "rx": VppEnum.vl_api_flowprobe_direction_t.FLOWPROBE_DIRECTION_RX,
 103             "tx": VppEnum.vl_api_flowprobe_direction_t.FLOWPROBE_DIRECTION_TX,
 104             "both": VppEnum.vl_api_flowprobe_direction_t.FLOWPROBE_DIRECTION_BOTH,
 105         }
 106         self._test.vapi.flowprobe_interface_add_del(
 107             is_add=is_add,
 108             which=which_map[self._datapath],
 109             direction=direction_map[self._direction],
 110             sw_if_index=self._intf_obj.sw_if_index,
 111         )
 112
 113     def enable_flowprobe_feature(self):
 114         self._enable_disable_flowprobe_feature(is_add=True)
 115
 116     def disable_exporter(self):
 117         self._test.vapi.cli("set ipfix exporter collector 0.0.0.0")
 118
 119     def disable_flowprobe_feature(self):
 120         self._enable_disable_flowprobe_feature(is_add=False)
 121
 122     def object_id(self):
 123         return "ipfix-collector-%s-%s" % (self._src, self.dst)
 124
 125     def query_vpp_config(self):
 126         return self._configured
 127
 128     def verify_templates(self, decoder=None, timeout=1, count=3):
 129         templates = []
 130         self._test.assertIn(count, (1, 2, 3))
 131         for _ in range(count):
 132             p = self._test.wait_for_cflow_packet(self._test.collector, 2, timeout)
 133             self._test.assertTrue(p.haslayer(IPFIX))
 134             if decoder is not None and p.haslayer(Template):
 135                 templates.append(p[Template].templateID)
 136                 decoder.add_template(p.getlayer(Template))
 137         return templates
 138
 139
 140 class MethodHolder(VppTestCase):
 141     """Flow-per-packet plugin: test L2, IP4, IP6 reporting"""
 142
 143     # Test variables
 144     debug_print = False
 145     max_number_of_packets = 10
 146     pkts = []
 147
 148     @classmethod
 149     def setUpClass(cls):
 150         """
 151         Perform standard class setup (defined by class method setUpClass in
 152         class VppTestCase) before running the test case, set test case related
 153         variables and configure VPP.
 154         """
 155         super(MethodHolder, cls).setUpClass()
 156         if (is_distro_ubuntu2204 == True or is_distro_debian11 == True) and not hasattr(
 157             cls, "vpp"
 158         ):
 159             return
 160         try:
 161             # Create pg interfaces
 162             cls.create_pg_interfaces(range(9))
 163
 164             # Packet sizes
 165             cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
 166
 167             # Create BD with MAC learning and unknown unicast flooding disabled
 168             # and put interfaces to this BD
 169             cls.vapi.bridge_domain_add_del_v2(
 170                 bd_id=1, uu_flood=1, learn=1, flood=1, forward=1, is_add=1
 171             )
 172             cls.vapi.sw_interface_set_l2_bridge(
 173                 rx_sw_if_index=cls.pg1._sw_if_index, bd_id=1
 174             )
 175             cls.vapi.sw_interface_set_l2_bridge(
 176                 rx_sw_if_index=cls.pg2._sw_if_index, bd_id=1
 177             )
 178
 179             # Set up all interfaces
 180             for i in cls.pg_interfaces:
 181                 i.admin_up()
 182
 183             cls.pg0.config_ip4()
 184             cls.pg0.configure_ipv4_neighbors()
 185             cls.collector = cls.pg0
 186
 187             cls.pg1.config_ip4()
 188             cls.pg1.resolve_arp()
 189             cls.pg2.config_ip4()
 190             cls.pg2.resolve_arp()
 191             cls.pg3.config_ip4()
 192             cls.pg3.resolve_arp()
 193             cls.pg4.config_ip4()
 194             cls.pg4.resolve_arp()
 195             cls.pg7.config_ip4()
 196             cls.pg8.config_ip4()
 197             cls.pg8.configure_ipv4_neighbors()
 198
 199             cls.pg5.config_ip6()
 200             cls.pg5.resolve_ndp()
 201             cls.pg5.disable_ipv6_ra()
 202             cls.pg6.config_ip6()
 203             cls.pg6.resolve_ndp()
 204             cls.pg6.disable_ipv6_ra()
 205         except Exception:
 206             super(MethodHolder, cls).tearDownClass()
 207             raise
 208
 209     @classmethod
 210     def tearDownClass(cls):
 211         super(MethodHolder, cls).tearDownClass()
 212
 213     def create_stream(
 214         self, src_if=None, dst_if=None, packets=None, size=None, ip_ver="v4"
 215     ):
 216         """Create a packet stream to tickle the plugin
 217
 218         :param VppInterface src_if: Source interface for packet stream
 219         :param VppInterface src_if: Dst interface for packet stream
 220         """
 221         if src_if is None:
 222             src_if = self.pg1
 223         if dst_if is None:
 224             dst_if = self.pg2
 225         self.pkts = []
 226         if packets is None:
 227             packets = random.randint(1, self.max_number_of_packets)
 228         pkt_size = size
 229         for p in range(0, packets):
 230             if size is None:
 231                 pkt_size = random.choice(self.pg_if_packet_sizes)
 232             info = self.create_packet_info(src_if, dst_if)
 233             payload = self.info_to_payload(info)
 234             p = Ether(src=src_if.remote_mac, dst=src_if.local_mac)
 235             if ip_ver == "v4":
 236                 p /= IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4)
 237             else:
 238                 p /= IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6)
 239             p /= UDP(sport=1234, dport=4321)
 240             p /= Raw(payload)
 241             info.data = p.copy()
 242             self.extend_packet(p, pkt_size)
 243             self.pkts.append(p)
 244
 245     def verify_cflow_data(self, decoder, capture, cflow):
 246         octets = 0
 247         packets = 0
 248         for p in capture:
 249             octets += p[IP].len
 250             packets += 1
 251         if cflow.haslayer(Data):
 252             data = decoder.decode_data_set(cflow.getlayer(Set))
 253             for record in data:
 254                 self.assertEqual(int(binascii.hexlify(record[1]), 16), octets)
 255                 self.assertEqual(int(binascii.hexlify(record[2]), 16), packets)
 256
 257     def send_packets(self, src_if=None, dst_if=None):
 258         if src_if is None:
 259             src_if = self.pg1
 260         if dst_if is None:
 261             dst_if = self.pg2
 262         self.pg_enable_capture([dst_if])
 263         src_if.add_stream(self.pkts)
 264         self.pg_start()
 265         return dst_if.get_capture(len(self.pkts))
 266
 267     def verify_cflow_data_detail(
 268         self, decoder, capture, cflow, data_set={1: "octets", 2: "packets"}, ip_ver="v4"
 269     ):
 270         if self.debug_print:
 271             print(capture[0].show())
 272         if cflow.haslayer(Data):
 273             data = decoder.decode_data_set(cflow.getlayer(Set))
 274             if self.debug_print:
 275                 print(data)
 276             if ip_ver == "v4":
 277                 ip_layer = capture[0][IP] if capture[0].haslayer(IP) else None
 278             else:
 279                 ip_layer = capture[0][IPv6] if capture[0].haslayer(IPv6) else None
 280             if data_set is not None:
 281                 for record in data:
 282                     # skip flow if ingress/egress interface is 0
 283                     if int(binascii.hexlify(record[10]), 16) == 0:
 284                         continue
 285                     if int(binascii.hexlify(record[14]), 16) == 0:
 286                         continue
 287
 288                     for field in data_set:
 289                         value = data_set[field]
 290                         if value == "octets":
 291                             value = ip_layer.len
 292                             if ip_ver == "v6":
 293                                 value += 40  # ??? is this correct
 294                         elif value == "packets":
 295                             value = 1
 296                         elif value == "src_ip":
 297                             if ip_ver == "v4":
 298                                 ip = socket.inet_pton(socket.AF_INET, ip_layer.src)
 299                             else:
 300                                 ip = socket.inet_pton(socket.AF_INET6, ip_layer.src)
 301                             value = int(binascii.hexlify(ip), 16)
 302                         elif value == "dst_ip":
 303                             if ip_ver == "v4":
 304                                 ip = socket.inet_pton(socket.AF_INET, ip_layer.dst)
 305                             else:
 306                                 ip = socket.inet_pton(socket.AF_INET6, ip_layer.dst)
 307                             value = int(binascii.hexlify(ip), 16)
 308                         elif value == "sport":
 309                             value = int(capture[0][UDP].sport)
 310                         elif value == "dport":
 311                             value = int(capture[0][UDP].dport)
 312                         self.assertEqual(
 313                             int(binascii.hexlify(record[field]), 16), value
 314                         )
 315
 316     def verify_cflow_data_notimer(self, decoder, capture, cflows):
 317         idx = 0
 318         for cflow in cflows:
 319             if cflow.haslayer(Data):
 320                 data = decoder.decode_data_set(cflow.getlayer(Set))
 321             else:
 322                 raise Exception("No CFLOW data")
 323
 324             for rec in data:
 325                 p = capture[idx]
 326                 idx += 1
 327                 self.assertEqual(p[IP].len, int(binascii.hexlify(rec[1]), 16))
 328                 self.assertEqual(1, int(binascii.hexlify(rec[2]), 16))
 329         self.assertEqual(len(capture), idx)
 330
 331     def wait_for_cflow_packet(self, collector_intf, set_id=2, timeout=1):
 332         """wait for CFLOW packet and verify its correctness
 333
 334         :param timeout: how long to wait
 335
 336         """
 337         self.logger.info("IPFIX: Waiting for CFLOW packet")
 338         # self.logger.debug(self.vapi.ppcli("show flow table"))
 339         p = collector_intf.wait_for_packet(timeout=timeout)
 340         self.assertEqual(p[Set].setID, set_id)
 341         # self.logger.debug(self.vapi.ppcli("show flow table"))
 342         self.logger.debug(ppp("IPFIX: Got packet:", p))
 343         return p
 344
 345
 346 @tag_run_solo
 347 @tag_fixme_vpp_workers
 348 @tag_fixme_ubuntu2204
 349 @tag_fixme_debian11
 350 class Flowprobe(MethodHolder):
 351     """Template verification, timer tests"""
 352
 353     @classmethod
 354     def setUpClass(cls):
 355         super(Flowprobe, cls).setUpClass()
 356
 357     @classmethod
 358     def tearDownClass(cls):
 359         super(Flowprobe, cls).tearDownClass()
 360
 361     def test_0001(self):
 362         """timer less than template timeout"""
 363         self.logger.info("FFP_TEST_START_0001")
 364         self.pg_enable_capture(self.pg_interfaces)
 365         self.pkts = []
 366
 367         ipfix = VppCFLOW(test=self, active=2)
 368         ipfix.add_vpp_config()
 369
 370         ipfix_decoder = IPFIXDecoder()
 371         # template packet should arrive immediately
 372         templates = ipfix.verify_templates(ipfix_decoder)
 373
 374         self.create_stream(packets=1)
 375         self.send_packets()
 376         capture = self.pg2.get_capture(1)
 377
 378         # make sure the one packet we expect actually showed up
 379         cflow = self.wait_for_cflow_packet(self.collector, templates[1], 15)
 380         self.verify_cflow_data(ipfix_decoder, capture, cflow)
 381
 382         ipfix.remove_vpp_config()
 383         self.logger.info("FFP_TEST_FINISH_0001")
 384
 385     def test_0002(self):
 386         """timer greater than template timeout"""
 387         self.logger.info("FFP_TEST_START_0002")
 388         self.pg_enable_capture(self.pg_interfaces)
 389         self.pkts = []
 390
 391         ipfix = VppCFLOW(test=self, timeout=3, active=4)
 392         ipfix.add_vpp_config()
 393
 394         ipfix_decoder = IPFIXDecoder()
 395         # template packet should arrive immediately
 396         ipfix.verify_templates()
 397
 398         self.create_stream(packets=2)
 399         self.send_packets()
 400         capture = self.pg2.get_capture(2)
 401
 402         # next set of template packet should arrive after 20 seconds
 403         # template packet should arrive within 20 s
 404         templates = ipfix.verify_templates(ipfix_decoder, timeout=5)
 405
 406         # make sure the one packet we expect actually showed up
 407         cflow = self.wait_for_cflow_packet(self.collector, templates[1], 15)
 408         self.verify_cflow_data(ipfix_decoder, capture, cflow)
 409
 410         ipfix.remove_vpp_config()
 411         self.logger.info("FFP_TEST_FINISH_0002")
 412
 413     def test_cflow_packet(self):
 414         """verify cflow packet fields"""
 415         self.logger.info("FFP_TEST_START_0000")
 416         self.pg_enable_capture(self.pg_interfaces)
 417         self.pkts = []
 418
 419         ipfix = VppCFLOW(
 420             test=self, intf="pg8", datapath="ip4", layer="l2 l3 l4", active=2
 421         )
 422         ipfix.add_vpp_config()
 423
 424         route_9001 = VppIpRoute(
 425             self,
 426             "9.0.0.0",
 427             24,
 428             [VppRoutePath(self.pg8._remote_hosts[0].ip4, self.pg8.sw_if_index)],
 429         )
 430         route_9001.add_vpp_config()
 431
 432         ipfix_decoder = IPFIXDecoder()
 433         templates = ipfix.verify_templates(ipfix_decoder, count=1)
 434
 435         self.pkts = [
 436             (
 437                 Ether(dst=self.pg7.local_mac, src=self.pg7.remote_mac)
 438                 / IP(src=self.pg7.remote_ip4, dst="9.0.0.100")
 439                 / TCP(sport=1234, dport=4321, flags=80)
 440                 / Raw(b"\xa5" * 100)
 441             )
 442         ]
 443
 444         nowUTC = int(time.time())
 445         nowUNIX = nowUTC + 2208988800
 446         self.send_packets(src_if=self.pg7, dst_if=self.pg8)
 447
 448         cflow = self.wait_for_cflow_packet(self.collector, templates[0], 10)
 449         self.collector.get_capture(2)
 450
 451         if cflow[0].haslayer(IPFIX):
 452             self.assertEqual(cflow[IPFIX].version, 10)
 453             self.assertEqual(cflow[IPFIX].observationDomainID, 1)
 454             self.assertEqual(cflow[IPFIX].sequenceNumber, 0)
 455             self.assertAlmostEqual(cflow[IPFIX].exportTime, nowUTC, delta=5)
 456         if cflow.haslayer(Data):
 457             record = ipfix_decoder.decode_data_set(cflow[0].getlayer(Set))[0]
 458             # ingress interface
 459             self.assertEqual(int(binascii.hexlify(record[10]), 16), 8)
 460             # egress interface
 461             self.assertEqual(int(binascii.hexlify(record[14]), 16), 9)
 462             # direction
 463             self.assertEqual(int(binascii.hexlify(record[61]), 16), 1)
 464             # packets
 465             self.assertEqual(int(binascii.hexlify(record[2]), 16), 1)
 466             # src mac
 467             self.assertEqual(mac_ntop(record[56]), self.pg8.local_mac)
 468             # dst mac
 469             self.assertEqual(mac_ntop(record[80]), self.pg8.remote_mac)
 470             flowTimestamp = int(binascii.hexlify(record[156]), 16) >> 32
 471             # flow start timestamp
 472             self.assertAlmostEqual(flowTimestamp, nowUNIX, delta=1)
 473             flowTimestamp = int(binascii.hexlify(record[157]), 16) >> 32
 474             # flow end timestamp
 475             self.assertAlmostEqual(flowTimestamp, nowUNIX, delta=1)
 476             # ethernet type
 477             self.assertEqual(int(binascii.hexlify(record[256]), 16), 8)
 478             # src ip
 479             self.assertEqual(inet_ntop(socket.AF_INET, record[8]), self.pg7.remote_ip4)
 480             # dst ip
 481             self.assertEqual(inet_ntop(socket.AF_INET, record[12]), "9.0.0.100")
 482             # protocol (TCP)
 483             self.assertEqual(int(binascii.hexlify(record[4]), 16), 6)
 484             # src port
 485             self.assertEqual(int(binascii.hexlify(record[7]), 16), 1234)
 486             # dst port
 487             self.assertEqual(int(binascii.hexlify(record[11]), 16), 4321)
 488             # tcp flags
 489             self.assertEqual(int(binascii.hexlify(record[6]), 16), 80)
 490
 491         ipfix.remove_vpp_config()
 492         self.logger.info("FFP_TEST_FINISH_0000")
 493
 494     def test_interface_dump(self):
 495         """Dump interfaces with IPFIX flow record generation enabled"""
 496         self.logger.info("FFP_TEST_START_0003")
 497
 498         # Enable feature for 3 interfaces
 499         ipfix1 = VppCFLOW(test=self, intf="pg1", datapath="l2", direction="rx")
 500         ipfix1.add_vpp_config()
 501
 502         ipfix2 = VppCFLOW(test=self, intf="pg2", datapath="ip4", direction="tx")
 503         ipfix2.enable_flowprobe_feature()
 504
 505         ipfix3 = VppCFLOW(test=self, intf="pg3", datapath="ip6", direction="both")
 506         ipfix3.enable_flowprobe_feature()
 507
 508         # When request "all", dump should contain all enabled interfaces
 509         dump = self.vapi.flowprobe_interface_dump()
 510         self.assertEqual(len(dump), 3)
 511
 512         # Verify 1st interface
 513         self.assertEqual(dump[0].sw_if_index, self.pg1.sw_if_index)
 514         self.assertEqual(
 515             dump[0].which, VppEnum.vl_api_flowprobe_which_t.FLOWPROBE_WHICH_L2
 516         )
 517         self.assertEqual(
 518             dump[0].direction,
 519             VppEnum.vl_api_flowprobe_direction_t.FLOWPROBE_DIRECTION_RX,
 520         )
 521
 522         # Verify 2nd interface
 523         self.assertEqual(dump[1].sw_if_index, self.pg2.sw_if_index)
 524         self.assertEqual(
 525             dump[1].which, VppEnum.vl_api_flowprobe_which_t.FLOWPROBE_WHICH_IP4
 526         )
 527         self.assertEqual(
 528             dump[1].direction,
 529             VppEnum.vl_api_flowprobe_direction_t.FLOWPROBE_DIRECTION_TX,
 530         )
 531
 532         # Verify 3rd interface
 533         self.assertEqual(dump[2].sw_if_index, self.pg3.sw_if_index)
 534         self.assertEqual(
 535             dump[2].which, VppEnum.vl_api_flowprobe_which_t.FLOWPROBE_WHICH_IP6
 536         )
 537         self.assertEqual(
 538             dump[2].direction,
 539             VppEnum.vl_api_flowprobe_direction_t.FLOWPROBE_DIRECTION_BOTH,
 540         )
 541
 542         # When request 2nd interface, dump should contain only the specified interface
 543         dump = self.vapi.flowprobe_interface_dump(sw_if_index=self.pg2.sw_if_index)
 544         self.assertEqual(len(dump), 1)
 545
 546         # Verify 2nd interface
 547         self.assertEqual(dump[0].sw_if_index, self.pg2.sw_if_index)
 548         self.assertEqual(
 549             dump[0].which, VppEnum.vl_api_flowprobe_which_t.FLOWPROBE_WHICH_IP4
 550         )
 551         self.assertEqual(
 552             dump[0].direction,
 553             VppEnum.vl_api_flowprobe_direction_t.FLOWPROBE_DIRECTION_TX,
 554         )
 555
 556         # When request 99th interface, dump should be empty
 557         dump = self.vapi.flowprobe_interface_dump(sw_if_index=99)
 558         self.assertEqual(len(dump), 0)
 559
 560         ipfix1.remove_vpp_config()
 561         ipfix2.remove_vpp_config()
 562         ipfix3.remove_vpp_config()
 563         self.logger.info("FFP_TEST_FINISH_0003")
 564
 565     def test_get_params(self):
 566         """Get IPFIX flow record generation parameters"""
 567         self.logger.info("FFP_TEST_START_0004")
 568
 569         # Enable feature for an interface with custom parameters
 570         ipfix = VppCFLOW(test=self, active=20, passive=40, layer="l2 l3 l4")
 571         ipfix.add_vpp_config()
 572
 573         # Get and verify parameters
 574         params = self.vapi.flowprobe_get_params()
 575         self.assertEqual(params.active_timer, 20)
 576         self.assertEqual(params.passive_timer, 40)
 577         record_flags = VppEnum.vl_api_flowprobe_record_flags_t.FLOWPROBE_RECORD_FLAG_L2
 578         record_flags |= VppEnum.vl_api_flowprobe_record_flags_t.FLOWPROBE_RECORD_FLAG_L3
 579         record_flags |= VppEnum.vl_api_flowprobe_record_flags_t.FLOWPROBE_RECORD_FLAG_L4
 580         self.assertEqual(params.record_flags, record_flags)
 581
 582         ipfix.remove_vpp_config()
 583         self.logger.info("FFP_TEST_FINISH_0004")
 584
 585
 586 class DatapathTestsHolder(object):
 587     """collect information on Ethernet, IP4 and IP6 datapath (no timers)"""
 588
 589     @classmethod
 590     def setUpClass(cls):
 591         super(DatapathTestsHolder, cls).setUpClass()
 592
 593     @classmethod
 594     def tearDownClass(cls):
 595         super(DatapathTestsHolder, cls).tearDownClass()
 596
 597     def test_templatesL2(self):
 598         """verify template on L2 datapath"""
 599         self.logger.info("FFP_TEST_START_0000")
 600         self.pg_enable_capture(self.pg_interfaces)
 601
 602         ipfix = VppCFLOW(
 603             test=self, intf=self.intf1, layer="l2", direction=self.direction
 604         )
 605         ipfix.add_vpp_config()
 606
 607         # template packet should arrive immediately
 608         self.vapi.ipfix_flush()
 609         ipfix.verify_templates(timeout=3, count=1)
 610         self.collector.get_capture(1)
 611
 612         ipfix.remove_vpp_config()
 613         self.logger.info("FFP_TEST_FINISH_0000")
 614
 615     def test_L2onL2(self):
 616         """L2 data on L2 datapath"""
 617         self.logger.info("FFP_TEST_START_0001")
 618         self.pg_enable_capture(self.pg_interfaces)
 619         self.pkts = []
 620
 621         ipfix = VppCFLOW(
 622             test=self, intf=self.intf1, layer="l2", direction=self.direction
 623         )
 624         ipfix.add_vpp_config()
 625
 626         ipfix_decoder = IPFIXDecoder()
 627         # template packet should arrive immediately
 628         templates = ipfix.verify_templates(ipfix_decoder, count=1)
 629
 630         self.create_stream(packets=1)
 631         capture = self.send_packets()
 632
 633         # make sure the one packet we expect actually showed up
 634         self.vapi.ipfix_flush()
 635         cflow = self.wait_for_cflow_packet(self.collector, templates[0])
 636         self.verify_cflow_data_detail(
 637             ipfix_decoder,
 638             capture,
 639             cflow,
 640             {2: "packets", 256: 8, 61: (self.direction == "tx")},
 641         )
 642         self.collector.get_capture(2)
 643
 644         ipfix.remove_vpp_config()
 645         self.logger.info("FFP_TEST_FINISH_0001")
 646
 647     def test_L3onL2(self):
 648         """L3 data on L2 datapath"""
 649         self.logger.info("FFP_TEST_START_0002")
 650         self.pg_enable_capture(self.pg_interfaces)
 651         self.pkts = []
 652
 653         ipfix = VppCFLOW(
 654             test=self, intf=self.intf1, layer="l3", direction=self.direction
 655         )
 656         ipfix.add_vpp_config()
 657
 658         ipfix_decoder = IPFIXDecoder()
 659         # template packet should arrive immediately
 660         templates = ipfix.verify_templates(ipfix_decoder, count=2)
 661
 662         self.create_stream(packets=1)
 663         capture = self.send_packets()
 664
 665         # make sure the one packet we expect actually showed up
 666         self.vapi.ipfix_flush()
 667         cflow = self.wait_for_cflow_packet(self.collector, templates[0])
 668         self.verify_cflow_data_detail(
 669             ipfix_decoder,
 670             capture,
 671             cflow,
 672             {
 673                 2: "packets",
 674                 4: 17,
 675                 8: "src_ip",
 676                 12: "dst_ip",
 677                 61: (self.direction == "tx"),
 678             },
 679         )
 680
 681         self.collector.get_capture(3)
 682
 683         ipfix.remove_vpp_config()
 684         self.logger.info("FFP_TEST_FINISH_0002")
 685
 686     def test_L23onL2(self):
 687         """L2/3 data on L2 datapath"""
 688         self.pg_enable_capture(self.pg_interfaces)
 689         self.pkts = []
 690
 691         ipfix = VppCFLOW(
 692             test=self, intf=self.intf1, layer="l2 l3", direction=self.direction
 693         )
 694         ipfix.add_vpp_config()
 695
 696         ipfix_decoder = IPFIXDecoder()
 697         # template packet should arrive immediately
 698         templates = ipfix.verify_templates(ipfix_decoder, count=3)
 699
 700         # verify IPv4 and IPv6 flows
 701         for ip_ver in ("v4", "v6"):
 702             self.create_stream(packets=1, ip_ver=ip_ver)
 703             capture = self.send_packets()
 704
 705             # make sure the one packet we expect actually showed up
 706             self.vapi.ipfix_flush()
 707             cflow = self.wait_for_cflow_packet(
 708                 self.collector, templates[1 if ip_ver == "v4" else 2]
 709             )
 710             src_ip_id = 8 if ip_ver == "v4" else 27
 711             dst_ip_id = 12 if ip_ver == "v4" else 28
 712             self.verify_cflow_data_detail(
 713                 ipfix_decoder,
 714                 capture,
 715                 cflow,
 716                 {
 717                     2: "packets",
 718                     256: 8 if ip_ver == "v4" else 56710,
 719                     4: 17,
 720                     src_ip_id: "src_ip",
 721                     dst_ip_id: "dst_ip",
 722                     61: (self.direction == "tx"),
 723                 },
 724                 ip_ver=ip_ver,
 725             )
 726
 727         # verify non-IP flow
 728         self.pkts = [
 729             (
 730                 Ether(dst=self.pg2.local_mac, src=self.pg1.remote_mac)
 731                 / SlowProtocol()
 732                 / LACP()
 733             )
 734         ]
 735         capture = self.send_packets()
 736
 737         # make sure the one packet we expect actually showed up
 738         self.vapi.ipfix_flush()
 739         cflow = self.wait_for_cflow_packet(self.collector, templates[0])
 740         self.verify_cflow_data_detail(
 741             ipfix_decoder,
 742             capture,
 743             cflow,
 744             {2: "packets", 256: 2440, 61: (self.direction == "tx")},
 745         )
 746
 747         self.collector.get_capture(6)
 748
 749         ipfix.remove_vpp_config()
 750
 751     def test_L4onL2(self):
 752         """L4 data on L2 datapath"""
 753         self.logger.info("FFP_TEST_START_0003")
 754         self.pg_enable_capture(self.pg_interfaces)
 755         self.pkts = []
 756
 757         ipfix = VppCFLOW(
 758             test=self, intf=self.intf1, layer="l4", direction=self.direction
 759         )
 760         ipfix.add_vpp_config()
 761
 762         ipfix_decoder = IPFIXDecoder()
 763         # template packet should arrive immediately
 764         templates = ipfix.verify_templates(ipfix_decoder, count=2)
 765
 766         self.create_stream(packets=1)
 767         capture = self.send_packets()
 768
 769         # make sure the one packet we expect actually showed up
 770         self.vapi.ipfix_flush()
 771         cflow = self.wait_for_cflow_packet(self.collector, templates[0])
 772         self.verify_cflow_data_detail(
 773             ipfix_decoder,
 774             capture,
 775             cflow,
 776             {2: "packets", 7: "sport", 11: "dport", 61: (self.direction == "tx")},
 777         )
 778
 779         self.collector.get_capture(3)
 780
 781         ipfix.remove_vpp_config()
 782         self.logger.info("FFP_TEST_FINISH_0003")
 783
 784     def test_templatesIp4(self):
 785         """verify templates on IP4 datapath"""
 786         self.logger.info("FFP_TEST_START_0000")
 787
 788         self.pg_enable_capture(self.pg_interfaces)
 789
 790         ipfix = VppCFLOW(
 791             test=self, intf=self.intf1, datapath="ip4", direction=self.direction
 792         )
 793         ipfix.add_vpp_config()
 794
 795         # template packet should arrive immediately
 796         self.vapi.ipfix_flush()
 797         ipfix.verify_templates(timeout=3, count=1)
 798         self.collector.get_capture(1)
 799
 800         ipfix.remove_vpp_config()
 801
 802         self.logger.info("FFP_TEST_FINISH_0000")
 803
 804     def test_L2onIP4(self):
 805         """L2 data on IP4 datapath"""
 806         self.logger.info("FFP_TEST_START_0001")
 807         self.pg_enable_capture(self.pg_interfaces)
 808         self.pkts = []
 809
 810         ipfix = VppCFLOW(
 811             test=self,
 812             intf=self.intf2,
 813             layer="l2",
 814             datapath="ip4",
 815             direction=self.direction,
 816         )
 817         ipfix.add_vpp_config()
 818
 819         ipfix_decoder = IPFIXDecoder()
 820         # template packet should arrive immediately
 821         templates = ipfix.verify_templates(ipfix_decoder, count=1)
 822
 823         self.create_stream(src_if=self.pg3, dst_if=self.pg4, packets=1)
 824         capture = self.send_packets(src_if=self.pg3, dst_if=self.pg4)
 825
 826         # make sure the one packet we expect actually showed up
 827         self.vapi.ipfix_flush()
 828         cflow = self.wait_for_cflow_packet(self.collector, templates[0])
 829         self.verify_cflow_data_detail(
 830             ipfix_decoder,
 831             capture,
 832             cflow,
 833             {2: "packets", 256: 8, 61: (self.direction == "tx")},
 834         )
 835
 836         # expected two templates and one cflow packet
 837         self.collector.get_capture(2)
 838
 839         ipfix.remove_vpp_config()
 840         self.logger.info("FFP_TEST_FINISH_0001")
 841
 842     def test_L3onIP4(self):
 843         """L3 data on IP4 datapath"""
 844         self.logger.info("FFP_TEST_START_0002")
 845         self.pg_enable_capture(self.pg_interfaces)
 846         self.pkts = []
 847
 848         ipfix = VppCFLOW(
 849             test=self,
 850             intf=self.intf2,
 851             layer="l3",
 852             datapath="ip4",
 853             direction=self.direction,
 854         )
 855         ipfix.add_vpp_config()
 856
 857         ipfix_decoder = IPFIXDecoder()
 858         # template packet should arrive immediately
 859         templates = ipfix.verify_templates(ipfix_decoder, count=1)
 860
 861         self.create_stream(src_if=self.pg3, dst_if=self.pg4, packets=1)
 862         capture = self.send_packets(src_if=self.pg3, dst_if=self.pg4)
 863
 864         # make sure the one packet we expect actually showed up
 865         self.vapi.ipfix_flush()
 866         cflow = self.wait_for_cflow_packet(self.collector, templates[0])
 867         self.verify_cflow_data_detail(
 868             ipfix_decoder,
 869             capture,
 870             cflow,
 871             {
 872                 1: "octets",
 873                 2: "packets",
 874                 8: "src_ip",
 875                 12: "dst_ip",
 876                 61: (self.direction == "tx"),
 877             },
 878         )
 879
 880         # expected two templates and one cflow packet
 881         self.collector.get_capture(2)
 882
 883         ipfix.remove_vpp_config()
 884         self.logger.info("FFP_TEST_FINISH_0002")
 885
 886     def test_L4onIP4(self):
 887         """L4 data on IP4 datapath"""
 888         self.logger.info("FFP_TEST_START_0003")
 889         self.pg_enable_capture(self.pg_interfaces)
 890         self.pkts = []
 891
 892         ipfix = VppCFLOW(
 893             test=self,
 894             intf=self.intf2,
 895             layer="l4",
 896             datapath="ip4",
 897             direction=self.direction,
 898         )
 899         ipfix.add_vpp_config()
 900
 901         ipfix_decoder = IPFIXDecoder()
 902         # template packet should arrive immediately
 903         templates = ipfix.verify_templates(ipfix_decoder, count=1)
 904
 905         self.create_stream(src_if=self.pg3, dst_if=self.pg4, packets=1)
 906         capture = self.send_packets(src_if=self.pg3, dst_if=self.pg4)
 907
 908         # make sure the one packet we expect actually showed up
 909         self.vapi.ipfix_flush()
 910         cflow = self.wait_for_cflow_packet(self.collector, templates[0])
 911         self.verify_cflow_data_detail(
 912             ipfix_decoder,
 913             capture,
 914             cflow,
 915             {2: "packets", 7: "sport", 11: "dport", 61: (self.direction == "tx")},
 916         )
 917
 918         # expected two templates and one cflow packet
 919         self.collector.get_capture(2)
 920
 921         ipfix.remove_vpp_config()
 922         self.logger.info("FFP_TEST_FINISH_0003")
 923
 924     def test_templatesIP6(self):
 925         """verify templates on IP6 datapath"""
 926         self.logger.info("FFP_TEST_START_0000")
 927         self.pg_enable_capture(self.pg_interfaces)
 928
 929         ipfix = VppCFLOW(
 930             test=self, intf=self.intf1, datapath="ip6", direction=self.direction
 931         )
 932         ipfix.add_vpp_config()
 933
 934         # template packet should arrive immediately
 935         ipfix.verify_templates(count=1)
 936         self.collector.get_capture(1)
 937
 938         ipfix.remove_vpp_config()
 939
 940         self.logger.info("FFP_TEST_FINISH_0000")
 941
 942     def test_L2onIP6(self):
 943         """L2 data on IP6 datapath"""
 944         self.logger.info("FFP_TEST_START_0001")
 945         self.pg_enable_capture(self.pg_interfaces)
 946         self.pkts = []
 947
 948         ipfix = VppCFLOW(
 949             test=self,
 950             intf=self.intf3,
 951             layer="l2",
 952             datapath="ip6",
 953             direction=self.direction,
 954         )
 955         ipfix.add_vpp_config()
 956
 957         ipfix_decoder = IPFIXDecoder()
 958         # template packet should arrive immediately
 959         templates = ipfix.verify_templates(ipfix_decoder, count=1)
 960
 961         self.create_stream(src_if=self.pg5, dst_if=self.pg6, packets=1, ip_ver="IPv6")
 962         capture = self.send_packets(src_if=self.pg5, dst_if=self.pg6)
 963
 964         # make sure the one packet we expect actually showed up
 965         self.vapi.ipfix_flush()
 966         cflow = self.wait_for_cflow_packet(self.collector, templates[0])
 967         self.verify_cflow_data_detail(
 968             ipfix_decoder,
 969             capture,
 970             cflow,
 971             {2: "packets", 256: 56710, 61: (self.direction == "tx")},
 972             ip_ver="v6",
 973         )
 974
 975         # expected two templates and one cflow packet
 976         self.collector.get_capture(2)
 977
 978         ipfix.remove_vpp_config()
 979         self.logger.info("FFP_TEST_FINISH_0001")
 980
 981     def test_L3onIP6(self):
 982         """L3 data on IP6 datapath"""
 983         self.logger.info("FFP_TEST_START_0002")
 984         self.pg_enable_capture(self.pg_interfaces)
 985         self.pkts = []
 986
 987         ipfix = VppCFLOW(
 988             test=self,
 989             intf=self.intf3,
 990             layer="l3",
 991             datapath="ip6",
 992             direction=self.direction,
 993         )
 994         ipfix.add_vpp_config()
 995
 996         ipfix_decoder = IPFIXDecoder()
 997         # template packet should arrive immediately
 998         templates = ipfix.verify_templates(ipfix_decoder, count=1)
 999
1000         self.create_stream(src_if=self.pg5, dst_if=self.pg6, packets=1, ip_ver="IPv6")
1001         capture = self.send_packets(src_if=self.pg5, dst_if=self.pg6)
1002
1003         # make sure the one packet we expect actually showed up
1004         self.vapi.ipfix_flush()
1005         cflow = self.wait_for_cflow_packet(self.collector, templates[0])
1006         self.verify_cflow_data_detail(
1007             ipfix_decoder,
1008             capture,
1009             cflow,
1010             {2: "packets", 27: "src_ip", 28: "dst_ip", 61: (self.direction == "tx")},
1011             ip_ver="v6",
1012         )
1013
1014         # expected two templates and one cflow packet
1015         self.collector.get_capture(2)
1016
1017         ipfix.remove_vpp_config()
1018         self.logger.info("FFP_TEST_FINISH_0002")
1019
1020     def test_L4onIP6(self):
1021         """L4 data on IP6 datapath"""
1022         self.logger.info("FFP_TEST_START_0003")
1023         self.pg_enable_capture(self.pg_interfaces)
1024         self.pkts = []
1025
1026         ipfix = VppCFLOW(
1027             test=self,
1028             intf=self.intf3,
1029             layer="l4",
1030             datapath="ip6",
1031             direction=self.direction,
1032         )
1033         ipfix.add_vpp_config()
1034
1035         ipfix_decoder = IPFIXDecoder()
1036         # template packet should arrive immediately
1037         templates = ipfix.verify_templates(ipfix_decoder, count=1)
1038
1039         self.create_stream(src_if=self.pg5, dst_if=self.pg6, packets=1, ip_ver="IPv6")
1040         capture = self.send_packets(src_if=self.pg5, dst_if=self.pg6)
1041
1042         # make sure the one packet we expect actually showed up
1043         self.vapi.ipfix_flush()
1044         cflow = self.wait_for_cflow_packet(self.collector, templates[0])
1045         self.verify_cflow_data_detail(
1046             ipfix_decoder,
1047             capture,
1048             cflow,
1049             {2: "packets", 7: "sport", 11: "dport", 61: (self.direction == "tx")},
1050             ip_ver="v6",
1051         )
1052
1053         # expected two templates and one cflow packet
1054         self.collector.get_capture(2)
1055
1056         ipfix.remove_vpp_config()
1057         self.logger.info("FFP_TEST_FINISH_0003")
1058
1059     def test_0001(self):
1060         """no timers, one CFLOW packet, 9 Flows inside"""
1061         self.logger.info("FFP_TEST_START_0001")
1062         self.pg_enable_capture(self.pg_interfaces)
1063         self.pkts = []
1064
1065         ipfix = VppCFLOW(test=self, intf=self.intf1, direction=self.direction)
1066         ipfix.add_vpp_config()
1067
1068         ipfix_decoder = IPFIXDecoder()
1069         # template packet should arrive immediately
1070         templates = ipfix.verify_templates(ipfix_decoder)
1071
1072         self.create_stream(packets=9)
1073         capture = self.send_packets()
1074
1075         # make sure the one packet we expect actually showed up
1076         self.vapi.ipfix_flush()
1077         cflow = self.wait_for_cflow_packet(self.collector, templates[1])
1078         self.verify_cflow_data_notimer(ipfix_decoder, capture, [cflow])
1079         self.collector.get_capture(4)
1080
1081         ipfix.remove_vpp_config()
1082         self.logger.info("FFP_TEST_FINISH_0001")
1083
1084     def test_0002(self):
1085         """no timers, two CFLOW packets (mtu=260), 3 Flows in each"""
1086         self.logger.info("FFP_TEST_START_0002")
1087         self.pg_enable_capture(self.pg_interfaces)
1088         self.pkts = []
1089
1090         ipfix = VppCFLOW(test=self, intf=self.intf1, direction=self.direction, mtu=260)
1091         ipfix.add_vpp_config()
1092
1093         ipfix_decoder = IPFIXDecoder()
1094         # template packet should arrive immediately
1095         self.vapi.ipfix_flush()
1096         templates = ipfix.verify_templates(ipfix_decoder)
1097
1098         self.create_stream(packets=6)
1099         capture = self.send_packets()
1100
1101         # make sure the one packet we expect actually showed up
1102         cflows = []
1103         self.vapi.ipfix_flush()
1104         cflows.append(self.wait_for_cflow_packet(self.collector, templates[1]))
1105         cflows.append(self.wait_for_cflow_packet(self.collector, templates[1]))
1106         self.verify_cflow_data_notimer(ipfix_decoder, capture, cflows)
1107         self.collector.get_capture(5)
1108
1109         ipfix.remove_vpp_config()
1110         self.logger.info("FFP_TEST_FINISH_0002")
1111
1112
1113 @tag_fixme_vpp_workers
1114 class DatapathTx(MethodHolder, DatapathTestsHolder):
1115     """Collect info on Ethernet, IP4 and IP6 datapath (TX) (no timers)"""
1116
1117     intf1 = "pg2"
1118     intf2 = "pg4"
1119     intf3 = "pg6"
1120     direction = "tx"
1121
1122
1123 @tag_fixme_vpp_workers
1124 class DatapathRx(MethodHolder, DatapathTestsHolder):
1125     """Collect info on Ethernet, IP4 and IP6 datapath (RX) (no timers)"""
1126
1127     intf1 = "pg1"
1128     intf2 = "pg3"
1129     intf3 = "pg5"
1130     direction = "rx"
1131
1132
1133 @unittest.skipUnless(config.extended, "part of extended tests")
1134 class DisableIPFIX(MethodHolder):
1135     """Disable IPFIX"""
1136
1137     @classmethod
1138     def setUpClass(cls):
1139         super(DisableIPFIX, cls).setUpClass()
1140
1141     @classmethod
1142     def tearDownClass(cls):
1143         super(DisableIPFIX, cls).tearDownClass()
1144
1145     def test_0001(self):
1146         """disable IPFIX after first packets"""
1147         self.logger.info("FFP_TEST_START_0001")
1148         self.pg_enable_capture(self.pg_interfaces)
1149         self.pkts = []
1150
1151         ipfix = VppCFLOW(test=self)
1152         ipfix.add_vpp_config()
1153
1154         ipfix_decoder = IPFIXDecoder()
1155         # template packet should arrive immediately
1156         templates = ipfix.verify_templates(ipfix_decoder)
1157
1158         self.create_stream()
1159         self.send_packets()
1160
1161         # make sure the one packet we expect actually showed up
1162         self.vapi.ipfix_flush()
1163         self.wait_for_cflow_packet(self.collector, templates[1])
1164         self.collector.get_capture(4)
1165
1166         # disable IPFIX
1167         ipfix.disable_exporter()
1168         self.pg_enable_capture([self.collector])
1169
1170         self.send_packets()
1171
1172         # make sure no one packet arrived in 1 minute
1173         self.vapi.ipfix_flush()
1174         self.sleep(1, "wait before verifying no packets sent")
1175         self.collector.assert_nothing_captured()
1176
1177         ipfix.remove_vpp_config()
1178         self.logger.info("FFP_TEST_FINISH_0001")
1179
1180
1181 @unittest.skipUnless(config.extended, "part of extended tests")
1182 class ReenableIPFIX(MethodHolder):
1183     """Re-enable IPFIX"""
1184
1185     @classmethod
1186     def setUpClass(cls):
1187         super(ReenableIPFIX, cls).setUpClass()
1188
1189     @classmethod
1190     def tearDownClass(cls):
1191         super(ReenableIPFIX, cls).tearDownClass()
1192
1193     def test_0011(self):
1194         """disable IPFIX after first packets and re-enable after few packets"""
1195         self.logger.info("FFP_TEST_START_0001")
1196         self.pg_enable_capture(self.pg_interfaces)
1197         self.pkts = []
1198
1199         ipfix = VppCFLOW(test=self)
1200         ipfix.add_vpp_config()
1201
1202         ipfix_decoder = IPFIXDecoder()
1203         # template packet should arrive immediately
1204         templates = ipfix.verify_templates(ipfix_decoder)
1205
1206         self.create_stream(packets=5)
1207         self.send_packets()
1208
1209         # make sure the one packet we expect actually showed up
1210         self.vapi.ipfix_flush()
1211         self.wait_for_cflow_packet(self.collector, templates[1])
1212         self.collector.get_capture(4)
1213
1214         # disable IPFIX
1215         ipfix.disable_exporter()
1216         self.vapi.ipfix_flush()
1217         self.pg_enable_capture([self.collector])
1218
1219         self.send_packets()
1220
1221         # make sure no one packet arrived in active timer span
1222         self.vapi.ipfix_flush()
1223         self.sleep(1, "wait before verifying no packets sent")
1224         self.collector.assert_nothing_captured()
1225         self.pg2.get_capture(5)
1226
1227         # enable IPFIX
1228         ipfix.enable_exporter()
1229
1230         capture = self.collector.get_capture(4)
1231         nr_templates = 0
1232         nr_data = 0
1233         for p in capture:
1234             self.assertTrue(p.haslayer(IPFIX))
1235             if p.haslayer(Template):
1236                 nr_templates += 1
1237         self.assertTrue(nr_templates, 3)
1238         for p in capture:
1239             self.assertTrue(p.haslayer(IPFIX))
1240             if p.haslayer(Data):
1241                 nr_data += 1
1242         self.assertTrue(nr_templates, 1)
1243
1244         ipfix.remove_vpp_config()
1245         self.logger.info("FFP_TEST_FINISH_0001")
1246
1247
1248 @unittest.skipUnless(config.extended, "part of extended tests")
1249 class DisableFP(MethodHolder):
1250     """Disable Flowprobe feature"""
1251
1252     @classmethod
1253     def setUpClass(cls):
1254         super(DisableFP, cls).setUpClass()
1255
1256     @classmethod
1257     def tearDownClass(cls):
1258         super(DisableFP, cls).tearDownClass()
1259
1260     def test_0001(self):
1261         """disable flowprobe feature after first packets"""
1262         self.logger.info("FFP_TEST_START_0001")
1263         self.pg_enable_capture(self.pg_interfaces)
1264         self.pkts = []
1265         ipfix = VppCFLOW(test=self)
1266         ipfix.add_vpp_config()
1267
1268         ipfix_decoder = IPFIXDecoder()
1269         # template packet should arrive immediately
1270         templates = ipfix.verify_templates(ipfix_decoder)
1271
1272         self.create_stream()
1273         self.send_packets()
1274
1275         # make sure the one packet we expect actually showed up
1276         self.vapi.ipfix_flush()
1277         self.wait_for_cflow_packet(self.collector, templates[1])
1278         self.collector.get_capture(4)
1279
1280         # disable IPFIX
1281         ipfix.disable_flowprobe_feature()
1282         self.pg_enable_capture([self.collector])
1283
1284         self.send_packets()
1285
1286         # make sure no one packet arrived in active timer span
1287         self.vapi.ipfix_flush()
1288         self.sleep(1, "wait before verifying no packets sent")
1289         self.collector.assert_nothing_captured()
1290
1291         ipfix.remove_vpp_config()
1292         self.logger.info("FFP_TEST_FINISH_0001")
1293
1294
1295 @unittest.skipUnless(config.extended, "part of extended tests")
1296 class ReenableFP(MethodHolder):
1297     """Re-enable Flowprobe feature"""
1298
1299     @classmethod
1300     def setUpClass(cls):
1301         super(ReenableFP, cls).setUpClass()
1302
1303     @classmethod
1304     def tearDownClass(cls):
1305         super(ReenableFP, cls).tearDownClass()
1306
1307     def test_0001(self):
1308         """disable flowprobe feature after first packets and re-enable
1309         after few packets"""
1310         self.logger.info("FFP_TEST_START_0001")
1311         self.pg_enable_capture(self.pg_interfaces)
1312         self.pkts = []
1313
1314         ipfix = VppCFLOW(test=self)
1315         ipfix.add_vpp_config()
1316
1317         ipfix_decoder = IPFIXDecoder()
1318         # template packet should arrive immediately
1319         self.vapi.ipfix_flush()
1320         templates = ipfix.verify_templates(ipfix_decoder, timeout=3)
1321
1322         self.create_stream()
1323         self.send_packets()
1324
1325         # make sure the one packet we expect actually showed up
1326         self.vapi.ipfix_flush()
1327         self.wait_for_cflow_packet(self.collector, templates[1], 5)
1328         self.collector.get_capture(4)
1329
1330         # disable FPP feature
1331         ipfix.disable_flowprobe_feature()
1332         self.pg_enable_capture([self.collector])
1333
1334         self.send_packets()
1335
1336         # make sure no one packet arrived in active timer span
1337         self.vapi.ipfix_flush()
1338         self.sleep(5, "wait before verifying no packets sent")
1339         self.collector.assert_nothing_captured()
1340
1341         # enable FPP feature
1342         ipfix.enable_flowprobe_feature()
1343         self.vapi.ipfix_flush()
1344         templates = ipfix.verify_templates(ipfix_decoder, timeout=3)
1345
1346         self.send_packets()
1347
1348         # make sure the next packets (templates and data) we expect actually
1349         # showed up
1350         self.vapi.ipfix_flush()
1351         self.wait_for_cflow_packet(self.collector, templates[1], 5)
1352         self.collector.get_capture(4)
1353
1354         ipfix.remove_vpp_config()
1355         self.logger.info("FFP_TEST_FINISH_0001")
1356
1357
1358 if __name__ == "__main__":
1359     unittest.main(testRunner=VppTestRunner)