X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_ping.py;h=c2eb8b75299ebfaf650f7a1c657ac64e572fa7c6;hb=34850e01876005422ba9523df5ae0400964e1c91;hp=4f3921e992e9de2af9ea9c854bbb03518ca0e50f;hpb=e3d52803149d58737aad38435361e692d70c9d7a;p=vpp.git diff --git a/test/test_ping.py b/test/test_ping.py index 4f3921e992e..c2eb8b75299 100644 --- a/test/test_ping.py +++ b/test/test_ping.py @@ -7,6 +7,8 @@ from scapy.packet import Raw from framework import VppTestCase from util import ppp +from vpp_ip_route import VppIpInterfaceAddress, VppIpRoute, VppRoutePath +from vpp_neighbor import VppNeighbor """ TestPing is a subclass of VPPTestCase classes. @@ -16,7 +18,7 @@ Basic test for sanity check of the ping. class TestPing(VppTestCase): - """ Ping Test Case """ + """Ping Test Case""" @classmethod def setUpClass(cls): @@ -36,17 +38,37 @@ class TestPing(VppTestCase): super(TestPing, cls).tearDownClass() raise + @classmethod + def tearDownClass(cls): + super(TestPing, cls).tearDownClass() + def tearDown(self): super(TestPing, self).tearDown() - if not self.vpp_dead: - self.vapi.cli("show hardware") + + def show_commands_at_teardown(self): + self.logger.info(self.vapi.cli("show hardware")) + + def verify_ping_request(self, p, src, dst, seq): + ip = p[IP] + self.assertEqual(ip.version, 4) + self.assertEqual(ip.flags, 0) + self.assertEqual(ip.src, src) + self.assertEqual(ip.dst, dst) + self.assertEqual(ip.proto, 1) + self.assertEqual(len(ip.options), 0) + self.assertGreaterEqual(ip.ttl, 254) + icmp = p[ICMP] + self.assertEqual(icmp.type, 8) + self.assertEqual(icmp.code, 0) + self.assertEqual(icmp.seq, seq) + return icmp def test_ping_basic(self): - """ basic ping test """ + """basic ping test""" try: self.pg_enable_capture(self.pg_interfaces) self.pg_start() - self.logger.info(self.vapi.cli("show ip arp")) + self.logger.info(self.vapi.cli("show ip4 neighbors")) self.logger.info(self.vapi.cli("show ip6 neighbors")) remote_ip4 = self.pg1.remote_ip4 @@ -57,18 +79,9 @@ class TestPing(VppTestCase): icmp_id = None icmp_seq = 1 for p in out: - ip = p[IP] - self.assertEqual(ip.version, 4) - self.assertEqual(ip.flags, 0) - self.assertEqual(ip.src, self.pg1.local_ip4) - self.assertEqual(ip.dst, self.pg1.remote_ip4) - self.assertEqual(ip.proto, 1) - self.assertEqual(len(ip.options), 0) - self.assertGreaterEqual(ip.ttl, 254) - icmp = p[ICMP] - self.assertEqual(icmp.type, 8) - self.assertEqual(icmp.code, 0) - self.assertEqual(icmp.seq, 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 @@ -78,34 +91,24 @@ 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() - self.logger.info(self.vapi.cli("show ip arp")) - self.logger.info(self.vapi.cli("show ip6 neighbors")) + self.logger.info(self.vapi.cli("show ip neighbors")) remote_ip4 = self.pg1.remote_ip4 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: - ip = p[IP] - self.assertEqual(ip.version, 4) - self.assertEqual(ip.flags, 0) - self.assertEqual(ip.src, self.pg1.local_ip4) - self.assertEqual(ip.dst, self.pg1.remote_ip4) - self.assertEqual(ip.proto, 1) - self.assertEqual(len(ip.options), 0) - self.assertGreaterEqual(ip.ttl, 254) - icmp = p[ICMP] - self.assertEqual(icmp.type, 8) - self.assertEqual(icmp.code, 0) - self.assertEqual(icmp.seq, 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 @@ -116,3 +119,63 @@ class TestPing(VppTestCase): self.assertEqual(icmp.id, icmp_id) finally: self.vapi.cli("show error") + + def test_ping_src(self): + """ping with source address set""" + + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + self.logger.info(self.vapi.cli("show ip4 neighbors")) + self.logger.info(self.vapi.cli("show ip6 neighbors")) + + 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() + + 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_seq = icmp_seq + 1 + + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + ping_cmd = "ping %s interval 0.01 repeat 3" % nbr_addr + 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, "10.0.0.1", nbr_addr, icmp_seq) + icmp_seq = icmp_seq + 1 + + def test_ping_fib_routed_dst(self): + """ping destination routed according to FIB table""" + + try: + self.pg1.generate_remote_hosts(1) + self.pg_enable_capture(self.pg_interfaces) + 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() + 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) + icmp_seq = icmp_seq + 1 + finally: + self.vapi.cli("show error")