From ab96134d82f0bcb7d14f86b7062cbf304a997126 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Tue, 27 Oct 2026 06:53:13 -0400 Subject: [PATCH] refactor: remove two unreachable branches, commented-out raises and unused test helpers (#2318) Problem: two guards could never run, because something earlier had already raised -- Header.read's FileError on an unknown magic number, which the schema's magic_number_callback beats to it; and _read_opt_calipso's two length guards, which calipso_pad_len beats to it while the option is unpacked. Beside them sat commented-out raise lines next to the warn that replaced them, a repeated pylint: disable=unused-argument, and three names nothing read: dispatch.py's _internet_enum, a FAKE_GH query no caller sends, and GATING. A third candidate, _read_opt_qs's unknown QS function arm, looked unreachable the same way but is not: the unassigned-option dispatch it depends on is by schema class, decided from the func a parse's own registry resolves it against, not from the func held in a directly-constructed schema's flags. A QuickStartRequestOption or QuickStartReportOption built directly with an unassigned func reaches this method's own check regardless, and a parse reaches it too once a caller registers a schema against that func, so it stays, with a Note on each of the three readers explaining why. Fix: the two genuinely unreachable guards are deleted, with the prose that described them corrected -- Header.read documents ProtocolError, and _read_opt_calipso documents what checked the length instead. The three fabricated-schema stanzas that reached the deleted guards by calling a private reader directly are replaced by the real parse paths, which raise as before. Test: tests/protocols/test_dead_branches_2318_unit.py parses what was supposed to reach each deleted branch and pins what happens instead, pins the restored QS guard against both schema kinds across all three protocols through the public construction path and through a parse once a schema is registered against the func, and pins the source as holding no such deleted branch, no commented-out raise beside a warn, and no repeated pragma; tests/project/test_dead_helpers_2318_unit.py pins the rule that made each helper dead. 8 of the 17 test methods across the two files fail on the parent commit; the other 9 pin unchanged behaviour by design. --- examples/generators/dispatch.py | 24 +- pcapkit/protocols/application/httpv2.py | 1 - pcapkit/protocols/internet/hip.py | 1 - pcapkit/protocols/internet/hopopt.py | 26 +- pcapkit/protocols/internet/ipv4.py | 10 +- pcapkit/protocols/internet/ipv6_opts.py | 26 +- pcapkit/protocols/internet/mh.py | 1 - pcapkit/protocols/misc/pcap/header.py | 17 +- pcapkit/protocols/misc/pcapng.py | 6 - pcapkit/protocols/schema/internet/hip.py | 4 - pcapkit/protocols/schema/misc/pcapng.py | 1 - tests/project/test_dead_helpers_2318_unit.py | 149 ++++++ .../test_release_tags_the_run_sha_1634.py | 10 +- .../test_sweeps_on_latest_only_1732.py | 3 - .../internet/test_ipv6_extension_unit.py | 12 +- .../misc/pcap/test_header_frame_unit.py | 19 +- .../protocols/test_dead_branches_2318_unit.py | 493 ++++++++++++++++++ 17 files changed, 709 insertions(+), 94 deletions(-) create mode 100644 tests/project/test_dead_helpers_2318_unit.py create mode 100644 tests/protocols/test_dead_branches_2318_unit.py diff --git a/examples/generators/dispatch.py b/examples/generators/dispatch.py index 805229ca80..cfc8037733 100644 --- a/examples/generators/dispatch.py +++ b/examples/generators/dispatch.py @@ -135,13 +135,16 @@ def _resolve(entry: 'Any') -> 'Any': def code_name(code: 'Any', enum: 'Optional[Any]' = None) -> 'str': """Render a registry key as a name usable in a test id. - Most ``__proto__`` tables are keyed entirely by enumeration members, whose - own ``.name`` is used directly. ``Link.__proto__`` holds an entry keyed by - a bare :class:`int` instead (0x8137 for IPX) -- :func:`_resolve` does not - care, since dict lookup does not distinguish an :class:`~enum.IntEnum` - member from the plain :class:`int` it equals, but a case label needs a - name. ``enum`` is consulted for that one, so the label reads - ``Novell_Inc_0x8137`` rather than the opaque ``33079``. + Many ``__proto__`` tables are keyed entirely by enumeration members, whose + own ``.name`` is used directly. ``Link.__proto__`` and the port tables hold + bare :class:`int` keys as well -- :func:`_resolve` does not care, since + dict lookup does not distinguish an :class:`~enum.IntEnum` member from the + plain :class:`int` it equals, but a case label needs a name. ``enum`` is + consulted where an enumeration member equals such a key, which is + ``Link.__proto__``'s 0x8137 for IPX, so that label reads + ``Novell_Inc_0x8137`` rather than the opaque ``33079``. A port number has + no enumeration to consult, so it keeps its plain string form and its family + is given no ``enum``. Args: code: Registry key. @@ -580,16 +583,11 @@ def _link_enum() -> 'Any': return EtherType -def _internet_enum() -> 'Any': - from pcapkit.const.reg.transtype import TransType - return TransType - - #: Every family this module exercises, one per ``__proto__`` table named in #: GitHub issue #496, and the loopback table #1574 added. FAMILIES = ( Family('link', _link_registry, _link_build, _link_enum), - Family('internet', _internet_registry, _internet_build, _internet_enum), + Family('internet', _internet_registry, _internet_build), Family('tcp', _tcp_registry, _tcp_build), Family('udp', _udp_registry, _udp_build), Family('sctp', _sctp_registry, _sctp_build), diff --git a/pcapkit/protocols/application/httpv2.py b/pcapkit/protocols/application/httpv2.py index 4e70f7f687..73c9134df4 100644 --- a/pcapkit/protocols/application/httpv2.py +++ b/pcapkit/protocols/application/httpv2.py @@ -575,7 +575,6 @@ def _read_http_none(self, schema: 'Schema_UnassignedFrame', *, """ if any(header.flags.values()): - #raise ProtocolError(f'HTTP/2: [Type {frame}] invalid format') warn(f'HTTP/2: [Type {header.type}] invalid format', ProtocolWarning) data = Data_UnassignedFrame( diff --git a/pcapkit/protocols/internet/hip.py b/pcapkit/protocols/internet/hip.py index a32f909306..fa1abe3cff 100644 --- a/pcapkit/protocols/internet/hip.py +++ b/pcapkit/protocols/internet/hip.py @@ -1508,7 +1508,6 @@ def _read_param_hip_cipher(self, schema: 'Schema_HIPCipherParameter', *, version ciphers = schema.ciphers if len(ciphers) > 6: warn(f'HIPv{version}: [ParamNo {schema.type}] invalid format', ProtocolWarning) - # raise ProtocolError(f'HIPv{version}: [ParamNo {schema.type}] invalid format') ciphers = ciphers[:6] hip_cipher = Data_HIPCipherParameter( diff --git a/pcapkit/protocols/internet/hopopt.py b/pcapkit/protocols/internet/hopopt.py index 81376f5027..99ee315a46 100644 --- a/pcapkit/protocols/internet/hopopt.py +++ b/pcapkit/protocols/internet/hopopt.py @@ -769,17 +769,14 @@ def _read_opt_calipso(self, schema: 'Schema_CALIPSOOption', *, options: 'Option' Returns: Parsed option data. - Raises: - ProtocolError: If the option is malformed. + Note: + ``Cmpt Length`` counts 32-bit words and may be odd + [:rfc:`5570#section-5.2`]; it only has to fit inside ``Opt Data + Len``, which + :func:`~pcapkit.protocols.schema.internet.ipv6_option.calipso_pad_len` + has already checked while the option was unpacked. """ - if schema.len < 8 and schema.len % 8 != 0: - raise ProtocolError(f'{self.alias}: [OptNo {schema.type}] invalid format') - # ``Cmpt Length`` counts 32-bit words and may be odd [:rfc:`5570#section-5.2`]; - # it only has to fit inside ``Opt Data Len`` - if schema.len < 8 + schema.cmpt_len * 4: - raise ProtocolError(f'{self.alias}: [OptNo {schema.type}] invalid format') - opt = Data_CALIPSOOption( type=schema.type, action=Enum_OptionAction.get(schema.type >> 6), @@ -930,7 +927,7 @@ def _read_opt_pdm(self, schema: 'Schema_PDMOption', *, options: 'Option') -> 'Da ) return opt - def _read_opt_qs(self, schema: 'Schema_QuickStartOption', *, options: 'Option') -> 'Data_QuickStartOption': # pylint: disable=unused-argument # pylint: disable=unused-argument + def _read_opt_qs(self, schema: 'Schema_QuickStartOption', *, options: 'Option') -> 'Data_QuickStartOption': # pylint: disable=unused-argument """Read HOPOPT Quick Start option. Structure of HOPOPT Quick-Start option [:rfc:`4782`]: @@ -974,7 +971,13 @@ def _read_opt_qs(self, schema: 'Schema_QuickStartOption', *, options: 'Option') Note: An option whose QS function is neither of the two above has no defined layout, so the schema keeps it as an unassigned option and - it is parsed as one, see :meth:`_read_opt_none`. + it is parsed as one, see :meth:`_read_opt_none`. That dispatch is + by the schema's own class, decided from the octets a file read + hands it; a :class:`~pcapkit.protocols.schema.internet.hopopt.QuickStartRequestOption` + or :class:`~pcapkit.protocols.schema.internet.hopopt.QuickStartReportOption` + built directly with an unassigned ``func`` reaches this method + regardless, which is why it still raises below rather than + returning early. """ if isinstance(schema, Schema_UnassignedOption): @@ -1533,7 +1536,6 @@ def _make_opt_pad(self, code: 'Enum_Option', opt: 'Optional[Data_PadOption]' = N pad = getattr(opt, 'pad', b'') if code == Enum_Option.Pad1 and length != 0: - #raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid format') warn(f'{self.alias}: [OptNo {code}] invalid format', ProtocolWarning) code = Enum_Option.PadN # type: ignore[assignment] diff --git a/pcapkit/protocols/internet/ipv4.py b/pcapkit/protocols/internet/ipv4.py index 4d5b0a9871..68cc3cb245 100644 --- a/pcapkit/protocols/internet/ipv4.py +++ b/pcapkit/protocols/internet/ipv4.py @@ -883,7 +883,6 @@ def _read_opt_sec(self, schema: 'Schema_SECOption', *, options: 'Option') -> 'Da flags.append(authority) if byte & 0x01 == 0 and base < schema.length - 4: - #raise ProtocolError(f'{self.alias}: [OptNo {kind}] invalid format: remaining data') warn(f'{self.alias}: [OptNo {schema.type}] invalid format: remaining data', ProtocolWarning) if schema.data[-1] & 0x01 == 1: @@ -1319,7 +1318,14 @@ def _read_opt_qs(self, schema: 'Schema_QSOption', *, options: 'Option') -> 'Data Note: An option whose QS function is neither of the two above has no defined layout, so the schema keeps it as an unassigned option and - it is parsed as one, see :meth:`_read_opt_unassigned`. + it is parsed as one, see :meth:`_read_opt_unassigned`. That + dispatch is by the schema's own class, decided from the octets a + file read hands it; a + :class:`~pcapkit.protocols.schema.internet.ipv4.QuickStartRequestOption` + or :class:`~pcapkit.protocols.schema.internet.ipv4.QuickStartReportOption` + built directly with an unassigned ``func`` reaches this method + regardless, which is why it still raises below rather than + returning early. """ if isinstance(schema, Schema_UnassignedOption): diff --git a/pcapkit/protocols/internet/ipv6_opts.py b/pcapkit/protocols/internet/ipv6_opts.py index fe62f868d4..703d6798d0 100644 --- a/pcapkit/protocols/internet/ipv6_opts.py +++ b/pcapkit/protocols/internet/ipv6_opts.py @@ -757,17 +757,14 @@ def _read_opt_calipso(self, schema: 'Schema_CALIPSOOption', *, options: 'Option' Returns: Parsed option data. - Raises: - ProtocolError: If the option is malformed. + Note: + ``Cmpt Length`` counts 32-bit words and may be odd + [:rfc:`5570#section-5.2`]; it only has to fit inside ``Opt Data + Len``, which + :func:`~pcapkit.protocols.schema.internet.ipv6_option.calipso_pad_len` + has already checked while the option was unpacked. """ - if schema.len < 8 and schema.len % 8 != 0: - raise ProtocolError(f'{self.alias}: [OptNo {schema.type}] invalid format') - # ``Cmpt Length`` counts 32-bit words and may be odd [:rfc:`5570#section-5.2`]; - # it only has to fit inside ``Opt Data Len`` - if schema.len < 8 + schema.cmpt_len * 4: - raise ProtocolError(f'{self.alias}: [OptNo {schema.type}] invalid format') - opt = Data_CALIPSOOption( type=schema.type, action=Enum_OptionAction.get(schema.type >> 6), @@ -918,7 +915,7 @@ def _read_opt_pdm(self, schema: 'Schema_PDMOption', *, options: 'Option') -> 'Da ) return opt - def _read_opt_qs(self, schema: 'Schema_QuickStartOption', *, options: 'Option') -> 'Data_QuickStartOption': # pylint: disable=unused-argument # pylint: disable=unused-argument + def _read_opt_qs(self, schema: 'Schema_QuickStartOption', *, options: 'Option') -> 'Data_QuickStartOption': # pylint: disable=unused-argument """Read IPv6-Opts Quick Start option. Structure of IPv6-Opts Quick-Start option [:rfc:`4782`]: @@ -962,7 +959,13 @@ def _read_opt_qs(self, schema: 'Schema_QuickStartOption', *, options: 'Option') Note: An option whose QS function is neither of the two above has no defined layout, so the schema keeps it as an unassigned option and - it is parsed as one, see :meth:`_read_opt_none`. + it is parsed as one, see :meth:`_read_opt_none`. That dispatch is + by the schema's own class, decided from the octets a file read + hands it; a :class:`~pcapkit.protocols.schema.internet.ipv6_opts.QuickStartRequestOption` + or :class:`~pcapkit.protocols.schema.internet.ipv6_opts.QuickStartReportOption` + built directly with an unassigned ``func`` reaches this method + regardless, which is why it still raises below rather than + returning early. """ if isinstance(schema, Schema_UnassignedOption): @@ -1521,7 +1524,6 @@ def _make_opt_pad(self, code: 'Enum_Option', opt: 'Optional[Data_PadOption]' = N pad = getattr(opt, 'pad', b'') if code == Enum_Option.Pad1 and length != 0: - #raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid format') warn(f'{self.alias}: [OptNo {code}] invalid format', ProtocolWarning) code = Enum_Option.PadN # type: ignore[assignment] diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index 787c1a18f7..f404287345 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -7921,7 +7921,6 @@ def _make_opt_pad(self, type: 'Enum_Option', option: 'Optional[Data_PadOption]' data = getattr(option, 'data', b'') if type == Enum_Option.Pad1 and length != 0: - # raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid format') warn(f'{self.alias}: [OptNo {type}] invalid format', ProtocolWarning) type = Enum_Option.PadN # type: ignore[assignment] diff --git a/pcapkit/protocols/misc/pcap/header.py b/pcapkit/protocols/misc/pcap/header.py index 6a2c1e0732..59901f2a80 100644 --- a/pcapkit/protocols/misc/pcap/header.py +++ b/pcapkit/protocols/misc/pcap/header.py @@ -42,7 +42,7 @@ from pcapkit.protocols.schema.misc.pcap.header import GLOBAL_HEADER_LEN, MAGIC_NUMBER from pcapkit.protocols.schema.misc.pcap.header import Header as Schema_Header from pcapkit.protocols.schema.schema import keep_short_read, replay_short_read -from pcapkit.utilities.exceptions import EndianError, FileError, UnsupportedCall +from pcapkit.utilities.exceptions import EndianError, UnsupportedCall if TYPE_CHECKING: from enum import IntEnum as StdlibEnum @@ -141,11 +141,10 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_Header' Parsed packet data. Raises: - FileError: If the magic number is invalid. Parsing does not get + ProtocolError: If the magic number is invalid. Parsing does not get this far with one: the schema's :func:`~pcapkit.protocols.schema.misc.pcap.header.magic_number_callback` - raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` for it - while the header is unpacked. + raises it while the header is unpacked. """ if TYPE_CHECKING: @@ -154,12 +153,10 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_Header' _magn = schema.magic_number # NOTE: looked up as :obj:`bytes`, so that a :obj:`bytearray` magic - # number is matched rather than rejected as unhashable. - _decl = (MAGIC_NUMBER.get(bytes(_magn)) - if isinstance(_magn, (bytes, bytearray, memoryview)) else None) - if _decl is None: - raise FileError(5, 'Unknown file format', self._file.name) # pylint: disable=no-member - self._byte, self._nsec = _decl + # number is matched rather than rejected as unhashable. The lookup + # cannot miss: ``magic_number_callback`` has already raised for a magic + # number this mapping does not hold. + self._byte, self._nsec = MAGIC_NUMBER[bytes(_magn)] header = Data_Header( magic_number=Data_MagicNumber( diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index e6a01b085a..56909829af 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -989,7 +989,6 @@ def nanosecond(self) -> 'bool': def ts_resolution(self) -> 'int': """Timestamp resolution of the current block, in units per second.""" if self._ctx is None: - #raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute 'ts_resolution'") warn(f"'{self.__class__.__name__}' object has no attribute 'ts_resolution'", AttributeWarning, stacklevel=stacklevel()) return DEFAULT_RESOLUTION @@ -1001,7 +1000,6 @@ def ts_resolution(self) -> 'int': def ts_offset(self) -> 'int': """Timestamp offset of the current block, in seconds.""" if self._ctx is None: - #raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute 'ts_offset'") warn(f"'{self.__class__.__name__}' object has no attribute 'ts_offset'", AttributeWarning, stacklevel=stacklevel()) return 0 @@ -1018,7 +1016,6 @@ def ts_timezone(self) -> 'timezone': """ if self._ctx is None: - #raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute 'ts_timezone'") warn(f"'{self.__class__.__name__}' object has no attribute 'ts_timezone'", AttributeWarning, stacklevel=stacklevel()) return datetime.timezone.utc @@ -1786,7 +1783,6 @@ def _get_resolution(self, interface_id: 'int' = 0) -> 'int': """ if self._ctx is None: - # raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute '_get_resolution'") warn(f"'{self.__class__.__name__}' object has no attribute '_get_resolution'", AttributeWarning, stacklevel=stacklevel()) return DEFAULT_RESOLUTION @@ -1809,7 +1805,6 @@ def _get_offset(self, interface_id: 'int' = 0) -> 'int': """ if self._ctx is None: - # raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute '_get_offset'") warn(f"'{self.__class__.__name__}' object has no attribute '_get_offset'", AttributeWarning, stacklevel=stacklevel()) return 0 @@ -1841,7 +1836,6 @@ def _get_timezone(self, interface_id: 'int' = 0) -> 'timezone': # and ``_get_offset`` fall back to the format's own defaults in this # situation; this is the matching one. if self._ctx is None: - # raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute '_get_timezone'") warn(f"'{self.__class__.__name__}' object has no attribute '_get_timezone'", AttributeWarning, stacklevel=stacklevel()) return datetime.timezone.utc diff --git a/pcapkit/protocols/schema/internet/hip.py b/pcapkit/protocols/schema/internet/hip.py index b39455261f..fe056a85d1 100644 --- a/pcapkit/protocols/schema/internet/hip.py +++ b/pcapkit/protocols/schema/internet/hip.py @@ -918,8 +918,6 @@ def pre_unpack(cls, packet: 'dict[str, Any]') -> 'None': if not cipher_list: warn(f'HIP: [ParamNo {Enum_Parameter.ENCRYPTED}] ' 'missing HIP_CIPHER parameter', ProtocolWarning) - # raise ProtocolError(f'HIPv{version}: [ParamNo {schema.type}] invalid format') - cipher_id = Enum_Cipher(0xffff) else: cipher_ids = [] # type: list[Enum_Cipher] @@ -933,8 +931,6 @@ def pre_unpack(cls, packet: 'dict[str, Any]') -> 'None': if encrypted_index >= len(cipher_ids): warn(f'HIP: [ParamNo {Enum_Parameter.ENCRYPTED}] ' 'too many ENCRYPTED parameters', ProtocolWarning) - # raise ProtocolError(f'HIPv{version}: [ParamNo {schema.type}] invalid format') - cipher_id = Enum_Cipher(0xfffe) else: cipher_id = cipher_ids[encrypted_index] diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index 278132e034..bd1421dbc2 100644 --- a/pcapkit/protocols/schema/misc/pcapng.py +++ b/pcapkit/protocols/schema/misc/pcapng.py @@ -1039,7 +1039,6 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': """ if self.length != self.length2: block_type = packet.get('__packet__', {}).get('type', 'N/A') - #raise ProtocolError(f'PCAP-NG: [Block {block_type}] block length mismatch: {self.length} != {self.length2}') warn(f'PCAP-NG: [Block {block_type}] block length mismatch: {self.length} != {self.length2}', ProtocolWarning, stacklevel=stacklevel()) return self diff --git a/tests/project/test_dead_helpers_2318_unit.py b/tests/project/test_dead_helpers_2318_unit.py new file mode 100644 index 0000000000..0a993b1c18 --- /dev/null +++ b/tests/project/test_dead_helpers_2318_unit.py @@ -0,0 +1,149 @@ +# -*- coding: utf-8 -*- +"""Every helper here is one something reads. C.f. #2318. + +A helper nothing reads answers a question nobody asks. Each is held to the rule +that makes it worth keeping rather than to its name. + +A ``Family`` in ``examples/generators/dispatch.py`` carries an ``enum`` only +where :func:`code_name` can fall back to one, i.e. where the family's registry +holds a key with no ``.name`` of its own; every key of ``Internet.__proto__`` +carries one, so the ``internet`` family needs no ``enum``. A ``FAKE_GH`` arm +answers only a query a caller sends, and :file:`util/release_upload.py` sends +``ASSETS_JQ`` rather than a bare-name ``assets`` one. And a module-level +constant of ``test_sweeps_on_latest_only_1732`` is worth keeping only where +something reads it, in its own module or in another. + +""" + +import ast +import importlib.util +import pathlib +import re +import sys +import types +import unittest + +#: Test tier (#1958): reads repository files and the dispatch case table, which +#: do not differ by interpreter. +__tier__ = 'LATEST' + +#: Repository root. +ROOT = pathlib.Path(__file__).resolve().parent.parent.parent +#: The generator whose ``Family`` table carries the ``enum`` fall-back. +DISPATCH = ROOT / 'examples' / 'generators' / 'dispatch.py' +#: The workflow test whose ``gh`` stand-in answers the release queries. +RELEASE_TEST = ROOT / 'tests' / 'project' / 'test_release_tags_the_run_sha_1634.py' +#: The workflow test whose module-level constants the other ones import. +SWEEPS_TEST = ROOT / 'tests' / 'project' / 'test_sweeps_on_latest_only_1732.py' +#: Where a query ``FAKE_GH`` answers has to be sent from. +CALLERS = (ROOT / 'util' / 'release_upload.py', *sorted((ROOT / '.github' / 'workflows').glob('*.yml'))) + + +def load_dispatch() -> 'types.ModuleType': + """Load :file:`examples/generators/dispatch.py` by path. + + Follows :func:`tests.protocols.test_dispatch_registry_unit._load_generator`: + the directory is not a package and its module name is too generic to put on + :data:`sys.path`. + + Raises: + RuntimeError: If the module cannot be found or loaded. + + """ + spec = importlib.util.spec_from_file_location('pcapkit_samples_dispatch_2318', DISPATCH) + if spec is None or spec.loader is None: # pragma: no cover + raise RuntimeError(f'cannot load the dispatch case table from {DISPATCH}') + + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def constants(path: 'pathlib.Path') -> 'list[str]': + """The module-level upper-case constants assigned in ``path``.""" + found = [] + for node in ast.parse(path.read_text(encoding='utf-8')).body: + if isinstance(node, ast.Assign): + found.extend(target.id for target in node.targets + if isinstance(target, ast.Name) and target.id.isupper()) + return found + + +def loaded(path: 'pathlib.Path') -> 'set[str]': + """The names ``path`` reads, whether bare or through a module attribute.""" + found = set() + for node in ast.walk(ast.parse(path.read_text(encoding='utf-8'))): + if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Load): + found.add(node.id) + elif isinstance(node, ast.Attribute): + found.add(node.attr) + return found + + +def imports(path: 'pathlib.Path') -> 'set[str]': + """The names ``path`` imports from another module.""" + return {alias.name for node in ast.walk(ast.parse(path.read_text(encoding='utf-8'))) + if isinstance(node, ast.ImportFrom) for alias in node.names} + + +class DispatchEnumTests(unittest.TestCase): + """A family carries an ``enum`` only where ``code_name`` falls back to one.""" + + @classmethod + def setUpClass(cls) -> None: + cls.dispatch = load_dispatch() + + def test_an_enum_is_given_only_where_a_key_needs_it(self) -> None: + for family in self.dispatch.FAMILIES: + with self.subTest(family=family.label): + nameless = [code for code in family.registry() + if getattr(code, 'name', None) is None] + if family.enum is None: + continue + # The fall-back is only reachable for a key with no ``.name``, + # and only useful where an enumeration member equals it. + self.assertTrue(nameless) + enum = family.enum() + self.assertTrue(any(enum(code).name for code in nameless)) + + def test_the_fall_back_names_a_key(self) -> None: + from pcapkit.const.reg.ethertype import EtherType + + self.assertEqual(self.dispatch.code_name(0x8137, EtherType), 'Novell_Inc_0x8137') + self.assertEqual(self.dispatch.code_name(20, None), '20') + + +class FakeGHQueryTests(unittest.TestCase): + """Every ``--json``/``--jq`` pair ``FAKE_GH`` answers is one a caller sends.""" + + def test_each_arm_answers_a_real_query(self) -> None: + source = RELEASE_TEST.read_text(encoding='utf-8') + asked = '\n'.join(path.read_text(encoding='utf-8') for path in CALLERS if path.is_file()) + arms = re.findall(r"^\s*'--json (\S+) --jq (.+?)'\)", source, re.MULTILINE) + self.assertTrue(arms) + for field, query in arms: + with self.subTest(query=query): + self.assertIn(field, asked) + self.assertIn(query, asked) + + +class SweepsConstantTests(unittest.TestCase): + """Every constant of the sweeps workflow test is read somewhere. + + The names are counted from the syntax rather than from the text, so that a + constant named in prose -- in this module's own docstring, for one -- does + not read as a use of it. + + """ + + def test_no_constant_goes_unread(self) -> None: + read = loaded(SWEEPS_TEST) + for path in sorted((ROOT / 'tests').rglob('*.py')): + if path != SWEEPS_TEST: + read |= loaded(path) | imports(path) + self.assertEqual([name for name in constants(SWEEPS_TEST) if name not in read], []) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/project/test_release_tags_the_run_sha_1634.py b/tests/project/test_release_tags_the_run_sha_1634.py index b439e81a75..0b5e2923a2 100644 --- a/tests/project/test_release_tags_the_run_sha_1634.py +++ b/tests/project/test_release_tags_the_run_sha_1634.py @@ -1497,11 +1497,10 @@ def test_the_apps_token_refused_names_its_permissions(self) -> None: #: ``namestate`` lines; absent, there is no release, and ``release view`` #: fails as the real one does ("release not found", exit 1, measured with gh #: 2.102.0). It answers the ``--json``/``--jq`` pairs the workflows ask -- -#: :data:`ASSETS_JQ`, and ``isDraft`` from ``$GH_DRAFT`` -- and a bare-name -#: ``assets`` pair no workflow asks. ``release upload`` appends the file as -#: ``uploaded`` and, like the real one, refuses a name already there unless -#: ``--clobber`` deletes it first. ``$GH_FAIL`` fails any call. Anything else -#: exits 96-99. +#: :data:`ASSETS_JQ`, and ``isDraft`` from ``$GH_DRAFT``. ``release upload`` +#: appends the file as ``uploaded`` and, like the real one, refuses a name +#: already there unless ``--clobber`` deletes it first. ``$GH_FAIL`` fails any +#: call. Anything else exits 96-99. FAKE_GH = r'''#!/bin/sh printf '%s\n' "$*" >> "$GH_LOG" [ -z "${GH_FAIL:-}" ] || { echo "$GH_FAIL" >&2; exit 1; } @@ -1512,7 +1511,6 @@ def test_the_apps_token_refused_names_its_permissions(self) -> None: shift 5 case "$*" in '--json assets --jq .assets[] | [.name, .state] | @tsv') cat "$GH_ASSETS" ;; - '--json assets --jq .assets[].name') cut -f 1 "$GH_ASSETS" ;; '--json isDraft --jq .isDraft') echo "${GH_DRAFT:-false}" ;; *) exit 96 ;; esac ;; diff --git a/tests/project/test_sweeps_on_latest_only_1732.py b/tests/project/test_sweeps_on_latest_only_1732.py index 7f63e9d69c..807312189e 100644 --- a/tests/project/test_sweeps_on_latest_only_1732.py +++ b/tests/project/test_sweeps_on_latest_only_1732.py @@ -71,9 +71,6 @@ #: main's, ``experimental``'s are sized for GraalPy (88-113 minutes, estimate). #: The watcher's interpreter and its requests are capped too (#1938). CAPS = {JOB: (45, {}), EXPERIMENTAL: (140, {'python': 5, 'install': 10, 'watcher-python': 1, 'watcher': 1, 'tests': 120})} -#: The jobs whose result reaches a gate: ``required-checks`` and what it needs, and -#: ``unittest-ordering``, which runs on main's pushes and gates publishing. -GATING = ('required-checks', 'unittest-ordering') #: The unit tier's own flags, which every leg passes. TIER_FLAGS = ('--ignore=tests/integration', '--ignore-glob=*_runtime.py', '--ignore-glob=*_regression.py') #: A ``python`` that appends each argument on a line of ``$ARGS_FILE``, then an diff --git a/tests/protocols/internet/test_ipv6_extension_unit.py b/tests/protocols/internet/test_ipv6_extension_unit.py index 3800128e77..c4b5be5c71 100644 --- a/tests/protocols/internet/test_ipv6_extension_unit.py +++ b/tests/protocols/internet/test_ipv6_extension_unit.py @@ -1312,14 +1312,10 @@ def assert_bad(reader, option_schema) -> None: type=Option.Router_Alert, len=1, alert=RouterAlert.Datagram_contains_RSVP_message, )) - assert_bad(proto._read_opt_calipso, schema.CALIPSOOption( - type=Option.CALIPSO, len=7, domain=1, cmpt_len=0, - level=3, checksum=b'\x01\x02', bitmap=b'', - )) - assert_bad(proto._read_opt_calipso, schema.CALIPSOOption( - type=Option.CALIPSO, len=8, domain=1, cmpt_len=1, - level=3, checksum=b'\x01\x02', bitmap=b'abcd', - )) + # NOTE: A CALIPSO option too short for its compartment bitmap is + # rejected while it is unpacked, by ``calipso_pad_len``, so the reader + # keeps no length guard of its own (#2318); see + # tests/protocols/test_dead_branches_2318_unit.py. bad_smf = schema.SMFHashBasedDPDOption(type=Option.SMF_DPD, len=3, hav=b'\x81\x02\x03') object.__setattr__(bad_smf, 'mode', 2) assert_bad(proto._read_opt_smf_dpd, bad_smf) diff --git a/tests/protocols/misc/pcap/test_header_frame_unit.py b/tests/protocols/misc/pcap/test_header_frame_unit.py index 1adba6eba7..217f0243df 100644 --- a/tests/protocols/misc/pcap/test_header_frame_unit.py +++ b/tests/protocols/misc/pcap/test_header_frame_unit.py @@ -36,7 +36,7 @@ def test_header_make_length_index_and_make_data(self) -> None: from pcapkit.corekit.version import VersionInfo from pcapkit.protocols.misc.pcap.header import Header from pcapkit.protocols.schema.misc.pcap.header import Header as Schema_Header - from pcapkit.utilities.exceptions import EndianError, FileError + from pcapkit.utilities.exceptions import EndianError, ProtocolError from pcapkit.utilities.exceptions import UnsupportedCall header = object.__new__(Header) @@ -106,19 +106,10 @@ def test_header_make_length_index_and_make_data(self) -> None: self.assertEqual(reader.nanosecond, nanosecond) self.assertEqual(reader.protocol, LinkType.NULL) - invalid = object.__new__(Header) - invalid.__header__ = Schema_Header( - magic_number=b'bad!', - version_major=2, - version_minor=4, - thiszone=0, - sigfigs=0, - snaplen=65535, - network=LinkType.NULL, - ) - invalid._file = SimpleNamespace(name='bad.pcap') - with self.assertRaises(FileError): - invalid.read() + # An unknown magic number is rejected while the header is unpacked, by + # the schema's ``magic_number_callback`` (#2318). + with self.assertRaises(ProtocolError): + Header(io.BytesIO(b'bad!' + bytes(20))) Header.__post_init__(header, byteorder='big', nanosecond=True, snaplen=128, network=LinkType.ETHERNET) diff --git a/tests/protocols/test_dead_branches_2318_unit.py b/tests/protocols/test_dead_branches_2318_unit.py new file mode 100644 index 0000000000..48b7ad136b --- /dev/null +++ b/tests/protocols/test_dead_branches_2318_unit.py @@ -0,0 +1,493 @@ +# -*- coding: utf-8 -*- +"""Which guards a reader needs, and which something earlier already runs. C.f. #2318. + +A guard is worth keeping only where some route reaches it. The guards removed +here are unreachable, because something earlier raises first: +``Header.read``'s ``FileError`` on an unknown magic number, which the schema's +:func:`~pcapkit.protocols.schema.misc.pcap.header.magic_number_callback` beats +to it; and the ``Opt Data Len`` guards of ``_read_opt_calipso``, which +:func:`~pcapkit.protocols.schema.internet.ipv6_option.calipso_pad_len` beats to +it whether the option is unpacked off the wire or packed from a caller's own +schema. For each, a test feeds the input that would have reached it, pins what +is raised -- or parsed -- instead, and pins the source as holding no such +guard. + +The ``unknown QS function`` arm of ``_read_opt_qs`` is reachable, so it stays. +A parse of the stock registry does not reach it, since that registry leaves an +unassigned function with no schema and the option arrives as an +``UnassignedOption``; *construction* reaches it, since the reader is handed the +caller's own option schemas and the schema's selector never runs. A *parse* +reaches it too once a caller has registered a schema of their own against an +unassigned function code, which is what the selector then hands the reader. See +:class:`QuickStartFunctionTests` and +:class:`RegisteredQuickStartFunctionTests`, which pin those routes. + +The commented-out ``raise`` lines that sit beside a ``warn`` standing in their +place are audited the same way, over the whole package, since they read as if +the stricter behaviour were one uncomment away from returning. + +:mod:`pcapkit` is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import ast +import inspect +import io +import pathlib +import re +import tempfile +import tokenize +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class + +if TYPE_CHECKING: + from typing import Any + +#: Test tier (#1958): in-memory parses and a source audit, neither of which +#: differs by interpreter. +__tier__ = 'LATEST' + +#: The package audited for a commented-out ``raise``. +PACKAGE = pathlib.Path(__file__).resolve().parent.parent.parent / 'pcapkit' + +#: How near a ``warn(`` has to be for a commented-out ``raise`` to read as the +#: stricter behaviour a ``warn`` stands in for, in lines either side. +NEARBY = 3 + + +def hopopt_bytes(option: 'bytes') -> 'bytes': + """A HOPOPT header carrying ``option``, padded to a multiple of 8 octets.""" + pad = -(len(option) + 2) % 8 + if pad == 1: + option += bytes(1) + elif pad: + option += bytes([1, pad - 2]) + bytes(pad - 2) + return bytes([59, (len(option) + 2) // 8 - 1]) + option + + +def ipv4_bytes(options: 'bytes') -> 'bytes': + """An IPv4 header carrying ``options``, padded to a multiple of 4 octets.""" + options += bytes(-len(options) % 4) + length = 20 + len(options) + return (bytes([0x40 | length // 4, 0]) + length.to_bytes(2, 'big') + bytes(4) + + bytes([64, 59, 0, 0]) + bytes([10, 0, 0, 1]) + bytes([10, 0, 0, 2]) + options) + + +def commented_raises(path: 'pathlib.Path') -> 'list[int]': + """The lines of ``path`` holding a commented-out ``raise`` beside a ``warn``. + + A comment counts only when it parses as a lone :keyword:`raise` statement, + so that prose opening with the word ("raise ``OSError`` on a host with no + ...") is left alone, and only when a ``warn(`` call sits within + :data:`NEARBY` lines of it. + + """ + lines = path.read_text(encoding='utf-8').splitlines() + found = [] + with open(path, 'rb') as stream: + for token in tokenize.tokenize(stream.readline): + if token.type != tokenize.COMMENT: + continue + try: + body = ast.parse(token.string.lstrip('#').strip()).body + except SyntaxError: + continue + if len(body) != 1 or not isinstance(body[0], ast.Raise): + continue + row = token.start[0] - 1 + if any('warn(' in line for line in lines[max(0, row - NEARBY):row + NEARBY + 1]): + found.append(row + 1) + return found + + +class MagicNumberTests(unittest.TestCase): + """An unknown magic number raises from the schema, as documented.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_unknown_magic_number_raises_protocol_error(self) -> None: + from pcapkit.protocols.misc.pcap.header import Header + from pcapkit.utilities.exceptions import ProtocolError + + with self.assertRaises(ProtocolError) as caught: + Header(io.BytesIO(b'\x00\x01\x02\x03' + bytes(20))) + self.assertIn('invalid magic number', str(caught.exception)) + + def test_the_header_module_names_protocol_error(self) -> None: + from pcapkit.protocols.misc.pcap import header + + docstring = header.Header.read.__doc__ or '' + self.assertIn('ProtocolError', docstring) + self.assertNotIn('FileError', docstring) + self.assertNotIn('FileError', inspect.getsource(header)) + + +class QuickStartFunctionTests(unittest.TestCase): + """An unassigned QS function parses as unassigned, and is rejected on + construction. + + A parse and a construction reach ``_read_opt_qs`` by different routes, and + what the parse route carries depends on what the ``QuickStartOption`` + registry holds. + + A parse runs + :func:`~pcapkit.protocols.schema.internet.ipv6_option.quick_start_data_selector` + over the octets first, and that selector reads the registry. Under the stock + registry, which assigns ``Quick_Start_Request`` (``0``) and + ``Report_of_Approved_Rate`` (``8``) and nothing else, any other ``func`` + nibble arrives as an ``UnassignedOption`` and is read by + ``_read_opt_none``, so the function arms never see it. A caller who + registers a schema of their own against an unassigned function code changes + that, and an ordinary parse then reaches the arms; see + :class:`RegisteredQuickStartFunctionTests`. + + Construction hands the reader the caller's own schema objects -- + ``_read_hopopt_options`` and its siblings iterate + ``self.__header__.options``, not a re-parse of the octets they wrote -- so + the selector never runs. A ``QuickStartRequestOption`` or + ``QuickStartReportOption`` whose ``func`` names neither assigned layout + therefore reaches the arms, and the ``unknown QS function`` guard is what + rejects it. Without that guard such an option is written to the wire for an + unassigned function, and a request schema is read as a report schema, + failing on the report layout's ``reserved`` field -- which the request + layout does not have -- with an :exc:`AttributeError` naming nothing the + caller got wrong. + + """ + + #: QS functions with no assigned layout: either side of the functions the + #: registry assigns, ``Quick_Start_Request`` (``0``) and + #: ``Report_of_Approved_Rate`` (``8``), and the top of the nibble. + UNASSIGNED = (1, 2, 7, 9, 15) + + def setUp(self) -> None: + reimport_once_per_class(self) + + def qs_cases(self, func: 'int') -> 'list[tuple[str, Any, dict[str, Any], Any]]': + """Each protocol, with a request and a report schema naming ``func``. + + Returns: + One ``(label, protocol, construction keywords, option schema)`` + tuple per protocol and schema class, for an option whose ``func`` + nibble is ``func``. + + """ + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.const.ipv6.option import Option + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + from pcapkit.protocols.schema.internet import hopopt as schema_hopopt + from pcapkit.protocols.schema.internet import ipv4 as schema_ipv4 + from pcapkit.protocols.schema.internet import ipv6_opts as schema_ipv6_opts + + flags = {'func': func, 'rate': 1} + nonce = {'nonce': 0x12345, 'reserved': 0} + cases = [] + for protocol, schema in ((HOPOPT, schema_hopopt), (IPv6_Opts, schema_ipv6_opts)): + # An extension header option carries ``Opt Data Len``, i.e. the six + # octets after the two header ones. + cases.append((f'{protocol.__name__}/request', protocol, {'next': 59, 'payload': b''}, + schema.QuickStartRequestOption( + type=Option.Quick_Start, len=6, flags=flags, ttl=10, nonce=nonce))) + cases.append((f'{protocol.__name__}/report', protocol, {'next': 59, 'payload': b''}, + schema.QuickStartReportOption( + type=Option.Quick_Start, len=6, flags=flags, + reserved=bytes(1), nonce=nonce))) + # An IPv4 option's ``length`` counts its own header octets too. + cases.append(('IPv4/request', IPv4, {'protocol': 59}, + schema_ipv4.QuickStartRequestOption( + type=OptionNumber.QS, length=8, flags=flags, ttl=10, nonce=nonce))) + cases.append(('IPv4/report', IPv4, {'protocol': 59}, + schema_ipv4.QuickStartReportOption( + type=OptionNumber.QS, length=8, flags=flags, reserved=0, nonce=nonce))) + return cases + + def test_a_constructed_option_with_an_unassigned_function_is_rejected(self) -> None: + """``make`` rejects an option schema naming a function it cannot be. + + This is the route the schema's selector does not stand in front of, so + it is the route the ``unknown QS function`` guard answers without a + caller having registered a schema of their own. The message names the + function, so that the option -- and not the option-length check the + reader runs before it -- is what the caller is told about. + + """ + from pcapkit.utilities.exceptions import ProtocolError + + for func in self.UNASSIGNED: + for label, protocol, keywords, schema in self.qs_cases(func): + with self.subTest(case=label, func=func): + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + with self.assertRaises(ProtocolError) as caught: + protocol(options=[schema], **keywords) + self.assertIn(f'unknown QS function: {func}', str(caught.exception)) + + def test_an_assigned_function_constructs(self) -> None: + """A schema naming the function its layout is the layout of is written. + + The guard rejects a self-contradictory option, not every constructed + one, so each schema class is built with its own function and the + octets it writes are read back. + + """ + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.const.ipv6.option import Option + + for func, kind in ((0, 'request'), (8, 'report')): + for label, protocol, keywords, schema in self.qs_cases(func): + if not label.endswith(kind): + continue + with self.subTest(case=label, func=func): + packet = protocol(options=[schema], **keywords) + code = OptionNumber.QS if protocol.__name__ == 'IPv4' else Option.Quick_Start + self.assertEqual(type(packet.info.options[code]).__name__, + type(schema).__name__) + + def test_every_function_nibble_parses(self) -> None: + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.const.ipv6.option import Option + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + + expected = {0: 'QuickStartRequestOption', 8: 'QuickStartReportOption'} + for func in range(16): + name = expected.get(func, 'UnassignedOption') + option = bytes([38, 6, (func << 4) | 1, 10]) + bytes(4) + for protocol in (HOPOPT, IPv6_Opts): + with self.subTest(protocol=protocol.__name__, func=func): + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + parsed = protocol(hopopt_bytes(option)).info.options[Option.Quick_Start] + self.assertEqual(type(parsed).__name__, name) + with self.subTest(protocol='IPv4', func=func): + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + option4 = bytes([25, 8, (func << 4) | 1, 10]) + bytes(4) + parsed = IPv4(ipv4_bytes(option4)).info.options[OptionNumber.QS] + self.assertEqual(type(parsed).__name__, name) + + def test_every_reader_keeps_the_unknown_function_guard(self) -> None: + """Each reader holds the guard, and documents that it raises. + + The source is pinned as well as the behaviour, because a parse of the + stock registry does not reach the guard: a reading of that path alone + makes it look dead. + + """ + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + + for protocol in (HOPOPT, IPv6_Opts, IPv4): + with self.subTest(protocol=protocol.__name__): + source = inspect.getsource(protocol._read_opt_qs) + self.assertIn('unknown QS function', source) + self.assertIn('ProtocolError', protocol._read_opt_qs.__doc__ or '') + # One ``# pylint: disable=`` pragma per line, not two. + for line in source.splitlines(): + self.assertLessEqual(line.count('pylint: disable='), 1, line) + + +class RegisteredQuickStartFunctionTests(unittest.TestCase): + """A parse reaches the guard once a caller registers an unassigned function. + + ``QuickStartOption`` is an + :class:`~pcapkit.protocols.schema.schema.EnumSchema`, so a caller may + register a layout of their own against a function code the stock registry + leaves empty. + :func:`~pcapkit.protocols.schema.internet.ipv6_option.quick_start_data_selector` + then hands the reader that schema rather than an ``UnassignedOption``, and + ordinary octets off a file read reach the ``unknown QS function`` arm -- + the arm is therefore not construction-only. + + The registration mutates a package-wide registry, so this class takes its + :mod:`pcapkit` import with ``restore=True`` and gives it back when it + finishes. + + """ + + #: A QS function the stock registry leaves without a layout. + FUNC = 3 + + def setUp(self) -> None: + reimport_once_per_class(self, restore=True) + + def register(self, base: 'Any', code: 'Any') -> 'None': + """Register a caller's own layout for ``code`` against ``base``.""" + from pcapkit.corekit.fields.numbers import UInt8Field + from pcapkit.corekit.fields.strings import BytesField + from pcapkit.protocols.schema.schema import schema_final + + self.assertIsNone(base.registry[code]) + + @schema_final + class CallerQuickStartOption(base, code=code): # type: ignore[misc,valid-type] + """A caller's own layout for an unassigned QS function.""" + + ttl: 'int' = UInt8Field() + rest: 'bytes' = BytesField(length=4) + + self.assertIs(base.registry[code], CallerQuickStartOption) + + def test_a_parse_reaches_the_guard(self) -> None: + """Ordinary octets reach the arm once the function has a layout.""" + from pcapkit.const.ipv4.qs_function import QSFunction as QSFunction4 + from pcapkit.const.ipv6.qs_function import QSFunction as QSFunction6 + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + from pcapkit.protocols.schema.internet.hopopt import QuickStartOption as QSOption_hopopt + from pcapkit.protocols.schema.internet.ipv4 import QSOption as QSOption_ipv4 + from pcapkit.protocols.schema.internet.ipv6_opts import \ + QuickStartOption as QSOption_ipv6_opts + from pcapkit.utilities.exceptions import ProtocolError + + flags = (self.FUNC << 4) | 1 + # An extension header option carries ``Opt Data Len``; an IPv4 option's + # ``length`` counts its own header octets too. + extension = hopopt_bytes(bytes([38, 6, flags, 10]) + bytes(4)) + cases = ( + (HOPOPT, QSOption_hopopt, QSFunction6.get(self.FUNC), extension), + (IPv6_Opts, QSOption_ipv6_opts, QSFunction6.get(self.FUNC), extension), + (IPv4, QSOption_ipv4, QSFunction4.get(self.FUNC), + ipv4_bytes(bytes([25, 8, flags, 10]) + bytes(4))), + ) + for protocol, base, code, packet in cases: + with self.subTest(protocol=protocol.__name__): + self.register(base, code) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + with self.assertRaises(ProtocolError) as caught: + protocol(packet) + self.assertIn(f'unknown QS function: {self.FUNC}', str(caught.exception)) + + +class CALIPSOLengthTests(unittest.TestCase): + """A CALIPSO option too short for its bitmap is rejected either way round. + + ``calipso_pad_len`` sizes ``pad`` on both routes that reach the reader -- + a parse, where the option is unpacked off the wire, and a construction, + where the caller's own option schema is packed -- so each is pinned. + + """ + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_short_option_raises_from_the_pad_length(self) -> None: + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + from pcapkit.utilities.exceptions import FieldValueError + + for length, cmpt_len in ((7, 0), (5, 0), (8, 1), (11, 1)): + option = (bytes([7, length]) + bytes(4) + bytes([cmpt_len, 0]) + b'\xab\xcd' + + bytes(max(0, length - 8))) + for protocol in (HOPOPT, IPv6_Opts): + with self.subTest(protocol=protocol.__name__, length=length): + with self.assertRaises(FieldValueError) as caught: + protocol(hopopt_bytes(option)) + self.assertIn(f'invalid CALIPSO option length: {length}', str(caught.exception)) + + def test_a_constructed_option_raises_from_the_pad_length(self) -> None: + """``make`` rejects the same option, before any reader runs. + + Construction packs the caller's own option schema, so + :func:`~pcapkit.protocols.schema.internet.ipv6_option.calipso_pad_len` + sizes ``pad`` there too and raises out of ``schema.pack()`` while the + header is being made. ``_read_opt_calipso`` never sees the option, so + a length guard of its own would be unreachable on this route as well + as on a parse. + + """ + from pcapkit.const.ipv6.option import Option + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + from pcapkit.protocols.schema.internet import hopopt as schema_hopopt + from pcapkit.protocols.schema.internet import ipv6_opts as schema_ipv6_opts + from pcapkit.utilities.exceptions import FieldValueError + + for length, cmpt_len in ((7, 0), (5, 0), (8, 1), (11, 1)): + # ``bitmap`` is conditional on a non-zero ``cmpt_len``, and is the + # four octets per compartment word that ``Opt Data Len`` is short of. + bitmap = bytes(cmpt_len * 4) if cmpt_len else None + for protocol, schema in ((HOPOPT, schema_hopopt), (IPv6_Opts, schema_ipv6_opts)): + with self.subTest(protocol=protocol.__name__, length=length): + option = schema.CALIPSOOption( + type=Option.CALIPSO, len=length, domain=0x12345678, + cmpt_len=cmpt_len, level=5, checksum=b'\xab\xcd', + bitmap=bitmap, pad=b'') + with self.assertRaises(FieldValueError) as caught: + option.pack() + self.assertIn(f'invalid CALIPSO option length: {length}', str(caught.exception)) + with self.assertRaises(FieldValueError) as caught: + protocol(options=[option], next=59, payload=b'') + self.assertIn(f'invalid CALIPSO option length: {length}', str(caught.exception)) + + def test_the_reader_keeps_no_length_guard(self) -> None: + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + + for protocol in (HOPOPT, IPv6_Opts): + with self.subTest(protocol=protocol.__name__): + source = inspect.getsource(protocol._read_opt_calipso) + self.assertNotIn('raise', source) + + def test_a_well_formed_option_parses(self) -> None: + from pcapkit.const.ipv6.option import Option + from pcapkit.protocols.internet.hopopt import HOPOPT + + option = (bytes([7, 12]) + (0x12345678).to_bytes(4, 'big') + bytes([1, 5]) + + b'\xab\xcd' + bytes([0xde, 0xad, 0xbe, 0xef])) + parsed = HOPOPT(hopopt_bytes(option)).info.options[Option.CALIPSO] + self.assertEqual(parsed.cmpt_len, 4) + self.assertEqual(parsed.cmpt_bitmap, (0xde, 0xad, 0xbe, 0xef)) + + +class CommentedOutRaiseTests(unittest.TestCase): + """No commented-out ``raise`` sits beside a ``warn`` standing in its place.""" + + def test_the_package_keeps_none(self) -> None: + offenders = {str(path.relative_to(PACKAGE)): rows + for path in sorted(PACKAGE.rglob('*.py')) + if (rows := commented_raises(path))} + self.assertEqual(offenders, {}) + + def test_the_audit_finds_one(self) -> None: + """The audit catches a commented-out ``raise``, and spares the prose.""" + with tempfile.TemporaryDirectory() as directory: + path = pathlib.Path(directory) / 'sample.py' + path.write_text( + 'def f(bad):\n' + ' if bad:\n' + " # raise ValueError('bad')\n" + " warn('bad', UserWarning)\n" + '\n' + 'def g():\n' + ' # raise ``OSError`` where the library is absent\n' + " warn('absent', UserWarning)\n", + encoding='utf-8') + self.assertEqual(commented_raises(path), [3]) + + +class PylintPragmaTests(unittest.TestCase): + """No source line repeats a ``pylint: disable=`` pragma.""" + + def test_no_line_carries_two(self) -> None: + offenders = [] + for path in sorted(PACKAGE.rglob('*.py')): + for row, line in enumerate(path.read_text(encoding='utf-8').splitlines(), 1): + if len(re.findall(r'pylint: disable=', line)) > 1: + offenders.append(f'{path.relative_to(PACKAGE)}:{row}') + self.assertEqual(offenders, []) + + +if __name__ == '__main__': + unittest.main()