tests: Revert "Tests: Example duplicate code refactoring."

This reverts commit 6955595a577e1b7d316b5b69267bf1d1d951a4ab. The result
is that test filtering for reassembly tests works again.

Type: fix

Change-Id: I4acb094b5b4aa264745986afa0bb0528789807b3
Signed-off-by: Klement Sekera <ksekera@cisco.com>

1 files changed, 277 insertions, 243 deletions

diff --git a/test/test_reassembly.py b/test/test_reassembly.py
index 6a356a00b0c..e95d533684e 100644
--- a/test/test_reassembly.py
+++ b/test/test_reassembly.py
@@ -1,18 +1,18 @@
 #!/usr/bin/env python
 
-from random import shuffle
 import six
 import unittest
+from random import shuffle
+
+from framework import VppTestCase, VppTestRunner
 
-from parameterized import parameterized
 import scapy.compat
 from scapy.packet import Raw
 from scapy.layers.l2 import Ether, GRE
 from scapy.layers.inet import IP, UDP, ICMP
-
+from util import ppp, fragment_rfc791, fragment_rfc8200
 from scapy.layers.inet6 import IPv6, IPv6ExtHdrFragment, ICMPv6ParamProblem,\
     ICMPv6TimeExceeded
-
 from framework import VppTestCase, VppTestRunner
 from util import ppp, fragment_rfc791, fragment_rfc8200
 from vpp_gre_interface import VppGreInterface
@@ -22,183 +22,8 @@ from vpp_ip_route import VppIpRoute, VppRoutePath, FibPathProto
 # 35 is enough to have >257 400-byte fragments
 test_packet_count = 35
 
-# <class 'scapy.layers.inet.IP'>
-# <class 'scapy.layers.inet6.IPv6'>
-_scapy_ip_family_types = (IP, IPv6)
-
-
-def validate_scapy_ip_family(scapy_ip_family):
-
-    if scapy_ip_family not in _scapy_ip_family_types:
-        raise ValueError("'scapy_ip_family' must be of type: %s. Got %s" %
-                         (_scapy_ip_family_types, scapy_ip_family))
-
-
-class TestIPReassemblyMixin(object):
-
-    def verify_capture(self, scapy_ip_family, capture,
-                       dropped_packet_indexes=None):
-        """Verify captured packet stream.
-
-        :param list capture: Captured packet stream.
-        """
-        validate_scapy_ip_family(scapy_ip_family)
-
-        if dropped_packet_indexes is None:
-            dropped_packet_indexes = []
-        info = None
-        seen = set()
-        for packet in capture:
-            try:
-                self.logger.debug(ppp("Got packet:", packet))
-                ip = packet[scapy_ip_family]
-                udp = packet[UDP]
-                payload_info = self.payload_to_info(packet[Raw])
-                packet_index = payload_info.index
-                self.assertTrue(
-                    packet_index not in dropped_packet_indexes,
-                    ppp("Packet received, but should be dropped:", packet))
-                if packet_index in seen:
-                    raise Exception(ppp("Duplicate packet received", packet))
-                seen.add(packet_index)
-                self.assertEqual(payload_info.dst, self.src_if.sw_if_index)
-                info = self._packet_infos[packet_index]
-                self.assertTrue(info is not None)
-                self.assertEqual(packet_index, info.index)
-                saved_packet = info.data
-                self.assertEqual(ip.src, saved_packet[scapy_ip_family].src)
-                self.assertEqual(ip.dst, saved_packet[scapy_ip_family].dst)
-                self.assertEqual(udp.payload, saved_packet[UDP].payload)
-            except Exception:
-                self.logger.error(ppp("Unexpected or invalid packet:", packet))
-                raise
-        for index in self._packet_infos:
-            self.assertTrue(index in seen or index in dropped_packet_indexes,
-                            "Packet with packet_index %d not received" % index)
-
-    def test_disabled(self, scapy_ip_family, stream,
-                      dropped_packet_indexes):
-        """ reassembly disabled """
-        validate_scapy_ip_family(scapy_ip_family)
-        is_ip6 = 1 if scapy_ip_family == IPv6 else 0
-
-        self.vapi.ip_reassembly_set(timeout_ms=1000, max_reassemblies=0,
-                                    max_reassembly_length=1000,
-                                    expire_walk_interval_ms=10000,
-                                    is_ip6=is_ip6)
-
-        self.pg_enable_capture()
-        self.src_if.add_stream(stream)
-        self.pg_start()
-
-        packets = self.dst_if.get_capture(
-            len(self.pkt_infos) - len(dropped_packet_indexes))
-        self.verify_capture(scapy_ip_family, packets, dropped_packet_indexes)
-        self.src_if.assert_nothing_captured()
-
-    def test_duplicates(self, scapy_ip_family, stream):
-        """ duplicate fragments """
-        validate_scapy_ip_family(scapy_ip_family)
-
-        self.pg_enable_capture()
-        self.src_if.add_stream(stream)
-        self.pg_start()
-
-        packets = self.dst_if.get_capture(len(self.pkt_infos))
-        self.verify_capture(scapy_ip_family, packets)
-        self.src_if.assert_nothing_captured()
-
-    def test_random(self, scapy_ip_family, stream):
-        """ random order reassembly """
-        validate_scapy_ip_family(scapy_ip_family)
-
-        fragments = list(stream)
-        shuffle(fragments)
-
-        self.pg_enable_capture()
-        self.src_if.add_stream(fragments)
-        self.pg_start()
-
-        packets = self.dst_if.get_capture(len(self.packet_infos))
-        self.verify_capture(scapy_ip_family, packets)
-        self.src_if.assert_nothing_captured()
-
-        # run it all again to verify correctness
-        self.pg_enable_capture()
-        self.src_if.add_stream(fragments)
-        self.pg_start()
-
-        packets = self.dst_if.get_capture(len(self.packet_infos))
-        self.verify_capture(scapy_ip_family, packets)
-        self.src_if.assert_nothing_captured()
-
-    def test_reassembly(self, scapy_ip_family, stream):
-        """ basic reassembly """
-        validate_scapy_ip_family(scapy_ip_family)
-
-        self.pg_enable_capture()
-        self.src_if.add_stream(stream)
-        self.pg_start()
-
-        packets = self.dst_if.get_capture(len(self.pkt_infos))
-        self.verify_capture(scapy_ip_family, packets)
-        self.src_if.assert_nothing_captured()
 
-        # run it all again to verify correctness
-        self.pg_enable_capture()
-        self.src_if.add_stream(stream)
-        self.pg_start()
-
-        packets = self.dst_if.get_capture(len(self.pkt_infos))
-        self.verify_capture(scapy_ip_family, packets)
-        self.src_if.assert_nothing_captured()
-
-    def test_reversed(self, scapy_ip_family, stream):
-        """ reverse order reassembly """
-        validate_scapy_ip_family(scapy_ip_family)
-
-        fragments = list(stream)
-        fragments.reverse()
-
-        self.pg_enable_capture()
-        self.src_if.add_stream(fragments)
-        self.pg_start()
-
-        packets = self.dst_if.get_capture(len(self.packet_infos))
-        self.verify_capture(scapy_ip_family, packets)
-        self.src_if.assert_nothing_captured()
-
-        # run it all again to verify correctness
-        self.pg_enable_capture()
-        self.src_if.add_stream(fragments)
-        self.pg_start()
-
-        packets = self.dst_if.get_capture(len(self.packet_infos))
-        self.verify_capture(scapy_ip_family, packets)
-        self.src_if.assert_nothing_captured()
-
-    def test_timeout_inline(self, scapy_ip_family, stream,
-                            dropped_packet_indexes):
-        """ timeout (inline) """
-        validate_scapy_ip_family(scapy_ip_family)
-        is_ip6 = 1 if scapy_ip_family == IPv6 else 0
-
-        self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
-                                    max_reassembly_length=1000,
-                                    expire_walk_interval_ms=10000,
-                                    is_ip6=is_ip6)
-
-        self.pg_enable_capture()
-        self.src_if.add_stream(stream)
-        self.pg_start()
-
-        packets = self.dst_if.get_capture(
-            len(self.pkt_infos) - len(dropped_packet_indexes))
-        self.verify_capture(scapy_ip_family, packets,
-                            dropped_packet_indexes)
-
-
-class TestIPv4Reassembly(TestIPReassemblyMixin, VppTestCase):
+class TestIPv4Reassembly(VppTestCase):
     """ IPv4 Reassembly """
 
     @classmethod
@@ -288,22 +113,83 @@ class TestIPv4Reassembly(TestIPReassemblyMixin, VppTestCase):
                          (len(infos), len(cls.fragments_400),
                              len(cls.fragments_300), len(cls.fragments_200)))
 
-    @parameterized.expand([(IP, None)])
-    def test_reassembly(self, family, stream):
+    def verify_capture(self, capture, dropped_packet_indexes=[]):
+        """Verify captured packet stream.
+
+        :param list capture: Captured packet stream.
+        """
+        info = None
+        seen = set()
+        for packet in capture:
+            try:
+                self.logger.debug(ppp("Got packet:", packet))
+                ip = packet[IP]
+                udp = packet[UDP]
+                payload_info = self.payload_to_info(packet[Raw])
+                packet_index = payload_info.index
+                self.assertTrue(
+                    packet_index not in dropped_packet_indexes,
+                    ppp("Packet received, but should be dropped:", packet))
+                if packet_index in seen:
+                    raise Exception(ppp("Duplicate packet received", packet))
+                seen.add(packet_index)
+                self.assertEqual(payload_info.dst, self.src_if.sw_if_index)
+                info = self._packet_infos[packet_index]
+                self.assertTrue(info is not None)
+                self.assertEqual(packet_index, info.index)
+                saved_packet = info.data
+                self.assertEqual(ip.src, saved_packet[IP].src)
+                self.assertEqual(ip.dst, saved_packet[IP].dst)
+                self.assertEqual(udp.payload, saved_packet[UDP].payload)
+            except Exception:
+                self.logger.error(ppp("Unexpected or invalid packet:", packet))
+                raise
+        for index in self._packet_infos:
+            self.assertTrue(index in seen or index in dropped_packet_indexes,
+                            "Packet with packet_index %d not received" % index)
+
+    def test_reassembly(self):
         """ basic reassembly """
-        stream = self.__class__.fragments_200
-        super(TestIPv4Reassembly, self).test_reassembly(family, stream)
 
-    @parameterized.expand([(IP, None)])
-    def test_reversed(self, family, stream):
+        self.pg_enable_capture()
+        self.src_if.add_stream(self.fragments_200)
+        self.pg_start()
+
+        packets = self.dst_if.get_capture(len(self.pkt_infos))
+        self.verify_capture(packets)
+        self.src_if.assert_nothing_captured()
+
+        # run it all again to verify correctness
+        self.pg_enable_capture()
+        self.src_if.add_stream(self.fragments_200)
+        self.pg_start()
+
+        packets = self.dst_if.get_capture(len(self.pkt_infos))
+        self.verify_capture(packets)
+        self.src_if.assert_nothing_captured()
+
+    def test_reversed(self):
         """ reverse order reassembly """
-        stream = self.__class__.fragments_200
-        super(TestIPv4Reassembly, self).test_reversed(family, stream)
 
-    @parameterized.expand([(IP, None)])
-    def test_random(self, family, stream):
-        stream = self.__class__.fragments_200
-        super(TestIPv4Reassembly, self).test_random(family, stream)
+        fragments = list(self.fragments_200)
+        fragments.reverse()
+
+        self.pg_enable_capture()
+        self.src_if.add_stream(fragments)
+        self.pg_start()
+
+        packets = self.dst_if.get_capture(len(self.packet_infos))
+        self.verify_capture(packets)
+        self.src_if.assert_nothing_captured()
+
+        # run it all again to verify correctness
+        self.pg_enable_capture()
+        self.src_if.add_stream(fragments)
+        self.pg_start()
+
+        packets = self.dst_if.get_capture(len(self.packet_infos))
+        self.verify_capture(packets)
+        self.src_if.assert_nothing_captured()
 
     def test_long_fragment_chain(self):
         """ long fragment chain """
@@ -432,16 +318,45 @@ class TestIPv4Reassembly(TestIPReassemblyMixin, VppTestCase):
         # self.assert_packet_counter_equal(
         #     "/err/ip4-reassembly-feature/malformed packets", 1)
 
-    @parameterized.expand([(IP, None)])
-    def test_duplicates(self, family, stream):
+    def test_random(self):
+        """ random order reassembly """
+
+        fragments = list(self.fragments_200)
+        shuffle(fragments)
+
+        self.pg_enable_capture()
+        self.src_if.add_stream(fragments)
+        self.pg_start()
+
+        packets = self.dst_if.get_capture(len(self.packet_infos))
+        self.verify_capture(packets)
+        self.src_if.assert_nothing_captured()
+
+        # run it all again to verify correctness
+        self.pg_enable_capture()
+        self.src_if.add_stream(fragments)
+        self.pg_start()
+
+        packets = self.dst_if.get_capture(len(self.packet_infos))
+        self.verify_capture(packets)
+        self.src_if.assert_nothing_captured()
+
+    def test_duplicates(self):
         """ duplicate fragments """
+
         fragments = [
-            # IPv4 uses 4 fields in pkt_infos, IPv6 uses 3.
             x for (_, frags, _, _) in self.pkt_infos
             for x in frags
             for _ in range(0, min(2, len(frags)))
         ]
-        super(TestIPv4Reassembly, self).test_duplicates(family, fragments)
+
+        self.pg_enable_capture()
+        self.src_if.add_stream(fragments)
+        self.pg_start()
+
+        packets = self.dst_if.get_capture(len(self.pkt_infos))
+        self.verify_capture(packets)
+        self.src_if.assert_nothing_captured()
 
     def test_overlap1(self):
         """ overlapping fragments case #1 """
@@ -460,7 +375,7 @@ class TestIPv4Reassembly(TestIPReassemblyMixin, VppTestCase):
         self.pg_start()
 
         packets = self.dst_if.get_capture(len(self.pkt_infos))
-        self.verify_capture(IP, packets)
+        self.verify_capture(packets)
         self.src_if.assert_nothing_captured()
 
         # run it all to verify correctness
@@ -469,7 +384,7 @@ class TestIPv4Reassembly(TestIPReassemblyMixin, VppTestCase):
         self.pg_start()
 
         packets = self.dst_if.get_capture(len(self.pkt_infos))
-        self.verify_capture(IP, packets)
+        self.verify_capture(packets)
         self.src_if.assert_nothing_captured()
 
     def test_overlap2(self):
@@ -495,7 +410,7 @@ class TestIPv4Reassembly(TestIPReassemblyMixin, VppTestCase):
         self.pg_start()
 
         packets = self.dst_if.get_capture(len(self.pkt_infos))
-        self.verify_capture(IP, packets)
+        self.verify_capture(packets)
         self.src_if.assert_nothing_captured()
 
         # run it all to verify correctness
@@ -504,20 +419,27 @@ class TestIPv4Reassembly(TestIPReassemblyMixin, VppTestCase):
         self.pg_start()
 
         packets = self.dst_if.get_capture(len(self.pkt_infos))
-        self.verify_capture(IP, packets)
+        self.verify_capture(packets)
         self.src_if.assert_nothing_captured()
 
-    @parameterized.expand([(IP, None, None)])
-    def test_timeout_inline(self, family, stream, dropped_packet_indexes):
+    def test_timeout_inline(self):
         """ timeout (inline) """
-        stream = self.fragments_400
 
         dropped_packet_indexes = set(
             index for (index, frags, _, _) in self.pkt_infos if len(frags) > 1
         )
-        super(TestIPv4Reassembly, self).test_timeout_inline(
-            family, stream, dropped_packet_indexes)
 
+        self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
+                                    max_reassembly_length=3,
+                                    expire_walk_interval_ms=10000)
+
+        self.pg_enable_capture()
+        self.src_if.add_stream(self.fragments_400)
+        self.pg_start()
+
+        packets = self.dst_if.get_capture(
+            len(self.pkt_infos) - len(dropped_packet_indexes))
+        self.verify_capture(packets, dropped_packet_indexes)
         self.src_if.assert_nothing_captured()
 
     def test_timeout_cleanup(self):
@@ -553,22 +475,31 @@ class TestIPv4Reassembly(TestIPReassemblyMixin, VppTestCase):
 
         packets = self.dst_if.get_capture(
             len(self.pkt_infos) - len(dropped_packet_indexes))
-        self.verify_capture(IP, packets, dropped_packet_indexes)
+        self.verify_capture(packets, dropped_packet_indexes)
         self.src_if.assert_nothing_captured()
 
-    @parameterized.expand([(IP, None, None)])
-    def test_disabled(self, family, stream, dropped_packet_indexes):
+    def test_disabled(self):
         """ reassembly disabled """
 
-        stream = self.__class__.fragments_400
         dropped_packet_indexes = set(
             index for (index, frags_400, _, _) in self.pkt_infos
             if len(frags_400) > 1)
-        super(TestIPv4Reassembly, self).test_disabled(
-            family, stream, dropped_packet_indexes)
 
+        self.vapi.ip_reassembly_set(timeout_ms=1000, max_reassemblies=0,
+                                    max_reassembly_length=3,
+                                    expire_walk_interval_ms=10000)
+
+        self.pg_enable_capture()
+        self.src_if.add_stream(self.fragments_400)
+        self.pg_start()
 
-class TestIPv6Reassembly(TestIPReassemblyMixin, VppTestCase):
+        packets = self.dst_if.get_capture(
+            len(self.pkt_infos) - len(dropped_packet_indexes))
+        self.verify_capture(packets, dropped_packet_indexes)
+        self.src_if.assert_nothing_captured()
+
+
+class TestIPv6Reassembly(VppTestCase):
     """ IPv6 Reassembly """
 
     @classmethod
@@ -655,35 +586,123 @@ class TestIPv6Reassembly(TestIPReassemblyMixin, VppTestCase):
                          (len(infos), len(cls.fragments_400),
                              len(cls.fragments_300)))
 
-    @parameterized.expand([(IPv6, None)])
-    def test_reassembly(self, family, stream):
+    def verify_capture(self, capture, dropped_packet_indexes=[]):
+        """Verify captured packet strea .
+
+        :param list capture: Captured packet stream.
+        """
+        info = None
+        seen = set()
+        for packet in capture:
+            try:
+                self.logger.debug(ppp("Got packet:", packet))
+                ip = packet[IPv6]
+                udp = packet[UDP]
+                payload_info = self.payload_to_info(packet[Raw])
+                packet_index = payload_info.index
+                self.assertTrue(
+                    packet_index not in dropped_packet_indexes,
+                    ppp("Packet received, but should be dropped:", packet))
+                if packet_index in seen:
+                    raise Exception(ppp("Duplicate packet received", packet))
+                seen.add(packet_index)
+                self.assertEqual(payload_info.dst, self.src_if.sw_if_index)
+                info = self._packet_infos[packet_index]
+                self.assertTrue(info is not None)
+                self.assertEqual(packet_index, info.index)
+                saved_packet = info.data
+                self.assertEqual(ip.src, saved_packet[IPv6].src)
+                self.assertEqual(ip.dst, saved_packet[IPv6].dst)
+                self.assertEqual(udp.payload, saved_packet[UDP].payload)
+            except Exception:
+                self.logger.error(ppp("Unexpected or invalid packet:", packet))
+                raise
+        for index in self._packet_infos:
+            self.assertTrue(index in seen or index in dropped_packet_indexes,
+                            "Packet with packet_index %d not received" % index)
+
+    def test_reassembly(self):
         """ basic reassembly """
-        stream = self.__class__.fragments_400
-        super(TestIPv6Reassembly, self).test_reassembly(family, stream)
 
-    @parameterized.expand([(IPv6, None)])
-    def test_reversed(self, family, stream):
+        self.pg_enable_capture()
+        self.src_if.add_stream(self.fragments_400)
+        self.pg_start()
+
+        packets = self.dst_if.get_capture(len(self.pkt_infos))
+        self.verify_capture(packets)
+        self.src_if.assert_nothing_captured()
+
+        # run it all again to verify correctness
+        self.pg_enable_capture()
+        self.src_if.add_stream(self.fragments_400)
+        self.pg_start()
+
+        packets = self.dst_if.get_capture(len(self.pkt_infos))
+        self.verify_capture(packets)
+        self.src_if.assert_nothing_captured()
+
+    def test_reversed(self):
         """ reverse order reassembly """
-        stream = self.__class__.fragments_400
-        super(TestIPv6Reassembly, self).test_reversed(family, stream)
 
-    @parameterized.expand([(IPv6, None)])
-    def test_random(self, family, stream):
+        fragments = list(self.fragments_400)
+        fragments.reverse()
+
+        self.pg_enable_capture()
+        self.src_if.add_stream(fragments)
+        self.pg_start()
+
+        packets = self.dst_if.get_capture(len(self.pkt_infos))
+        self.verify_capture(packets)
+        self.src_if.assert_nothing_captured()
+
+        # run it all again to verify correctness
+        self.pg_enable_capture()
+        self.src_if.add_stream(fragments)
+        self.pg_start()
+
+        packets = self.dst_if.get_capture(len(self.pkt_infos))
+        self.verify_capture(packets)
+        self.src_if.assert_nothing_captured()
+
+    def test_random(self):
         """ random order reassembly """
-        stream = self.__class__.fragments_400
-        super(TestIPv6Reassembly, self).test_random(family, stream)
 
-    @parameterized.expand([(IPv6, None)])
-    def test_duplicates(self, family, stream):
+        fragments = list(self.fragments_400)
+        shuffle(fragments)
+
+        self.pg_enable_capture()
+        self.src_if.add_stream(fragments)
+        self.pg_start()
+
+        packets = self.dst_if.get_capture(len(self.pkt_infos))
+        self.verify_capture(packets)
+        self.src_if.assert_nothing_captured()
+
+        # run it all again to verify correctness
+        self.pg_enable_capture()
+        self.src_if.add_stream(fragments)
+        self.pg_start()
+
+        packets = self.dst_if.get_capture(len(self.pkt_infos))
+        self.verify_capture(packets)
+        self.src_if.assert_nothing_captured()
+
+    def test_duplicates(self):
         """ duplicate fragments """
 
         fragments = [
-            # IPv4 uses 4 fields in pkt_infos, IPv6 uses 3.
             x for (_, frags, _) in self.pkt_infos
             for x in frags
             for _ in range(0, min(2, len(frags)))
         ]
-        super(TestIPv6Reassembly, self).test_duplicates(family, fragments)
+
+        self.pg_enable_capture()
+        self.src_if.add_stream(fragments)
+        self.pg_start()
+
+        packets = self.dst_if.get_capture(len(self.pkt_infos))
+        self.verify_capture(packets)
+        self.src_if.assert_nothing_captured()
 
     def test_long_fragment_chain(self):
         """ long fragment chain """
@@ -712,7 +731,7 @@ class TestIPv6Reassembly(TestIPReassemblyMixin, VppTestCase):
         self.assert_error_counter_equal(error_cnt_str, error_cnt + 1)
 
     def test_overlap1(self):
-        """ overlapping fragments case #1 (differs from IP test case)"""
+        """ overlapping fragments case #1 """
 
         fragments = []
         for _, frags_400, frags_300 in self.pkt_infos:
@@ -733,11 +752,11 @@ class TestIPv6Reassembly(TestIPReassemblyMixin, VppTestCase):
 
         packets = self.dst_if.get_capture(
             len(self.pkt_infos) - len(dropped_packet_indexes))
-        self.verify_capture(IPv6, packets, dropped_packet_indexes)
+        self.verify_capture(packets, dropped_packet_indexes)
         self.src_if.assert_nothing_captured()
 
     def test_overlap2(self):
-        """ overlapping fragments case #2 (differs from IP test case)"""
+        """ overlapping fragments case #2 """
 
         fragments = []
         for _, frags_400, frags_300 in self.pkt_infos:
@@ -764,20 +783,27 @@ class TestIPv6Reassembly(TestIPReassemblyMixin, VppTestCase):
 
         packets = self.dst_if.get_capture(
             len(self.pkt_infos) - len(dropped_packet_indexes))
-        self.verify_capture(IPv6, packets, dropped_packet_indexes)
+        self.verify_capture(packets, dropped_packet_indexes)
         self.src_if.assert_nothing_captured()
 
-    @parameterized.expand([(IPv6, None, None)])
-    def test_timeout_inline(self, family, stream, dropped_packets_index):
+    def test_timeout_inline(self):
         """ timeout (inline) """
-        stream = self.__class__.fragments_400
 
         dropped_packet_indexes = set(
             index for (index, frags, _) in self.pkt_infos if len(frags) > 1
         )
-        super(TestIPv6Reassembly, self).test_timeout_inline(
-            family, stream, dropped_packet_indexes)
 
+        self.vapi.ip_reassembly_set(timeout_ms=0, max_reassemblies=1000,
+                                    max_reassembly_length=3,
+                                    expire_walk_interval_ms=10000, is_ip6=1)
+
+        self.pg_enable_capture()
+        self.src_if.add_stream(self.fragments_400)
+        self.pg_start()
+
+        packets = self.dst_if.get_capture(
+            len(self.pkt_infos) - len(dropped_packet_indexes))
+        self.verify_capture(packets, dropped_packet_indexes)
         pkts = self.src_if.get_capture(
             expected_count=len(dropped_packet_indexes))
         for icmp in pkts:
@@ -823,7 +849,7 @@ class TestIPv6Reassembly(TestIPReassemblyMixin, VppTestCase):
 
         packets = self.dst_if.get_capture(
             len(self.pkt_infos) - len(dropped_packet_indexes))
-        self.verify_capture(IPv6, packets, dropped_packet_indexes)
+        self.verify_capture(packets, dropped_packet_indexes)
         pkts = self.src_if.get_capture(
             expected_count=len(dropped_packet_indexes))
         for icmp in pkts:
@@ -832,16 +858,24 @@ class TestIPv6Reassembly(TestIPReassemblyMixin, VppTestCase):
             self.assertIn(icmp[IPv6ExtHdrFragment].id, dropped_packet_indexes)
             dropped_packet_indexes.remove(icmp[IPv6ExtHdrFragment].id)
 
-    @parameterized.expand([(IPv6, None, None)])
-    def test_disabled(self, family, stream, dropped_packet_indexes):
+    def test_disabled(self):
         """ reassembly disabled """
 
-        stream = self.__class__.fragments_400
         dropped_packet_indexes = set(
             index for (index, frags_400, _) in self.pkt_infos
             if len(frags_400) > 1)
-        super(TestIPv6Reassembly, self).test_disabled(
-            family, stream, dropped_packet_indexes)
+
+        self.vapi.ip_reassembly_set(timeout_ms=1000, max_reassemblies=0,
+                                    max_reassembly_length=3,
+                                    expire_walk_interval_ms=10000, is_ip6=1)
+
+        self.pg_enable_capture()
+        self.src_if.add_stream(self.fragments_400)
+        self.pg_start()
+
+        packets = self.dst_if.get_capture(
+            len(self.pkt_infos) - len(dropped_packet_indexes))
+        self.verify_capture(packets, dropped_packet_indexes)
         self.src_if.assert_nothing_captured()
 
     def test_missing_upper(self):