1 from config import config
2 from asfframework import VppTestRunner
4 from framework import VppTestCase
5 from scapy.layers.l2 import Ether
6 from scapy.layers.inet import IP, UDP
7 from random import randint
11 def create_stream(src_if, dst_if, count):
13 for i in range(count):
15 Ether(dst=src_if.local_mac, src=src_if.remote_mac)
16 / IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4)
17 / UDP(sport=randint(49152, 65535), dport=5678)
25 "tracedump" in config.excluded_plugins, "Exclude tracedump plugin tests"
27 class TestTracedump(VppTestCase):
28 """Tracedump plugin tests"""
32 super(TestTracedump, cls).setUpClass()
33 cls.create_pg_interfaces(range(2))
34 for i in cls.pg_interfaces:
40 def tearDownClass(cls):
41 for i in cls.pg_interfaces:
44 super(TestTracedump, cls).tearDownClass()
46 def test_tracedump_include(self):
47 """Check API/CLI output + include node"""
48 packets = create_stream(self.pg0, self.pg1, 5)
49 self.pg0.add_stream(packets)
51 self.vapi.trace_clear_cache()
52 self.vapi.trace_clear_capture()
53 # get pg-input node index
54 reply = self.vapi.graph_node_get(
59 self.assertTrue(reply[1][0].name == "pg-input")
60 pg_input_index = reply[1][0].index
61 self.vapi.trace_set_filters(flag=1, node_index=pg_input_index, count=5)
62 self.vapi.trace_capture_packets(
63 node_index=pg_input_index,
67 pre_capture_clear=True,
70 reply = self.vapi.cli(
71 "show graph node want_arcs input drop output punt handoff no_free polling interrupt"
73 self.assertIn("af-packet-input", reply)
76 reply = self.vapi.trace_v2_dump(
77 thread_id=0xFFFFFFFF, position=0, clear_cache=False
79 self.assertTrue(reply)
80 reply = self.vapi.trace_filter_function_dump()
81 self.assertTrue(reply[1].selected)
83 def test_tracedump_exclude(self):
84 """Exclude node (no trace output)"""
85 self.vapi.trace_clear_cache()
86 self.vapi.trace_clear_capture()
88 packets = create_stream(self.pg0, self.pg1, 5)
89 self.pg0.add_stream(packets)
92 reply = self.vapi.graph_node_get(
97 self.assertTrue(reply[1][0].name == "pg-input")
98 pg_input_index = reply[1][0].index
99 self.vapi.trace_set_filters(flag=2, node_index=pg_input_index, count=5)
100 self.vapi.trace_capture_packets(
101 node_index=pg_input_index,
105 pre_capture_clear=True,
108 reply = self.vapi.trace_v2_dump(thread_id=0, position=1, clear_cache=False)
109 self.assertFalse(reply)
112 if __name__ == "__main__":
113 unittest.main(testRunner=VppTestRunner)