X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_ipip.py;h=00721ec90a097ab070e54d39c2e5f29de9df0fd5;hb=refs%2Fchanges%2F97%2F13197%2F1;hp=1ae4de0adf8254a03ceb579d011b0d443baa86ae;hpb=298c69510ff4b64a262d465eb8877c4e7f4e60e0;p=vpp.git diff --git a/test/test_ipip.py b/test/test_ipip.py index 1ae4de0adf8..00721ec90a0 100644 --- a/test/test_ipip.py +++ b/test/test_ipip.py @@ -1,17 +1,11 @@ -# -# IP{4,6} over IP{v,6} tunnel functional tests -# +#!/usr/bin/env python +"""IP{4,6} over IP{v,6} tunnel functional tests""" import unittest -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 framework import VppTestCase +from scapy.layers.inet6 import IPv6, Ether, IP, UDP +from framework import VppTestCase, VppTestRunner from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto -from util import ppp -from ipaddress import * +from socket import AF_INET, AF_INET6, inet_pton """ Testipip is a subclass of VPPTestCase classes. @@ -26,71 +20,49 @@ class TestIPIP(VppTestCase): @classmethod def setUpClass(cls): super(TestIPIP, cls).setUpClass() - try: - cls.create_pg_interfaces(range(2)) - cls.interfaces = list(cls.pg_interfaces) - except Exception: - super(TestIPIP, cls).tearDownClass() - raise + cls.create_pg_interfaces(range(2)) + cls.interfaces = list(cls.pg_interfaces) def setUp(cls): super(TestIPIP, cls).setUp() - try: - for i in cls.interfaces: - i.admin_up() - i.config_ip4() - i.config_ip6() - i.disable_ipv6_ra() - i.resolve_arp() - i.resolve_ndp() - except Exception: - super(TestIPIP, cls).tearDown() - raise + for i in cls.interfaces: + i.admin_up() + i.config_ip4() + i.config_ip6() + i.disable_ipv6_ra() + i.resolve_arp() + i.resolve_ndp() def tearDown(self): super(TestIPIP, self).tearDown() if not self.vpp_dead: - self.vapi.cli("show hardware") - for i in self.pg_interfaces: - i.unconfig_ip4() - i.unconfig_ip6() - i.admin_down() - self.vapi.cli("show error") + for i in self.pg_interfaces: + i.unconfig_ip4() + i.unconfig_ip6() + i.admin_down() def validate(self, rx, expected): - expected = expected.__class__(str(expected)) - if rx != expected: - print('RX packet:') - print(rx.show()) - print('EXPECTED packet:') - print(expected.show()) - self.assertDictEqual(rx, expected) - - def payload(self, len): - return 'x' * len + self.assertEqual(rx, expected.__class__(str(expected))) def test_ipip4(self): """ ip{v4,v6} over ip4 test """ p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) - p_ip6 = IPv6(src="1::1", dst="DEAD::1", nh='UDP') - p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1") + p_ip6 = IPv6(src="1::1", dst="DEAD::1", nh='UDP', tc=42) + p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42) p_payload = UDP(sport=1234, dport=1234) # IPv4 transport rv = self.vapi.ipip_add_tunnel( - src_address=str(ip_address(self.pg0.local_ip4).packed), - dst_address=str(ip_address(self.pg1.remote_ip4).packed), - is_ipv6=0) - self.assertEqual(rv.retval, 0) + src_address=self.pg0.local_ip4n, + dst_address=self.pg1.remote_ip4n, + is_ipv6=0, tc_tos=0xFF) sw_if_index = rv.sw_if_index # Set interface up and enable IP on it - rv = self.vapi.sw_interface_set_flags(sw_if_index, 1) - self.assertEqual(rv.retval, 0) - rv = self.vapi.sw_interface_set_unnumbered( + self.vapi.sw_interface_set_flags(sw_if_index, 1) + self.vapi.sw_interface_set_unnumbered( ip_sw_if_index=self.pg0.sw_if_index, sw_if_index=sw_if_index) - self.assertEqual(rv.retval, 0) # Add IPv4 and IPv6 routes via tunnel interface ip4_via_tunnel = VppIpRoute( @@ -112,7 +84,7 @@ class TestIPIP(VppTestCase): p_inner_ip6 = p_ip6 p_inner_ip6.hlim -= 1 p6_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4, - proto='ipv6', id=0) / p_inner_ip6 / p_payload) + proto='ipv6', id=0, tos=42) / p_inner_ip6 / p_payload) p6_reply.ttl -= 1 rx = self.send_and_expect(self.pg0, p6*10, self.pg1) for p in rx: @@ -122,8 +94,9 @@ class TestIPIP(VppTestCase): p4 = (p_ether / p_ip4 / p_payload) p_ip4_inner = p_ip4 p_ip4_inner.ttl -= 1 - p4_reply = (IP(src=self.pg0.local_ip4, - dst=self.pg1.remote_ip4) / p_ip4_inner / p_payload) + p4_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4, + tos=42) / + p_ip4_inner / p_payload) p4_reply.ttl -= 1 p4_reply.id = 0 rx = self.send_and_expect(self.pg0, p4*10, self.pg1) @@ -156,23 +129,20 @@ class TestIPIP(VppTestCase): def test_ipip6(self): """ ip{v4,v6} over ip6 test """ p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) - p_ip6 = IPv6(src="1::1", dst="DEAD::1", nh='UDP') - p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1") + p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP') + p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42) p_payload = UDP(sport=1234, dport=1234) # IPv6 transport rv = self.vapi.ipip_add_tunnel( - src_address=str(ip_address(self.pg0.local_ip6).packed), - dst_address=str(ip_address(self.pg1.remote_ip6).packed)) - self.assertEqual(rv.retval, 0) + src_address=self.pg0.local_ip6n, + dst_address=self.pg1.remote_ip6n, tc_tos=255) sw_if_index = rv.sw_if_index - rv = self.vapi.sw_interface_set_flags(sw_if_index, 1) - self.assertEqual(rv.retval, 0) - rv = self.vapi.sw_interface_set_unnumbered( + self.vapi.sw_interface_set_flags(sw_if_index, 1) + self.vapi.sw_interface_set_unnumbered( ip_sw_if_index=self.pg0.sw_if_index, sw_if_index=sw_if_index) - self.assertEqual(rv.retval, 0) # Add IPv4 and IPv6 routes via tunnel interface ip4_via_tunnel = VppIpRoute( @@ -193,19 +163,21 @@ class TestIPIP(VppTestCase): # IPv6 in to IPv6 tunnel p6 = (p_ether / p_ip6 / p_payload) - p6_reply = (IPv6(src=self.pg0.local_ip6, - dst=self.pg1.remote_ip6, hlim=63) / p_ip6 / p_payload) + p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6, + hlim=63, tc=42) / + p_ip6 / p_payload) p6_reply[1].hlim -= 1 - rx = self.send_and_expect(self.pg0, p6*10, self.pg1) + rx = self.send_and_expect(self.pg0, p6*11, self.pg1) for p in rx: self.validate(p[1], p6_reply) # IPv4 in to IPv6 tunnel p4 = (p_ether / p_ip4 / p_payload) p4_reply = (IPv6(src=self.pg0.local_ip6, - dst=self.pg1.remote_ip6, hlim=63) / p_ip4 / p_payload) + dst=self.pg1.remote_ip6, hlim=63, tc=42) / + p_ip4 / p_payload) p4_reply[1].ttl -= 1 - rx = self.send_and_expect(self.pg0, p4*10, self.pg1) + rx = self.send_and_expect(self.pg0, p4*11, self.pg1) for p in rx: self.validate(p[1], p4_reply) @@ -219,7 +191,7 @@ class TestIPIP(VppTestCase): dst=self.pg0.local_ip6) / p_ip4 / p_payload) p4_reply = (p_ip4 / p_payload) p4_reply.ttl -= 1 - rx = self.send_and_expect(self.pg1, p4*10, self.pg0) + rx = self.send_and_expect(self.pg1, p4*11, self.pg0) for p in rx: self.validate(p[1], p4_reply) @@ -229,19 +201,17 @@ class TestIPIP(VppTestCase): dst=self.pg0.local_ip6) / p_ip6 / p_payload) p6_reply = (p_ip6 / p_payload) p6_reply.hlim = 63 - rx = self.send_and_expect(self.pg1, p6*10, self.pg0) + rx = self.send_and_expect(self.pg1, p6*11, self.pg0) for p in rx: self.validate(p[1], p6_reply) def test_ipip_create(self): """ ipip create / delete interface test """ rv = self.vapi.ipip_add_tunnel( - src_address=str(ip_address('1.2.3.4').packed), - dst_address=str(ip_address('2.3.4.5').packed), is_ipv6=0) - self.assertEqual(rv.retval, 0) + src_address=inet_pton(AF_INET, '1.2.3.4'), + dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0) sw_if_index = rv.sw_if_index - rv = self.vapi.ipip_del_tunnel(sw_if_index) - self.assertEqual(rv.retval, 0) + self.vapi.ipip_del_tunnel(sw_if_index) if __name__ == '__main__':