From 4294f275223972fb40c614653051694d0d9ef631 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Thu, 17 Sep 2026 13:52:22 -0400 Subject: [PATCH 1/4] tests: exercise construction as well as parsing, over every registered option code The sample corpus covered the parse path for common protocols and almost none of the option, chunk, parameter and block space -- and none of the *construction* side of it, which is why defects there keep being found by hand. - Add `examples/generators/options.py`. It enumerates 258 codes from the dispatch registries themselves (TCP, MPTCP, IPv4, HOPOPT, IPv6-Opts, IPv6-Route, SCTP chunks/parameters/causes, MH messages/options/extensions, HIP parameters, HTTP/2 frames, PCAP-NG blocks/options/records/secrets), then for each one constructs it, parses it back, constructs it again from what was parsed, and compares the octets. Registries are iterated, never subscripted, so no lookup grows a shared `defaultdict`; `handler()` goes through `_lookup_registry`. - Write the 166 cases whose parse terminates into five `options-*.pcap` fixtures, wired into `make_samples.py`. They regenerate byte-identically. - Add `tests/protocols/test_option_roundtrip_unit.py`: 172 of 258 cases close the cycle; the other 86 are recorded case by case against the `file:line` that stops each, and are asserted to still fail in the recorded way so a fix turns the tier red rather than leaving a stale entry. - Add `tests/protocols/test_option_coverage_runtime.py`, reading the fixtures through the extraction interface and both dumpers. - Vendor `time_limit` into `tests/_support.py`, byte-identical to #431's copy; two HOPOPT/IPv6-Opts cases hang rather than fail, so the sweep is deadlined. No library file is touched: the 86 failures are reported, not fixed. 886 passed, 17 skipped (877 before, adjusting for 18 tier-guard tests that only skip in a non-git tree). Every pre-existing capture, and its `tree` and `json` dump under `ip=True, tcp=True, reassembly=True`, is byte-identical to origin/main. --- README.rst | 11 + examples/generators/make_samples.py | 9 +- examples/generators/options.py | 1766 +++++++++++++++++ tests/_support.py | 53 +- .../protocols/test_option_coverage_runtime.py | 224 +++ tests/protocols/test_option_roundtrip_unit.py | 681 +++++++ 6 files changed, 2741 insertions(+), 3 deletions(-) create mode 100644 examples/generators/options.py create mode 100644 tests/protocols/test_option_coverage_runtime.py create mode 100644 tests/protocols/test_option_roundtrip_unit.py diff --git a/README.rst b/README.rst index 152922b299..b58f0eeb37 100644 --- a/README.rst +++ b/README.rst @@ -386,6 +386,17 @@ and ``make test-all`` regenerates them before running the whole suite: The same fixtures back the demonstration scripts in ``examples/legacy_smoke/``, which read them as ``../captures/…``. +One of the generators works differently from the rest and is worth knowing about. +``examples/generators/options.py`` does not describe packets it wants; it asks the +library which option, chunk, parameter, message, frame and block codes it +registers, and then constructs one of each through the public construction API, +parses it back, and constructs it again from what was parsed. The +``options-*.pcap`` captures are the cases that survive that round trip, so they +record what this version of ``pcapkit`` builds rather than what a third-party +tool builds. ``tests/protocols/test_option_roundtrip_unit.py`` runs the same +cases without needing a fixture at all, and carries a table of the ones that do +not yet close the cycle, each named against the defect that stops it. + Continuous integration runs the ``make test`` selection, since the fixtures are not in the repository. ``tshark`` is only required to exercise the PyShark engine, and is not needed by the test suite. diff --git a/examples/generators/make_samples.py b/examples/generators/make_samples.py index d05c715d75..c07c24a452 100644 --- a/examples/generators/make_samples.py +++ b/examples/generators/make_samples.py @@ -20,6 +20,7 @@ :file:`pcap.py` the ``.pcap`` captures the unit and runtime tests read :file:`pcapng.py` the ``.pcapng`` captures the regression tests read :file:`legacy.py` the extra captures ``examples/legacy_smoke/`` reads +:file:`options.py` the ``options-*.pcap`` option-coverage captures =================== ========================================================== They are loaded by path rather than imported by name, since this directory is @@ -44,8 +45,12 @@ HERE = pathlib.Path(__file__).resolve().parent #: Destination directory for every generated capture. DEST = ROOT / 'examples' / 'captures' -#: Generator modules, in the order they are run. -GENERATORS = ('pcap', 'pcapng', 'legacy') +#: Generator modules, in the order they are run. :file:`options.py` runs last +#: because it is the only one that builds its captures out of :mod:`pcapkit`'s +#: own construction output, so a failure in it is a statement about the library +#: rather than about the fixture -- and reading it after the others have already +#: printed keeps that distinction visible in the log. +GENERATORS = ('pcap', 'pcapng', 'legacy', 'options') def load(name: 'str') -> 'ModuleType': diff --git a/examples/generators/options.py b/examples/generators/options.py new file mode 100644 index 0000000000..28521c5ad7 --- /dev/null +++ b/examples/generators/options.py @@ -0,0 +1,1766 @@ +# -*- coding: utf-8 -*- +"""Generate the option-coverage sample captures, and the case table behind them. + +The other generators in this directory cover the *parse* path for common +protocols. What they do not cover is the option, chunk, parameter and block +space -- and in particular they do not cover the **construction** side of it at +all, the ``_make_*`` methods and the schema ``pack`` path. This module closes +that gap, and it is deliberately built the other way round from its siblings: +they know which packets they want and spell them out, whereas this one asks the +library which codes it claims to support and then exercises every one of them. + +=============================== ============================================= +Capture Codes +=============================== ============================================= +:file:`options-tcp.pcap` TCP options and Multipath TCP subtypes +:file:`options-ipv4.pcap` IPv4 options +:file:`options-ipv6.pcap` HOPOPT, IPv6-Opts and IPv6-Route +:file:`options-transport.pcap` SCTP chunks, parameters and error causes, + and HTTP/2 frames +:file:`options-internet.pcap` Mobility Header messages, options and + extensions, and HIP parameters +=============================== ============================================= + +Each frame carries exactly one item of exactly one code, so a parse failure +names the option that caused it instead of a soup of several. Every frame is an +Ethernet envelope around octets that :mod:`pcapkit` itself constructed, rather +than a packet assembled by :mod:`scapy` from its own protocol model. That is the +point: the octets under test are the library's own output, so a capture is a +record of what this version of ``pcapkit`` builds, and re-parsing it exercises +``_read_*`` against ``_make_*``. + +Only the codes that construct *successfully* can appear in a capture -- there +are no octets for the ones that do not. Roughly a fifth of the space does not, +and that shortfall is not swept up: it is recorded case by case in +:file:`tests/protocols/test_option_roundtrip_unit.py`, against the defect that +causes it. + +Why the case table is not a list of options +------------------------------------------- + +:func:`cases` walks the dispatch registries -- ``TCP.__option__``, +``SCTP.__chunk__``, ``MH.__message__`` and the rest -- so a code registered +tomorrow appears here tomorrow, and the unit test fails until somebody gives it +a case. What the tables below hold is only the *arguments*, which cannot be +derived: each option means something different, so each needs its own value. +Most need nothing at all, since nearly every ``_make_*`` in the tree is fully +defaulted; an entry exists where the default is degenerate (a zero-length +variable field), machine-dependent, or outright rejected. + +Every one of these registries is a :class:`collections.defaultdict` held on a +*class* attribute, so ``registry[code]`` inserts on a miss and permanently grows +a registry shared by every instance in the process. Nothing here subscripts one: +:func:`cases` only ever *iterates*, which touches nothing, and :func:`handler` -- +which resolves one particular code, and so cannot iterate -- goes through +:meth:`ProtocolBase._lookup_registry +` instead. + +IPv4 and HIP have no such registry yet -- they dispatch on the attribute name +``_make_opt_${name}`` / ``_make_param_${name}`` -- so for those two the source +of truth is the enumeration crossed with the presence of the handler, which is +what :func:`_named_registry` builds. + +Determinism +----------- + +The captures have to regenerate byte-identically; a fixture that churns is +worse than none, because every unrelated change then shows up as a diff. Four +things would otherwise vary between runs and are pinned: + +1. Frame timestamps, from :data:`EPOCH` rather than the clock. +2. The order cases are emitted in, from :func:`sort_key` rather than from + registry iteration order -- which is insertion order, and therefore an + implementation detail of whichever module last registered a code. +3. ``MH``'s ``MESG_ID_OPTION_TYPE``, whose ``_make_opt_mesg_id`` falls back to + :meth:`datetime.datetime.now` when given neither ``timestamp`` nor + ``interval``. It is given one below. +4. Anything reading the host. No case here uses PCAP-NG's + ``_make_option_if_os``, which defaults to the live :mod:`platform` string. + +Two deliberate deviations from the sibling generators +---------------------------------------------------- + +Almost every import of :mod:`pcapkit` and of :mod:`scapy` in this module is +inside the function that needs it, which is why :mod:`pylint`'s +``import-outside-toplevel`` is switched off for the file rather than argued with +forty times. Both are load-bearing. :mod:`scapy` is needed only by +:func:`generate`, while the case table and :func:`roundtrip` are useful without +it -- the unit-tier test uses both and has to run on a checkout that installed no +:mod:`scapy`, so a top-level import would make that test skip for a dependency it +never touches. The :mod:`pcapkit` imports are deferred so that importing this +module does not drag in every protocol in the tree, which is what lets the unit +test purge :mod:`pcapkit` and *then* load the case table without the two +orderings fighting. + +:func:`roundtrip` puts a :func:`signal.alarm` deadline around each case. That +is not defensive dressing: GitHub issue #431 is a parser defect that +degenerates into a loop making no progress, ``HOPOPT``'s ``SMF_DPD`` reaches it, +and without a deadline ``make samples`` does not fail -- it *hangs*, taking the +whole build with it. A signal is what interrupts it, because the loop is pure +Python and holds the GIL for the whole of an iteration, so no watchdog thread +would ever get to run. + +""" + +# Deferring these is the design, not an oversight -- see the module docstring. +# pylint: disable=import-outside-toplevel + +from __future__ import annotations + +import collections +import datetime +import pathlib +import signal +import warnings +from typing import TYPE_CHECKING, NamedTuple + +from pcapkit.protocols.protocol import ProtocolBase + +if TYPE_CHECKING: + from typing import Any, Callable, Iterator, Optional + +__all__ = ['generate', 'cases', 'roundtrip', 'outcomes', 'Case', 'Outcome', 'FAMILIES'] + +#: Repository root, i.e. the grandparent of the directory holding this file. +ROOT = pathlib.Path(__file__).resolve().parents[2] +#: Default destination directory for the generated captures. +SAMPLE = ROOT / 'examples' / 'captures' + +#: Capture start time, fixed so that regenerating gives identical files. +EPOCH = 1500000000.0 + +#: Source MAC for every generated frame. +SRC_MAC = '02:00:00:00:00:01' +#: Destination MAC for every generated frame. +DST_MAC = '02:00:00:00:00:02' +#: Source address for every generated IPv4 envelope. +SRC_IP = '192.0.2.1' +#: Destination address for every generated IPv4 envelope. +DST_IP = '198.51.100.1' +#: Source address for every generated IPv6 envelope. +SRC_IP6 = '2001:db8::1' +#: Destination address for every generated IPv6 envelope. +DST_IP6 = '2001:db8::2' + +#: Seconds a single case may take before :func:`roundtrip` gives up on it and +#: records ``'TIMEOUT'``. Generous next to a working case, which takes low +#: single-digit milliseconds, and short enough that a whole sweep still ends. +DEADLINE = 5 + +#: Fixed instant handed to any constructor that would otherwise read the clock. +#: Timezone-aware and in the past, so it is stable and unambiguous. +FIXED_TIME = datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc) + + +class Case(NamedTuple): + """One option-like code, and the arguments that construct it.""" + + #: Family label, e.g. ``'tcp-option'``. Names the registry the code came + #: from and, with :attr:`name`, forms the case's unique label. + family: 'str' + #: Registry key, i.e. the wire code. Usually an enumeration member; for + #: ``PCAPNG.__option__`` a ``(str, int)`` tuple, since the same numeric + #: option code means different things in different block types. + code: 'Any' + #: Human-readable code name, used in the case label and in test ids. + name: 'str' + #: Keyword arguments handed to the constructor. ``{}`` means "every + #: default", which is what most codes need. + kwargs: 'dict[str, Any]' + + @property + def label(self) -> 'str': + """``family/name``, unique across every family.""" + return f'{self.family}/{self.name}' + + +class Outcome(NamedTuple): + """What one construct -> parse -> construct cycle did.""" + + #: The case this describes. + case: 'Case' + #: The step that failed, or ``'OK'`` -- see :data:`STATUSES`. + status: 'str' + #: Exception type and message, or for ``'MISMATCH'`` the two hex strings. + #: Empty for ``'OK'``. + detail: 'str' + #: The octets construction produced, or :data:`None` if it never got that + #: far. Kept as :obj:`bytes` so a caller can put them straight in a frame. + octets: 'Optional[bytes]' + #: Warning messages raised anywhere in the cycle, deduplicated but in + #: order. A case can be ``'OK'`` and still warn, and that is worth seeing: + #: a silent ``SchemaWarning: packet length < 0`` means an option over-read + #: and the octets around it are not what they appear to be. + warnings: 'tuple[str, ...]' + + +#: Statuses whose octets may go into a capture, i.e. the ones whose *parse* +#: step ran to completion. +#: +#: This is a safety property, not a tidiness one. ``HOPOPT``'s ``SMF_DPD`` +#: constructs perfectly well and then spins forever on the way back in, so its +#: octets are available and putting them in a fixture would produce a capture +#: that wedges every reader of it -- including the rest of this suite. A +#: ``CONSTRUCT`` or ``PARSE`` failure is milder but no more useful: the frame +#: would be one no reader can get through. So a capture carries only frames +#: that are known to parse, and the cases that do not are recorded in +#: :file:`tests/protocols/test_option_roundtrip_unit.py` instead, where a +#: failure is an assertion rather than a hang. +CAPTURABLE = ('OK', 'RECONSTRUCT', 'MISMATCH') + +#: Every value :attr:`Outcome.status` can take. +#: +#: * ``'OK'`` -- constructed, parsed, reconstructed, and the octets matched. +#: * ``'CONSTRUCT'`` -- the constructor raised. Note that +#: :meth:`ProtocolBase.__post_init__` packs *and then* unpacks, so a +#: ``_read_*`` fault on a perfectly good pack also lands here. +#: * ``'PARSE'`` -- the octets would not parse back. +#: * ``'RECONSTRUCT'`` -- ``_make_*`` could not consume what ``_read_*`` +#: produced. This is the step a construct-then-parse test cannot see. +#: * ``'MISMATCH'`` -- both directions worked and the octets differed. +#: * ``'TIMEOUT'`` -- the case did not finish inside :data:`DEADLINE`. +STATUSES = ('OK', 'CONSTRUCT', 'PARSE', 'RECONSTRUCT', 'MISMATCH', 'TIMEOUT') + + +class Family(NamedTuple): + """A registry, and how to exercise the codes in it.""" + + #: Family label, copied into every :class:`Case` it yields. + label: 'str' + #: Zero-argument callable returning the dispatch registry, or a mapping + #: standing in for one. Deferred so that importing this module does not + #: import every protocol in the tree. + registry: 'Callable[[], Any]' + #: Zero-argument callable returning per-code keyword overrides, keyed by + #: registry key. A code absent from the mapping is constructed with no + #: arguments at all. Deferred for the same reason as :attr:`registry`. + overrides: 'Callable[[], dict[Any, dict[str, Any]]]' + #: ``build(code, kwargs)`` -> a constructed protocol carrying exactly one + #: item of ``code``. + build: 'Callable[[Any, dict[str, Any]], Any]' + #: ``parse(octets)`` -> the protocol, parsed back from its own octets. + parse: 'Callable[[bytes], Any]' + #: ``extract(parsed)`` -> the parsed collection, in whatever form + #: :attr:`rebuild` takes. + extract: 'Callable[[Any], Any]' + #: ``rebuild(collection)`` -> a protocol constructed from the *parsed* + #: collection rather than from keyword arguments. This is the step that + #: catches a ``_make_*`` unable to consume what its own ``_read_*`` made. + rebuild: 'Callable[[Any], Any]' + #: Name of the capture this family's frames go into, or :data:`None` for a + #: family that is exercised but not captured -- see :data:`FAMILIES`. + capture: 'Optional[str]' + #: How the octets are wrapped into a frame -- see :func:`_frame`. + envelope: 'Optional[str]' + #: Next-layer protocol number for the envelope, where it needs one. + proto: 'Optional[int]' = None + + +def sort_key(code: 'Any') -> 'tuple[int, str, int]': + """Total order over registry keys, stable across runs. + + Registry iteration order is insertion order, which is an implementation + detail of whichever module last registered a code -- so ordering emitted + frames by it would let an unrelated import rewrite every capture. This + orders by the numeric code where there is one, and falls back to the string + form for the ``(str, int)`` keys of ``PCAPNG.__option__``. + + Args: + code: Registry key: an enumeration member, an :obj:`int`, or a + ``(str, int)`` tuple. + + Returns: + A tuple safe to compare against this function's output for any other + key, with numeric-keyed codes sorted ahead of tuple-keyed ones. + + """ + if isinstance(code, tuple): + name, number = code + return (1, str(name), int(number)) + try: + return (0, '', int(code)) + except (TypeError, ValueError): # pragma: no cover + return (2, str(code), 0) + + +def handler(registry: 'Any', code: 'Any') -> 'Any': + """The handler a registry holds for ``code``, without recording a miss. + + Enumerating a registry never needs this -- iterating a + :class:`collections.defaultdict` touches nothing, which is why :func:`cases` + only ever iterates. Asking what a *particular* code resolves to does need + it, because ``registry[code]`` on a miss inserts the key and permanently + grows a registry shared by every instance of the class in the process. The + inserted value is whatever the default factory would have produced anyway, + so it buys nothing and costs a spurious "already registered" warning from + the next genuine ``register`` call for that code. + + Args: + registry: A dispatch registry, or a mapping standing in for one. + code: Registry key to resolve. + + Returns: + The handler registered for ``code``, or the registry's fallback. + + """ + return ProtocolBase._lookup_registry(registry, code) # pylint: disable=protected-access + + +def code_name(code: 'Any') -> 'str': + """Render a registry key as a name usable in a test id. + + Args: + code: Registry key. + + Returns: + ``code.name`` for an enumeration member, ``'_'`` for a + ``(str, int)`` key, and the plain string form for anything else. + + """ + if isinstance(code, tuple): + name, number = code + return f'{name}_{number}' + return getattr(code, 'name', None) or str(code) + + +def _named_registry(owner: 'type', enum: 'Any', prefix: 'str', + attribute: 'str') -> 'Any': + """A protocol's dispatch registry, or a stand-in derived from its handlers. + + ``IPv4`` and ``HIP`` are the two protocols that still dispatch their options + and parameters by attribute name rather than through a registry, so for them + there is nothing to enumerate. The equivalent source of truth is the + enumeration crossed with the presence of the handler, which is what this + builds when the registry is absent. + + The real registry is preferred whenever it exists, so that the migration + landing is a no-op here rather than something to come back and finish. Both + shapes are a :class:`collections.defaultdict`, so :func:`cases` and + :func:`handler` cannot tell which one they were given -- and in particular + :meth:`ProtocolBase._lookup_registry + ` still has a + ``default_factory`` to fall back on. + + Args: + owner: Protocol class holding the handler methods. + enum: Enumeration of wire codes to consider. + prefix: Handler-name prefix, e.g. ``'_make_opt_'``. + attribute: Name the real registry would be held under, e.g. + ``'__option__'``. + + Returns: + The registry, real or derived. + + """ + existing = getattr(owner, attribute, None) + if existing is not None: + return existing + + found = collections.defaultdict(lambda: None) # type: Any + for code in enum: + name = code.name.lower() + if hasattr(owner, f'{prefix}{name}'): + found[code] = name + return found + + +############################################################################### +# TCP -- options, and Multipath TCP subtypes +############################################################################### + +#: Header fields shared by every constructed TCP segment. Only ``options`` +#: varies between cases, so a difference in the octets is a difference in the +#: option and nothing else. +TCP_BASE = { + 'srcport': 50000, 'dstport': 80, 'seq': 1, 'ack': 0, + 'ns': False, 'cwr': False, 'ece': False, 'urg': False, 'ack_flag': False, + 'psh': False, 'rst': False, 'syn': True, 'fin': False, + 'window': 8192, 'checksum': b'\x00\x00', 'urgent_pointer': 0, + 'payload': b'', +} + + +def _tcp_registry() -> 'Any': + from pcapkit.protocols.transport.tcp import TCP + return TCP.__option__ + + +def _tcp_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.tcp.option import Option as Enum_Option + return { + # Each of these four defaults to an empty variable-length field, which + # is the one shape that never runs the field's packing loop. + Enum_Option.SACK: {'sack': [(1, 2), (3, 4)]}, + Enum_Option.TCP_Alternate_Checksum_Data: {'data': b'\x01\x02'}, + Enum_Option.TCP_Authentication_Option: {'mac': b'\x01\x02\x03\x04'}, + Enum_Option.TCP_Fast_Open_Cookie: {'cookie': b'\x01\x02\x03\x04\x05\x06\x07\x08'}, + } + + +def _tcp_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.transport.tcp import TCP + return TCP(options=[(code, kwargs)], **TCP_BASE) + + +def _tcp_parse(octets: 'bytes') -> 'Any': + from pcapkit.protocols.transport.tcp import TCP + return TCP(octets, len(octets)) + + +def _tcp_extract(parsed: 'Any') -> 'Any': + return parsed.info.options + + +def _tcp_rebuild(options: 'Any') -> 'Any': + from pcapkit.protocols.transport.tcp import TCP + return TCP(options=options, **TCP_BASE) + + +def _mptcp_registry() -> 'Any': + from pcapkit.protocols.transport.tcp import TCP + return TCP.__mp_option__ + + +def _mptcp_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.tcp.mp_tcp_option import MPTCPOption as Enum_MPTCPOption + return { + Enum_MPTCPOption.MP_CAPABLE: {'skey': 0x0102030405060708}, + Enum_MPTCPOption.DSS: { + 'ack': 1, 'dsn': 2, 'ssn': 3, 'dl_len': 4, 'checksum': b'\x00\x00'}, + Enum_MPTCPOption.ADD_ADDR: {'addr_id': 1, 'addr': '192.0.2.1'}, + Enum_MPTCPOption.REMOVE_ADDR: {'addr_id': [1]}, + Enum_MPTCPOption.MP_PRIO: {'addr_id': 1}, + Enum_MPTCPOption.MP_FAIL: {'dsn': 7}, + Enum_MPTCPOption.MP_FASTCLOSE: {'key': 9}, + } + + +def _mptcp_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.const.tcp.option import Option as Enum_Option + from pcapkit.protocols.transport.tcp import TCP + + # The subtype travels as an argument of the enclosing Multipath TCP option + # rather than as a registry key of its own, so it is spliced in here rather + # than repeated in every entry of the table above. + args = dict(kwargs) + args['subtype'] = code + return TCP(options=[(Enum_Option.Multipath_TCP, args)], **TCP_BASE) + + +############################################################################### +# IPv4 -- options +############################################################################### + +#: Header fields shared by every constructed IPv4 packet. +IPV4_BASE = {'protocol': 6, 'src': '192.0.2.1', 'dst': '198.51.100.1', 'payload': b''} + + +def _ipv4_registry() -> 'Any': + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.protocols.internet.ipv4 import IPv4 + return _named_registry(IPv4, OptionNumber, '_make_opt_', '__option__') + + +def _ipv4_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.const.ipv4.protection_authority import ProtectionAuthority + return { + # A single authority whose value is 0 makes ``_make_opt_sec`` compute a + # zero-octet bitmap and then index into it; two keeps it non-empty. + OptionNumber.SEC: {'authorities': [ProtectionAuthority.GENSER, + ProtectionAuthority.NSA]}, + # ``counts=10``, the default, needs 43 option octets, which overflows + # the 4-bit ``ihl``. Nine is the largest an IPv4 header can carry. + OptionNumber.LSR: {'counts': 9}, + OptionNumber.RR: {'counts': 9}, + OptionNumber.SSR: {'counts': 9}, + # ``timestamp=None``, the default, is rejected outright. + OptionNumber.TS: {'counts': 1, 'timestamp': [1]}, + OptionNumber.E_SEC: {'info': b'\x00'}, + } + + +def _ipv4_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.internet.ipv4 import IPv4 + return IPv4(options=[(code, kwargs)], **IPV4_BASE) + + +def _ipv4_parse(octets: 'bytes') -> 'Any': + from pcapkit.protocols.internet.ipv4 import IPv4 + return IPv4(octets, len(octets)) + + +def _ipv4_extract(parsed: 'Any') -> 'Any': + from pcapkit.corekit.multidict import OrderedMultiDict + + # EOOL and NOP are dropped by ``_make_ipv4_options`` as padding, so the + # header they produce has no options at all and ``info`` carries no + # ``options`` attribute. An empty collection is the honest reading of that, + # and it keeps the two padding codes from being reported as parse failures. + return getattr(parsed.info, 'options', OrderedMultiDict()) + + +def _ipv4_rebuild(options: 'Any') -> 'Any': + from pcapkit.protocols.internet.ipv4 import IPv4 + return IPv4(options=options, **IPV4_BASE) + + +############################################################################### +# IPv6 extension headers -- HOPOPT, IPv6-Opts, IPv6-Route +############################################################################### + + +def _hopopt_registry() -> 'Any': + from pcapkit.protocols.internet.hopopt import HOPOPT + return HOPOPT.__option__ + + +def _ipv6_opts_registry() -> 'Any': + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + return IPv6_Opts.__option__ + + +def _ipv6_option_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.ipv6.option import Option as Enum_Option + + # HOPOPT and IPv6-Opts key on the same enumeration and behave identically + # on every code, so one table serves both. + return { + # ``nonce=0`` gives a zero-octet nonce. Only widths 3, 5, 6 and 7 pack + # at all, because a callable-length NumberField never clears the flag + # that says "hand struct a bytes"; 0xFFFFFF is three octets. + Enum_Option.ILNP_Nonce: {'nonce': 0xFFFFFF}, + Enum_Option.Line_Identification_Option: {'id': b'line-1'}, + } + + +def _hopopt_base() -> 'dict[str, Any]': + from pcapkit.const.reg.transtype import TransType + return {'next': TransType.UDP, 'payload': b''} + + +def _hopopt_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.internet.hopopt import HOPOPT + return HOPOPT(options=[(code, kwargs)], **_hopopt_base()) + + +def _hopopt_parse(octets: 'bytes') -> 'Any': + from pcapkit.protocols.internet.hopopt import HOPOPT + return HOPOPT(octets, len(octets), extension=True) + + +def _ipv6_extract(parsed: 'Any') -> 'Any': + from pcapkit.corekit.multidict import OrderedMultiDict + return getattr(parsed.info, 'options', OrderedMultiDict()) + + +def _hopopt_rebuild(options: 'Any') -> 'Any': + from pcapkit.protocols.internet.hopopt import HOPOPT + return HOPOPT(options=options, **_hopopt_base()) + + +def _ipv6_opts_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + return IPv6_Opts(options=[(code, kwargs)], **_hopopt_base()) + + +def _ipv6_opts_parse(octets: 'bytes') -> 'Any': + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + return IPv6_Opts(octets, len(octets), extension=True) + + +def _ipv6_opts_rebuild(options: 'Any') -> 'Any': + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + return IPv6_Opts(options=options, **_hopopt_base()) + + +def _ipv6_route_registry() -> 'Any': + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + return IPv6_Route.__routing__ + + +def _ipv6_route_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.ipv6.routing import Routing as Enum_Routing + return { + Enum_Routing.Source_Route: {'ip': ['2001:db8::1', '2001:db8::2']}, + Enum_Routing.Type_2_Routing_Header: {'ip': '2001:db8::2'}, + Enum_Routing.RPL_Source_Route_Header: {'ip': ['2001:db8::1', '2001:db8::2']}, + } + + +def _ipv6_route_base() -> 'dict[str, Any]': + from pcapkit.const.reg.transtype import TransType + return {'next': TransType.UDP, 'seg_left': 0, 'payload': b''} + + +def _ipv6_route_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + return IPv6_Route(type=code, data=dict(kwargs), **_ipv6_route_base()) + + +def _ipv6_route_parse(octets: 'bytes') -> 'Any': + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + return IPv6_Route(octets, len(octets), extension=True) + + +def _ipv6_route_extract(parsed: 'Any') -> 'Any': + # A routing header carries one routing type, not a collection of them, so + # the "collection" handed on is the message itself. + return parsed.info + + +def _ipv6_route_rebuild(info: 'Any') -> 'Any': + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + return IPv6_Route(type=info.type, data=info, next=info.next, + seg_left=info.seg_left, payload=b'') + + +############################################################################### +# SCTP -- chunks, parameters and error causes +############################################################################### + +#: Header fields shared by every constructed SCTP packet. ``chksum`` is pinned +#: rather than left to the CRC32c path, so a mismatch points at the chunk under +#: test and not at a recomputed checksum. +SCTP_BASE = { + 'srcport': 50000, 'dstport': 80, 'vtag': 0x11223344, + 'chksum': b'\x00\x00\x00\x00', +} + + +def _sctp_chunk_registry() -> 'Any': + from pcapkit.protocols.transport.sctp import SCTP + return SCTP.__chunk__ + + +def _sctp_chunk_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.sctp.cause_code import CauseCode as Enum_CauseCode + from pcapkit.const.sctp.chunk import Chunk as Enum_Chunk + from pcapkit.const.sctp.parameter import Parameter as Enum_Parameter + + # The three-octet payloads are deliberate: they make each item's length + # 3 (mod 4), so every case runs the PaddingField path instead of landing + # already aligned. + return { + # ``data=b''`` is rejected outright -- RFC 9260 s3.3.1 wants at least + # one octet of user data -- so this override is required, not cosmetic. + Enum_Chunk.Payload_Data: {'data': b'\x01\x02\x03'}, + Enum_Chunk.Initiation: { + 'parameters': [(Enum_Parameter.State_Cookie, {'cookie': b'\x01\x02\x03'})]}, + Enum_Chunk.Initiation_Acknowledgement: { + 'parameters': [(Enum_Parameter.State_Cookie, {'cookie': b'\x01\x02\x03'})]}, + Enum_Chunk.Selective_Acknowledgement: {'gap_blocks': [(1, 2)], 'dup_tsn': [3]}, + # RFC 9260 s3.3.5 and s3.3.6 mandate exactly one Heartbeat Info + # parameter, so the default -- none at all -- is structurally invalid. + Enum_Chunk.Heartbeat_Request: { + 'parameters': [(Enum_Parameter.Heartbeat_Info, {'info': b'\x01\x02\x03'})]}, + Enum_Chunk.Heartbeat_Acknowledgement: { + 'parameters': [(Enum_Parameter.Heartbeat_Info, {'info': b'\x01\x02\x03'})]}, + Enum_Chunk.Abort: { + 'error': [(Enum_CauseCode.User_Initiated_Abort, {'info': b'\x01\x02\x03'})]}, + # ``_make_chunk_error`` documents "one or more error causes"; the + # default gives zero. + Enum_Chunk.Operation_Error: { + 'error': [(Enum_CauseCode.Protocol_Violation, {'info': b'\x01\x02\x03'})]}, + Enum_Chunk.State_Cookie: {'cookie': b'\x01\x02\x03'}, + } + + +def _sctp_chunk_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.transport.sctp import SCTP + return SCTP(chunks=[(code, kwargs)], **SCTP_BASE) + + +def _sctp_parse(octets: 'bytes') -> 'Any': + from pcapkit.protocols.transport.sctp import SCTP + return SCTP(octets, len(octets)) + + +def _sctp_extract(parsed: 'Any') -> 'Any': + return parsed.info.chunks + + +def _sctp_rebuild(chunks: 'Any') -> 'Any': + from pcapkit.protocols.transport.sctp import SCTP + return SCTP(chunks=chunks, **SCTP_BASE) + + +def _sctp_param_registry() -> 'Any': + from pcapkit.protocols.transport.sctp import SCTP + return SCTP.__parameter__ + + +def _sctp_param_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.sctp.parameter import Parameter as Enum_Parameter + return { + Enum_Parameter.Heartbeat_Info: {'info': b'\x01\x02\x03'}, + Enum_Parameter.IPv4_Address: {'address': '192.0.2.1'}, + Enum_Parameter.IPv6_Address: {'address': '2001:db8::1'}, + Enum_Parameter.State_Cookie: {'cookie': b'\x01\x02\x03'}, + # A complete unrecognised TLV -- type 0xffff, length 5, one value + # octet -- rather than the empty default, which is not a TLV at all. + Enum_Parameter.Unrecognized_Parameter: {'value': b'\xff\xff\x00\x05\x01'}, + Enum_Parameter.Host_Name_Address: {'name': b'localhost\x00'}, + Enum_Parameter.Supported_Address_Types: { + 'types': [Enum_Parameter.IPv4_Address, Enum_Parameter.IPv6_Address]}, + } + + +def _sctp_param_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.const.sctp.chunk import Chunk as Enum_Chunk + from pcapkit.protocols.transport.sctp import SCTP + + # A parameter is not a top-level item: it travels inside a chunk, and + # HEARTBEAT is the smallest chunk that carries an arbitrary one. + return SCTP(chunks=[(Enum_Chunk.Heartbeat_Request, + {'parameters': [(code, kwargs)]})], **SCTP_BASE) + + +def _sctp_cause_registry() -> 'Any': + from pcapkit.protocols.transport.sctp import SCTP + return SCTP.__cause__ + + +def _sctp_cause_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.sctp.cause_code import CauseCode as Enum_CauseCode + from pcapkit.const.sctp.parameter import Parameter as Enum_Parameter + return { + Enum_CauseCode.Missing_Mandatory_Parameter: { + 'types': [Enum_Parameter.State_Cookie]}, + # The four causes below carry an embedded TLV, and each one's default + # is a zero-length value, i.e. not a TLV. + Enum_CauseCode.Unresolvable_Address: { + 'value': b'\x00\x0b\x00\x0elocalhost\x00'}, + Enum_CauseCode.Unrecognized_Chunk_Type: {'value': b'\xff\x00\x00\x05\x01'}, + Enum_CauseCode.Unrecognized_Parameters: {'value': b'\xff\xff\x00\x05\x01'}, + Enum_CauseCode.Restart_of_an_Association_with_New_Addresses: { + 'value': b'\x00\x05\x00\x08\xc0\x00\x02\x01'}, + Enum_CauseCode.User_Initiated_Abort: {'info': b'\x01\x02\x03'}, + Enum_CauseCode.Protocol_Violation: {'info': b'\x01\x02\x03'}, + } + + +def _sctp_cause_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.const.sctp.chunk import Chunk as Enum_Chunk + from pcapkit.protocols.transport.sctp import SCTP + + # An error cause travels inside an ERROR (or ABORT) chunk, under the + # keyword ``error`` rather than ``causes``. + return SCTP(chunks=[(Enum_Chunk.Operation_Error, + {'error': [(code, kwargs)]})], **SCTP_BASE) + + +############################################################################### +# Mobility Header -- messages, options and extensions +############################################################################### + +#: Header fields shared by every constructed Mobility Header. ``next`` is +#: IPv6-NoNxt, so nothing downstream has to be constructed as well. +MH_BASE = {'next': 59, 'chksum': b'\x00\x00', 'payload': b''} + + +def _mh_message_registry() -> 'Any': + from pcapkit.protocols.internet.mh import MH + return MH.__message__ + + +def _mh_message_overrides() -> 'dict[Any, dict[str, Any]]': + return {} + + +def _mh_message_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.internet.mh import MH + + # ``data`` has to be a mapping for ``make`` to dispatch on the message + # type; a Data object takes the other branch, which is what rebuild uses. + return MH(type=code, data=dict(kwargs), **MH_BASE) + + +def _mh_parse(octets: 'bytes') -> 'Any': + from pcapkit.protocols.internet.mh import MH + return MH(octets, len(octets), extension=True) + + +def _mh_message_extract(parsed: 'Any') -> 'Any': + return parsed.info + + +def _mh_message_rebuild(message: 'Any') -> 'Any': + from pcapkit.protocols.internet.mh import MH + return MH(type=message.type, data=message, **MH_BASE) + + +def _mh_option_registry() -> 'Any': + from pcapkit.protocols.internet.mh import MH + return MH.__option__ + + +def _mh_option_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.mh.option import Option as Enum_Option + import ipaddress + return { + # ``{}`` makes ``_make_opt_pad`` warn and silently emit a Pad1, so the + # PadN case would otherwise be a duplicate of the Pad1 one. + Enum_Option.PadN: {'length': 4}, + # The authenticator has to be a multiple of 8 octets. + Enum_Option.Authorization_Data: {'data': b'\xa5' * 8}, + Enum_Option.Mobility_Header_Link_Layer_Address_option: { + 'address': b'\x00\x11\x22\x33\x44\x55'}, + # The default identifier is the *string* ``'::'``, whose ``len()`` is 2 + # rather than the 16 octets the field packs -- so the declared length + # is 3 going out and 17 coming back. An address object avoids it. + Enum_Option.MN_ID_OPTION_TYPE: {'identifier': ipaddress.IPv6Address('::1')}, + # ``(len + 6) % 4`` has to be 0. + Enum_Option.AUTH_OPTION_TYPE: {'data': b'\xa5\xa5'}, + # Without this ``_make_opt_mesg_id`` reads the clock, and the capture + # would differ on every regeneration. + Enum_Option.MESG_ID_OPTION_TYPE: {'interval': FIXED_TIME}, + Enum_Option.Signature: {'signature': b'\xaa' * 4}, + Enum_Option.Permanent_Home_Keygen_Token: {'token': b'\xbb' * 8}, + Enum_Option.Experimental_Mobility_Option: {'data': b'\xcc' * 4}, + Enum_Option.Binding_Authorization_Data_for_FMIPv6: {'data': b'\xb5'}, + } + + +def _mh_carrier() -> 'Any': + from pcapkit.const.mh.packet import Packet as Enum_Packet + + # Binding Refresh Request is the smallest message that carries an + # arbitrary option list: two reserved octets and then the options. + return Enum_Packet.Binding_Refresh_Request + + +def _mh_option_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.internet.mh import MH + return MH(type=_mh_carrier(), data={'options': [(code, dict(kwargs))]}, **MH_BASE) + + +def _mh_option_extract(parsed: 'Any') -> 'Any': + from pcapkit.corekit.multidict import OrderedMultiDict + return getattr(parsed.info, 'options', OrderedMultiDict()) + + +def _mh_option_rebuild(options: 'Any') -> 'Any': + from pcapkit.protocols.internet.mh import MH + return MH(type=_mh_carrier(), data={'options': options}, **MH_BASE) + + +def _mh_extension_registry() -> 'Any': + from pcapkit.protocols.internet.mh import MH + return MH.__extension__ + + +def _mh_extension_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.mh.cga_extension import CGAExtension as Enum_CGAExtension + return {Enum_CGAExtension.Multi_Prefix: {'prefixes': [0x20010db800000001]}} + + +def _mh_extension_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.const.mh.cga_type import CGAType as Enum_CGAType + from pcapkit.const.mh.option import Option as Enum_Option + from pcapkit.protocols.data.internet.mh import CGAParameter as Data_CGAParameter + from pcapkit.protocols.internet.mh import MH + + # A CGA extension has exactly one carrier: the ``extensions`` list of a CGA + # parameter, inside a CGA Parameters option. It has to be a *Data* model + # rather than a schema, because ``_make_opt_cga_param`` keeps only the + # length of the extensions it builds for a schema and throws the schemas + # themselves away. + # The data model annotates ``modifier`` as a CGAType and ``extensions`` as an + # already-built OrderedMultiDict, but the constructor this feeds takes the + # tuple form and resolves both -- so the arguments go through an untyped + # mapping rather than being spelled as keywords a checker would reject. + fields = { + 'modifier': Enum_CGAType.Tag_086F_CA5E_10B2_00C9_9C8C_E001_6427_7C08, + 'prefix': 0x20010db800000000, + 'collision_count': 0, + 'public_key': b'\x30\x02\xaa\xbb', + 'extensions': [(code, dict(kwargs))], + } # type: dict[str, Any] + parameter = Data_CGAParameter(**fields) + return MH(type=_mh_carrier(), + data={'options': [(Enum_Option.CGA_Parameters, + {'parameters': [parameter]})]}, + **MH_BASE) + + +############################################################################### +# HIP -- parameters +############################################################################### + +#: Header fields shared by every constructed HIP packet. +HIP_BASE = { + 'next': 6, 'packet': 1, 'version': 2, 'checksum': b'\x00\x00', + 'controls_anonymous': False, 'shit': 0, 'rhit': 0, 'payload': b'', +} + +#: Codes that are HIPv1-only, and the version their constructor demands. +HIP_VERSION = {129: 2, 128: 1} + +#: How many copies of the parameter under test go in one packet. +#: +#: Two, not one, and this is not padding for its own sake. ``HIP.make`` +#: computes the header's ``len`` field as ``total_length // 8 + 4``, which is +#: only lossless when the parameter octets are a multiple of eight -- and the +#: parameter padding rule pads the *contents* to eight, ignoring the four-octet +#: type-and-length header, so a single parameter is always ``4 (mod 8)`` and +#: always loses those four octets. ``_read_hip_param`` then checks the value +#: exactly and rejects the packet. Two copies sum to a multiple of eight, so +#: the arithmetic is exact and the parameter constructors become reachable: +#: measured, this is the difference between 4 and 29 of the 49 codes +#: round-tripping. The single-parameter case is not lost -- it is what +#: ``hip-parameter-single`` in the test's expected-failure table records. +HIP_COPIES = 2 + + +def _hip_registry() -> 'Any': + from pcapkit.const.hip.parameter import Parameter + from pcapkit.protocols.internet.hip import HIP + return _named_registry(HIP, Parameter, '_make_param_', '__param__') + + +def _hip_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.hip.ecdsa_curve import ECDSACurve + from pcapkit.const.hip.parameter import Parameter + return { + # ``lifetime=0`` reaches ``math.log2(0)``. + Parameter.PUZZLE: {'lifetime': 1}, + Parameter.SOLUTION: {'lifetime': 1}, + # ``hi_curve=None`` falls through to an explicit raise. + Parameter.HOST_ID: {'hi_curve': ECDSACurve.NIST_P_256}, + } + + +def _hip_base(code: 'Any') -> 'dict[str, Any]': + base = dict(HIP_BASE) + base['version'] = HIP_VERSION.get(int(code), 2) + return base + + +def _hip_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.internet.hip import HIP + + # ``extension=True`` is not decoration. ``HIP.alias`` reads ``self._info``, + # which ``__post_init__`` only assigns after ``read`` returns, and the sole + # path that returns before touching it is the ``if extension`` early exit. + # Without it every one of the 49 codes fails with ``AttributeError: 'HIP' + # object has no attribute '_info'``. + return HIP(parameters=[(code, kwargs)] * HIP_COPIES, + extension=True, **_hip_base(code)) + + +def _hip_parse(octets: 'bytes') -> 'Any': + from pcapkit.protocols.internet.hip import HIP + return HIP(octets, len(octets), extension=True) + + +def _hip_extract(parsed: 'Any') -> 'Any': + from pcapkit.corekit.multidict import OrderedMultiDict + return getattr(parsed.info, 'parameters', OrderedMultiDict()) + + +def _hip_rebuild(parameters: 'Any') -> 'Any': + from pcapkit.protocols.internet.hip import HIP + + # ``parameters`` is the parsed collection, so the version has to come off + # one of its members rather than from the code being tested. + first = next(iter(parameters), None) + base = dict(HIP_BASE) + if first is not None: + base['version'] = HIP_VERSION.get(int(first), 2) + return HIP(parameters=parameters, extension=True, **base) + + +############################################################################### +# HTTP/2 -- frames +############################################################################### + + +def _httpv2_registry() -> 'Any': + from pcapkit.protocols.application.httpv2 import HTTP + return HTTP.__frame__ + + +def _httpv2_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.http.frame import Frame as Enum_Frame + from pcapkit.const.http.setting import Setting as Enum_Setting + return { + # ``settings=None``, the default, matches none of the accepted forms. + Enum_Frame.SETTINGS: {'settings': [(Enum_Setting.HEADER_TABLE_SIZE, 4096)]}, + Enum_Frame.GOAWAY: {'debug_data': b'\xde\xad'}, + } + + +def _httpv2_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.application.httpv2 import HTTP + + # SETTINGS and PING both reject a non-zero stream identifier, so every + # frame uses stream 0 rather than each one picking its own. + return HTTP(type=code, sid=0, frame=dict(kwargs)) + + +def _httpv2_parse(octets: 'bytes') -> 'Any': + from pcapkit.protocols.application.httpv2 import HTTP + return HTTP(octets, len(octets)) + + +def _httpv2_extract(parsed: 'Any') -> 'Any': + return parsed.info + + +def _httpv2_rebuild(frame: 'Any') -> 'Any': + from pcapkit.protocols.application.httpv2 import HTTP + + # ``flags`` is deliberately not passed on: ``HTTP.make`` rebinds it from + # the ``_make_http_*`` return whenever ``frame`` is a mapping or a Data + # object, so passing it would be silently ignored. + return HTTP(type=frame.type, sid=frame.sid, frame=frame) + + +############################################################################### +# PCAP-NG -- blocks, block options, name records and decryption secrets +############################################################################### +# +# These four families are exercised by :func:`roundtrip` but deliberately write +# no capture, and the reason is not that a PCAP-NG capture would be +# uninteresting -- it is that the construction API cannot produce a +# *reproducible* one. ``_make_block_shb`` takes the section's byte order from +# :data:`sys.byteorder` with no way to override it, and the TLS and WireGuard +# key-log writers stamp :meth:`datetime.datetime.now` into the block body with +# no keyword to pin it. A fixture built this way would differ between a +# little-endian and a big-endian host, and the two key-log types would differ +# between one run and the next. +# +# The PCAP-NG option space is not left uncovered by that decision: +# :file:`examples/generators/pcapng.py` already builds ``profile.pcapng``, +# ``test.pcapng`` and ``many_interfaces.pcapng`` over it, hand-rolling every +# octet with :func:`struct.pack` for exactly this reason. What is added here is +# the half that generator cannot reach, since it never calls the construction +# API at all: whether ``_make_block_*`` and ``_make_option_*`` can produce what +# ``_read_block_*`` and ``_read_option_*`` consume. + +#: Fixed PCAP-NG timestamp, the same instant :file:`pcapng.py` uses. Not +#: optional: leaving ``timestamp`` unset reaches ``self._info`` before +#: ``__post_init__`` has assigned it, so the "now" default is unreachable as +#: well as non-deterministic. +PCAPNG_TIME = 1500000000 + +#: A minimal Ethernet frame for the packet-carrying blocks: destination and +#: source address, an unassigned EtherType, and two octets of body. +#: +#: It has to be non-empty. A ``PayloadField`` whose computed length is +#: legitimately zero reads the whole remainder of the block instead of nothing, +#: which swallows the option area -- so an empty packet would silently take the +#: whole ``epb_*`` and ``pack_*`` option space with it. Sixteen octets rather +#: than the bare 14-octet header, so the Ethernet dissector does not under-run. +#: 6 octets of destination, 6 of source, the EtherType, then the body. +PCAPNG_PACKET = bytes.fromhex('02000000000102000000000288b50000') + +#: A systemd journal export entry whose length is a multiple of four. +#: ``_make_block_systemd`` adds no padding, unlike every other block, so an +#: unaligned entry makes the block length invalid. +PCAPNG_JOURNAL = (b'__REALTIME_TIMESTAMP=1500000000000000\n_TRANSPORT=journal\n' + b'PRIORITY=6\nMESSAGE=probe.....\n\n') + + +def _pcapng_context() -> 'tuple[Any, Any]': + """A one-section PCAP-NG context with one Ethernet interface in it. + + Every block other than a Section Header Block is read and written relative + to a section, and the packet-carrying blocks resolve their snap length + through the section's interface list, so there has to be one. + + Returns: + The :class:`~pcapkit.foundation.engines.pcapng.Context` and the + interface-description block registered in it. + + """ + from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType + from pcapkit.const.reg.linktype import LinkType as Enum_LinkType + from pcapkit.foundation.engines.pcapng import Context + from pcapkit.protocols.misc.pcapng import PCAPNG + + section = PCAPNG(num=0, sct=1, ctx=None, + type=Enum_BlockType.Section_Header_Block, block={}) + # ``Protocol.info`` is annotated as the generic data model for the protocol, + # so a checker cannot see that a Section Header Block's is the one ``Context`` + # wants. The engine does exactly this, at + # pcapkit/foundation/engines/pcapng.py:148. + section_info = section.info # type: Any + context = Context(section_info) + interface = PCAPNG(num=1, sct=1, ctx=context, + type=Enum_BlockType.Interface_Description_Block, + block={'linktype': Enum_LinkType.ETHERNET, 'snaplen': 0x40000}) + interface_info = interface.info # type: Any + context.interfaces.append(interface_info) + return context, interface_info + + +def _pcapng_block(code: 'Any', block: 'dict[str, Any] | Any') -> 'Any': + """Construct one PCAP-NG block of ``code``. + + A fresh context per call, deliberately. ``PCAPNG.__post_init__`` packs and + then re-reads *on the same instance*, and the two halves share the + per-instance option counter, so state left by one case would change the + next one's result. + + Args: + code: Block type. + block: Block body, as a mapping of constructor arguments or as the + parsed data model. + + Returns: + The constructed block. + + """ + from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType + from pcapkit.protocols.misc.pcapng import PCAPNG + + # A Section Header Block is what *starts* a section, so it is the one block + # the engine reads with no context at all. + context = None if code == Enum_BlockType.Section_Header_Block else _pcapng_context()[0] + return PCAPNG(num=2, sct=1, ctx=context, type=code, block=block) + + +#: Block type of a Section Header Block, as it appears on the wire. The value +#: is byte-order independent by design -- that is what lets a reader work out a +#: section's endianness -- so it can be recognised before the byte order is +#: known, which is exactly what :func:`_pcapng_parse` needs. +PCAPNG_SHB_MAGIC = bytes.fromhex('0a0d0d0a') + + +def _pcapng_parse(octets: 'bytes') -> 'Any': + """Parse one PCAP-NG block back from its own octets. + + The block type is read out of the octets rather than passed in, so that + this keeps the one-argument shape every other family's ``parse`` has. It is + the first four octets of any block, and a Section Header Block -- the one + block read with no section context, because it is what *starts* a + section -- is recognisable by :data:`PCAPNG_SHB_MAGIC`. + + Args: + octets: The constructed block. + + Returns: + The parsed block. + + """ + from pcapkit.protocols.misc.pcapng import PCAPNG + + context = None if octets[:4] == PCAPNG_SHB_MAGIC else _pcapng_context()[0] + return PCAPNG(octets, len(octets), num=2, sct=1, ctx=context) + + +def _pcapng_block_registry() -> 'Any': + from pcapkit.protocols.misc.pcapng import PCAPNG + return PCAPNG.__block__ + + +def _pcapng_block_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType + from pcapkit.const.pcapng.record_type import RecordType as Enum_RecordType + from pcapkit.const.pcapng.secrets_type import SecretsType as Enum_SecretsType + from pcapkit.const.reg.linktype import LinkType as Enum_LinkType + return { + Enum_BlockType.Interface_Description_Block: { + 'linktype': Enum_LinkType.ETHERNET, 'snaplen': 0x40000}, + Enum_BlockType.Enhanced_Packet_Block: { + 'timestamp': PCAPNG_TIME, 'packet_data': PCAPNG_PACKET}, + Enum_BlockType.Simple_Packet_Block: {'packet_data': PCAPNG_PACKET}, + # Without the terminating record the records field consumes the whole + # block and the option area disappears. + Enum_BlockType.Name_Resolution_Block: { + 'records': [(Enum_RecordType.nrb_record_end, {})]}, + Enum_BlockType.Interface_Statistics_Block: {'timestamp': PCAPNG_TIME}, + Enum_BlockType.systemd_Journal_Export_Block: {'entries': PCAPNG_JOURNAL}, + # ZigBee rather than the default TLS key log, whose writer stamps the + # current time into the body and so can never round-trip. + Enum_BlockType.Decryption_Secrets_Block: { + 'secrets_type': Enum_SecretsType.ZigBee_NWK_Key, + 'secrets_data': {'nwk_key': bytes(range(16)), 'pan_id': 0x1234}}, + Enum_BlockType.Custom_Block_that_rewriters_can_copy_into_new_files: { + 'pen': 32473, 'data': b'\x01\x02\x03\x04'}, + Enum_BlockType.Custom_Block_that_rewriters_should_not_copy_into_new_files: { + 'pen': 32473, 'data': b'\x01\x02\x03\x04'}, + Enum_BlockType.Packet_Block: { + 'timestamp': PCAPNG_TIME, 'packet_data': PCAPNG_PACKET}, + } + + +def _pcapng_block_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + return _pcapng_block(code, dict(kwargs)) + + +def _pcapng_block_extract(parsed: 'Any') -> 'Any': + return parsed.info + + +def _pcapng_block_rebuild(info: 'Any') -> 'Any': + return _pcapng_block(info.type, info) + + +#: Which block hosts each block-option namespace, and that host's own +#: arguments. The namespace is the part of the ``_option_key`` tuple's first +#: element before the first underscore. +PCAPNG_OPTION_HOSTS = { + 'opt': 'Section_Header_Block', + 'if': 'Interface_Description_Block', + 'epb': 'Enhanced_Packet_Block', + 'ns': 'Name_Resolution_Block', + 'isb': 'Interface_Statistics_Block', + 'pack': 'Packet_Block', +} + + +def _pcapng_option_registry() -> 'Any': + from pcapkit.protocols.misc.pcapng import PCAPNG + return PCAPNG.__option__ + + +def _pcapng_option_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.pcapng.option_type import OptionType as Enum_OptionType + from pcapkit.protocols.misc.pcapng import _option_key + return { + _option_key(Enum_OptionType.opt_comment): {'comment': 'probe'}, + _option_key(Enum_OptionType.opt_custom_2988): {'pen': 32473, 'data': b'\x01\x02\x03\x04'}, + _option_key(Enum_OptionType.opt_custom_2989): {'pen': 32473, 'data': b'\x01\x02\x03\x04'}, + _option_key(Enum_OptionType.opt_custom_19372): {'pen': 32473, 'data': b'\x01\x02\x03\x04'}, + _option_key(Enum_OptionType.opt_custom_19373): {'pen': 32473, 'data': b'\x01\x02\x03\x04'}, + _option_key(Enum_OptionType.if_name): {'name': 'eth0'}, + _option_key(Enum_OptionType.if_description): {'description': 'probe interface'}, + _option_key(Enum_OptionType.if_IPv4addr): {'interface': '10.0.0.1/255.255.255.0'}, + _option_key(Enum_OptionType.if_IPv6addr): { + 'interface': '2001:db8:85a3:8d3:1319:8a2e:370:7344/64'}, + _option_key(Enum_OptionType.if_MACaddr): {'interface': '02:00:00:00:00:01'}, + _option_key(Enum_OptionType.if_EUIaddr): {'interface': '02:00:00:FF:FE:00:00:01'}, + _option_key(Enum_OptionType.if_speed): {'speed': 1000000000}, + _option_key(Enum_OptionType.if_tsresol): {'resolution': 1000000}, + _option_key(Enum_OptionType.if_tzone): {'tzone': 0}, + _option_key(Enum_OptionType.if_filter): {'expression': b'udp port 67'}, + # ``if_os`` and ``if_hardware`` default to the live platform strings, so + # leaving them unset would make the case machine-dependent. + _option_key(Enum_OptionType.if_os): {'os': 'probe OS'}, + _option_key(Enum_OptionType.if_hardware): {'hardware': 'probe adapter'}, + _option_key(Enum_OptionType.if_fcslen): {'fcs_length': 4}, + _option_key(Enum_OptionType.if_tsoffset): {'offset': 0}, + _option_key(Enum_OptionType.if_txspeed): {'speed': 1000000000}, + _option_key(Enum_OptionType.if_rxspeed): {'speed': 1000000000}, + _option_key(Enum_OptionType.epb_hash): {'hash': b'\x01\x02\x03\x04'}, + _option_key(Enum_OptionType.epb_verdict): {'value': b'\x01'}, + _option_key(Enum_OptionType.ns_dnsname): {'name': 'resolver.example'}, + _option_key(Enum_OptionType.ns_dnsIP4addr): {'ip': '10.0.0.53'}, + # The default is the IPv4 literal ``'8.8.8.8'``, which the v6 field + # rejects outright. + _option_key(Enum_OptionType.ns_dnsIP6addr): {'ip': '2001:db8::35'}, + _option_key(Enum_OptionType.isb_starttime): {'timestamp': PCAPNG_TIME}, + _option_key(Enum_OptionType.isb_endtime): {'timestamp': PCAPNG_TIME}, + _option_key(Enum_OptionType.isb_ifrecv): {'packets': 4}, + _option_key(Enum_OptionType.isb_ifdrop): {'packets': 1}, + _option_key(Enum_OptionType.isb_filteraccept): {'packets': 4}, + _option_key(Enum_OptionType.isb_osdrop): {'packets': 0}, + _option_key(Enum_OptionType.isb_usrdeliv): {'packets': 3}, + _option_key(Enum_OptionType.pack_hash): {'hash': b'\x01\x02\x03\x04'}, + } + + +def _pcapng_option_enum(key: 'Any') -> 'Any': + """The enumeration member behind an ``_option_key`` tuple. + + ``PCAPNG.__option__`` is keyed by ``(name, value)`` tuples because the same + numeric option code means different things in different block types, but + the construction API takes the enumeration member. + + Args: + key: The ``(str, int)`` registry key. + + Returns: + The matching :class:`~pcapkit.const.pcapng.option_type.OptionType`. + + Raises: + KeyError: If no member matches, which would mean the registry and the + enumeration have diverged. + + """ + from pcapkit.const.pcapng.option_type import OptionType as Enum_OptionType + from pcapkit.protocols.misc.pcapng import _option_key + + # The enumeration is an :mod:`aenum` one, whose metaclass does not advertise + # ``__iter__`` to a type checker even though it iterates perfectly well. + members = Enum_OptionType # type: Any + for member in members: + if _option_key(member) == key: + return member + raise KeyError(f'no OptionType matches registry key {key!r}') + + +def _pcapng_option_host(key: 'Any') -> 'tuple[Any, dict[str, Any]]': + """The block type that carries option ``key``, and its own arguments.""" + from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType + + namespace = str(key[0]).split('_', 1)[0] + name = PCAPNG_OPTION_HOSTS[namespace] + code = getattr(Enum_BlockType, name) + return code, dict(_pcapng_block_overrides().get(code, {})) + + +def _pcapng_option_build(key: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + code, block = _pcapng_option_host(key) + block['options'] = [(_pcapng_option_enum(key), dict(kwargs))] + return _pcapng_block(code, block) + + +def _pcapng_option_rebuild(info: 'Any') -> 'Any': + """Rebuild an option's host block, taking only the options from the parse. + + The host's own fields come from :func:`_pcapng_block_overrides` rather than + from the parsed block, and that is the point. Reconstructing the whole block + from ``info`` would drop the captured payload of a packet-carrying block -- + ``_make_block_epb`` and its siblings never restore ``packet_data``, and the + data model has nowhere to keep it -- so every ``epb_*`` and ``pack_*`` + option would be reported as a mismatch on account of a defect that has + nothing to do with the option. Re-supplying the host fields isolates the + option, which is what this family is measuring. + + Args: + info: The parsed block. + + Returns: + The reconstructed block. + + """ + block = dict(_pcapng_block_overrides().get(info.type, {})) + block['options'] = info.options + return _pcapng_block(info.type, block) + + +def _pcapng_record_registry() -> 'Any': + from pcapkit.protocols.misc.pcapng import PCAPNG + return PCAPNG.__record__ + + +def _pcapng_record_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.pcapng.record_type import RecordType as Enum_RecordType + return { + Enum_RecordType.nrb_record_ipv4: {'ip': '10.0.0.1', 'names': ['gateway.example']}, + # The default is the IPv4 literal ``'127.0.0.1'``, which the v6 field + # rejects outright. + Enum_RecordType.nrb_record_ipv6: {'ip': '2001:db8::1', 'names': ['v6.example']}, + } + + +def _pcapng_record_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType + from pcapkit.const.pcapng.record_type import RecordType as Enum_RecordType + + records = [(code, dict(kwargs))] + if code != Enum_RecordType.nrb_record_end: + # A record list has to be terminated, or the records field eats the + # rest of the block. + records.append((Enum_RecordType.nrb_record_end, {})) + return _pcapng_block(Enum_BlockType.Name_Resolution_Block, {'records': records}) + + +def _pcapng_record_extract(parsed: 'Any') -> 'Any': + return parsed.info.records + + +def _pcapng_record_rebuild(records: 'Any') -> 'Any': + from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType + return _pcapng_block(Enum_BlockType.Name_Resolution_Block, {'records': records}) + + +def _pcapng_secrets_registry() -> 'Any': + from pcapkit.protocols.misc.pcapng import PCAPNG + return PCAPNG.__secrets__ + + +def _pcapng_secrets_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.pcapng.secrets_type import SecretsType as Enum_SecretsType + from pcapkit.corekit.multidict import OrderedMultiDict + from pcapkit.protocols.misc.pcapng import TLSKeyLabel, WireGuardKeyLabel + return { + Enum_SecretsType.TLS_Key_Log: { + 'entries': {TLSKeyLabel.CLIENT_RANDOM: OrderedMultiDict( + [(bytes(range(0x20, 0x40)), bytes(range(0x40, 0x70)))])}}, + Enum_SecretsType.WireGuard_Key_Log: { + 'entries': OrderedMultiDict( + [(WireGuardKeyLabel.LOCAL_STATIC_PRIVATE_KEY, bytes(range(32)))])}, + Enum_SecretsType.ZigBee_NWK_Key: {'nwk_key': bytes(range(16)), 'pan_id': 0x1234}, + Enum_SecretsType.ZigBee_APS_Key: { + 'aps_key': bytes(range(16)), 'pan_id': 0x1234, 'short_address': 0xABCD}, + } + + +def _pcapng_secrets_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType + return _pcapng_block(Enum_BlockType.Decryption_Secrets_Block, + {'secrets_type': code, 'secrets_data': dict(kwargs)}) + + +############################################################################### +# The families, and the cycle +############################################################################### + +#: Every family this module exercises, in the order their frames are emitted. +FAMILIES = ( + Family('tcp-option', _tcp_registry, _tcp_overrides, + _tcp_build, _tcp_parse, _tcp_extract, _tcp_rebuild, + capture='options-tcp.pcap', envelope='ipv4', proto=6), + Family('tcp-mptcp', _mptcp_registry, _mptcp_overrides, + _mptcp_build, _tcp_parse, _tcp_extract, _tcp_rebuild, + capture='options-tcp.pcap', envelope='ipv4', proto=6), + Family('ipv4-option', _ipv4_registry, _ipv4_overrides, + _ipv4_build, _ipv4_parse, _ipv4_extract, _ipv4_rebuild, + capture='options-ipv4.pcap', envelope='ethernet-ipv4'), + Family('hopopt-option', _hopopt_registry, _ipv6_option_overrides, + _hopopt_build, _hopopt_parse, _ipv6_extract, _hopopt_rebuild, + capture='options-ipv6.pcap', envelope='ipv6', proto=0), + Family('ipv6-opts-option', _ipv6_opts_registry, _ipv6_option_overrides, + _ipv6_opts_build, _ipv6_opts_parse, _ipv6_extract, _ipv6_opts_rebuild, + capture='options-ipv6.pcap', envelope='ipv6', proto=60), + Family('ipv6-route-type', _ipv6_route_registry, _ipv6_route_overrides, + _ipv6_route_build, _ipv6_route_parse, _ipv6_route_extract, + _ipv6_route_rebuild, + capture='options-ipv6.pcap', envelope='ipv6', proto=43), + Family('sctp-chunk', _sctp_chunk_registry, _sctp_chunk_overrides, + _sctp_chunk_build, _sctp_parse, _sctp_extract, _sctp_rebuild, + capture='options-transport.pcap', envelope='ipv4', proto=132), + Family('sctp-parameter', _sctp_param_registry, _sctp_param_overrides, + _sctp_param_build, _sctp_parse, _sctp_extract, _sctp_rebuild, + capture='options-transport.pcap', envelope='ipv4', proto=132), + Family('sctp-cause', _sctp_cause_registry, _sctp_cause_overrides, + _sctp_cause_build, _sctp_parse, _sctp_extract, _sctp_rebuild, + capture='options-transport.pcap', envelope='ipv4', proto=132), + Family('httpv2-frame', _httpv2_registry, _httpv2_overrides, + _httpv2_build, _httpv2_parse, _httpv2_extract, _httpv2_rebuild, + capture='options-transport.pcap', envelope='tcp'), + Family('mh-message', _mh_message_registry, _mh_message_overrides, + _mh_message_build, _mh_parse, _mh_message_extract, _mh_message_rebuild, + capture='options-internet.pcap', envelope='ipv6', proto=135), + Family('mh-option', _mh_option_registry, _mh_option_overrides, + _mh_option_build, _mh_parse, _mh_option_extract, _mh_option_rebuild, + capture='options-internet.pcap', envelope='ipv6', proto=135), + Family('mh-extension', _mh_extension_registry, _mh_extension_overrides, + _mh_extension_build, _mh_parse, _mh_option_extract, _mh_option_rebuild, + capture='options-internet.pcap', envelope='ipv6', proto=135), + Family('hip-parameter', _hip_registry, _hip_overrides, + _hip_build, _hip_parse, _hip_extract, _hip_rebuild, + capture='options-internet.pcap', envelope='ipv6', proto=139), + # The four PCAP-NG families write no capture -- see the section comment + # above ``PCAPNG_TIME`` for why the construction API cannot make a + # reproducible one, and what already covers that space instead. + Family('pcapng-block', _pcapng_block_registry, _pcapng_block_overrides, + _pcapng_block_build, _pcapng_parse, _pcapng_block_extract, + _pcapng_block_rebuild, capture=None, envelope=None), + # Both of these reconstruct from the parsed *block* rather than from the + # option or secrets collection alone, because neither collection records + # which block type carried it -- but the option family re-supplies the + # host's own fields on the way, for the reason ``_pcapng_option_rebuild`` + # gives. + Family('pcapng-option', _pcapng_option_registry, _pcapng_option_overrides, + _pcapng_option_build, _pcapng_parse, _pcapng_block_extract, + _pcapng_option_rebuild, capture=None, envelope=None), + Family('pcapng-record', _pcapng_record_registry, _pcapng_record_overrides, + _pcapng_record_build, _pcapng_parse, _pcapng_record_extract, + _pcapng_record_rebuild, capture=None, envelope=None), + Family('pcapng-secrets', _pcapng_secrets_registry, _pcapng_secrets_overrides, + _pcapng_secrets_build, _pcapng_parse, _pcapng_block_extract, + _pcapng_block_rebuild, capture=None, envelope=None), +) + +#: The families keyed by label, for a caller that wants just one. +FAMILY_MAP = {family.label: family for family in FAMILIES} + +#: Codes deliberately left out of the case list, with the reason. +#: +#: The padding options are dropped by every ``_make_*_options`` in the tree -- +#: it discards them and inserts its own padding to reach the alignment the +#: protocol wants -- so a case for one would assert only that construction +#: ignores its own input, which it does. They are kept for TCP, where the +#: dropping is symmetric and the case is a genuine round trip of an +#: option-less header, and skipped where a second padding code would merely +#: duplicate the first. +SKIP = { + ('tcp-option', 'End_of_Option_List'): 'padding, dropped by _make_tcp_options', + ('tcp-option', 'No_Operation'): 'padding, dropped by _make_tcp_options', + # Multipath TCP is the envelope for the whole tcp-mptcp family, so a case + # here would silently duplicate whichever subtype it defaulted to. + ('tcp-option', 'Multipath_TCP'): 'envelope for the tcp-mptcp family', + # Pad1 and PadN are both discarded by _make_hopopt_options, so the two + # cases emit byte-identical option-less headers. + ('hopopt-option', 'PadN'): 'padding, indistinguishable from Pad1 here', + ('ipv6-opts-option', 'PadN'): 'padding, indistinguishable from Pad1 here', +} + + +def cases(families: 'Optional[tuple[Family, ...]]' = None) -> 'list[Case]': + """Every code every family registry claims to support, as a case. + + The list comes from the registries rather than from a table, which is what + lets :file:`tests/protocols/test_option_roundtrip_unit.py` notice that a + newly registered code has no case yet. + + Args: + families: Families to enumerate; :data:`FAMILIES` if not given. + + Returns: + The cases, ordered by family and then by :func:`sort_key`, so two runs + produce the same list and therefore the same captures. + + """ + out = [] # type: list[Case] + for family in (FAMILIES if families is None else families): + registry = family.registry() + overrides = family.overrides() + for code in sorted(registry, key=sort_key): + name = code_name(code) + if (family.label, name) in SKIP: + continue + # A code missing from ``overrides`` means "every default", which is + # what most of them want. + out.append(Case(family.label, code, name, dict(overrides.get(code, {})))) + return out + + +def roundtrip(case: 'Case', deadline: 'int' = DEADLINE) -> 'Outcome': + """Construct ``case``, parse it back, construct it again, compare. + + The third step is the one that earns its keep. A ``_make_*`` that takes + only keyword arguments, and cannot consume the data model its own + ``_read_*`` produced, still passes a construct-then-parse test and fails + here. + + Args: + case: The case to exercise. + deadline: Whole seconds to allow, or ``0`` for no deadline. See the + module docstring for why there is one at all. + + Returns: + An :class:`Outcome` naming the step that failed, or ``'OK'``. + + """ + family = FAMILY_MAP[case.family] + # Bound inside the ``with`` below and read after it, so it is declared here. + # Non-optional, unlike the ``octets`` field of the ``Outcome`` it ends up in: + # every path that reaches the comparison at the bottom has been through a + # successful construction, and saying so keeps the ``.hex()`` there honest. + octets = b'' + again = b'' + + with _deadline(deadline), warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + + def seen() -> 'tuple[str, ...]': + """Warning messages so far, deduplicated but in order. + + :exc:`ResourceWarning` is dropped. It is raised by the garbage + collector rather than by the code under test, so whichever case + happens to be running when a file object from somewhere else is + finalised would otherwise be blamed for it -- which is exactly what + happened: a ``.pcapng`` handle left open by a sibling generator was + reported against ``hopopt-option/Router_Alert``. + """ + return tuple(dict.fromkeys( + str(entry.message) for entry in caught + if not issubclass(entry.category, ResourceWarning))) + + try: + octets = bytes(family.build(case.code, case.kwargs)) + except TimeoutError as exc: + return Outcome(case, 'TIMEOUT', str(exc), None, seen()) + except Exception as exc: # pylint: disable=broad-except + # Broad on purpose: the point is to find out *what* a constructor + # does when it is wrong, and narrowing this to ProtocolError would + # let an AttributeError or a KeyError -- which is what most of the + # current defects raise -- abort the sweep rather than be recorded + # as this case's result. + return Outcome(case, 'CONSTRUCT', f'{type(exc).__name__}: {exc}', None, seen()) + + try: + parsed = family.extract(family.parse(octets)) + except TimeoutError as exc: + return Outcome(case, 'TIMEOUT', str(exc), octets, seen()) + except Exception as exc: # pylint: disable=broad-except + return Outcome(case, 'PARSE', f'{type(exc).__name__}: {exc}', octets, seen()) + + try: + again = bytes(family.rebuild(parsed)) + except TimeoutError as exc: + return Outcome(case, 'TIMEOUT', str(exc), octets, seen()) + except Exception as exc: # pylint: disable=broad-except + return Outcome(case, 'RECONSTRUCT', f'{type(exc).__name__}: {exc}', + octets, seen()) + + warned = seen() + + if octets != again: + return Outcome(case, 'MISMATCH', f'{octets.hex()} != {again.hex()}', + octets, warned) + return Outcome(case, 'OK', '', octets, warned) + + +class _deadline: # pylint: disable=invalid-name + """Raise :exc:`TimeoutError` in the body after ``seconds`` seconds. + + Written as a class rather than a :func:`contextlib.contextmanager` so that + it can be entered alongside :func:`warnings.catch_warnings` in one ``with`` + statement without the generator machinery swallowing the alarm. + + A no-op where :data:`signal.SIGALRM` is missing (Windows) or where the + caller asked for no deadline. That is a deliberate degradation rather than + a hard failure: without an interval timer the only alternative would be to + refuse to generate the captures at all. + + """ + + def __init__(self, seconds: 'int') -> 'None': + #: Whole seconds to allow. :func:`signal.alarm` counts in whole + #: seconds, so this cannot usefully be fractional. + self.seconds = seconds + #: Whether the alarm was actually armed, and so needs disarming. + self.armed = False + #: Handler displaced by :meth:`__enter__`, restored by :meth:`__exit__`. + self.previous = None # type: Any + + def __enter__(self) -> '_deadline': + if self.seconds <= 0 or not hasattr(signal, 'SIGALRM'): + return self + + def expire(signum: 'int', frame: 'Any') -> 'None': + raise TimeoutError(f'did not finish within {self.seconds}s') + + self.previous = signal.signal(signal.SIGALRM, expire) + signal.alarm(self.seconds) + self.armed = True + return self + + def __exit__(self, *exc_info: 'Any') -> 'None': + if self.armed: + # Cancel before restoring, so an alarm firing between the two + # cannot be delivered to whatever handler was installed before. + signal.alarm(0) + signal.signal(signal.SIGALRM, self.previous) + self.armed = False + + +def outcomes(families: 'Optional[tuple[Family, ...]]' = None, + deadline: 'int' = DEADLINE) -> 'Iterator[Outcome]': + """Run :func:`roundtrip` over :func:`cases`. + + Args: + families: Families to enumerate; :data:`FAMILIES` if not given. + deadline: Whole seconds to allow each case. + + Yields: + One :class:`Outcome` per case, in case order. + + """ + for case in cases(families): + yield roundtrip(case, deadline) + + +def _frame(family: 'Family', octets: 'bytes', index: 'int') -> 'Any': + """Wrap constructed octets in the envelope that makes them decodable. + + Args: + family: The family the octets came from, which decides the envelope. + octets: The octets :func:`roundtrip` constructed. + index: Position within the capture, used as the IPv4 identification so + that two frames carrying identical options still differ somewhere. + + Returns: + A :mod:`scapy` packet, with its timestamp already pinned. + + Raises: + ValueError: If ``family`` names an envelope this function does not know. + + """ + from scapy.all import IP, IPv6, Ether, Raw, TCP # pylint: disable=no-name-in-module + + link = Ether(src=SRC_MAC, dst=DST_MAC) + if family.envelope == 'ethernet-ipv4': + # The octets *are* the IPv4 header, options and all, so nothing may be + # put between them and the link layer -- and because the payload is a + # bare ``Raw``, scapy has nothing to infer the EtherType from and falls + # back to its default of 0x9000. That decodes as Loopback, not IPv4, so + # the type is stated here rather than left to be guessed. + frame = link / Raw(octets) + frame.type = 0x0800 + elif family.envelope == 'ipv4': + frame = link / IP(src=SRC_IP, dst=DST_IP, proto=family.proto, + id=index, ttl=64) / Raw(octets) + elif family.envelope == 'ipv6': + frame = link / IPv6(src=SRC_IP6, dst=DST_IP6, nh=family.proto, + hlim=64) / Raw(octets) + elif family.envelope == 'tcp': + frame = (link / IP(src=SRC_IP, dst=DST_IP, id=index, ttl=64) + / TCP(sport=50000, dport=80, flags='A', seq=1, ack=1) + / Raw(octets)) + else: # pragma: no cover + raise ValueError(f'unknown envelope {family.envelope!r}') + + # A fixed epoch plus the frame's own index, rather than the clock. + frame.time = EPOCH + index + return frame + + +def generate(dest: 'pathlib.Path | None' = None) -> 'list[pathlib.Path]': + """Write the option-coverage sample captures. + + Args: + dest: Destination directory; ``examples/captures/`` under the + repository root, if not given. Created if it does not exist. + + Returns: + The paths written, in the order they were written. + + """ + from scapy.all import wrpcap # pylint: disable=no-name-in-module + + dest = SAMPLE if dest is None else pathlib.Path(dest) + dest.mkdir(parents=True, exist_ok=True) + + buckets = {} # type: dict[str, list[Any]] + unreachable = {} # type: dict[str, str] + warned = {} # type: dict[str, tuple[str, ...]] + + for outcome in outcomes(): + family = FAMILY_MAP[outcome.case.family] + if family.capture is None: + # Exercised, but deliberately not captured. + continue + if outcome.status not in CAPTURABLE: + unreachable[outcome.case.label] = f'{outcome.status}: {outcome.detail}' + continue + if outcome.warnings: + warned[outcome.case.label] = outcome.warnings + + octets = outcome.octets + if octets is None: # pragma: no cover + # Unreachable: every status in CAPTURABLE got past construction, so + # it has octets. Narrowed explicitly rather than asserted, because a + # generator should not abort a whole run over a bookkeeping slip. + continue + + frames = buckets.setdefault(family.capture, []) + frames.append(_frame(family, octets, len(frames))) + + written = [] # type: list[pathlib.Path] + for name in sorted(buckets): + path = dest / name + wrpcap(str(path), buckets[name]) + written.append(path) + + total = sum(len(frames) for frames in buckets.values()) + print(f'options: {total} capturable case(s) across {len(written)} capture(s); ' + f'{len(unreachable)} case(s) left out') + for label in sorted(unreachable): + print(f' [left out] {label}: {unreachable[label]}') + for label in sorted(warned): + print(f' [warned] {label}: {"; ".join(warned[label])}') + return written + + +if __name__ == '__main__': + for sample in generate(): + print(f'{str(sample.relative_to(ROOT)):<32s} {sample.stat().st_size:8d} octets') diff --git a/tests/_support.py b/tests/_support.py index a0e6d5082d..e7be92c219 100644 --- a/tests/_support.py +++ b/tests/_support.py @@ -2,17 +2,68 @@ import abc import collections.abc +import contextlib import importlib.util import inspect import pathlib +import signal import sys import types -from typing import Iterable +import unittest +from typing import Iterable, Iterator from tests._tiers import (ROOT, SAMPLE_ROOT, REGENERATE_SAMPLES_CMD, GeneratedFixtureInUnitTierError, check_unit_tier_read) +@contextlib.contextmanager +def time_limit(seconds: int = 5) -> Iterator[None]: + """Fail the calling test if its body has not finished in ``seconds`` seconds. + + A parser defect that degenerates into a loop making no progress -- GitHub + issue #431 is one -- offers a test nothing to assert on: the call under test + simply never returns. A test written for it without a deadline does not fail, + it *wedges*, taking the rest of the run with it, so the deadline is as much a + part of the regression test as the assertion is. + + :func:`signal.alarm` is what interrupts the body, rather than a watchdog + thread: the loops this guards are pure Python and hold the GIL for the whole + of an iteration, so nothing in another thread gets to run and stop them, + whereas a signal is delivered between bytecodes. That also rules out + :data:`signal.SIGTERM` from an outer :program:`timeout`, which such a loop + likewise never gets around to handling. + + Args: + seconds: Whole seconds to allow the body. :func:`signal.alarm` counts in + whole seconds, so this cannot usefully be fractional. + + Yields: + Nothing. The deadline applies to the body of the ``with`` statement. + + Raises: + TimeoutError: If the body has not finished within ``seconds`` seconds. + + """ + # An interval timer is a POSIX facility, and the deadline is the whole point + # of this helper: silently running the body without one would restore exactly + # the wedged run it exists to prevent, so the test is skipped instead. + if not hasattr(signal, 'SIGALRM'): + raise unittest.SkipTest('signal.alarm is unavailable on this platform') + + def expire(signum: int, frame: object) -> None: + raise TimeoutError(f'did not finish within {seconds}s') + + previous = signal.signal(signal.SIGALRM, expire) + signal.alarm(seconds) + try: + yield + finally: + # Cancel before restoring, so that an alarm which fires between the two + # cannot be delivered to whatever handler was installed before. + signal.alarm(0) + signal.signal(signal.SIGALRM, previous) + + def sample_path(name: str) -> str: """Resolve a sample capture file name to its absolute path. diff --git a/tests/protocols/test_option_coverage_runtime.py b/tests/protocols/test_option_coverage_runtime.py new file mode 100644 index 0000000000..13575143ab --- /dev/null +++ b/tests/protocols/test_option_coverage_runtime.py @@ -0,0 +1,224 @@ +# -*- coding: utf-8 -*- +"""The ``options-*.pcap`` fixtures decode, and the octets in them are pcapkit's own. + +:file:`examples/generators/options.py` constructs one item of every option-like +code the library registers, and writes the ones that survive a construct -> +parse -> construct cycle into five captures. This module reads them back through +the public extraction interface. + +The point is not to re-assert the round trip -- +:file:`tests/protocols/test_option_roundtrip_unit.py` does that, and does it +without needing a fixture. The point is that these captures go through the +*whole* stack: the pcap reader, the link and internet layer dispatch, the +option-area walk, and the dumpers. A defect that only shows up once an option is +inside a real frame -- an option area whose length disagrees with the header, a +next-layer lookup that fires on the wrong protocol number -- is invisible to a +direct round trip and lands here. + +This module is fixture tier, so it may read generated captures; it is named +``*_runtime.py`` for exactly that reason, and it handles the captures being +absent so that a checkout which has not run ``make samples`` skips rather than +fails. See :mod:`tests._tiers` for the rule. + +Every extraction is wrapped in :func:`tests._support.time_limit`. That is not +belt-and-braces: the generator deliberately leaves out the cases whose parse does +not terminate -- ``HOPOPT``'s ``SMF_DPD`` is one -- and this deadline is what +turns a regression that puts one back into a failure rather than a wedged CI run. + +""" + +from __future__ import annotations + +import importlib.util +import os +import tempfile +import unittest +from typing import TYPE_CHECKING + +from tests._support import close_extractor, purge_modules, time_limit +from tests._tiers import SAMPLE_ROOT + +if TYPE_CHECKING: + from typing import Any, Optional + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: Whole seconds one capture may take to extract. The largest of them is a few +#: hundred frames of a few dozen octets, so this is orders of magnitude of slack; +#: it exists for the non-terminating case, not for slowness. +EXTRACT_TIMEOUT = 30 + +#: The captures, and the outermost protocol chain every frame in each should +#: decode to. Spelled as a prefix rather than the whole chain because the +#: interesting part is that the envelope dispatched to the right layer -- what +#: the option itself decodes to is the round-trip test's business. +CAPTURES = { + 'options-tcp.pcap': 'Ethernet:IPv4:TCP', + 'options-ipv4.pcap': 'Ethernet:IPv4', + 'options-ipv6.pcap': 'Ethernet:IPv6', + 'options-transport.pcap': 'Ethernet:IPv4', + 'options-internet.pcap': 'Ethernet:IPv6', +} + + +def _capture(name: str) -> 'Optional[str]': + """Path to a generated option capture, or :data:`None` if it is not there. + + Deliberately not :func:`tests._support.sample_path`: that raises + :exc:`FileNotFoundError` for a missing capture, and every caller here wants + to skip instead. The tier guard is satisfied either way, since this module is + fixture tier and may read a generated capture at all. + + Args: + name: Bare capture file name. + + Returns: + The absolute path, or :data:`None` if the capture has not been generated. + + """ + path = SAMPLE_ROOT / name + return str(path) if path.is_file() else None + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class OptionCoverageCaptureTests(unittest.TestCase): + """The generated option captures extract cleanly.""" + + def setUp(self) -> None: + purge_modules(['pcapkit']) + + def test_every_option_capture_extracts(self) -> None: + """Each capture reads back, with every frame reaching its expected layer. + + One subtest per capture, so a single broken fixture names itself rather + than stopping at the first one. + + """ + import pcapkit + + found = 0 + for name, chain in sorted(CAPTURES.items()): + path = _capture(name) + if path is None: + continue + found += 1 + + with self.subTest(capture=name): + with time_limit(EXTRACT_TIMEOUT): + extractor = pcapkit.extract(fin=path, nofile=True, format='tree', + store=True, ip=True, tcp=True) + self.addCleanup(close_extractor, extractor) + + self.assertGreater(extractor.length, 0, + f'{name} carries no frames') + self.assertEqual(len(extractor.frame), extractor.length) + + # ``Extractor.frame`` is annotated as a union that includes the + # raw ``(timestamp, bytes)`` tuple the ``pcap_ct``-style engines + # yield; the default engine always gives a protocol object. + frames = list(extractor.frame) # type: list[Any] + for index, frame in enumerate(frames): + protochain = str(frame.protochain) + self.assertTrue( + protochain.startswith(chain), + f'{name} frame {index}: expected a {chain} envelope, ' + f'got {protochain}' + ) + + if not found: + self.skipTest( + 'no options-*.pcap fixtures present; run ' + 'examples/generators/make_samples.py to build them' + ) + + def test_option_captures_dump_to_tree_and_json(self) -> None: + """Each capture survives both dumpers, with reassembly turned on. + + The dumpers walk every field of every parsed option, so this reaches + representation code that extraction alone does not -- a field whose value + cannot be rendered fails here and nowhere else. + + """ + import pcapkit + + found = 0 + for name in sorted(CAPTURES): + path = _capture(name) + if path is None: + continue + found += 1 + + for fmt, suffix in (('tree', 'txt'), ('json', 'json')): + with self.subTest(capture=name, format=fmt): + with tempfile.TemporaryDirectory() as tmpdir: + out = os.path.join(tmpdir, f'{name}.{suffix}') + # ``extract`` types ``format`` as a Literal, and a value + # read out of a loop is a plain ``str`` to a checker. + fmt_arg = fmt # type: Any + with time_limit(EXTRACT_TIMEOUT): + extractor = pcapkit.extract( + fin=path, fout=out, format=fmt_arg, store=False, + ip=True, tcp=True, reassembly=True) + self.addCleanup(close_extractor, extractor) + self.assertGreater(extractor.length, 0) + self.assertTrue(os.path.isfile(out), + f'{fmt} dump of {name} wrote no file') + self.assertGreater(os.path.getsize(out), 0, + f'{fmt} dump of {name} is empty') + + if not found: + self.skipTest('no options-*.pcap fixtures present') + + def test_option_captures_are_what_the_generator_says_they_are(self) -> None: + """Frame counts match the case table, so a stale fixture is caught. + + A capture regenerated from a different revision of the case table would + otherwise sit on disk indefinitely and quietly test the wrong thing. The + count is derived from the same table the generator uses, not written down + here. + + """ + from tests._tiers import ROOT + + spec = importlib.util.spec_from_file_location( + 'pcapkit_samples_options_runtime', + ROOT / 'examples' / 'generators' / 'options.py') + if spec is None or spec.loader is None: # pragma: no cover + self.skipTest('cannot load the option case table') + options = importlib.util.module_from_spec(spec) + spec.loader.exec_module(options) + + import pcapkit + + expected = {} # type: dict[str, int] + for outcome in options.outcomes(): + family = options.FAMILY_MAP[outcome.case.family] + if family.capture is None or outcome.status not in options.CAPTURABLE: + continue + expected[family.capture] = expected.get(family.capture, 0) + 1 + + found = 0 + for name, count in sorted(expected.items()): + path = _capture(name) + if path is None: + continue + found += 1 + with self.subTest(capture=name): + with time_limit(EXTRACT_TIMEOUT): + extractor = pcapkit.extract(fin=path, nofile=True, format='tree', + store=False) + self.addCleanup(close_extractor, extractor) + self.assertEqual( + extractor.length, count, + f'{name} holds {extractor.length} frames but the case table ' + f'yields {count}; regenerate it with ' + f'examples/generators/make_samples.py' + ) + + if not found: + self.skipTest('no options-*.pcap fixtures present') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/test_option_roundtrip_unit.py b/tests/protocols/test_option_roundtrip_unit.py new file mode 100644 index 0000000000..41f1e565a7 --- /dev/null +++ b/tests/protocols/test_option_roundtrip_unit.py @@ -0,0 +1,681 @@ +# -*- coding: utf-8 -*- +"""Construct -> parse -> construct identity, over every option-like code. + +Every option, chunk, parameter, message type, frame type and block type this +library claims to support is enumerated *from the dispatch registries* and put +through one cycle: construct it through the public construction API, parse the +result back, construct it again from what was parsed, and require the octets to +be identical across the round trip. + +That third step is what this module exists for. A construct-then-parse test +passes for a ``_make_*`` that takes only keyword arguments and cannot consume +the data model its own ``_read_*`` produced -- which is a defect that ships, and +one this suite had no way to see. Sixteen HIP parameters are in exactly that +state today. + +The case list and the cycle both live in +:file:`examples/generators/options.py`, next to the generator that turns the +same cases into the ``options-*.pcap`` fixtures, so that the captures and these +assertions can never describe different things. See that module for why the +enumeration is registry-driven and how the per-code arguments were chosen. + +What this module adds is the *judgement*: :data:`EXPECTED_FAILURES` records, case +by case, which cycles do not close today and which defect stops each one. A case +absent from that table has to come back ``'OK'``. + +Why the table is asserted in both directions +-------------------------------------------- + +An expected failure that is only allowed to fail is a test that rots. So each +entry is checked to still fail, *and* to fail in the recorded way -- so fixing +one of these defects turns this module red, which is the reminder to delete the +entry. The reverse check matters as much: :meth:`test_expected_failures_name_real_cases` +fails if the table names a case that no longer exists, which is what happens when +a registry entry is renamed or removed. + +Nothing here is a workaround for the defects it records. Every ``_make_*`` +argument in the generator's tables is a legitimate value for that option, and +none of the assertions below has been loosened to make a failing case pass. The +one place where an argument was chosen to route *around* a defect rather than +into it is HIP's ``HIP_COPIES``, which puts two copies of each parameter in a +packet because one is unrepresentable; the defect that forces it is not lost, +:meth:`OptionRoundTripTests.test_a_single_hip_parameter_cannot_be_constructed` +pins it directly. + +This module is unit tier: it constructs its own octets and reads no capture, so +it runs on a fresh checkout with nothing generated. + +""" + +from __future__ import annotations + +import importlib.util +import sys +import types +import unittest +from typing import TYPE_CHECKING, NamedTuple + +from tests._support import purge_modules, time_limit +from tests._tiers import ROOT + +if TYPE_CHECKING: + from typing import Any + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: Whole seconds one case may take. Generous next to a working case, which takes +#: low single-digit milliseconds; the point of the deadline is the cases that +#: never finish at all. +CASE_TIMEOUT = 10 + + +class Gap(NamedTuple): + """One cycle that does not close, and the defect that stops it.""" + + #: Expected :attr:`~examples.generators.options.Outcome.status`. + status: 'str' + #: Substring the failure detail must contain, or ``''`` to assert only the + #: status. Empty is used where the detail is a pair of long hex strings, or + #: where it embeds a timestamp and so is not stable between runs. + fragment: 'str' + #: The defect, named with the ``file:line`` that causes it. This is the + #: field that makes the entry worth keeping rather than just silencing. + defect: 'str' + + +#: Every case whose cycle does not close today. +#: +#: Grouped by defect rather than by family, because the failures cluster far more +#: tightly by cause than by protocol: four lines of ``schema/transport/tcp.py`` +#: account for four Multipath TCP subtypes, and one line of +#: ``pcapkit/protocols/misc/pcapng.py`` accounts for twenty-seven PCAP-NG +#: options. +EXPECTED_FAILURES = { + + # -- TCP ------------------------------------------------------------------ + + # ``_make_mode_timeout`` writes ``length=3`` into an option that packs to + # four octets (kind, length, and a two-octet bitfield), and + # ``_read_mode_timeout`` checks the length exactly. Measured: the schema + # packs ``1c03003c`` where a correct one is ``1c04003c``. + 'tcp-option/User_Timeout_Option': Gap( + 'CONSTRUCT', 'invalid format', + 'pcapkit/protocols/transport/tcp.py:2506 -- _make_mode_timeout sets ' + 'length=3 for a 4-octet option'), + + # ``_make_mode_qs`` computes ``rate_val`` as a floor of a logarithm that is + # negative for any rate under 40 kbps, and a negative value then fails to + # pack into the 4-bit field with ``ValueError: invalid literal for int() + # with base 2: b'0000-110'``. At the default ``rate=0`` it is guarded, and + # the option constructs -- and then will not parse back. + 'tcp-option/Quick_Start_Response': Gap( + 'PARSE', 'StructError', + 'pcapkit/protocols/transport/tcp.py:2465 -- _make_mode_qs; rate_val is ' + 'negative below 40 kbps, and the option it emits at rate=0 does not ' + 'parse back'), + + # Four lambdas read ``pkt['length']`` while packing, but the enclosing + # ``_make_mptcp_*`` never puts a ``length`` in the packet, so the key is + # simply absent on the construction path. + 'tcp-mptcp/MP_CAPABLE': Gap( + 'CONSTRUCT', "KeyError: 'length'", + 'pcapkit/protocols/schema/transport/tcp.py:658'), + 'tcp-mptcp/ADD_ADDR': Gap( + 'CONSTRUCT', "KeyError: 'length'", + 'pcapkit/protocols/schema/transport/tcp.py:790'), + 'tcp-mptcp/REMOVE_ADDR': Gap( + 'CONSTRUCT', "KeyError: 'length'", + 'pcapkit/protocols/schema/transport/tcp.py:807'), + 'tcp-mptcp/MP_PRIO': Gap( + 'CONSTRUCT', "KeyError: 'length'", + 'pcapkit/protocols/schema/transport/tcp.py:827'), + + # ``_read_tcp_options`` reads ``schema.kind`` off every option schema, but + # the nested Multipath TCP subtype schemas do not declare one -- the field + # belongs to the enclosing option. + 'tcp-mptcp/DSS': Gap( + 'CONSTRUCT', "no attribute 'kind'", + 'pcapkit/protocols/transport/tcp.py:668 -- MPTCPDSS declares no kind'), + 'tcp-mptcp/MP_FAIL': Gap( + 'CONSTRUCT', "no attribute 'kind'", + 'pcapkit/protocols/transport/tcp.py:668 -- MPTCPFallback declares no kind'), + 'tcp-mptcp/MP_FASTCLOSE': Gap( + 'CONSTRUCT', "no attribute 'kind'", + 'pcapkit/protocols/transport/tcp.py:668 -- MPTCPFastclose declares no kind'), + + # ``_make_mptcp_join`` branches on ``self._flags``, which only the parse + # path ever sets, so the constructor cannot be called at all. + 'tcp-mptcp/MP_JOIN': Gap( + 'CONSTRUCT', "no attribute '_flags'", + 'pcapkit/protocols/transport/tcp.py:2675 -- _make_mptcp_join reads ' + 'self._flags, which exists only while parsing'), + + # -- IPv4 ----------------------------------------------------------------- + + # ``_make_ipv4_options`` appends a bare enumeration member to the option + # list as its end-of-list padding, and ``OptionField.pack`` accepts only + # bytes or a Schema. It fires for every option whose packed length is not a + # multiple of four, which is what these four have in common. LSR, RR and SSR + # cannot be brought to a multiple of four by any argument: their length is + # ``3 + counts * 4``. + 'ipv4-option/LSR': Gap( + 'CONSTRUCT', 'Field options has invalid value', + 'pcapkit/protocols/internet/ipv4.py:1225 and :1252 -- a bare ' + 'Enum_OptionNumber.EOOL is appended to the option list'), + 'ipv4-option/RR': Gap( + 'CONSTRUCT', 'Field options has invalid value', + 'pcapkit/protocols/internet/ipv4.py:1225 and :1252'), + 'ipv4-option/SSR': Gap( + 'CONSTRUCT', 'Field options has invalid value', + 'pcapkit/protocols/internet/ipv4.py:1225 and :1252'), + # SID reaches the same padding branch for a second reason of its own: + # ``_make_opt_sid`` declares ``length=4`` while ``SIDOption.sid`` is a + # 32-bit field, so the option packs to six octets (``880400000000``). + 'ipv4-option/SID': Gap( + 'CONSTRUCT', 'Field options has invalid value', + 'pcapkit/protocols/internet/ipv4.py:1683 -- _make_opt_sid declares ' + 'length=4 but packs 6 octets, which then trips the :1225 padding branch'), + + # ``_make_opt_ts`` passes ``data=`` where the schema field is ``ts_data``. + # ``Schema.__init__`` only warns about an unknown field name and carries on, + # so the value is dropped and the attribute stays bound to the class-level + # descriptor -- which ``post_process`` then tries to iterate. + 'ipv4-option/TS': Gap( + 'CONSTRUCT', "'ListField' object is not iterable", + 'pcapkit/protocols/internet/ipv4.py:1488 -- data= should be ts_data=, ' + 'dropped with UnknownFieldWarning and surfacing at ' + 'pcapkit/protocols/schema/internet/ipv4.py:262'), + + # -- Quick-Start, in all three protocols that carry it -------------------- + + # ``_make_opt_qs`` returns a bare nested schema, but ``func`` is set only by + # ``_QSOption.post_process``, which runs on the parse path. Since + # ``__post_init__`` packs and then re-reads, construction fails. + # + # There is a second, independent defect in the same option that this case + # never reaches, and it is the more serious of the two: + # ``quick_start_data_selector`` hands the nested schema a hardcoded + # ``SchemaField(length=5)`` where the schema needs eight octets + # (type 1 + length 1 + flags 1 + ttl 1 + nonce 4). Measured on a + # hand-built, well-formed 8-octet IPv4 Quick-Start option + # ``1908002adeadbee0``: it parses "successfully" with + # ``SchemaWarning: packet length < 0: -3`` and decodes ``nonce`` as **55** + # instead of 933982136 -- silent corruption rather than a failure -- and the + # three unconsumed octets are then read as a further, fabricated option, + # which makes the enclosing IPv4 packet unparseable. The identical + # ``SchemaField(length=5)`` is at hopopt.py:224 and ipv6_opts.py:224, with + # the same measured nonce of 55. Fixing the ``func`` defect alone will not + # make these cases pass. + 'ipv4-option/QS': Gap( + 'CONSTRUCT', "no attribute 'func'", + 'pcapkit/protocols/internet/ipv4.py:1144 -- func is set only by ' + 'post_process; and separately ' + 'pcapkit/protocols/schema/internet/ipv4.py:128 -- SchemaField(length=5) ' + 'for an 8-octet option, which decodes nonce as 55'), + 'hopopt-option/Quick_Start': Gap( + 'CONSTRUCT', "no attribute 'func'", + 'pcapkit/protocols/internet/hopopt.py:869; and separately ' + 'pcapkit/protocols/schema/internet/hopopt.py:224 -- SchemaField(length=5)'), + 'ipv6-opts-option/Quick_Start': Gap( + 'CONSTRUCT', "no attribute 'func'", + 'pcapkit/protocols/internet/ipv6_opts.py:881; and separately ' + 'pcapkit/protocols/schema/internet/ipv6_opts.py:224 -- ' + 'SchemaField(length=5)'), + + # -- The non-progress loop ------------------------------------------------ + + # These two do not fail, they *hang*, which is why the cycle is run under a + # deadline at all. ``_MFTCP``-style forward-match sizing gives the nested + # SMF_DPD schema a length that swallows the whole option area, so + # ``OptionField.unpack`` is left at end-of-file while its own accounting + # still says octets remain; every further iteration then reads ``b''``, + # decodes a zero-length option, and subtracts nothing from the remaining + # length. GitHub issue #431 is the same shape, fixed for one case on an + # unmerged branch. + 'hopopt-option/SMF_DPD': Gap( + 'TIMEOUT', 'did not finish', + 'pcapkit/corekit/fields/collections.py:359 -- length -= len(data) makes ' + 'no progress; the over-long nested length comes from ' + 'pcapkit/protocols/schema/internet/hopopt.py:163 and the ' + "'len': (1, 8) namespace at :374, which decodes 16 for a length of 1"), + 'ipv6-opts-option/SMF_DPD': Gap( + 'TIMEOUT', 'did not finish', + 'pcapkit/corekit/fields/collections.py:359, via ' + 'pcapkit/protocols/schema/internet/ipv6_opts.py:163 and :374'), + + # -- IPv6-Route ----------------------------------------------------------- + + # ``IPv6_Route.make`` writes a raw octet count into ``length``, which is an + # 8-octet-unit field, on the dict and Data paths -- while the bytes and + # Schema paths divide by eight. The read-side guards then reject it, and + # those guards compare the unit field against octet counts too, so the + # RFC-correct value would fail as well. + 'ipv6-route-type/Source_Route': Gap( + 'CONSTRUCT', 'invalid format', + 'pcapkit/protocols/internet/ipv6_route.py:276 -- length in octets, not ' + 'in 8-octet units; guard at :461'), + 'ipv6-route-type/Type_2_Routing_Header': Gap( + 'CONSTRUCT', 'invalid format', + 'pcapkit/protocols/internet/ipv6_route.py:276; guard at :505'), + # RPL fails earlier still: ``post_process`` assumes ``addresses`` is bytes, + # which is true after unpacking and false while packing, where it is still + # the list the constructor was handed. + 'ipv6-route-type/RPL_Source_Route_Header': Gap( + 'CONSTRUCT', 'does not appear to be an IPv4 or IPv6 address', + 'pcapkit/protocols/schema/internet/ipv6_route.py:156 -- post_process ' + 'assumes bytes; it runs on the pack path too, from schema.py:647'), + + # -- Mobility Header ------------------------------------------------------ + + # ``CGAParameter``'s nested length callback reads ``pkt['length']``, which + # is present while packing and absent while unpacking: ``SchemaField.unpack`` + # starts the nested schema with a fresh context whose parent is under + # ``__packet__``. A CGA extension has no other carrier, so the whole + # ``MH.__extension__`` registry is unreachable through the public API. + 'mh-extension/Multi_Prefix': Gap( + 'PARSE', "KeyError: 'length'", + 'pcapkit/protocols/schema/internet/mh.py:516 -- needs ' + "pkt['__packet__']['length'] on the unpack path"), + + # -- HIP ------------------------------------------------------------------ + + # ``_read_param_*`` and ``_make_param_*`` are found by enumeration member + # name, so both exist for code 128; but the *schema* registry is keyed by + # the ``code=`` of the class statement, and ``R1CounterParameter`` declares + # only 129. So code 128 parses as an ``UnassignedParameter``. + 'hip-parameter/R1_Counter': Gap( + 'PARSE', "no attribute 'counter'", + 'pcapkit/protocols/internet/hip.py:822 -- Parameter.registry[128] is ' + 'UnassignedParameter, because R1CounterParameter declares code=129 only'), + + # Sixteen parameters whose ``_read_param_*`` stores a list-valued field as a + # tuple, which ``_make_param_*`` passes straight back to a ``ListField`` + # that accepts only a list. This is the class of defect the reconstruct step + # exists to find: each one constructs and parses perfectly. + **{ + f'hip-parameter/{name}': Gap( + 'RECONSTRUCT', "unsupported type ", + 'pcapkit/protocols/schema/schema.py:624 -- _read_param_* returns a ' + 'tuple where _make_param_* needs a list') + for name in ( + 'ACK', 'DH_GROUP_LIST', 'HIP_CIPHER', 'NAT_TRAVERSAL_MODE', + 'HIT_SUITE_LIST', 'REG_INFO', 'REG_REQUEST', 'REG_RESPONSE', + 'REG_FAILED', 'TRANSPORT_FORMAT_LIST', 'ESP_TRANSFORM', 'ACK_DATA', + 'ROUTE_DST', 'HIP_TRANSPORT_MODE', 'ROUTE_VIA', 'VIA_RVS', + ) + }, + + # ``_make_param_encrypted`` passes ``cipher=``, which is not a field of + # ``EncryptedParameter`` -- so the cipher id is dropped with an + # ``UnknownFieldWarning`` and never reaches the wire. The mismatch itself + # comes from a second defect in the same parameter: the ``data`` length + # callback omits the four octets ``reserved`` already consumed out of + # ``len``, so ``len`` grows by four on every round trip (measured: 4 -> 8). + 'hip-parameter/ENCRYPTED': Gap( + 'MISMATCH', '', + 'pcapkit/protocols/internet/hip.py:3445 -- cipher= is not a field of ' + 'EncryptedParameter and is silently dropped; and ' + 'pcapkit/protocols/schema/internet/hip.py:463 -- the data length ' + "callback omits the 4 octets 'reserved' took out of len"), + + # Two parameters whose own packed length is not what the header arithmetic + # can represent even in pairs -- see HIP_COPIES in the generator for why the + # pair is used at all. + 'hip-parameter/HIP_TRANSFORM': Gap( + 'CONSTRUCT', 'invalid parameter', + 'pcapkit/protocols/internet/hip.py:698 -- the len check; HIP_TRANSFORM ' + 'packs to a length the make-side arithmetic cannot express'), + 'hip-parameter/HOST_ID': Gap( + 'CONSTRUCT', 'invalid format', + 'pcapkit/protocols/internet/hip.py:698 -- HOST_ID packs to 14 octets ' + 'with len=8, so it is not even 4-aligned'), + + # -- HTTP/2 --------------------------------------------------------------- + + # ``SchemaField.pack`` gives a nested frame schema a fresh packet context + # whose only link to the parent is ``__packet__``, but six frame schemas + # reach for the HTTP/2 header's ``flags`` bitfield directly -- either from a + # ConditionalField test or from ``FrameType.post_process``. The three frames + # that pass are exactly the three declaring no flag members at all: + # RST_STREAM, GOAWAY and WINDOW_UPDATE. + **{ + f'httpv2-frame/{name}': Gap( + 'CONSTRUCT', "KeyError: 'flags'", + 'pcapkit/protocols/schema/application/httpv2.py:144 ' + '(FrameType.post_process) and the pad_len ConditionalField tests at ' + ':175, :204, :305 -- the nested context reaches for the parent ' + "header's flags") + for name in ('DATA', 'HEADERS', 'SETTINGS', 'PUSH_PROMISE', 'PING', + 'CONTINUATION') + }, + + # ``make`` writes ``length = payload + 9`` and a PRIORITY payload is five + # octets, so the constructed header always says 14 -- while the reader + # demands exactly 9. Its siblings all check payload+9 (RST_STREAM 13, + # WINDOW_UPDATE 13, PING 17), so 9 looks like the outlier. + 'httpv2-frame/PRIORITY': Gap( + 'CONSTRUCT', 'invalid format', + 'pcapkit/protocols/application/httpv2.py:572 -- reads length != 9 for a ' + 'frame make() always builds with length 14'), + + # -- PCAP-NG -------------------------------------------------------------- + + # ``PCAPNG.__post_init__`` packs and then re-parses *on the same instance*, + # and both halves share the per-instance option counter: the make path + # increments it, the read path then trips its own "only one of these" + # guard on the option it just built. Twenty-seven options have such a + # guard; the ten that construct are the ones that do not. + **{ + f'pcapng-option/{name}': Gap( + 'CONSTRUCT', 'option must be only one', + 'pcapkit/protocols/misc/pcapng.py:1078 packs and :1088 re-parses on ' + 'one instance, sharing self._opt; incremented at :3941, checked at ' + ':2175') + for name in ( + 'if_name_2', 'if_description_3', 'if_MACaddr_6', 'if_EUIaddr_7', + 'if_speed_8', 'if_tsresol_9', 'if_tzone_10', 'if_filter_11', + 'if_os_12', 'if_fcslen_13', 'if_tsoffset_14', 'if_hardware_15', + 'if_txspeed_16', 'if_rxspeed_17', + 'epb_flags_2', 'epb_dropcount_4', 'epb_packetid_5', 'epb_queue_6', + 'ns_dnsname_2', 'ns_dnsIP4addr_3', 'ns_dnsIP6addr_4', + 'isb_ifrecv_4', 'isb_ifdrop_5', 'isb_filteraccept_6', + 'isb_osdrop_7', 'isb_usrdeliv_8', + 'pack_flags_2', + ) + }, + + # ``_make_option_if_ipv6`` hardcodes ``length=8``, copied from its IPv4 + # sibling where 8 is right, while ``IPv6InterfaceField`` is 17 octets. The + # option packs to 21, the block total becomes 41, and no argument the caller + # can pass changes it. + 'pcapng-option/if_IPv6addr_5': Gap( + 'CONSTRUCT', 'invalid length: 41', + 'pcapkit/protocols/misc/pcapng.py:4190 -- length=8 for a 17-octet ' + 'IPv6InterfaceField'), + + # ``_isb_interface_id`` is read by these two make-side constructors but + # assigned only by ``_read_block_isb``. + 'pcapng-option/isb_starttime_2': Gap( + 'CONSTRUCT', "no attribute '_isb_interface_id'", + 'pcapkit/protocols/misc/pcapng.py:4991 -- reads an attribute set only ' + 'at :1790, on the parse path'), + 'pcapng-option/isb_endtime_3': Gap( + 'CONSTRUCT', "no attribute '_isb_interface_id'", + 'pcapkit/protocols/misc/pcapng.py:5024 -- as above'), + + # The three packet-carrying blocks lose their payload on the way back: + # ``_make_block_*`` never restores ``packet_data``, and it could not, since + # the data model has no field to keep it in -- the octets go to the + # next-layer dissector and survive only as the decoded chain. The rebuilt + # block keeps ``captured_len`` while carrying no data, so it is malformed + # rather than merely shorter. + **{ + f'pcapng-block/{name}': Gap( + 'MISMATCH', '', + 'pcapkit/protocols/misc/pcapng.py:3514, :3565, :3851 -- ' + 'packet_data is not restored, and ' + 'pcapkit/protocols/data/misc/pcapng.py:442, :474, :895 have no ' + 'field to restore it from') + for name in ('Enhanced_Packet_Block', 'Simple_Packet_Block', 'Packet_Block') + }, + + # The two key-log secrets writers terminate each line with ``os.sep`` -- a + # forward slash on POSIX -- where a newline is meant, while the readers + # split on newlines. So the whole log parses as one comment line and every + # entry is lost. They also stamp the current time into the body with no way + # to override it, which is why these two are the only cases in the suite + # whose failure detail is not stable, and why the generator's + # Decryption Secrets Block case uses a ZigBee key instead. + 'pcapng-secrets/TLS_Key_Log': Gap( + 'MISMATCH', '', + 'pcapkit/protocols/misc/pcapng.py:5553 and :5556 -- os.sep as a line ' + 'terminator, against splitlines() at ' + 'pcapkit/protocols/schema/misc/pcapng.py:1480; and datetime.now() in ' + 'the payload'), + 'pcapng-secrets/WireGuard_Key_Log': Gap( + 'MISMATCH', '', + 'pcapkit/protocols/misc/pcapng.py:5585 and :5587, against ' + 'pcapkit/protocols/schema/misc/pcapng.py:1519; and datetime.now()'), +} + + +def _load_generator() -> 'types.ModuleType': + """Load :file:`examples/generators/options.py` by path. + + That directory is not a package, and its module names are too generic to put + on :data:`sys.path` -- which is exactly why + :file:`examples/generators/make_samples.py` loads its siblings by path too. + This follows it. + + Returns: + The generator module, which exposes ``cases``, ``roundtrip``, ``FAMILIES`` + and ``SKIP``. + + Raises: + RuntimeError: If the module cannot be found or loaded. + + """ + path = ROOT / 'examples' / 'generators' / 'options.py' + spec = importlib.util.spec_from_file_location('pcapkit_samples_options', path) + if spec is None or spec.loader is None: # pragma: no cover + raise RuntimeError(f'cannot load the option case table from {path}') + + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class OptionRoundTripTests(unittest.TestCase): + """One construct -> parse -> construct cycle per registered code.""" + + #: The generator module, loaded once for the whole class. Loading it imports + #: :mod:`pcapkit`, so it must happen after :meth:`setUp` has purged the + #: previous test's copy -- hence a class attribute filled in + #: :meth:`setUpClass` rather than a module-level import. + options = None # type: Any + + @classmethod + def setUpClass(cls) -> None: + purge_modules(['pcapkit']) + cls.options = _load_generator() + + def _run(self, case: 'Any') -> 'Any': + """One cycle, under this suite's deadline rather than the generator's. + + The generator applies its own :func:`signal.alarm` deadline, because + ``make samples`` must not hang either. Letting both arm the alarm would + break the outer one: the inner context cancels the pending alarm on the + way out, which is the outer deadline's countdown. So the generator's is + switched off here and :func:`tests._support.time_limit` owns the clock, + which is what the rest of the suite uses. + + Args: + case: The case to exercise. + + Returns: + The cycle's outcome, with a timeout reported as ``'TIMEOUT'`` rather + than raised, so that it is asserted on like any other failure. + + """ + try: + with time_limit(CASE_TIMEOUT): + return self.options.roundtrip(case, deadline=0) + except TimeoutError as exc: + return self.options.Outcome(case, 'TIMEOUT', str(exc), None, ()) + + def test_every_registered_code_has_a_case(self) -> None: + """No registry entry is left without a case. + + This is what makes the coverage self-maintaining. Registering a new + option and forgetting to exercise it fails here, rather than going + unnoticed until the option turns out not to construct. + + """ + for family in self.options.FAMILIES: + with self.subTest(family=family.label): + registry = family.registry() + covered = {case.code for case in self.options.cases((family,))} + skipped = { + code for code in registry + if (family.label, self.options.code_name(code)) in self.options.SKIP + } + self.assertEqual( + set(registry), covered | skipped, + f'{family.label}: every code in the registry needs a case in ' + f'examples/generators/options.py, or an entry in its SKIP table' + ) + + def test_every_registered_code_resolves_without_growing_the_registry(self) -> None: + """Each code resolves to a handler, and asking does not mutate the registry. + + Two things at once, because they are the same lookup. + + A registry entry naming a handler that does not exist is a live bug that + nothing else catches: dispatch falls through to the fallback and the + option is quietly parsed as unassigned. Two of the ``MH`` registries' + own docstrings name prefixes that were never implemented + (``_read_option_*`` and ``_read_extension_*``, where the code uses + ``_read_opt_*`` and ``_read_ext_*``), so the hazard is not theoretical. + + And every one of these registries is a :class:`collections.defaultdict` + on a class attribute, so a lookup that misses *inserts*, permanently, for + the whole process. The sizes are compared either side to prove the + non-recording path was used. + + """ + for family in self.options.FAMILIES: + with self.subTest(family=family.label): + registry = family.registry() + before = len(registry) + + for code in list(registry): + resolved = self.options.handler(registry, code) + self.assertIsNotNone( + resolved, + f'{family.label}: {self.options.code_name(code)} resolves ' + f'to nothing' + ) + + self.assertEqual( + len(registry), before, + f'{family.label}: the registry grew while being read; a ' + f'lookup went through registry[code] instead of ' + f'ProtocolBase._lookup_registry' + ) + + def test_expected_failures_name_real_cases(self) -> None: + """The expected-failure table names only cases that exist. + + Without this the table rots silently: a renamed or removed registry + entry leaves an entry here that can never be checked, and which then + reads as documentation of a defect nobody can find. + + """ + labels = {case.label for case in self.options.cases()} + stale = sorted(set(EXPECTED_FAILURES) - labels) + self.assertEqual( + stale, [], + 'these entries of EXPECTED_FAILURES name cases that no longer exist; ' + 'delete them, or fix the label' + ) + + def test_round_trip_is_identity_or_a_recorded_gap(self) -> None: + """Every case either closes the cycle, or fails exactly as recorded. + + Both halves matter. A case absent from :data:`EXPECTED_FAILURES` must + come back ``'OK'``; a case present in it must still fail, and with the + recorded status and detail, so that fixing the defect turns this red and + the entry gets deleted rather than left behind. + + """ + cases = self.options.cases() + self.assertGreater(len(cases), 200, + 'the registries should yield a few hundred codes; a ' + 'much smaller number means the enumeration broke') + + for case in cases: + with self.subTest(case=case.label): + outcome = self._run(case) + gap = EXPECTED_FAILURES.get(case.label) + + if gap is None: + self.assertEqual( + outcome.status, 'OK', + f'{case.label} no longer round-trips: {outcome.detail}. ' + f'If this is a newly found defect, add it to ' + f'EXPECTED_FAILURES with the file:line that causes it -- ' + f'do not change the case to avoid it.' + ) + continue + + self.assertEqual( + outcome.status, gap.status, + f'{case.label} was recorded as failing with ' + f'{gap.status} ({gap.defect}) but came back ' + f'{outcome.status}: {outcome.detail}. If the defect is fixed, ' + f'delete its EXPECTED_FAILURES entry.' + ) + if gap.fragment: + self.assertIn( + gap.fragment, outcome.detail, + f'{case.label} still fails with {gap.status}, but not in ' + f'the recorded way ({gap.defect})' + ) + + def test_a_single_hip_parameter_cannot_be_constructed(self) -> None: + """A HIP packet carrying exactly one parameter is rejected by its own reader. + + This is the defect the generator's ``HIP_COPIES = 2`` routes around, and + it is pinned here so that routing around it does not also bury it. + + ``HIP.make`` computes the header's ``len`` as ``total_length // 8 + 4``, + which is lossless only when the parameter octets are a multiple of eight. + The parameter padding rule pads the *contents* to eight and ignores the + four-octet type-and-length header, so one parameter is always + ``4 (mod 8)``; the floor division drops those four octets, and + ``_read_hip_param`` compares the recovered length exactly and raises. + + Two copies sum to a multiple of eight, so the same parameter that fails + alone succeeds in a pair -- which is the control that makes this a + statement about the header arithmetic rather than about ``SEQ``. + + """ + from pcapkit.const.hip.parameter import Parameter + from pcapkit.protocols.internet.hip import HIP + from pcapkit.utilities.exceptions import ProtocolError + + base = dict(self.options.HIP_BASE) + one = [(Parameter.SEQ, {})] # type: list[tuple[Any, dict[str, Any]]] + + with self.assertRaises(ProtocolError) as caught: + HIP(parameters=one, extension=True, **base) + self.assertIn('invalid format', str(caught.exception)) + + # The control: the identical parameter, twice, round-trips exactly. + paired = bytes(HIP(parameters=one * 2, extension=True, **base)) + reparsed = HIP(paired, len(paired), extension=True) + again = bytes(HIP(parameters=reparsed.info.parameters, extension=True, **base)) + self.assertEqual(paired, again) + + def test_recorded_gaps_are_a_minority(self) -> None: + """Most of the option space round-trips, and the rest is accounted for. + + A guard on the shape of the result rather than on any one case: if a + change makes the failures outnumber the successes, something systemic + broke and the per-case assertions above will be too noisy to read. + + """ + total = len(self.options.cases()) + self.assertLess( + len(EXPECTED_FAILURES), total // 2, + f'{len(EXPECTED_FAILURES)} of {total} cases are recorded as failing; ' + f'that is more than half, which suggests the harness rather than the ' + f'library is at fault' + ) + + +if __name__ == '__main__': + unittest.main() From 543217b79f65f189fbc670ef32d90648abce6a37 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Thu, 17 Sep 2026 16:28:03 -0400 Subject: [PATCH 2/4] tests: record the seven option cases whose outcome depends on the interpreter, not the code Three things, all consequences of #432, #434 and #439 landing under the branch. - `INTERPRETER_GAPS`: seven PCAP-NG name-resolution cases round-trip from Python 3.11 on and fail to construct on 3.10, so one recorded outcome per case no longer suffices. The root cause is #439 in the *schema* hierarchy: `NameResolutionBlock.post_process` asks `isinstance(record, (IPv4Record, IPv6Record))` at `pcapkit/protocols/schema/misc/pcapng.py:1248`, every `Schema` subclass shares one `_abc_impl` on <= 3.10 (measured: `EndRecord._abc_impl is IPv4Record._abc_impl` is True on 3.10.20, False on 3.14.7), and the block's terminating `EndRecord` therefore tests True and has `.names` read off it. Measured in the real path, not inferred. The `Info` data models are unaffected on both interpreters, and that boundary is asserted too. The table overrides rather than sits beside `EXPECTED_FAILURES`, because the three `ns_dns*` options fail on every interpreter but for different reasons. - `test_schema_isinstance_is_interpreter_dependent` pins that mechanism, so the seven are excused by evidence about a named library bug rather than by a version comparison. On >= 3.11 they are still held to `OK`; fixing #439 turns 3.10 red. - The two SMF_DPD entries: `hopopt-option/SMF_DPD` now round-trips and its entry is deleted, since #429's over-read fix landed with #432's progress guard. `ipv6-opts-option/SMF_DPD` raises instead of hanging and is re-recorded as `PARSE`. The two schema modules are line-for-line duplicates, so that is one fix applied once where it was needed twice. The sweep keeps its deadline: it guards the next non-progress defect, not this one. Also: #434 gave IPv4 and HIP real registries, so the generator now reads `HIP.__parameter__` instead of falling back to enum-crossed-with-handler. No library file is touched. 3.14.7: 258 cases, 173 round-trip. 3.10.20: 258 cases, 169 round-trip -- the difference is exactly the four cases in `INTERPRETER_GAPS` that pass on 3.14. --- examples/generators/options.py | 13 +- tests/protocols/test_option_roundtrip_unit.py | 259 +++++++++++++++--- 2 files changed, 233 insertions(+), 39 deletions(-) diff --git a/examples/generators/options.py b/examples/generators/options.py index 28521c5ad7..b7b09b57da 100644 --- a/examples/generators/options.py +++ b/examples/generators/options.py @@ -327,14 +327,15 @@ def _named_registry(owner: 'type', enum: 'Any', prefix: 'str', attribute: 'str') -> 'Any': """A protocol's dispatch registry, or a stand-in derived from its handlers. - ``IPv4`` and ``HIP`` are the two protocols that still dispatch their options - and parameters by attribute name rather than through a registry, so for them - there is nothing to enumerate. The equivalent source of truth is the + ``IPv4`` and ``HIP`` were the last two protocols to dispatch their options + and parameters by attribute name rather than through a registry, and for + those there was nothing to enumerate. The equivalent source of truth is the enumeration crossed with the presence of the handler, which is what this builds when the registry is absent. - The real registry is preferred whenever it exists, so that the migration - landing is a no-op here rather than something to come back and finish. Both + GitHub pull request #434 gave both of them a real registry, so the fallback + is no longer taken for either -- it is kept because it costs nothing and is + the only thing that would notice a protocol added tomorrow without one. Both shapes are a :class:`collections.defaultdict`, so :func:`cases` and :func:`handler` cannot tell which one they were given -- and in particular :meth:`ProtocolBase._lookup_registry @@ -916,7 +917,7 @@ def _mh_extension_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': def _hip_registry() -> 'Any': from pcapkit.const.hip.parameter import Parameter from pcapkit.protocols.internet.hip import HIP - return _named_registry(HIP, Parameter, '_make_param_', '__param__') + return _named_registry(HIP, Parameter, '_make_param_', '__parameter__') def _hip_overrides() -> 'dict[Any, dict[str, Any]]': diff --git a/tests/protocols/test_option_roundtrip_unit.py b/tests/protocols/test_option_roundtrip_unit.py index 41f1e565a7..a422f4bc7f 100644 --- a/tests/protocols/test_option_roundtrip_unit.py +++ b/tests/protocols/test_option_roundtrip_unit.py @@ -23,6 +23,25 @@ by case, which cycles do not close today and which defect stops each one. A case absent from that table has to come back ``'OK'``. +When the outcome depends on the interpreter, not the code +--------------------------------------------------------- + +One recorded expectation per case assumes the outcome is a property of the +library. Seven PCAP-NG name-resolution cases show it can be a property of the +*interpreter*: they round-trip from Python 3.11 on and fail to construct on 3.10 +and earlier, because :class:`~pcapkit.protocols.schema.schema.Schema` subclasses +share one ``_abc_impl`` there and an ``isinstance`` against a schema class +therefore answers by cache order rather than by type (GitHub issue #439). + +Those seven live in :data:`INTERPRETER_GAPS` rather than in +:data:`EXPECTED_FAILURES`, and the separation is deliberate. Behaving differently +on two supported Pythons is a defect and not a variant, so the mechanism is +asserted directly by +:meth:`OptionRoundTripTests.test_schema_isinstance_is_interpreter_dependent`: +the record is evidence about a named library bug, not an exemption keyed on a +version number. On a modern interpreter the seven are held to ``'OK'`` like +anything else, and fixing #439 turns 3.10 red. + Why the table is asserted in both directions -------------------------------------------- @@ -59,7 +78,7 @@ from tests._tiers import ROOT if TYPE_CHECKING: - from typing import Any + from typing import Any, Optional RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) @@ -223,26 +242,27 @@ class Gap(NamedTuple): 'pcapkit/protocols/schema/internet/ipv6_opts.py:224 -- ' 'SchemaField(length=5)'), - # -- The non-progress loop ------------------------------------------------ - - # These two do not fail, they *hang*, which is why the cycle is run under a - # deadline at all. ``_MFTCP``-style forward-match sizing gives the nested - # SMF_DPD schema a length that swallows the whole option area, so - # ``OptionField.unpack`` is left at end-of-file while its own accounting - # still says octets remain; every further iteration then reads ``b''``, - # decodes a zero-length option, and subtracts nothing from the remaining - # length. GitHub issue #431 is the same shape, fixed for one case on an - # unmerged branch. - 'hopopt-option/SMF_DPD': Gap( - 'TIMEOUT', 'did not finish', - 'pcapkit/corekit/fields/collections.py:359 -- length -= len(data) makes ' - 'no progress; the over-long nested length comes from ' - 'pcapkit/protocols/schema/internet/hopopt.py:163 and the ' - "'len': (1, 8) namespace at :374, which decodes 16 for a length of 1"), + # -- The non-progress loop, now half fixed -------------------------------- + + # Both of these used to *hang* rather than fail, which is why the cycle is + # run under a deadline at all. #432 landed the progress guard in + # ``OptionField``/``ListField``, and #429's ``_SMFDPDOption`` over-read fix + # went in with it -- so ``hopopt-option/SMF_DPD`` now round-trips and has no + # entry here at all. + # + # ``IPv6-Opts`` did not get the same treatment, and that asymmetry is the + # finding: the two modules are line-for-line duplicates of each other (the + # over-read is at :163 and the ``'len': (1, 8)`` namespace at :374 in *both* + # schema files), so this is one fix that was applied once where it was needed + # twice. It now raises instead of hanging, which is the guard doing its job. + # + # The deadline in the sweep stays regardless. It is protection against the + # *next* non-progress defect, not against this one. 'ipv6-opts-option/SMF_DPD': Gap( - 'TIMEOUT', 'did not finish', - 'pcapkit/corekit/fields/collections.py:359, via ' - 'pcapkit/protocols/schema/internet/ipv6_opts.py:163 and :374'), + 'PARSE', 'invalid format', + 'pcapkit/protocols/schema/internet/ipv6_opts.py:163 over-reads and :374 ' + "declares 'len': (1, 8) where the octet is at bits 8-15; hopopt.py got " + 'the fix in #429/#432 and ipv6_opts.py did not'), # -- IPv6-Route ----------------------------------------------------------- @@ -440,6 +460,114 @@ class Gap(NamedTuple): } +#: Whether this interpreter gives every :class:`Schema` subclass its own +#: ``_abc_impl``, and therefore answers ``isinstance`` against schema classes +#: correctly. +#: +#: On CPython 3.10 and earlier they all share :class:`Schema`'s, which is GitHub +#: issue #439. Measured on 3.10.20 and 3.14.7: +#: ``EndRecord._abc_impl is IPv4Record._abc_impl`` is :data:`True` on 3.10 and +#: :data:`False` on 3.14, for the schema classes in +#: :mod:`pcapkit.protocols.schema.misc.pcapng`. Note that the *data* models +#: (:class:`~pcapkit.corekit.infoclass.Info` subclasses) are **not** affected on +#: either: ``Data_EndRecord._abc_impl is Info._abc_impl`` is :data:`False` on +#: both, so it is specifically the schema hierarchy. +SCHEMA_ABC_IS_PER_CLASS = sys.version_info >= (3, 11) + +#: Cases whose outcome is a property of the *interpreter* rather than of the +#: option, applying only where :data:`SCHEMA_ABC_IS_PER_CLASS` is false. +#: +#: This table exists reluctantly, and it is worth being explicit about why it is +#: not the same thing as :data:`EXPECTED_FAILURES`. +#: +#: A case that behaves differently on two supported Pythons is a defect, not a +#: variant, and the honest thing is to surface it rather than to encode it as a +#: shrug. So it is surfaced twice over: the entries below name the library defect +#: and the ``file:line`` that causes it, exactly as the main table does, and +#: :meth:`OptionRoundTripTests.test_schema_isinstance_is_interpreter_dependent` +#: pins the *mechanism* directly so the record is evidence and not an assertion +#: about a version number. What the table does not do is let the cases pass +#: quietly: on a modern interpreter they are held to ``'OK'`` by the main table's +#: absence of an entry, and on an old one they are held to failing in exactly the +#: recorded way. Fixing #439 turns this red on 3.10, which is the point. +#: +#: An entry here **overrides** :data:`EXPECTED_FAILURES` for the same label +#: rather than being disjoint from it, and three of the seven need that. The +#: ``ns_dns*`` options fail on every interpreter, but for different reasons: from +#: 3.11 on it is the shared ``self._opt`` counter rejecting the option the same +#: instance just built, and on 3.10 the block never gets that far because +#: ``post_process`` dies first. Same status, different cause, so both are recorded +#: and the interpreter decides which one is checked. The remaining four have no +#: entry in the main table at all, because they round-trip cleanly from 3.11. +#: +#: All seven are PCAP-NG name resolution, and they share one cause on 3.10. +#: ``NameResolutionBlock.post_process`` asks +#: ``isinstance(record, (IPv4Record, IPv6Record))`` at +#: :file:`pcapkit/protocols/schema/misc/pcapng.py:1248` and then reads +#: ``record.names``. Because every schema class shares one ``_abc_impl`` on 3.10, +#: that question returns :data:`True` for the block's terminating ``EndRecord`` +#: -- measured in the real path, with ``type(record) is EndRecord`` :data:`True` +#: and ``isinstance(record, (IPv4Record, IPv6Record))`` also :data:`True` in the +#: same breath -- so ``post_process`` reads ``names`` off a record that has none. +#: Every NRB carries a terminating ``EndRecord``, which is why the block, its +#: three ``ns_dns*`` options and its three records all go together. +INTERPRETER_GAPS = { + label: Gap( + 'CONSTRUCT', "'EndRecord' object has no attribute 'names'", + 'pcapkit/protocols/schema/misc/pcapng.py:1248 -- isinstance against ' + 'schema classes, which share one _abc_impl on CPython <= 3.10 (#439), so ' + 'the terminating EndRecord tests True as an IPv4Record/IPv6Record') + for label in ( + 'pcapng-block/Name_Resolution_Block', + 'pcapng-option/ns_dnsname_2', + 'pcapng-option/ns_dnsIP4addr_3', + 'pcapng-option/ns_dnsIP6addr_4', + 'pcapng-record/nrb_record_end', + 'pcapng-record/nrb_record_ipv4', + 'pcapng-record/nrb_record_ipv6', + ) +} + + +def _abc_impl(cls: 'type') -> 'Any': + """The ABC implementation object a class dispatches ``isinstance`` through. + + ``_abc_impl`` is a CPython internal and is not in the type stubs, so it is + read by name. Present on both the C ``_abc`` and the pure-python ``_py_abc`` + backends; :func:`tests._support._reset_abc_caches` relies on the same + attribute. + + Args: + cls: The class to inspect. + + Returns: + The class's ``_abc_impl``, or :data:`None` if it has none. + + """ + return getattr(cls, '_abc_impl', None) + + +def _gap_for(label: 'str') -> 'Optional[Gap]': + """The recorded expectation for ``label``, or :data:`None` if it should pass. + + Args: + label: Case label, e.g. ``'tcp-mptcp/MP_JOIN'``. + + Returns: + The recorded :class:`Gap`, or :data:`None` when the case is expected to + round-trip on this interpreter. + + """ + # The interpreter-conditional record wins where there is one, because on an + # affected interpreter the case fails there before it can reach whatever the + # main table describes. + if not SCHEMA_ABC_IS_PER_CLASS: + gap = INTERPRETER_GAPS.get(label) + if gap is not None: + return gap + return EXPECTED_FAILURES.get(label) + + def _load_generator() -> 'types.ModuleType': """Load :file:`examples/generators/options.py` by path. @@ -567,19 +695,81 @@ def test_every_registered_code_resolves_without_growing_the_registry(self) -> No ) def test_expected_failures_name_real_cases(self) -> None: - """The expected-failure table names only cases that exist. + """Both tables name only cases that exist. - Without this the table rots silently: a renamed or removed registry - entry leaves an entry here that can never be checked, and which then - reads as documentation of a defect nobody can find. + Without this they rot silently: a renamed or removed registry entry + leaves behind an entry that can never be checked, and which then reads as + documentation of a defect nobody can find. """ labels = {case.label for case in self.options.cases()} - stale = sorted(set(EXPECTED_FAILURES) - labels) + for name, table in (('EXPECTED_FAILURES', EXPECTED_FAILURES), + ('INTERPRETER_GAPS', INTERPRETER_GAPS)): + with self.subTest(table=name): + stale = sorted(set(table) - labels) + self.assertEqual( + stale, [], + f'these entries of {name} name cases that no longer exist; ' + f'delete them, or fix the label' + ) + + # An override that records the same thing as the entry it overrides is + # noise, and would quietly outlive the reason it was added. + for label in sorted(set(EXPECTED_FAILURES) & set(INTERPRETER_GAPS)): + with self.subTest(overridden=label): + general, specific = EXPECTED_FAILURES[label], INTERPRETER_GAPS[label] + self.assertNotEqual( + (general.status, general.fragment), + (specific.status, specific.fragment), + f'{label} records the same status and fragment in both tables, ' + f'so the interpreter-conditional entry adds nothing; delete it' + ) + + def test_schema_isinstance_is_interpreter_dependent(self) -> None: + """Pin the mechanism behind :data:`INTERPRETER_GAPS`, so it is evidence. + + The seven cases in that table are excused on old interpreters, and an + excuse resting on a version comparison would be indistinguishable from + giving up. So the *cause* is asserted here directly: every + :class:`~pcapkit.protocols.schema.schema.Schema` subclass shares one + ``_abc_impl`` on CPython 3.10 and earlier, and has its own from 3.11 -- + which is what makes ``isinstance`` against a schema class answer by cache + order rather than by type. + + This is GitHub issue #439 and belongs to ``SchemaMeta``, not here. When it + is fixed this test fails on 3.10, which is the prompt to delete both it + and :data:`INTERPRETER_GAPS`. + + The data models are checked too, and asserted *unaffected* on both + interpreters -- that is the boundary of the defect, and getting it wrong + in either direction would send the next reader down the wrong path. + + """ + from pcapkit.corekit.infoclass import Info + from pcapkit.protocols.data.misc.pcapng import EndRecord as Data_EndRecord + from pcapkit.protocols.schema.misc.pcapng import EndRecord, IPv4Record + from pcapkit.protocols.schema.schema import Schema + + shared = _abc_impl(EndRecord) is _abc_impl(IPv4Record) self.assertEqual( - stale, [], - 'these entries of EXPECTED_FAILURES name cases that no longer exist; ' - 'delete them, or fix the label' + shared, not SCHEMA_ABC_IS_PER_CLASS, + f'on Python {sys.version_info[0]}.{sys.version_info[1]}, schema ' + f'classes sharing one _abc_impl is {shared}; INTERPRETER_GAPS assumes ' + f'{not SCHEMA_ABC_IS_PER_CLASS}. If #439 is fixed, delete that table ' + f'and this test.' + ) + if shared: + self.assertIs( + _abc_impl(EndRecord), _abc_impl(Schema), + "the shared _abc_impl should be Schema's own" + ) + + # The boundary: Info subclasses are unaffected on every interpreter, so + # nothing that dispatches on a *data* model is implicated in #439. + self.assertIsNot( + _abc_impl(Data_EndRecord), _abc_impl(Info), + 'the data models were unaffected by #439 on both 3.10 and 3.14; if ' + 'that has changed, the scope of the defect is wider than recorded' ) def test_round_trip_is_identity_or_a_recorded_gap(self) -> None: @@ -599,7 +789,7 @@ def test_round_trip_is_identity_or_a_recorded_gap(self) -> None: for case in cases: with self.subTest(case=case.label): outcome = self._run(case) - gap = EXPECTED_FAILURES.get(case.label) + gap = _gap_for(case.label) if gap is None: self.assertEqual( @@ -669,11 +859,14 @@ def test_recorded_gaps_are_a_minority(self) -> None: """ total = len(self.options.cases()) + recorded = len(EXPECTED_FAILURES) + if not SCHEMA_ABC_IS_PER_CLASS: + recorded += len(INTERPRETER_GAPS) self.assertLess( - len(EXPECTED_FAILURES), total // 2, - f'{len(EXPECTED_FAILURES)} of {total} cases are recorded as failing; ' - f'that is more than half, which suggests the harness rather than the ' - f'library is at fault' + recorded, total // 2, + f'{recorded} of {total} cases are recorded as failing; that is more ' + f'than half, which suggests the harness rather than the library is at ' + f'fault' ) From d7240dc0c21a1c618a6c88741ba00325e12fa5c9 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Thu, 17 Sep 2026 17:25:27 -0400 Subject: [PATCH 3/4] tests: correct the recorded cause of the ipv6-opts SMF_DPD failure The entry claimed #432's fix "was applied once where it was needed twice". That was wrong, and it was an inference from the behaviour rather than something measured: #432 touched both schema modules symmetrically, 26 lines each, changing `'len': (1, 8)` to `(8, 8)` and adding the `+ 2` to the selector's `SchemaField` in each. Neither file still carries `(1, 8)`. The real cause is one line, older than #432. Normalising the two modules for their protocol names leaves exactly one structural difference: `ipv6_opts.SMFIdentificationBasedDPDOption` declares a second, redundant `test` `ForwardMatchField` at :434 that `hopopt`'s equivalent does not. The enclosing `_SMFDPDOption` already has one in both modules, and it is that outer field the selector reads -- nothing reads the nested copy, and it is the only field in the class with no `#:` comment. A `ForwardMatchField` consumes nothing but still occupies a slot in `__buffer__`, so the nested schema over-reports its size by an octet and `OptionField` mis-counts the option area. Measured on the same octets, `1100080100010100`, identically on 3.10.20 and 3.14.7 -- so this one is not interpreter-dependent: hopopt __fields__ = [type, len, info, tid, id] len(schema) = 3 ipv6_opts __fields__ = [type, len, test, info, tid, id] len(schema) = 4 HOPOPT(...) -> options=[SMF_DPD, PadN] IPv6_Opts(...) -> ProtocolError: IPv6-Opts: invalid format at pcapkit/protocols/internet/ipv6_opts.py:497 Comment and `Gap.defect` text only; the recorded status and fragment are unchanged, and no library file is touched. --- tests/protocols/test_option_roundtrip_unit.py | 38 +++++++++++++------ 1 file changed, 27 insertions(+), 11 deletions(-) diff --git a/tests/protocols/test_option_roundtrip_unit.py b/tests/protocols/test_option_roundtrip_unit.py index a422f4bc7f..dc6dea040a 100644 --- a/tests/protocols/test_option_roundtrip_unit.py +++ b/tests/protocols/test_option_roundtrip_unit.py @@ -246,23 +246,39 @@ class Gap(NamedTuple): # Both of these used to *hang* rather than fail, which is why the cycle is # run under a deadline at all. #432 landed the progress guard in - # ``OptionField``/``ListField``, and #429's ``_SMFDPDOption`` over-read fix - # went in with it -- so ``hopopt-option/SMF_DPD`` now round-trips and has no - # entry here at all. + # ``OptionField``/``ListField`` and fixed the ``_SMFDPDOption`` sizing in + # *both* schema modules symmetrically -- 26 lines each, ``'len': (1, 8)`` to + # ``(8, 8)`` and the ``+ 2`` on the selector's ``SchemaField`` -- so + # ``hopopt-option/SMF_DPD`` now round-trips and has no entry here at all. # - # ``IPv6-Opts`` did not get the same treatment, and that asymmetry is the - # finding: the two modules are line-for-line duplicates of each other (the - # over-read is at :163 and the ``'len': (1, 8)`` namespace at :374 in *both* - # schema files), so this is one fix that was applied once where it was needed - # twice. It now raises instead of hanging, which is the guard doing its job. + # ``IPv6-Opts`` still fails, and not because it missed that fix. The two + # modules differ in exactly one line of code, and it is older than #432: + # ``ipv6_opts.SMFIdentificationBasedDPDOption`` declares a second, redundant + # ``test`` ``ForwardMatchField`` that ``hopopt``'s does not. The enclosing + # ``_SMFDPDOption`` already has one, in both modules, and it is that outer + # field the selector reads -- nothing reads the nested copy. But a + # ``ForwardMatchField`` does not consume the stream while still occupying a + # slot in ``__buffer__``, so the nested schema over-reports its own size by + # one octet, and ``OptionField`` then mis-counts the option area against the + # header. Measured on the *same* octets, ``1100080100010100``: + # + # hopopt __fields__ = [type, len, info, tid, id] len(schema) = 3 + # ipv6_opts __fields__ = [type, len, test, info, tid, id] len(schema) = 4 + # + # HOPOPT(...) -> options=[SMF_DPD, PadN] + # IPv6_Opts(...) -> ProtocolError: IPv6-Opts: invalid format + # + # Identical on 3.10.20 and 3.14.7, so this one is not interpreter-dependent. # # The deadline in the sweep stays regardless. It is protection against the # *next* non-progress defect, not against this one. 'ipv6-opts-option/SMF_DPD': Gap( 'PARSE', 'invalid format', - 'pcapkit/protocols/schema/internet/ipv6_opts.py:163 over-reads and :374 ' - "declares 'len': (1, 8) where the octet is at bits 8-15; hopopt.py got " - 'the fix in #429/#432 and ipv6_opts.py did not'), + 'pcapkit/protocols/schema/internet/ipv6_opts.py:434 -- a redundant second ' + "'test' ForwardMatchField that hopopt.py's equivalent does not have; it " + 'consumes nothing but is counted in __buffer__, so the nested schema ' + 'reports 4 octets where it read 3, and the threshold check at ' + 'pcapkit/protocols/internet/ipv6_opts.py:497 rejects the option area'), # -- IPv6-Route ----------------------------------------------------------- From 4535f72f2134b75d645131c652215b3f26226f1a Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Thu, 17 Sep 2026 18:11:07 -0400 Subject: [PATCH 4/4] tests: pin the round-trip gaps to messages that identify their site The recorded fragment is what makes an EXPECTED_FAILURES entry self-maintaining: it has to fail again *in the recorded way*, or the entry gets revisited. Seven entries recorded only `invalid format` or `invalid parameter`, and those are substrings of 205 and 3 messages respectively across 13 modules -- 31 in internet/hip.py, 26 in transport/tcp.py -- so a regression at any of them satisfied the check. Each now carries the alias and whatever bracketed code the message prints, measured from the real detail rather than guessed: - tcp-option/User_Timeout_Option TCP: [OptNo 28] invalid format - ipv6-opts-option/SMF_DPD IPv6-Opts: invalid format - hip-parameter/HIP_TRANSFORM HIPv2: [ParamNo 577] invalid parameter - hip-parameter/HOST_ID HIPv2: invalid format - httpv2-frame/PRIORITY HTTP/2: [Type 2] invalid format Five of those now match exactly one raise site: the bracketed code picks out the one handler that can print it, and the two bare-alias forms are the only sites in their module that omit the bracket at all. `fragment` also accepts a tuple, all of whose members must appear. That is for the two ipv6-route entries, whose message interpolates a bare `type` in a method that has no such parameter and so renders `[TypeNo ]` (#442). Matching the literal rendering would bake that bug into the table and turn it red when #442 is fixed, which says nothing about the round trip; the stable parts either side still narrow it to the three `[TypeNo` lines of ipv6_route.py. The assertion failure now also prints the detail it got. Verified on 3.14.7 and 3.10.20: 7 passed, 299 subtests each. Negative control -- deliberately wrong fragments in both the string and the tuple form -- fails 3 subtests, so the assertion is reached rather than vacuous. --- tests/protocols/test_option_roundtrip_unit.py | 53 ++++++++++++++----- 1 file changed, 39 insertions(+), 14 deletions(-) diff --git a/tests/protocols/test_option_roundtrip_unit.py b/tests/protocols/test_option_roundtrip_unit.py index dc6dea040a..f7830b7a65 100644 --- a/tests/protocols/test_option_roundtrip_unit.py +++ b/tests/protocols/test_option_roundtrip_unit.py @@ -94,10 +94,21 @@ class Gap(NamedTuple): #: Expected :attr:`~examples.generators.options.Outcome.status`. status: 'str' - #: Substring the failure detail must contain, or ``''`` to assert only the - #: status. Empty is used where the detail is a pair of long hex strings, or - #: where it embeds a timestamp and so is not stable between runs. - fragment: 'str' + #: Substring the failure detail must contain, or a tuple of substrings *all* + #: of which it must contain, or ``''`` to assert only the status. Empty is + #: used where the detail is a pair of long hex strings, or where it embeds a + #: timestamp and so is not stable between runs. + #: + #: Make it as specific as the message allows, because a fragment that matches + #: half the tree does not pin anything: ``'invalid format'`` alone occurs 205 + #: times across 13 modules (31 in :file:`internet/hip.py`, 26 in + #: :file:`transport/tcp.py`), so it is satisfied by a regression at any of + #: them. Prefer the alias and whatever bracketed code the message carries -- + #: ``'TCP: [OptNo 28] invalid format'`` narrows those 26 sites to the one + #: option that can print ``28``. The tuple form is for messages whose stable + #: parts are not contiguous, so that a fragment does not have to bake in a + #: rendering that is itself defective. + fragment: 'str | tuple[str, ...]' #: The defect, named with the ``file:line`` that causes it. This is the #: field that makes the entry worth keeping rather than just silencing. defect: 'str' @@ -119,7 +130,7 @@ class Gap(NamedTuple): # ``_read_mode_timeout`` checks the length exactly. Measured: the schema # packs ``1c03003c`` where a correct one is ``1c04003c``. 'tcp-option/User_Timeout_Option': Gap( - 'CONSTRUCT', 'invalid format', + 'CONSTRUCT', 'TCP: [OptNo 28] invalid format', 'pcapkit/protocols/transport/tcp.py:2506 -- _make_mode_timeout sets ' 'length=3 for a 4-octet option'), @@ -273,7 +284,7 @@ class Gap(NamedTuple): # The deadline in the sweep stays regardless. It is protection against the # *next* non-progress defect, not against this one. 'ipv6-opts-option/SMF_DPD': Gap( - 'PARSE', 'invalid format', + 'PARSE', 'IPv6-Opts: invalid format', 'pcapkit/protocols/schema/internet/ipv6_opts.py:434 -- a redundant second ' "'test' ForwardMatchField that hopopt.py's equivalent does not have; it " 'consumes nothing but is counted in __buffer__, so the nested schema ' @@ -287,12 +298,21 @@ class Gap(NamedTuple): # Schema paths divide by eight. The read-side guards then reject it, and # those guards compare the unit field against octet counts too, so the # RFC-correct value would fail as well. + # Both fragments are tuples rather than the whole message, because the + # message itself is defective: these two guards interpolate a bare ``type`` + # into an f-string in a method that has no ``type`` parameter, so the name + # resolves to the builtin and the detail reads ``[TypeNo ]`` + # (issue #442). Matching the literal rendering would pin that bug into this + # table and turn it red when #442 is fixed, which says nothing about whether + # the round trip closes. The stable parts either side of it do pin the site: + # ``[TypeNo`` occurs at exactly three lines of ``ipv6_route.py`` (:462, :506, + # :546), against 205 occurrences of ``'invalid format'`` tree-wide. 'ipv6-route-type/Source_Route': Gap( - 'CONSTRUCT', 'invalid format', + 'CONSTRUCT', ('IPv6-Route', '[TypeNo', 'invalid format'), 'pcapkit/protocols/internet/ipv6_route.py:276 -- length in octets, not ' 'in 8-octet units; guard at :461'), 'ipv6-route-type/Type_2_Routing_Header': Gap( - 'CONSTRUCT', 'invalid format', + 'CONSTRUCT', ('IPv6-Route', '[TypeNo', 'invalid format'), 'pcapkit/protocols/internet/ipv6_route.py:276; guard at :505'), # RPL fails earlier still: ``post_process`` assumes ``addresses`` is bytes, # which is true after unpacking and false while packing, where it is still @@ -359,11 +379,11 @@ class Gap(NamedTuple): # can represent even in pairs -- see HIP_COPIES in the generator for why the # pair is used at all. 'hip-parameter/HIP_TRANSFORM': Gap( - 'CONSTRUCT', 'invalid parameter', + 'CONSTRUCT', 'HIPv2: [ParamNo 577] invalid parameter', 'pcapkit/protocols/internet/hip.py:698 -- the len check; HIP_TRANSFORM ' 'packs to a length the make-side arithmetic cannot express'), 'hip-parameter/HOST_ID': Gap( - 'CONSTRUCT', 'invalid format', + 'CONSTRUCT', 'HIPv2: invalid format', 'pcapkit/protocols/internet/hip.py:698 -- HOST_ID packs to 14 octets ' 'with len=8, so it is not even 4-aligned'), @@ -391,7 +411,7 @@ class Gap(NamedTuple): # demands exactly 9. Its siblings all check payload+9 (RST_STREAM 13, # WINDOW_UPDATE 13, PING 17), so 9 looks like the outlier. 'httpv2-frame/PRIORITY': Gap( - 'CONSTRUCT', 'invalid format', + 'CONSTRUCT', 'HTTP/2: [Type 2] invalid format', 'pcapkit/protocols/application/httpv2.py:572 -- reads length != 9 for a ' 'frame make() always builds with length 14'), @@ -824,11 +844,16 @@ def test_round_trip_is_identity_or_a_recorded_gap(self) -> None: f'{outcome.status}: {outcome.detail}. If the defect is fixed, ' f'delete its EXPECTED_FAILURES entry.' ) - if gap.fragment: + fragments = ((gap.fragment,) if isinstance(gap.fragment, str) + else gap.fragment) + for fragment in fragments: + if not fragment: + continue self.assertIn( - gap.fragment, outcome.detail, + fragment, outcome.detail, f'{case.label} still fails with {gap.status}, but not in ' - f'the recorded way ({gap.defect})' + f'the recorded way ({gap.defect}); detail was ' + f'{outcome.detail!r}' ) def test_a_single_hip_parameter_cannot_be_constructed(self) -> None: