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         # pg-input node = 425
  54         self.vapi.trace_set_filters(flag=1, node_index=425, count=5)
  55         self.vapi.trace_capture_packets(
  56             node_index=425,
  57             max_packets=5,
  58             use_filter=True,
  59             verbose=True,
  60             pre_capture_clear=True,
  61         )
  62
  63         reply = self.vapi.cli(
  64             "show graph node want_arcs input drop output punt handoff no_free polling interrupt"
  65         )
  66         self.assertIn("af-packet-input", reply)
  67
  68         self.pg_start()
  69         reply = self.vapi.graph_node_get(cursor=ctypes.c_uint32(~0).value, index=425)
  70         self.assertTrue(reply[1][0].name == "pg-input")
  71         reply = self.vapi.trace_v2_dump(
  72             thread_id=ctypes.c_uint32(~0).value, position=0, clear_cache=False
  73         )
  74         self.assertTrue(reply)
  75         reply = self.vapi.trace_filter_function_dump()
  76         self.assertTrue(reply[1].selected)
  77
  78     def test_tracedump_exclude(self):
  79         """Exclude node (no trace output)"""
  80         self.vapi.trace_clear_cache()
  81         self.vapi.trace_clear_capture()
  82
  83         packets = create_stream(self.pg0, self.pg1, 5)
  84         self.pg0.add_stream(packets)
  85
  86         # exclude node
  87         self.vapi.trace_set_filters(flag=2, node_index=425, count=5)
  88         self.vapi.trace_capture_packets(
  89             node_index=425,
  90             max_packets=5,
  91             use_filter=True,
  92             verbose=True,
  93             pre_capture_clear=True,
  94         )
  95         self.pg_start()
  96         reply = self.vapi.trace_v2_dump(thread_id=0, position=1, clear_cache=False)
  97         self.assertFalse(reply)
  98
  99
 100 if __name__ == "__main__":
 101     unittest.main(testRunner=VppTestRunner)