test/test_mdata.py

   1 from framework import VppTestCase, VppTestRunner
   2 import unittest
   3 from config import config
   4 from scapy.layers.l2 import Ether
   5 from scapy.packet import Raw
   6 from scapy.layers.inet import IP, UDP
   7 from random import randint
   8 from util import ppp
   9
  10
  11 @unittest.skipIf("mdata" in config.excluded_plugins, "Exclude mdata plugin tests")
  12 class TestMdataCli(VppTestCase):
  13     """mdata plugin test"""
  14
  15     @classmethod
  16     def setUpClass(cls):
  17         super(TestMdataCli, cls).setUpClass()
  18         try:
  19             cls.create_pg_interfaces(range(2))
  20             for i in cls.pg_interfaces:
  21                 i.config_ip4()
  22                 i.resolve_arp()
  23                 i.admin_up()
  24         except Exception:
  25             cls.tearDownClass()
  26             raise
  27
  28     @classmethod
  29     def tearDownClass(cls):
  30         for i in cls.pg_interfaces:
  31             i.unconfig_ip4()
  32             i.admin_down()
  33         super(TestMdataCli, cls).tearDownClass()
  34
  35     # https://fd.io/docs/vpp/master/developer/tests/overview.html#example-how-to-add-a-new-test
  36     def create_stream(self, src_if, dst_if, count):
  37         packets = []
  38         for i in range(count):
  39             info = self.create_packet_info(src_if, dst_if)
  40             payload = self.info_to_payload(info)
  41
  42             p = (
  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)
  46                 / Raw(payload)
  47             )
  48
  49             info.data = p.copy()
  50             packets.append(p)
  51
  52         return packets
  53
  54     def verify_capture(self, src_if, dst_if, capture):
  55         packet_info = None
  56         for packet in capture:
  57             try:
  58                 ip = packet[IP]
  59                 udp = packet[UDP]
  60                 # convert the payload to packet info object
  61                 payload_info = self.payload_to_info(packet[Raw])
  62                 # make sure the indexes match
  63                 self.assert_equal(
  64                     payload_info.src, src_if.sw_if_index, "source sw_if_index"
  65                 )
  66                 self.assert_equal(
  67                     payload_info.dst, dst_if.sw_if_index, "destination sw_if_index"
  68                 )
  69                 packet_info = self.get_next_packet_info_for_interface2(
  70                     src_if.sw_if_index, dst_if.sw_if_index, packet_info
  71                 )
  72                 # make sure we didn't run out of saved packets
  73                 self.assertIsNotNone(packet_info)
  74                 self.assert_equal(
  75                     payload_info.index, packet_info.index, "packet info index"
  76                 )
  77                 saved_packet = packet_info.data  # fetch the saved packet
  78                 # assert the values match
  79                 self.assert_equal(ip.src, saved_packet[IP].src, "IP source address")
  80                 # ... more assertions here
  81                 self.assert_equal(udp.sport, saved_packet[UDP].sport, "UDP source port")
  82             except Exception:
  83                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
  84                 raise
  85         remaining_packet = self.get_next_packet_info_for_interface2(
  86             src_if.sw_if_index, dst_if.sw_if_index, packet_info
  87         )
  88         self.assertIsNone(
  89             remaining_packet,
  90             "Interface %s: Packet expected from interface "
  91             "%s didn't arrive" % (dst_if.name, src_if.name),
  92         )
  93
  94     def test_mdata_cli(self):
  95         """turn on mdata tracking, send packets, verify, check CLI output"""
  96         self.vapi.cli("buffer metadata tracking on")
  97
  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()
 102         self.pg_start()
 103
 104         capture = self.pg1.get_capture()
 105         self.pg0.assert_nothing_captured()
 106         self.verify_capture(self.pg0, self.pg1, capture)
 107
 108         result = self.vapi.cli("show buffer metadata")
 109         expected = [
 110             "ip4-input",
 111             "ip4-rewrite",
 112             "ip4-lookup",
 113             "ethernet-input",
 114             "pg1-tx",
 115             "pg1-output",
 116         ]
 117         for entry in expected:
 118             self.assertIn(entry, result)
 119         self.vapi.cli("buffer metadata tracking off")
 120
 121
 122 if __name__ == "__main__":
 123     unittest.main(testRunner=VppTestRunner)