diff options
Diffstat (limited to 'test/test_ping.py')
| -rw-r--r-- | test/test_ping.py | 87 |
1 files changed, 60 insertions, 27 deletions
diff --git a/test/test_ping.py b/test/test_ping.py index 8c5c087b0c5..f3da7eb96ca 100644 --- a/test/test_ping.py +++ b/test/test_ping.py @@ -1,12 +1,6 @@ -import socket - -from scapy.layers.inet import IP, UDP, ICMP -from scapy.layers.inet6 import IPv6 -from scapy.layers.l2 import Ether, GRE -from scapy.packet import Raw +from scapy.layers.inet import IP, ICMP from framework import VppTestCase -from util import ppp from vpp_ip_route import VppIpInterfaceAddress, VppIpRoute, VppRoutePath from vpp_neighbor import VppNeighbor @@ -18,7 +12,7 @@ Basic test for sanity check of the ping. class TestPing(VppTestCase): - """ Ping Test Case """ + """Ping Test Case""" @classmethod def setUpClass(cls): @@ -64,7 +58,7 @@ class TestPing(VppTestCase): return icmp def test_ping_basic(self): - """ basic ping test """ + """basic ping test""" try: self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -79,8 +73,9 @@ class TestPing(VppTestCase): icmp_id = None icmp_seq = 1 for p in out: - icmp = self.verify_ping_request(p, self.pg1.local_ip4, - self.pg1.remote_ip4, icmp_seq) + icmp = self.verify_ping_request( + p, self.pg1.local_ip4, self.pg1.remote_ip4, icmp_seq + ) icmp_seq = icmp_seq + 1 if icmp_id is None: icmp_id = icmp.id @@ -90,7 +85,7 @@ class TestPing(VppTestCase): self.vapi.cli("show error") def test_ping_burst(self): - """ burst ping test """ + """burst ping test""" try: self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -100,13 +95,14 @@ class TestPing(VppTestCase): ping_cmd = "ping " + remote_ip4 + " interval 0.01 burst 3" ret = self.vapi.cli(ping_cmd) self.logger.info(ret) - out = self.pg1.get_capture(3*5) + out = self.pg1.get_capture(3 * 5) icmp_id = None icmp_seq = 1 count = 0 for p in out: - icmp = self.verify_ping_request(p, self.pg1.local_ip4, - self.pg1.remote_ip4, icmp_seq) + icmp = self.verify_ping_request( + p, self.pg1.local_ip4, self.pg1.remote_ip4, icmp_seq + ) count = count + 1 if count >= 3: icmp_seq = icmp_seq + 1 @@ -119,7 +115,7 @@ class TestPing(VppTestCase): self.vapi.cli("show error") def test_ping_src(self): - """ ping with source address set """ + """ping with source address set""" self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -128,17 +124,18 @@ class TestPing(VppTestCase): nbr_addr = "10.0.0.2" VppIpInterfaceAddress(self, self.pg1, "10.0.0.1", 24).add_vpp_config() - VppNeighbor(self, self.pg1.sw_if_index, - "00:11:22:33:44:55", - nbr_addr).add_vpp_config() + VppNeighbor( + self, self.pg1.sw_if_index, "00:11:22:33:44:55", nbr_addr + ).add_vpp_config() ping_cmd = "ping %s interval 0.01 repeat 3" % self.pg1.remote_ip4 ret = self.vapi.cli(ping_cmd) out = self.pg1.get_capture(3) icmp_seq = 1 for p in out: - icmp = self.verify_ping_request(p, self.pg1.local_ip4, - self.pg1.remote_ip4, icmp_seq) + icmp = self.verify_ping_request( + p, self.pg1.local_ip4, self.pg1.remote_ip4, icmp_seq + ) icmp_seq = icmp_seq + 1 self.pg_enable_capture(self.pg_interfaces) @@ -152,7 +149,7 @@ class TestPing(VppTestCase): icmp_seq = icmp_seq + 1 def test_ping_fib_routed_dst(self): - """ ping destination routed according to FIB table """ + """ping destination routed according to FIB table""" try: self.pg1.generate_remote_hosts(1) @@ -160,17 +157,53 @@ class TestPing(VppTestCase): self.pg_start() routed_dst = "10.0.2.0" self.logger.info(self.vapi.cli("show ip4 neighbors")) - VppIpRoute(self, routed_dst, 24, - [VppRoutePath(self.pg1.remote_hosts[0].ip4, - self.pg1.sw_if_index)]).add_vpp_config() + VppIpRoute( + self, + routed_dst, + 24, + [VppRoutePath(self.pg1.remote_hosts[0].ip4, self.pg1.sw_if_index)], + ).add_vpp_config() ping_cmd = "ping %s interval 0.01 repeat 3" % routed_dst ret = self.vapi.cli(ping_cmd) self.logger.info(ret) out = self.pg1.get_capture(3) icmp_seq = 1 for p in out: - self.verify_ping_request(p, self.pg1.local_ip4, routed_dst, - icmp_seq) + self.verify_ping_request(p, self.pg1.local_ip4, routed_dst, icmp_seq) + icmp_seq = icmp_seq + 1 + finally: + self.vapi.cli("show error") + + def test_ping_api(self): + """ping api""" + + try: + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + ret = self.vapi.want_ping_finished_events( + address=self.pg1.remote_ip4, + repeat=4, + interval=0.2, + ) + self.logger.info(ret) + timeout = 1 + + ev = self.vapi.wait_for_event(timeout, "ping_finished_event") + self.logger.info(ev) + self.assertEqual(ev.request_count, 4) + + out = self.pg1.get_capture(4) + icmp_id = None + icmp_seq = 1 + for p in out: + icmp = self.verify_ping_request( + p, self.pg1.local_ip4, self.pg1.remote_ip4, icmp_seq + ) icmp_seq = icmp_seq + 1 + if icmp_id is None: + icmp_id = icmp.id + else: + self.assertEqual(icmp.id, icmp_id) finally: self.vapi.cli("show error") |