test/test_ping.py

   1 import socket
   2
   3 from scapy.layers.inet import IP, UDP, ICMP
   4 from scapy.layers.inet6 import IPv6
   5 from scapy.layers.l2 import Ether, GRE
   6 from scapy.packet import Raw
   7
   8 from framework import VppTestCase
   9 from util import ppp
  10
  11 """ TestPing is a subclass of  VPPTestCase classes.
  12
  13 Basic test for sanity check of the ping.
  14
  15 """
  16
  17
  18 class TestPing(VppTestCase):
  19     """ Ping Test Case """
  20
  21     @classmethod
  22     def setUpClass(cls):
  23         super(TestPing, cls).setUpClass()
  24         try:
  25             cls.create_pg_interfaces(range(2))
  26             cls.interfaces = list(cls.pg_interfaces)
  27
  28             for i in cls.interfaces:
  29                 i.admin_up()
  30                 i.config_ip4()
  31                 i.config_ip6()
  32                 i.disable_ipv6_ra()
  33                 i.resolve_arp()
  34                 i.resolve_ndp()
  35         except Exception:
  36             super(TestPing, cls).tearDownClass()
  37             raise
  38
  39     @classmethod
  40     def tearDownClass(cls):
  41         super(TestPing, cls).tearDownClass()
  42
  43     def tearDown(self):
  44         super(TestPing, self).tearDown()
  45         if not self.vpp_dead:
  46             self.vapi.cli("show hardware")
  47
  48     def test_ping_basic(self):
  49         """ basic ping test """
  50         try:
  51             self.pg_enable_capture(self.pg_interfaces)
  52             self.pg_start()
  53             self.logger.info(self.vapi.cli("show ip arp"))
  54             self.logger.info(self.vapi.cli("show ip6 neighbors"))
  55
  56             remote_ip4 = self.pg1.remote_ip4
  57             ping_cmd = "ping " + remote_ip4 + " interval 0.01 repeat 10"
  58             ret = self.vapi.cli(ping_cmd)
  59             self.logger.info(ret)
  60             out = self.pg1.get_capture(10)
  61             icmp_id = None
  62             icmp_seq = 1
  63             for p in out:
  64                 ip = p[IP]
  65                 self.assertEqual(ip.version, 4)
  66                 self.assertEqual(ip.flags, 0)
  67                 self.assertEqual(ip.src, self.pg1.local_ip4)
  68                 self.assertEqual(ip.dst, self.pg1.remote_ip4)
  69                 self.assertEqual(ip.proto, 1)
  70                 self.assertEqual(len(ip.options), 0)
  71                 self.assertGreaterEqual(ip.ttl, 254)
  72                 icmp = p[ICMP]
  73                 self.assertEqual(icmp.type, 8)
  74                 self.assertEqual(icmp.code, 0)
  75                 self.assertEqual(icmp.seq, icmp_seq)
  76                 icmp_seq = icmp_seq + 1
  77                 if icmp_id is None:
  78                     icmp_id = icmp.id
  79                 else:
  80                     self.assertEqual(icmp.id, icmp_id)
  81         finally:
  82             self.vapi.cli("show error")
  83
  84     def test_ping_burst(self):
  85         """ burst ping test """
  86         try:
  87             self.pg_enable_capture(self.pg_interfaces)
  88             self.pg_start()
  89             self.logger.info(self.vapi.cli("show ip arp"))
  90             self.logger.info(self.vapi.cli("show ip6 neighbors"))
  91
  92             remote_ip4 = self.pg1.remote_ip4
  93             ping_cmd = "ping " + remote_ip4 + " interval 0.01 burst 3"
  94             ret = self.vapi.cli(ping_cmd)
  95             self.logger.info(ret)
  96             out = self.pg1.get_capture(3*5)
  97             icmp_id = None
  98             icmp_seq = 1
  99             count = 0
 100             for p in out:
 101                 ip = p[IP]
 102                 self.assertEqual(ip.version, 4)
 103                 self.assertEqual(ip.flags, 0)
 104                 self.assertEqual(ip.src, self.pg1.local_ip4)
 105                 self.assertEqual(ip.dst, self.pg1.remote_ip4)
 106                 self.assertEqual(ip.proto, 1)
 107                 self.assertEqual(len(ip.options), 0)
 108                 self.assertGreaterEqual(ip.ttl, 254)
 109                 icmp = p[ICMP]
 110                 self.assertEqual(icmp.type, 8)
 111                 self.assertEqual(icmp.code, 0)
 112                 self.assertEqual(icmp.seq, icmp_seq)
 113                 count = count + 1
 114                 if count >= 3:
 115                     icmp_seq = icmp_seq + 1
 116                     count = 0
 117                 if icmp_id is None:
 118                     icmp_id = icmp.id
 119                 else:
 120                     self.assertEqual(icmp.id, icmp_id)
 121         finally:
 122             self.vapi.cli("show error")