1 from framework import VppTestCase, VppTestRunner
3 from config import config
4 from scapy.layers.l2 import Ether
5 from scapy.packet import Raw
6 from scapy.layers.inet import IP, UDP
7 from random import randint
11 @unittest.skipIf("mdata" in config.excluded_plugins, "Exclude mdata plugin tests")
12 class TestMdataCli(VppTestCase):
13 """mdata plugin test"""
17 super(TestMdataCli, cls).setUpClass()
19 cls.create_pg_interfaces(range(2))
20 for i in cls.pg_interfaces:
29 def tearDownClass(cls):
30 for i in cls.pg_interfaces:
33 super(TestMdataCli, cls).tearDownClass()
35 # https://fd.io/docs/vpp/master/developer/tests/overview.html#example-how-to-add-a-new-test
36 def create_stream(self, src_if, dst_if, count):
38 for i in range(count):
39 info = self.create_packet_info(src_if, dst_if)
40 payload = self.info_to_payload(info)
43 Ether(dst=src_if.local_mac, src=src_if.remote_mac)
44 / IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4)
45 / UDP(sport=randint(49152, 65535), dport=5678)
54 def verify_capture(self, src_if, dst_if, capture):
56 for packet in capture:
60 # convert the payload to packet info object
61 payload_info = self.payload_to_info(packet[Raw])
62 # make sure the indexes match
64 payload_info.src, src_if.sw_if_index, "source sw_if_index"
67 payload_info.dst, dst_if.sw_if_index, "destination sw_if_index"
69 packet_info = self.get_next_packet_info_for_interface2(
70 src_if.sw_if_index, dst_if.sw_if_index, packet_info
72 # make sure we didn't run out of saved packets
73 self.assertIsNotNone(packet_info)
75 payload_info.index, packet_info.index, "packet info index"
77 saved_packet = packet_info.data # fetch the saved packet
78 # assert the values match
79 self.assert_equal(ip.src, saved_packet[IP].src, "IP source address")
80 # ... more assertions here
81 self.assert_equal(udp.sport, saved_packet[UDP].sport, "UDP source port")
83 self.logger.error(ppp("Unexpected or invalid packet:", packet))
85 remaining_packet = self.get_next_packet_info_for_interface2(
86 src_if.sw_if_index, dst_if.sw_if_index, packet_info
90 "Interface %s: Packet expected from interface "
91 "%s didn't arrive" % (dst_if.name, src_if.name),
94 def test_mdata_cli(self):
95 """turn on mdata tracking, send packets, verify, check CLI output"""
96 self.vapi.cli("buffer metadata tracking on")
98 packets = self.create_stream(self.pg0, self.pg1, 5)
99 self.pg0.add_stream(packets)
100 self.pg0.enable_capture()
101 self.pg1.enable_capture()
104 capture = self.pg1.get_capture()
105 self.pg0.assert_nothing_captured()
106 self.verify_capture(self.pg0, self.pg1, capture)
108 result = self.vapi.cli("show buffer metadata")
117 for entry in expected:
118 self.assertIn(entry, result)
119 self.vapi.cli("buffer metadata tracking off")
122 if __name__ == "__main__":
123 unittest.main(testRunner=VppTestRunner)