|
| 1 | +# -*- coding: utf-8 -*- |
| 2 | +"""GitHub issue #2308: a trial parse that raises and is kept as ``Raw`` warns nothing. |
| 3 | +
|
| 4 | +:meth:`~pcapkit.protocols.protocol.ProtocolBase._parse_next_layer` holds the |
| 5 | +warnings of a next layer captured shorter than its length hint (#1580). When |
| 6 | +that parse raises, :func:`~pcapkit.corekit.decorators.beholder` replaces the |
| 7 | +layer, and without the fix the held warnings were reported anyway: an Ethernet |
| 8 | +frame cut inside its IPv4 header warned ``packet length < 0`` three times for an |
| 9 | +IPv4 layer the result keeps as ``Raw``. They are dropped with the layer, whose |
| 10 | +``error`` names the cut. A held layer that is kept still warns, and so does one |
| 11 | +that raises where no :func:`~pcapkit.corekit.decorators.beholder` replaces it. |
| 12 | +
|
| 13 | +:mod:`pcapkit` is imported inside each test, after |
| 14 | +:func:`~tests._support.reimport_once_per_class`. |
| 15 | +
|
| 16 | +""" |
| 17 | + |
| 18 | +import re |
| 19 | +import unittest |
| 20 | +import warnings |
| 21 | + |
| 22 | +from tests._support import reimport_once_per_class |
| 23 | +from tests.utilities._harness import capture |
| 24 | + |
| 25 | +#: Test tier (#1958): exercises the warning machinery, which free-threaded |
| 26 | +#: builds make context-aware. |
| 27 | +__tier__ = 'ALL' |
| 28 | + |
| 29 | +#: The first 10 of the 20 octets of an IPv4 header. |
| 30 | +IPV4_CUT = bytes.fromhex('4500008c000100004011') |
| 31 | +#: An Ethernet frame carrying :data:`IPV4_CUT`. |
| 32 | +FRAME = bytes(6) + bytes(range(1, 7)) + b'\x08\x00' + IPV4_CUT |
| 33 | +#: The ``error`` of the ``Raw`` that :data:`FRAME`'s IPv4 is kept as. |
| 34 | +ERROR = 'IPv4: header length 20 runs past the end of the data (10 octets)' |
| 35 | + |
| 36 | + |
| 37 | +def _ipv6(next_header: 'int') -> 'bytes': |
| 38 | + """An Ethernet frame carrying IPv6 whose ``next_header`` is cut to 1 octet.""" |
| 39 | + return (bytes(6) + bytes(range(1, 7)) + b'\x86\xdd' |
| 40 | + + bytes.fromhex('600000000001') + bytes([next_header, 64]) |
| 41 | + + bytes(15) + b'\x01' + bytes(15) + b'\x02' + b'\x00') |
| 42 | + |
| 43 | + |
| 44 | +#: The warnings of the kept, cut ``IPv6_Ext`` that each :func:`_ipv6` frame ends in. |
| 45 | +IPV6_EXT_CUT = ['packet length < 0: -1', |
| 46 | + 'IPv6: extension header declares a length of 8 octet(s) with 1 octet(s) ' |
| 47 | + 'left in the chain; stopping the walk instead of skipping past it'] |
| 48 | + |
| 49 | + |
| 50 | +class TestTrialParseRaises(unittest.TestCase): |
| 51 | + """Pin that a raising trial parse replaced by ``beholder`` reports nothing.""" |
| 52 | + |
| 53 | + def setUp(self) -> None: |
| 54 | + reimport_once_per_class(self) |
| 55 | + |
| 56 | + def _parse(self, frame: 'bytes', action: 'str' = 'always') -> 'tuple[object, list[str]]': |
| 57 | + """Parse ``frame`` as Ethernet, with the :class:`SchemaWarning` messages it reports.""" |
| 58 | + from pcapkit.protocols.link.ethernet import Ethernet |
| 59 | + from pcapkit.utilities.warnings import SchemaWarning |
| 60 | + |
| 61 | + with warnings.catch_warnings(record=True) as caught: |
| 62 | + warnings.simplefilter(action) |
| 63 | + parsed = Ethernet(frame, len(frame)) |
| 64 | + return parsed, [str(item.message) for item in caught if issubclass(item.category, SchemaWarning)] |
| 65 | + |
| 66 | + def test_cut_ipv4_header_is_raw_and_warns_nothing(self) -> None: |
| 67 | + """IPv4 cut to 10 octets raises, is kept as ``Raw`` naming the cut, and reports nothing.""" |
| 68 | + from pcapkit.protocols.misc.raw import Raw |
| 69 | + from pcapkit.utilities.logging import logger |
| 70 | + |
| 71 | + with capture(logger) as recorder: |
| 72 | + parsed, messages = self._parse(FRAME) |
| 73 | + ipv4 = parsed.payload # type: ignore[attr-defined] |
| 74 | + self.assertIsInstance(ipv4, Raw) |
| 75 | + self.assertEqual(ipv4.info.error, ERROR) |
| 76 | + self.assertEqual(messages, []) |
| 77 | + self.assertEqual([record.getMessage() for record in recorder.records |
| 78 | + if record.getMessage().startswith('packet length')], []) |
| 79 | + |
| 80 | + def test_error_filter_keeps_the_cut_as_the_error(self) -> None: |
| 81 | + """Under ``error``, the ``Raw`` names the cut header, not a warning it dropped.""" |
| 82 | + from pcapkit.protocols.misc.raw import Raw |
| 83 | + |
| 84 | + parsed, messages = self._parse(FRAME, 'error') |
| 85 | + self.assertIsInstance(parsed.payload, Raw) # type: ignore[attr-defined] |
| 86 | + self.assertEqual(parsed.payload.info.error, ERROR) # type: ignore[attr-defined] |
| 87 | + self.assertEqual(messages, []) |
| 88 | + |
| 89 | + def test_replaced_by_ipv6_ext_reports_only_the_replacement(self) -> None: |
| 90 | + """A cut Hop-by-Hop header raises and is parsed as ``IPv6_Ext``, which alone warns.""" |
| 91 | + parsed, messages = self._parse(_ipv6(0)) |
| 92 | + self.assertEqual(str(parsed.protochain), 'Ethernet:IPv6:IPv6-Ext') # type: ignore[attr-defined] |
| 93 | + self.assertEqual(messages, IPV6_EXT_CUT) |
| 94 | + |
| 95 | + def test_kept_trial_still_warns(self) -> None: |
| 96 | + """A cut Shim6 header is held, kept, and reports its warnings as before.""" |
| 97 | + parsed, messages = self._parse(_ipv6(140)) |
| 98 | + self.assertEqual(str(parsed.protochain), 'Ethernet:IPv6:IPv6-Ext') # type: ignore[attr-defined] |
| 99 | + self.assertEqual(messages, IPV6_EXT_CUT) |
| 100 | + |
| 101 | + def test_raise_outside_beholder_reports(self) -> None: |
| 102 | + """With no ``beholder`` to replace it, the layer's error and its warnings both reach the caller.""" |
| 103 | + from pcapkit.protocols.internet.ipv4 import IPv4 |
| 104 | + from pcapkit.protocols.protocol import ProtocolBase |
| 105 | + from pcapkit.utilities.exceptions import ProtocolError |
| 106 | + from pcapkit.utilities.warnings import SchemaWarning |
| 107 | + |
| 108 | + with warnings.catch_warnings(record=True) as caught: |
| 109 | + warnings.simplefilter('always') |
| 110 | + with self.assertRaisesRegex(ProtocolError, re.escape(ERROR)): |
| 111 | + ProtocolBase._parse_next_layer(IPv4, IPV4_CUT, len(IPV4_CUT)) # pylint: disable=protected-access |
| 112 | + self.assertEqual([str(item.message) for item in caught if issubclass(item.category, SchemaWarning)], |
| 113 | + ['packet length < 0: -2', 'packet length < 0: -6', 'packet length < 0: -10', |
| 114 | + 'packet length < 0: -10']) |
| 115 | + |
| 116 | + |
| 117 | +if __name__ == '__main__': |
| 118 | + unittest.main() |
0 commit comments