1 from asfframework import VppTestRunner
2 from framework import VppTestCase
4 from config import config
5 from scapy.layers.l2 import Ether
6 from scapy.packet import Raw
7 from scapy.layers.inet import IP, UDP
8 from random import randint
12 @unittest.skipIf("mdata" in config.excluded_plugins, "Exclude mdata plugin tests")
13 class TestMdataCli(VppTestCase):
14 """mdata plugin test"""
18 super(TestMdataCli, cls).setUpClass()
20 cls.create_pg_interfaces(range(2))
21 for i in cls.pg_interfaces:
30 def tearDownClass(cls):
31 for i in cls.pg_interfaces:
34 super(TestMdataCli, cls).tearDownClass()
36 # https://fd.io/docs/vpp/master/developer/tests/overview.html#example-how-to-add-a-new-test
37 def create_stream(self, src_if, dst_if, count):
39 for i in range(count):
40 info = self.create_packet_info(src_if, dst_if)
41 payload = self.info_to_payload(info)
44 Ether(dst=src_if.local_mac, src=src_if.remote_mac)
45 / IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4)
46 / UDP(sport=randint(49152, 65535), dport=5678)
55 def verify_capture(self, src_if, dst_if, capture):
57 for packet in capture:
61 self.logger.debug(f"Converting payload to info for {packet[Raw]}")
62 # convert the payload to packet info object
63 payload_info = self.payload_to_info(packet[Raw])
64 # make sure the indexes match
66 payload_info.src, src_if.sw_if_index, "source sw_if_index"
69 payload_info.dst, dst_if.sw_if_index, "destination sw_if_index"
71 packet_info = self.get_next_packet_info_for_interface2(
72 src_if.sw_if_index, dst_if.sw_if_index, packet_info
74 # make sure we didn't run out of saved packets
75 self.assertIsNotNone(packet_info)
77 payload_info.index, packet_info.index, "packet info index"
79 saved_packet = packet_info.data # fetch the saved packet
80 # assert the values match
81 self.assert_equal(ip.src, saved_packet[IP].src, "IP source address")
82 # ... more assertions here
83 self.assert_equal(udp.sport, saved_packet[UDP].sport, "UDP source port")
85 self.logger.error(ppp("Unexpected or invalid packet:", packet))
87 remaining_packet = self.get_next_packet_info_for_interface2(
88 src_if.sw_if_index, dst_if.sw_if_index, packet_info
92 "Interface %s: Packet expected from interface "
93 "%s didn't arrive" % (dst_if.name, src_if.name),
96 def test_mdata_cli(self):
97 """turn on mdata tracking, send packets, verify, check CLI output"""
98 self.vapi.cli("buffer metadata tracking on")
100 packets = self.create_stream(self.pg0, self.pg1, 5)
101 self.pg0.add_stream(packets)
102 self.pg0.enable_capture()
103 self.pg1.enable_capture()
106 capture = self.pg1.get_capture()
107 self.pg0.assert_nothing_captured()
108 self.verify_capture(self.pg0, self.pg1, capture)
110 result = self.vapi.cli("show buffer metadata")
119 for entry in expected:
120 self.assertIn(entry, result)
121 self.vapi.cli("buffer metadata tracking off")
124 if __name__ == "__main__":
125 unittest.main(testRunner=VppTestRunner)