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("bufmon" in config.excluded_plugins, "Exclude bufmon plugin tests")
13 class TestBufmon(VppTestCase):
14 """bufmon plugin test"""
18 super(TestBufmon, 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(TestBufmon, 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)
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)
52 def verify_capture(self, src_if, dst_if, capture):
54 for packet in capture:
58 # convert the payload to packet info object
59 payload_info = self.payload_to_info(packet[Raw])
60 # make sure the indexes match
62 payload_info.src, src_if.sw_if_index, "source sw_if_index"
65 payload_info.dst, dst_if.sw_if_index, "destination sw_if_index"
67 packet_info = self.get_next_packet_info_for_interface2(
68 src_if.sw_if_index, dst_if.sw_if_index, packet_info
70 # make sure we didn't run out of saved packets
71 self.assertIsNotNone(packet_info)
73 payload_info.index, packet_info.index, "packet info index"
75 saved_packet = packet_info.data # fetch the saved packet
76 # assert the values match
77 self.assert_equal(ip.src, saved_packet[IP].src, "IP source address")
78 # ... more assertions here
79 self.assert_equal(udp.sport, saved_packet[UDP].sport, "UDP source port")
81 self.logger.error(ppp("Unexpected or invalid packet:", packet))
83 remaining_packet = self.get_next_packet_info_for_interface2(
84 src_if.sw_if_index, dst_if.sw_if_index, packet_info
88 "Interface %s: Packet expected from interface "
89 "%s didn't arrive" % (dst_if.name, src_if.name),
92 def test_bufmon(self):
93 self.vapi.cli("set buffer traces on")
95 reply = self.vapi.cli("show buffer traces status")
96 self.assertIn("buffers tracing is on", reply)
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)
117 reply = self.vapi.cli("show buffer traces verbose")
118 for entry in expected:
119 self.assertIn(entry, reply)
121 # not verbose, skips nodes w/o buffered buffers
122 reply = self.vapi.cli("show buffer traces")
123 self.assertNotIn("pg-input", reply)
125 self.vapi.cli("clear buffer traces")
126 reply = self.vapi.cli("show buffer traces verbose")
127 self.assertNotIn("pg-input", reply)
130 if __name__ == "__main__":
131 unittest.main(testRunner=VppTestRunner)