import scapy.compat
from scapy.packet import Raw
from scapy.layers.l2 import Ether, GRE
-from scapy.layers.inet import IP, UDP, ICMP
+from scapy.layers.inet import IP, UDP, ICMP, icmptypes
from scapy.layers.inet6 import HBHOptUnknown, ICMPv6ParamProblem,\
ICMPv6TimeExceeded, IPv6, IPv6ExtHdrFragment,\
- IPv6ExtHdrHopByHop, IPv6ExtHdrDestOpt, PadN, ICMPv6EchoRequest
+ IPv6ExtHdrHopByHop, IPv6ExtHdrDestOpt, PadN, ICMPv6EchoRequest,\
+ ICMPv6EchoReply
from framework import VppTestCase, VppTestRunner
from util import ppp, ppc, fragment_rfc791, fragment_rfc8200
from vpp_gre_interface import VppGreInterface
@classmethod
def setUpClass(cls):
- super(TestIPv4Reassembly, cls).setUpClass()
+ super().setUpClass()
cls.create_pg_interfaces([0, 1])
cls.src_if = cls.pg0
@classmethod
def tearDownClass(cls):
- super(TestIPv4Reassembly, cls).tearDownClass()
+ super().tearDownClass()
def setUp(self):
""" Test setup - force timeout on existing reassemblies """
- super(TestIPv4Reassembly, self).setUp()
+ super().setUp()
self.vapi.ip_reassembly_enable_disable(
sw_if_index=self.src_if.sw_if_index, enable_ip4=True)
self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
expire_walk_interval_ms=10000)
def tearDown(self):
- super(TestIPv4Reassembly, self).tearDown()
+ self.vapi.ip_reassembly_enable_disable(
+ sw_if_index=self.src_if.sw_if_index, enable_ip4=False)
+ super().tearDown()
def show_commands_at_teardown(self):
self.logger.debug(self.vapi.ppcli("show ip4-full-reassembly details"))
self.verify_capture(packets, dropped_packet_indexes)
self.src_if.assert_nothing_captured()
+ def test_local_enable_disable(self):
+ """ local reassembly enabled/disable """
+ self.vapi.ip_reassembly_enable_disable(
+ sw_if_index=self.src_if.sw_if_index, enable_ip4=False)
+ self.vapi.ip_local_reass_enable_disable(enable_ip4=True)
+ p = (Ether(src=self.src_if.remote_mac, dst=self.src_if.local_mac) /
+ IP(src=self.src_if.remote_ip4, dst=self.src_if.local_ip4) /
+ ICMP(id=1234, type='echo-request') /
+ Raw('x' * 1000))
+ frags = fragment_rfc791(p, 400)
+ r = self.send_and_expect(self.src_if, frags, self.src_if, n_rx=1)[0]
+ self.assertEqual(1234, r[ICMP].id)
+ self.assertEqual(icmptypes[r[ICMP].type], 'echo-reply')
+ self.vapi.ip_local_reass_enable_disable()
+
+ self.send_and_assert_no_replies(self.src_if, frags)
+ self.vapi.ip_local_reass_enable_disable(enable_ip4=True)
+
class TestIPv4SVReassembly(VppTestCase):
""" IPv4 Shallow Virtual Reassembly """
@classmethod
def setUpClass(cls):
- super(TestIPv4SVReassembly, cls).setUpClass()
+ super().setUpClass()
cls.create_pg_interfaces([0, 1])
cls.src_if = cls.pg0
def setUp(self):
""" Test setup - force timeout on existing reassemblies """
- super(TestIPv4SVReassembly, self).setUp()
+ super().setUp()
self.vapi.ip_reassembly_enable_disable(
sw_if_index=self.src_if.sw_if_index, enable_ip4=True,
type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL)
expire_walk_interval_ms=10000)
def tearDown(self):
- super(TestIPv4SVReassembly, self).tearDown()
+ super().tearDown()
self.logger.debug(self.vapi.ppcli("show ip4-sv-reassembly details"))
self.logger.debug(self.vapi.ppcli("show buffers"))
@classmethod
def setUpClass(cls):
- super(TestIPv4MWReassembly, cls).setUpClass()
+ super().setUpClass()
cls.create_pg_interfaces(range(cls.vpp_worker_count+1))
cls.src_if = cls.pg0
@classmethod
def tearDownClass(cls):
- super(TestIPv4MWReassembly, cls).tearDownClass()
+ super().tearDownClass()
def setUp(self):
""" Test setup - force timeout on existing reassemblies """
- super(TestIPv4MWReassembly, self).setUp()
+ super().setUp()
for intf in self.send_ifs:
self.vapi.ip_reassembly_enable_disable(
sw_if_index=intf.sw_if_index, enable_ip4=True)
expire_walk_interval_ms=10000)
def tearDown(self):
- super(TestIPv4MWReassembly, self).tearDown()
+ for intf in self.send_ifs:
+ self.vapi.ip_reassembly_enable_disable(
+ sw_if_index=intf.sw_if_index, enable_ip4=False)
+ super().tearDown()
def show_commands_at_teardown(self):
self.logger.debug(self.vapi.ppcli("show ip4-full-reassembly details"))
@classmethod
def setUpClass(cls):
- super(TestIPv6Reassembly, cls).setUpClass()
+ super().setUpClass()
cls.create_pg_interfaces([0, 1])
cls.src_if = cls.pg0
@classmethod
def tearDownClass(cls):
- super(TestIPv6Reassembly, cls).tearDownClass()
+ super().tearDownClass()
def setUp(self):
""" Test setup - force timeout on existing reassemblies """
- super(TestIPv6Reassembly, self).setUp()
+ super().setUp()
self.vapi.ip_reassembly_enable_disable(
sw_if_index=self.src_if.sw_if_index, enable_ip6=True)
self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
self.logger.debug(self.vapi.ppcli("show buffers"))
def tearDown(self):
- super(TestIPv6Reassembly, self).tearDown()
+ self.vapi.ip_reassembly_enable_disable(
+ sw_if_index=self.src_if.sw_if_index, enable_ip6=False)
+ super().tearDown()
def show_commands_at_teardown(self):
self.logger.debug(self.vapi.ppcli("show ip6-full-reassembly details"))
rx = self.send_and_expect(self.pg0, [pkt], self.pg0)
self.assertNotIn(IPv6ExtHdrFragment, rx)
+ def test_local_enable_disable(self):
+ """ local reassembly enabled/disable """
+ self.vapi.ip_reassembly_enable_disable(
+ sw_if_index=self.src_if.sw_if_index, enable_ip6=False)
+ self.vapi.ip_local_reass_enable_disable(enable_ip6=True)
+ pkt = (Ether(src=self.src_if.local_mac, dst=self.src_if.remote_mac) /
+ IPv6(src=self.src_if.remote_ip6, dst=self.src_if.local_ip6) /
+ ICMPv6EchoRequest(id=1234)/Raw('X' * 1600))
+ frags = fragment_rfc8200(pkt, 1, 400)
+ r = self.send_and_expect(self.src_if, frags, self.src_if, n_rx=1)[0]
+ self.assertEqual(1234, r[ICMPv6EchoReply].id)
+ self.vapi.ip_local_reass_enable_disable()
+
+ self.send_and_assert_no_replies(self.src_if, frags)
+ self.vapi.ip_local_reass_enable_disable(enable_ip6=True)
+
class TestIPv6MWReassembly(VppTestCase):
""" IPv6 Reassembly (multiple workers) """
@classmethod
def setUpClass(cls):
- super(TestIPv6MWReassembly, cls).setUpClass()
+ super().setUpClass()
cls.create_pg_interfaces(range(cls.vpp_worker_count+1))
cls.src_if = cls.pg0
@classmethod
def tearDownClass(cls):
- super(TestIPv6MWReassembly, cls).tearDownClass()
+ super().tearDownClass()
def setUp(self):
""" Test setup - force timeout on existing reassemblies """
- super(TestIPv6MWReassembly, self).setUp()
+ super().setUp()
for intf in self.send_ifs:
self.vapi.ip_reassembly_enable_disable(
sw_if_index=intf.sw_if_index, enable_ip6=True)
expire_walk_interval_ms=1000, is_ip6=1)
def tearDown(self):
- super(TestIPv6MWReassembly, self).tearDown()
+ for intf in self.send_ifs:
+ self.vapi.ip_reassembly_enable_disable(
+ sw_if_index=intf.sw_if_index, enable_ip6=False)
+ super().tearDown()
def show_commands_at_teardown(self):
self.logger.debug(self.vapi.ppcli("show ip6-full-reassembly details"))
@classmethod
def setUpClass(cls):
- super(TestIPv6SVReassembly, cls).setUpClass()
+ super().setUpClass()
cls.create_pg_interfaces([0, 1])
cls.src_if = cls.pg0
def setUp(self):
""" Test setup - force timeout on existing reassemblies """
- super(TestIPv6SVReassembly, self).setUp()
+ super().setUp()
self.vapi.ip_reassembly_enable_disable(
sw_if_index=self.src_if.sw_if_index, enable_ip6=True,
type=VppEnum.vl_api_ip_reass_type_t.IP_REASS_TYPE_SHALLOW_VIRTUAL)
expire_walk_interval_ms=10000, is_ip6=1)
def tearDown(self):
- super(TestIPv6SVReassembly, self).tearDown()
+ super().tearDown()
self.logger.debug(self.vapi.ppcli("show ip6-sv-reassembly details"))
self.logger.debug(self.vapi.ppcli("show buffers"))
@classmethod
def setUpClass(cls):
- super(TestIPv4ReassemblyLocalNode, cls).setUpClass()
+ super().setUpClass()
cls.create_pg_interfaces([0])
cls.src_dst_if = cls.pg0
@classmethod
def tearDownClass(cls):
- super(TestIPv4ReassemblyLocalNode, cls).tearDownClass()
+ super().tearDownClass()
def setUp(self):
""" Test setup - force timeout on existing reassemblies """
- super(TestIPv4ReassemblyLocalNode, self).setUp()
+ super().setUp()
self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
max_reassembly_length=1000,
expire_walk_interval_ms=10)
expire_walk_interval_ms=10000)
def tearDown(self):
- super(TestIPv4ReassemblyLocalNode, self).tearDown()
+ super().tearDown()
def show_commands_at_teardown(self):
self.logger.debug(self.vapi.ppcli("show ip4-full-reassembly details"))
@classmethod
def setUpClass(cls):
- super(TestFIFReassembly, cls).setUpClass()
+ super().setUpClass()
cls.create_pg_interfaces([0, 1])
cls.src_if = cls.pg0
@classmethod
def tearDownClass(cls):
- super(TestFIFReassembly, cls).tearDownClass()
+ super().tearDownClass()
def setUp(self):
""" Test setup - force timeout on existing reassemblies """
- super(TestFIFReassembly, self).setUp()
+ super().setUp()
self.vapi.ip_reassembly_enable_disable(
sw_if_index=self.src_if.sw_if_index, enable_ip4=True,
enable_ip6=True)
expire_walk_interval_ms=10000, is_ip6=1)
def tearDown(self):
- super(TestFIFReassembly, self).tearDown()
+ super().tearDown()
def show_commands_at_teardown(self):
self.logger.debug(self.vapi.ppcli("show ip4-full-reassembly details"))