test/test_gro.py

   1 #!/usr/bin/env python3
   2 """GRO functional tests"""
   3
   4 #
   5 # Add tests for:
   6 # - GRO
   7 # - Verify that sending 1500 Bytes frame without GRO enabled correctly
   8 # - Verify that sending 1500 Bytes frame with GRO enabled correctly
   9 #
  10 import unittest
  11
  12 from scapy.packet import Raw
  13 from scapy.layers.inet6 import IPv6, Ether, IP, UDP, ICMPv6PacketTooBig
  14 from scapy.layers.inet6 import ipv6nh, IPerror6
  15 from scapy.layers.inet import TCP, ICMP
  16 from scapy.data import ETH_P_IP, ETH_P_IPV6, ETH_P_ARP
  17
  18 from framework import VppTestCase, VppTestRunner
  19 from vpp_object import VppObject
  20 from vpp_interface import VppInterface
  21
  22
  23 """ Test_gro is a subclass of VPPTestCase classes.
  24     GRO tests.
  25 """
  26
  27
  28 class TestGRO(VppTestCase):
  29     """ GRO Test Case """
  30
  31     @classmethod
  32     def setUpClass(self):
  33         super(TestGRO, self).setUpClass()
  34         res = self.create_pg_interfaces(range(2))
  35         res_gro = self.create_pg_interfaces(range(2, 3), 1, 1460)
  36         self.create_pg_interfaces(range(3, 4), 1, 8940)
  37         self.pg_interfaces.append(res[0])
  38         self.pg_interfaces.append(res[1])
  39         self.pg_interfaces.append(res_gro[0])
  40         self.pg2.coalesce_enable()
  41         self.pg3.coalesce_enable()
  42
  43     @classmethod
  44     def tearDownClass(self):
  45         super(TestGRO, self).tearDownClass()
  46
  47     def setUp(self):
  48         super(TestGRO, self).setUp()
  49         for i in self.pg_interfaces:
  50             i.admin_up()
  51             i.config_ip4()
  52             i.config_ip6()
  53             i.disable_ipv6_ra()
  54             i.resolve_arp()
  55             i.resolve_ndp()
  56
  57     def tearDown(self):
  58         super(TestGRO, self).tearDown()
  59         if not self.vpp_dead:
  60             for i in self.pg_interfaces:
  61                 i.unconfig_ip4()
  62                 i.unconfig_ip6()
  63                 i.admin_down()
  64
  65     def test_gro(self):
  66         """ GRO test """
  67
  68         n_packets = 124
  69         #
  70         # Send 1500 bytes frame with gro disabled
  71         #
  72         p4 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
  73               IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4,
  74                  flags='DF') /
  75               TCP(sport=1234, dport=4321) /
  76               Raw(b'\xa5' * 1460))
  77
  78         rxs = self.send_and_expect(self.pg0, n_packets * p4, self.pg1)
  79         for rx in rxs:
  80             self.assertEqual(rx[Ether].src, self.pg1.local_mac)
  81             self.assertEqual(rx[Ether].dst, self.pg1.remote_mac)
  82             self.assertEqual(rx[IP].src, self.pg0.remote_ip4)
  83             self.assertEqual(rx[IP].dst, self.pg1.remote_ip4)
  84             self.assertEqual(rx[TCP].sport, 1234)
  85             self.assertEqual(rx[TCP].dport, 4321)
  86
  87         #
  88         # Send 1500 bytes frame with gro enabled on
  89         # output interfaces support GRO
  90         #
  91         p = []
  92         s = 0
  93         for n in range(0, n_packets):
  94             p.append((Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
  95                       IP(src=self.pg0.remote_ip4, dst=self.pg2.remote_ip4,
  96                          flags='DF') /
  97                       TCP(sport=1234, dport=4321, seq=s, ack=n, flags='A') /
  98                       Raw(b'\xa5' * 1460)))
  99             s += 1460
 100
 101         rxs = self.send_and_expect(self.pg0, p, self.pg2, n_rx=2)
 102
 103         i = 0
 104         for rx in rxs:
 105             i += 1
 106             self.assertEqual(rx[Ether].src, self.pg2.local_mac)
 107             self.assertEqual(rx[Ether].dst, self.pg2.remote_mac)
 108             self.assertEqual(rx[IP].src, self.pg0.remote_ip4)
 109             self.assertEqual(rx[IP].dst, self.pg2.remote_ip4)
 110             self.assertEqual(rx[IP].len, 64280)  # 1460 * 44 + 40 < 65536
 111             self.assertEqual(rx[TCP].sport, 1234)
 112             self.assertEqual(rx[TCP].dport, 4321)
 113             self.assertEqual(rx[TCP].ack, (44*i - 1))
 114
 115         p4_temp = (Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac) /
 116                    IP(src=self.pg2.remote_ip4, dst=self.pg0.remote_ip4,
 117                       flags='DF') /
 118                    TCP(sport=1234, dport=4321, flags='F'))
 119
 120         rxs = self.send_and_expect(self.pg2, 100*[p4_temp], self.pg0, n_rx=100)
 121         rx_coalesce = self.pg2.get_capture(1, timeout=1)
 122
 123         rx0 = rx_coalesce[0]
 124         self.assertEqual(rx0[Ether].src, self.pg2.local_mac)
 125         self.assertEqual(rx0[Ether].dst, self.pg2.remote_mac)
 126         self.assertEqual(rx0[IP].src, self.pg0.remote_ip4)
 127         self.assertEqual(rx0[IP].dst, self.pg2.remote_ip4)
 128         self.assertEqual(rx0[IP].len, 52600)  # 1460 * 36 + 40
 129         self.assertEqual(rx0[TCP].sport, 1234)
 130         self.assertEqual(rx0[TCP].dport, 4321)
 131
 132         for rx in rxs:
 133             self.assertEqual(rx[Ether].src, self.pg0.local_mac)
 134             self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
 135             self.assertEqual(rx[IP].src, self.pg2.remote_ip4)
 136             self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
 137             self.assertEqual(rx[IP].len, 40)
 138             self.assertEqual(rx[TCP].sport, 1234)
 139             self.assertEqual(rx[TCP].dport, 4321)
 140
 141 if __name__ == '__main__':
 142     unittest.main(testRunner=VppTestRunner)