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()