test/test_tracedump.py

   1 from config import config
   2 from asfframework import VppTestRunner
   3 import unittest
   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
   8 import ctypes
   9
  10
  11 def create_stream(src_if, dst_if, count):
  12     packets = []
  13     for i in range(count):
  14         p = (
  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)
  18         )
  19         packets.append(p)
  20
  21     return packets
  22
  23
  24 @unittest.skipIf(
  25     "tracedump" in config.excluded_plugins, "Exclude tracedump plugin tests"
  26 )
  27 class TestTracedump(VppTestCase):
  28     """Tracedump plugin tests"""
  29
  30     @classmethod
  31     def setUpClass(cls):
  32         super(TestTracedump, cls).setUpClass()
  33         cls.create_pg_interfaces(range(2))
  34         for i in cls.pg_interfaces:
  35             i.admin_up()
  36             i.config_ip4()
  37             i.resolve_arp()
  38
  39     @classmethod
  40     def tearDownClass(cls):
  41         for i in cls.pg_interfaces:
  42             i.unconfig_ip4()
  43             i.admin_down()
  44         super(TestTracedump, cls).tearDownClass()
  45
  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)
  50
  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(
  55             cursor=0xFFFFFFFF,
  56             index=0xFFFFFFFF,
  57             name="pg-input",
  58         )
  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,
  64             max_packets=5,
  65             use_filter=True,
  66             verbose=True,
  67             pre_capture_clear=True,
  68         )
  69
  70         reply = self.vapi.cli(
  71             "show graph node want_arcs input drop output punt handoff no_free polling interrupt"
  72         )
  73         self.assertIn("af-packet-input", reply)
  74
  75         self.pg_start()
  76         reply = self.vapi.trace_v2_dump(
  77             thread_id=0xFFFFFFFF, position=0, clear_cache=False
  78         )
  79         self.assertTrue(reply)
  80         reply = self.vapi.trace_filter_function_dump()
  81         self.assertTrue(reply[1].selected)
  82
  83     def test_tracedump_exclude(self):
  84         """Exclude node (no trace output)"""
  85         self.vapi.trace_clear_cache()
  86         self.vapi.trace_clear_capture()
  87
  88         packets = create_stream(self.pg0, self.pg1, 5)
  89         self.pg0.add_stream(packets)
  90
  91         # exclude node
  92         reply = self.vapi.graph_node_get(
  93             cursor=0xFFFFFFFF,
  94             index=0xFFFFFFFF,
  95             name="pg-input",
  96         )
  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,
 102             max_packets=5,
 103             use_filter=True,
 104             verbose=True,
 105             pre_capture_clear=True,
 106         )
 107         self.pg_start()
 108         reply = self.vapi.trace_v2_dump(thread_id=0, position=1, clear_cache=False)
 109         self.assertFalse(reply)
 110
 111
 112 if __name__ == "__main__":
 113     unittest.main(testRunner=VppTestRunner)