test: generate FRAD's NLPID map and the ICMP type registries

examples/generators/dispatch.py

@@ -615,6 +615,36 @@ def _loopback_build(family: 'int') -> 'Any':
     return _extract_first(_pcap_bytes((_loopback(family, payload),), linktype=0), '.pcap')
 
 
+###############################################################################
+# FRAD -- NLPID.
+###############################################################################
+
+
+def _frad_registry() -> 'Any':
+    from pcapkit.protocols.link.frad import FRAD
+    return FRAD.__proto__
+
+
+def _frad_payload(nlpid: 'int') -> 'bytes':
+    """The registered target's own wire bytes for ``nlpid``."""
+    if nlpid == 0xCC:
+        return _ip4(6, _tcp(9999))
+    if nlpid == 0x8E:
+        return _ip6(6, _tcp(9999))
+    raise LookupError(f'frad: no payload builder for NLPID {nlpid!r}')  # pragma: no cover
+
+
+def _frad_build(nlpid: 'int') -> 'Any':
+    """A direct-NLPID FRAD frame, reached as LinuxSLL dispatches it in a capture."""
+    from pcapkit.const.linux.arphrd import ARPHRD
+    from pcapkit.protocols.link.frad import FRAD
+    from pcapkit.protocols.link.linux_sll1 import LinuxSLL1
+
+    frad = bytes(FRAD(nlpid=int(nlpid), payload=_frad_payload(nlpid)))
+    record = bytes(LinuxSLL1(arphrd_type=ARPHRD.FRAD, protocol_type=0, payload=frad))
+    return _extract_first(_pcap_bytes((record,), linktype=113), '.pcap')
+
+
 ###############################################################################
 # The families, the cases, and the probe.
 ###############################################################################
@@ -642,7 +672,8 @@ def _link_enum() -> 'Any':
 
 
 #: Every family this module exercises, one per ``__proto__`` table named in
-#: GitHub issue #496, and the loopback table #1574 added.
+#: GitHub issue #496, the loopback table #1574 added, and FRAD's NLPID table
+#: (#2560).
 FAMILIES = (
     Family('link', _link_registry, _link_build, _link_enum),
     Family('internet', _internet_registry, _internet_build),
@@ -652,6 +683,7 @@ def _link_enum() -> 'Any':
     Family('pcap-frame', _pcap_frame_registry, _pcap_frame_build),
     Family('pcapng-frame', _pcapng_registry, _pcapng_build),
     Family('loopback', _loopback_registry, _loopback_build),
+    Family('frad', _frad_registry, _frad_build),
 )
 
 #: The families keyed by label, for a caller -- :func:`probe` -- that wants
@@ -762,6 +794,10 @@ def _link_enum() -> 'Any':
     'loopback/24': ('pcapkit.protocols.internet.ipv6', 'IPv6'),
     'loopback/28': ('pcapkit.protocols.internet.ipv6', 'IPv6'),
     'loopback/30': ('pcapkit.protocols.internet.ipv6', 'IPv6'),
+
+    # -- FRAD.__proto__ (NLPID, keyed in decimal: 0x8E, 0xCC) ------------------
+    'frad/142': ('pcapkit.protocols.internet.ipv6', 'IPv6'),
+    'frad/204': ('pcapkit.protocols.internet.ipv4', 'IPv4'),
 }
 
 

examples/generators/options.py

@@ -1165,6 +1165,103 @@ def _dccp_rebuild(options: 'Any') -> 'Any':
     return DCCP(options=options, **DCCP_BASE)
 
 
+###############################################################################
+# ICMP and ICMPv6 -- message types
+###############################################################################
+
+#: The invoking packet an ICMP error message quotes: an IPv4 header and the
+#: first eight octets of its UDP datagram [:rfc:`792`].
+ICMPV4_QUOTED = (bytes.fromhex('4500001c00010000401100000a0000010a000002')
+                 + bytes.fromhex('3039003500080000'))
+
+#: The invoking packet an ICMPv6 error message quotes: an IPv6 header and its
+#: UDP header [:rfc:`4443#section-3`].
+ICMPV6_QUOTED = (bytes.fromhex('6000000000081140') + bytes(15) + b'\x01' + bytes(15) + b'\x02'
+                 + bytes.fromhex('3039003500080000'))
+
+#: Echo arguments shared by both versions. The defaults are all zero and empty,
+#: which cannot tell a swapped ``id`` and ``seq``, or a dropped ``data``, apart.
+ICMP_ECHO = {'id': 0x1234, 'seq': 0x5678, 'data': b'ping'}
+
+
+def _icmpv4_registry() -> 'Any':
+    from pcapkit.protocols.internet.icmpv4 import ICMPv4
+    return ICMPv4.__type__
+
+
+def _icmpv4_overrides() -> 'dict[Any, dict[str, Any]]':
+    from pcapkit.const.icmp.type import Type as Enum_Type
+    return {
+        Enum_Type.Echo_Reply: dict(ICMP_ECHO),
+        Enum_Type.Echo: dict(ICMP_ECHO),
+        Enum_Type.Destination_Unreachable: {'data': ICMPV4_QUOTED},
+        Enum_Type.Time_Exceeded: {'data': ICMPV4_QUOTED},
+        # Octet 8 of the quoted header is its TTL.
+        Enum_Type.Parameter_Problem: {'pointer': 8, 'data': ICMPV4_QUOTED},
+    }
+
+
+def _icmpv4_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any':
+    from pcapkit.protocols.internet.icmpv4 import ICMPv4
+
+    # ``body`` has to be a mapping for ``make`` to dispatch on ``__type__``;
+    # a ``bytes`` body is parsed against the schema registry instead.
+    return ICMPv4(type=code, code=0, body=dict(kwargs))
+
+
+def _icmpv4_parse(octets: 'bytes') -> 'Any':
+    from pcapkit.protocols.internet.icmpv4 import ICMPv4
+    return ICMPv4(octets, len(octets))
+
+
+def _icmp_extract(parsed: 'Any') -> 'Any':
+    return parsed.info
+
+
+def _icmpv4_rebuild(message: 'Any') -> 'Any':
+    from pcapkit.protocols.internet.icmpv4 import ICMPv4
+    return ICMPv4(type=message.type, code=message.code, checksum=message.checksum,
+                  body=message)
+
+
+def _icmpv6_registry() -> 'Any':
+    from pcapkit.protocols.internet.icmpv6 import ICMPv6
+    return ICMPv6.__type__
+
+
+def _icmpv6_overrides() -> 'dict[Any, dict[str, Any]]':
+    from pcapkit.const.icmpv6.type import Type as Enum_Type
+    return {
+        Enum_Type.Destination_Unreachable: {'data': ICMPV6_QUOTED},
+        # The IPv6 minimum link MTU [RFC 8200 section 5].
+        Enum_Type.Packet_Too_Big: {'mtu': 1280, 'data': ICMPV6_QUOTED},
+        Enum_Type.Time_Exceeded: {'data': ICMPV6_QUOTED},
+        # Octet 6 of the quoted header is its Next Header.
+        Enum_Type.Parameter_Problem: {'pointer': 6, 'data': ICMPV6_QUOTED},
+        Enum_Type.Echo_Request: dict(ICMP_ECHO),
+        Enum_Type.Echo_Reply: dict(ICMP_ECHO),
+    }
+
+
+def _icmpv6_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any':
+    from pcapkit.protocols.internet.icmpv6 import ICMPv6
+
+    # With no enclosing IPv6 layer to supply the pseudo-header, the checksum is
+    # left zero, and the rebuild passes the parsed zero back.
+    return ICMPv6(type=code, code=0, body=dict(kwargs))
+
+
+def _icmpv6_parse(octets: 'bytes') -> 'Any':
+    from pcapkit.protocols.internet.icmpv6 import ICMPv6
+    return ICMPv6(octets, len(octets))
+
+
+def _icmpv6_rebuild(message: 'Any') -> 'Any':
+    from pcapkit.protocols.internet.icmpv6 import ICMPv6
+    return ICMPv6(type=message.type, code=message.code, checksum=message.checksum,
+                  body=message)
+
+
 ###############################################################################
 # HTTP/2 -- frames
 ###############################################################################
@@ -1660,6 +1757,14 @@ def _pcapng_secrets_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any':
     Family('hip-parameter', _hip_registry, _hip_overrides,
            _hip_build, _hip_parse, _hip_extract, _hip_rebuild,
            capture='options-internet.pcap', envelope='ipv6', proto=139),
+    # NOTE: The ICMP families write no capture; their wire form is pinned against
+    # RFC 792 and RFC 4443 by test_option_generator_icmp_2560_unit instead.
+    Family('icmpv4-type', _icmpv4_registry, _icmpv4_overrides,
+           _icmpv4_build, _icmpv4_parse, _icmp_extract, _icmpv4_rebuild,
+           capture=None, envelope=None),
+    Family('icmpv6-type', _icmpv6_registry, _icmpv6_overrides,
+           _icmpv6_build, _icmpv6_parse, _icmp_extract, _icmpv6_rebuild,
+           capture=None, envelope=None),
     # The PCAP-NG families write no capture -- see the section comment
     # above ``PCAPNG_TIME`` for why the construction API's defaults do not
     # make a reproducible one, and what already covers that space instead.

tests/protocols/test_dispatch_registry_unit.py

@@ -4,7 +4,8 @@
 GitHub issue #496: the ``__proto__`` dispatch registries decide which
 :class:`~pcapkit.protocols.protocol.Protocol` subclass parses the next layer.
 #496 names the tables, joined by #904's
-``Internet.__proto__[TransType.Shim6]`` and #1574's loopback table.
+``Internet.__proto__[TransType.Shim6]``, #1574's loopback table and #2560's
+FRAD table.
 :file:`test_registry_runtime.py` checks that a table entry *resolves* to the
 right class object, and
 :file:`test_dispatch_bindings_unit.py` checks that a hand-picked set of codes
@@ -215,7 +216,7 @@ def test_registry_matches_pinned_target(self) -> None:
                 )
 
     def test_cases_cover_every_table_named_in_the_issue(self) -> None:
-        """The enumeration finds every entry, across the tables of #496 and #1574's loopback table.
+        """The enumeration finds every entry, across the tables of #496, #1574 and #2560.
 
         A guard on the shape of the result rather than on any one case: if the
         registries grow or shrink without this module noticing, the per-family
@@ -229,20 +230,22 @@ def test_cases_cover_every_table_named_in_the_issue(self) -> None:
         # address families of Loopback.__proto__, #2438's
         # Frame.__proto__[LinkType.LINUX_SLL]/[LinkType.LINUX_SLL2] in both the
         # pcap and pcapng frame tables, #2439's IGMP, #2436's
-        # Internet.__proto__[TransType.ICMP] and [TransType.IPv6_ICMP], and
-        # #2489's Internet.__proto__[TransType.DCCP].
-        self.assertEqual(len(cases), 57,
-                         'expected exactly 57 entries across the seven __proto__ '
-                         'tables named in GitHub issue #496 and the loopback table '
-                         'of #1574; a different count means a registry changed '
-                         'shape and this module was not updated to match')
+        # Internet.__proto__[TransType.ICMP] and [TransType.IPv6_ICMP], #2489's
+        # Internet.__proto__[TransType.DCCP], and #2560's two NLPIDs of
+        # FRAD.__proto__.
+        self.assertEqual(len(cases), 59,
+                         'expected exactly 59 entries across the seven __proto__ '
+                         'tables named in GitHub issue #496, the loopback table '
+                         'of #1574 and the FRAD table of #2560; a different count '
+                         'means a registry changed shape and this module was not '
+                         'updated to match')
 
         by_family = {}  # type: dict[str, int]
         for case in cases:
             by_family[case.family] = by_family.get(case.family, 0) + 1
         self.assertEqual(by_family, {
             'link': 7, 'internet': 21, 'tcp': 5, 'udp': 4, 'sctp': 2,
-            'pcap-frame': 7, 'pcapng-frame': 7, 'loopback': 4,
+            'pcap-frame': 7, 'pcapng-frame': 7, 'loopback': 4, 'frad': 2,
         })
 
     def test_dispatch_reaches_target_or_is_a_recorded_degrade(self) -> None: