timeout (inline)
(self)
| 1621 | self.verify_capture(packets, dropped_packet_indexes) |
| 1622 | |
| 1623 | def test_timeout_inline(self): |
| 1624 | """timeout (inline)""" |
| 1625 | |
| 1626 | dropped_packet_indexes = set( |
| 1627 | index for (index, frags, _) in self.pkt_infos if len(frags) > 1 |
| 1628 | ) |
| 1629 | |
| 1630 | self.vapi.ip_reassembly_set( |
| 1631 | timeout_ms=0, |
| 1632 | max_reassemblies=1000, |
| 1633 | max_reassembly_length=3, |
| 1634 | expire_walk_interval_ms=10000, |
| 1635 | is_ip6=1, |
| 1636 | ) |
| 1637 | |
| 1638 | self.pg_enable_capture() |
| 1639 | self.src_if.add_stream(self.fragments_400) |
| 1640 | self.pg_start() |
| 1641 | |
| 1642 | packets = self.dst_if.get_capture( |
| 1643 | len(self.pkt_infos) - len(dropped_packet_indexes) |
| 1644 | ) |
| 1645 | self.verify_capture(packets, dropped_packet_indexes) |
| 1646 | pkts = self.src_if._get_capture() |
| 1647 | for icmp in pkts: |
| 1648 | self.assertIn(ICMPv6TimeExceeded, icmp) |
| 1649 | self.assertIn(IPv6ExtHdrFragment, icmp) |
| 1650 | self.assertIn(icmp[IPv6ExtHdrFragment].id, dropped_packet_indexes) |
| 1651 | dropped_packet_indexes.remove(icmp[IPv6ExtHdrFragment].id) |
| 1652 | |
| 1653 | def test_timeout_cleanup(self): |
| 1654 | """timeout (cleanup)""" |
nothing calls this directly
no test coverage detected