X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_mtu.py;h=922d83dc5ef7f1a76b33ba2d4cc33d413b31b8bc;hb=683bdb6743e0568bbf02c1f4953d3f73e6f2b544;hp=e9b4ffed9c1b41f84c3427e41d8e716ada79c36f;hpb=282093f1fe783b5d36a014d4495995cd64e2e3fb;p=vpp.git diff --git a/test/test_mtu.py b/test/test_mtu.py index e9b4ffed9c1..922d83dc5ef 100644 --- a/test/test_mtu.py +++ b/test/test_mtu.py @@ -1,4 +1,4 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 """IP4 and IP6 MTU functional tests""" # @@ -13,32 +13,20 @@ from scapy.layers.inet6 import IPv6, Ether, IP, UDP, ICMPv6PacketTooBig from scapy.layers.inet import ICMP from framework import VppTestCase, VppTestRunner from vpp_ip import DpoProto -from vpp_ip_route import VppIpRoute, VppRoutePath +from vpp_ip_route import VppIpRoute, VppRoutePath, FibPathProto from socket import AF_INET, AF_INET6, inet_pton -import StringIO +from util import reassemble4 + """ Test_mtu is a subclass of VPPTestCase classes. MTU tests. """ -def reassemble(listoffragments): - buffer = StringIO.StringIO() - first = listoffragments[0] - buffer.seek(20) - for pkt in listoffragments: - # pkt.show2() - buffer.seek(pkt[IP].frag*8) - buffer.write(pkt[IP].payload) - first.len = len(buffer.getvalue()) + 20 - first.flags = 0 - del(first.chksum) - header = str(first[IP])[:20] - return first[IP].__class__(header + buffer.getvalue()) - - class TestMTU(VppTestCase): - """ MTU Test Case """ + """MTU Test Case""" + + maxDiff = None @classmethod def setUpClass(cls): @@ -46,9 +34,13 @@ class TestMTU(VppTestCase): cls.create_pg_interfaces(range(2)) cls.interfaces = list(cls.pg_interfaces) - def setUp(cls): - super(TestMTU, cls).setUp() - for i in cls.interfaces: + @classmethod + def tearDownClass(cls): + super(TestMTU, cls).tearDownClass() + + def setUp(self): + super(TestMTU, self).setUp() + for i in self.interfaces: i.admin_up() i.config_ip4() i.config_ip6() @@ -65,37 +57,35 @@ class TestMTU(VppTestCase): i.admin_down() def validate(self, rx, expected): - self.assertEqual(rx, expected.__class__(str(expected))) + self.assertEqual(rx, expected.__class__(expected)) def validate_bytes(self, rx, expected): self.assertEqual(rx, expected) def payload(self, len): - return 'x' * len + return "x" * len def get_mtu(self, sw_if_index): - rv = self.vapi.sw_interface_dump() + rv = self.vapi.sw_interface_dump(sw_if_index=sw_if_index) for i in rv: if i.sw_if_index == sw_if_index: return i.mtu[0] return 0 def test_ip4_mtu(self): - """ IP4 MTU test """ + """IP4 MTU test""" p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) - p_ip4 = IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, - flags='DF') + p_ip4 = IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, flags="DF") current_mtu = self.get_mtu(self.pg1.sw_if_index) - p_payload = UDP(sport=1234, dport=1234) / self.payload( - current_mtu - 20 - 8) + p_payload = UDP(sport=1234, dport=1234) / self.payload(current_mtu - 20 - 8) p4 = p_ether / p_ip4 / p_payload p4_reply = p_ip4 / p_payload p4_reply.ttl -= 1 - rx = self.send_and_expect(self.pg0, p4*11, self.pg1) + rx = self.send_and_expect(self.pg0, p4 * 11, self.pg1) for p in rx: self.validate(p[1], p4_reply) @@ -104,40 +94,44 @@ class TestMTU(VppTestCase): self.assertEqual(576, self.get_mtu(self.pg1.sw_if_index)) # Should fail. Too large MTU - p_icmp4 = ICMP(type='dest-unreach', code='fragmentation-needed', - nexthopmtu=576, chksum=0x2dbb) - icmp4_reply = (IP(src=self.pg0.local_ip4, - dst=self.pg0.remote_ip4, - ttl=254, len=576, id=0) / - p_icmp4 / p_ip4 / p_payload) - icmp4_reply[1].ttl -= 1 - n = icmp4_reply.__class__(str(icmp4_reply)) - s = str(icmp4_reply) + p_icmp4 = ICMP( + type="dest-unreach", + code="fragmentation-needed", + nexthopmtu=576, + chksum=0x2DBB, + ) + icmp4_reply = ( + IP(src=self.pg0.local_ip4, dst=self.pg0.remote_ip4, ttl=254, len=576, id=0) + / p_icmp4 + / p_ip4 + / p_payload + ) + n = icmp4_reply.__class__(icmp4_reply) + s = bytes(icmp4_reply) icmp4_reply = s[0:576] - rx = self.send_and_expect(self.pg0, p4*11, self.pg0) + rx = self.send_and_expect_some(self.pg0, p4 * 11, self.pg0) for p in rx: # p.show2() # n.show2() - self.validate_bytes(str(p[1]), icmp4_reply) + self.validate_bytes(bytes(p[1]), icmp4_reply) # Now with DF off. Expect fragments. # First go with 1500 byte packets. - p_payload = UDP(sport=1234, dport=1234) / self.payload( - 1500 - 20 - 8) + p_payload = UDP(sport=1234, dport=1234) / self.payload(1500 - 20 - 8) p4 = p_ether / p_ip4 / p_payload p4.flags = 0 p4_reply = p_ip4 / p_payload - p4_reply.ttl = 62 # check this + p4_reply.ttl = p_ip4.ttl - 1 p4_reply.flags = 0 p4_reply.id = 256 self.pg_enable_capture() - self.pg0.add_stream(p4*1) + self.pg0.add_stream(p4 * 1) self.pg_start() rx = self.pg1.get_capture(3) - reass_pkt = reassemble(rx) + reass_pkt = reassemble4(rx) self.validate(reass_pkt, p4_reply) - ''' + """ # Now what happens with a 9K frame p_payload = UDP(sport=1234, dport=1234) / self.payload( current_mtu - 20 - 8) @@ -152,31 +146,29 @@ class TestMTU(VppTestCase): self.pg0.add_stream(p4*1) self.pg_start() rx = self.pg1.get_capture(16) - reass_pkt = reassemble(rx) + reass_pkt = reassemble4(rx) reass_pkt.show2() p4_reply.show2() self.validate(reass_pkt, p4_reply) - ''' + """ # Reset MTU - self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, - [current_mtu, 0, 0, 0]) + self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [current_mtu, 0, 0, 0]) def test_ip6_mtu(self): - """ IP6 MTU test """ + """IP6 MTU test""" current_mtu = self.get_mtu(self.pg1.sw_if_index) p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) p_ip6 = IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6) - p_payload = UDP(sport=1234, dport=1234) / self.payload( - current_mtu - 40 - 8) + p_payload = UDP(sport=1234, dport=1234) / self.payload(current_mtu - 40 - 8) p6 = p_ether / p_ip6 / p_payload p6_reply = p_ip6 / p_payload p6_reply.hlim -= 1 - rx = self.send_and_expect(self.pg0, p6*9, self.pg1) + rx = self.send_and_expect(self.pg0, p6 * 9, self.pg1) for p in rx: self.validate(p[1], p6_reply) @@ -185,24 +177,25 @@ class TestMTU(VppTestCase): self.assertEqual(1280, self.get_mtu(self.pg1.sw_if_index)) # Should fail. Too large MTU - p_icmp6 = ICMPv6PacketTooBig(mtu=1280, cksum=0x4c7a) - icmp6_reply = (IPv6(src=self.pg0.local_ip6, - dst=self.pg0.remote_ip6, - hlim=255, plen=1240) / - p_icmp6 / p_ip6 / p_payload) + p_icmp6 = ICMPv6PacketTooBig(mtu=1280, cksum=0x4C7A) + icmp6_reply = ( + IPv6(src=self.pg0.local_ip6, dst=self.pg0.remote_ip6, hlim=255, plen=1240) + / p_icmp6 + / p_ip6 + / p_payload + ) icmp6_reply[2].hlim -= 1 - n = icmp6_reply.__class__(str(icmp6_reply)) - s = str(icmp6_reply) + n = icmp6_reply.__class__(icmp6_reply) + s = bytes(icmp6_reply) icmp6_reply_str = s[0:1280] - rx = self.send_and_expect(self.pg0, p6*9, self.pg0) + rx = self.send_and_expect_some(self.pg0, p6 * 9, self.pg0) for p in rx: - self.validate_bytes(str(p[1]), icmp6_reply_str) + self.validate_bytes(bytes(p[1]), icmp6_reply_str) # Reset MTU - self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, - [current_mtu, 0, 0, 0]) + self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [current_mtu, 0, 0, 0]) -if __name__ == '__main__': +if __name__ == "__main__": unittest.main(testRunner=VppTestRunner)