+ 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)
+ # direction
+ self.assertEqual(int(binascii.hexlify(record[61]), 16), 1)
+ # 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")
+
+ def test_interface_dump(self):
+ """Dump interfaces with IPFIX flow record generation enabled"""
+ self.logger.info("FFP_TEST_START_0003")
+
+ # Enable feature for 3 interfaces
+ ipfix1 = VppCFLOW(test=self, intf="pg1", datapath="l2", direction="rx")
+ ipfix1.add_vpp_config()
+
+ ipfix2 = VppCFLOW(test=self, intf="pg2", datapath="ip4", direction="tx")
+ ipfix2.enable_flowprobe_feature()
+
+ ipfix3 = VppCFLOW(test=self, intf="pg3", datapath="ip6", direction="both")
+ ipfix3.enable_flowprobe_feature()
+
+ # When request "all", dump should contain all enabled interfaces
+ dump = self.vapi.flowprobe_interface_dump()
+ self.assertEqual(len(dump), 3)
+
+ # Verify 1st interface
+ self.assertEqual(dump[0].sw_if_index, self.pg1.sw_if_index)
+ self.assertEqual(
+ dump[0].which, VppEnum.vl_api_flowprobe_which_t.FLOWPROBE_WHICH_L2
+ )
+ self.assertEqual(
+ dump[0].direction,
+ VppEnum.vl_api_flowprobe_direction_t.FLOWPROBE_DIRECTION_RX,
+ )
+
+ # Verify 2nd interface
+ self.assertEqual(dump[1].sw_if_index, self.pg2.sw_if_index)
+ self.assertEqual(
+ dump[1].which, VppEnum.vl_api_flowprobe_which_t.FLOWPROBE_WHICH_IP4
+ )
+ self.assertEqual(
+ dump[1].direction,
+ VppEnum.vl_api_flowprobe_direction_t.FLOWPROBE_DIRECTION_TX,
+ )
+
+ # Verify 3rd interface
+ self.assertEqual(dump[2].sw_if_index, self.pg3.sw_if_index)
+ self.assertEqual(
+ dump[2].which, VppEnum.vl_api_flowprobe_which_t.FLOWPROBE_WHICH_IP6
+ )
+ self.assertEqual(
+ dump[2].direction,
+ VppEnum.vl_api_flowprobe_direction_t.FLOWPROBE_DIRECTION_BOTH,
+ )
+
+ # When request 2nd interface, dump should contain only the specified interface
+ dump = self.vapi.flowprobe_interface_dump(sw_if_index=self.pg2.sw_if_index)
+ self.assertEqual(len(dump), 1)
+
+ # Verify 2nd interface
+ self.assertEqual(dump[0].sw_if_index, self.pg2.sw_if_index)
+ self.assertEqual(
+ dump[0].which, VppEnum.vl_api_flowprobe_which_t.FLOWPROBE_WHICH_IP4
+ )
+ self.assertEqual(
+ dump[0].direction,
+ VppEnum.vl_api_flowprobe_direction_t.FLOWPROBE_DIRECTION_TX,
+ )
+
+ # When request 99th interface, dump should be empty
+ dump = self.vapi.flowprobe_interface_dump(sw_if_index=99)
+ self.assertEqual(len(dump), 0)
+
+ ipfix1.remove_vpp_config()
+ ipfix2.remove_vpp_config()
+ ipfix3.remove_vpp_config()