fix(link): honour a stop at LinuxSLL on its ARPHRD registry path

pcapkit/protocols/link/linux_sll.py

@@ -238,8 +238,9 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *,  #
             <LinuxSLL.__proto__>` registers for ``arphrd_type`` when it has a key
             for it, and otherwise the class :attr:`Link.__proto__
             <pcapkit.protocols.link.link.Link.__proto__>` registers for ``proto``
-            as an EtherType. A :class:`~pcapkit.protocols.misc.raw.Raw` fallback
-            records ``proto`` either way.
+            as an EtherType. Where parsing stops at this layer, either path
+            keeps the payload :class:`~pcapkit.protocols.misc.raw.Raw`, and a
+            :class:`~pcapkit.protocols.misc.raw.Raw` records ``proto`` either way.
 
         Note:
             :attr:`self.__proto__ <LinuxSLL.__proto__>` holds the
@@ -261,13 +262,11 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *,  #
         if not min(length, len(file_)):
             from pcapkit.protocols.misc.null import \
                 NoPayload as protocol  # isort: skip # pylint: disable=import-outside-toplevel
-        # NOTE: ahead of ``_sigterm``, so a registered ``arphrd_type`` is parsed
-        # even where parsing stops at this layer (#2510).
-        elif arphrd_type in self.__proto__:
-            protocol = self._lookup_next_layer(self.__proto__, arphrd_type)
         elif self._sigterm:
             from pcapkit.protocols.misc.raw import \
                 Raw as protocol  # isort: skip # pylint: disable=import-outside-toplevel
+        elif arphrd_type in self.__proto__:
+            protocol = self._lookup_next_layer(self.__proto__, arphrd_type)
         else:
             protocol = self._lookup_next_layer(Link.__proto__, proto)
 

tests/protocols/link/test_linux_sll_arphrd_registry_unit.py

@@ -4,6 +4,7 @@
 
 Covers :attr:`LinuxSLL.__proto__ <pcapkit.protocols.link.linux_sll.LinuxSLL.__proto__>`
 as shipped, dispatch through it and through the EtherType path from both versions,
+a parse stopping at either version on either path (#2532),
 the ``__index__`` of the classes it reaches, and the registrars of it and of
 :class:`~pcapkit.protocols.link.frad.FRAD`'s and
 :class:`~pcapkit.protocols.link.netlink.Netlink`'s registries.
@@ -138,25 +139,73 @@ def test_an_empty_payload_is_no_payload_on_either_path(self) -> None:
                     raw = header(0x0800, arphrd)
                     self.assertIsInstance(version(raw, len(raw)).payload, NoPayload)
 
-    def test_a_stop_at_this_layer_still_parses_a_registered_arphrd_type(self) -> None:
-        """The order the NOTE in ``LinuxSLL._import_next_layer`` records.
+    def _stops(self, version):  # type: ignore[no-untyped-def]
+        """Each spelling of a stop at ``version``: its layer, family and own name or class."""
+        from pcapkit.protocols.link.linux_sll import LinuxSLL
+
+        return ({'layer': 'Link'}, {'layer': 'link'}, {'protocol': 'LinuxSLL'},
+                {'protocol': version.__name__}, {'protocol': LinuxSLL}, {'protocol': version})
 
-        Stopping at the link layer or at the version keeps an EtherType payload
-        :class:`~pcapkit.protocols.misc.raw.Raw`, while a registered
-        ``arphrd_type`` is still dispatched.
-        """
+    def test_a_stop_at_this_layer_keeps_a_registered_arphrd_type_raw(self) -> None:
+        """The registry path stops as the EtherType path does (#2532)."""
         from pcapkit.const.linux.arphrd import ARPHRD
+        from pcapkit.protocols.misc.raw import Raw
+
+        for version, header, alias in self._versions():
+            for name, (payload, _) in CASES.items():
+                for stop in self._stops(version):
+                    with self.subTest(version=version.__name__, arphrd=name, stop=stop):
+                        raw = header(0x0800, getattr(ARPHRD, name)) + payload
+                        parsed = version(raw, len(raw), **stop)
+                        self.assertEqual(parsed.protochain.chain, f'{alias}:Raw')
+                        self.assertIsInstance(parsed.payload, Raw)
+                        self.assertEqual(parsed.payload.data, payload)
+                        self.assertEqual(parsed.payload.info.protocol, 0x0800)
+                        self.assertEqual(parsed.data, raw)
+                        self.assertEqual(version.from_data(parsed.info).data, raw)
+
+                        # Without a length, the whole payload is the next layer's.
+                        again = parsed._import_next_layer(parsed.protocol)
+                        self.assertIsInstance(again, Raw)
+                        self.assertEqual(again.data, payload)
+
+    def test_a_stop_at_this_layer_keeps_an_ethertype_raw(self) -> None:
+        """The Raw is named after the EtherType it would have dispatched on."""
+        from pcapkit.const.reg.ethertype import EtherType
+        from pcapkit.protocols.misc.raw import Raw
 
         for version, header, alias in self._versions():
-            for stop in ({'layer': 'Link'}, {'protocol': version.__name__}):
+            for stop in self._stops(version):
                 with self.subTest(version=version.__name__, stop=stop):
                     raw = header(0x0800, 1) + IPV4
-                    self.assertEqual(version(raw, len(raw), **stop).protochain.chain,
+                    parsed = version(raw, len(raw), **stop)
+                    self.assertEqual(parsed.protochain.chain,
                                      f'{alias}:Internet_Protocol_version_4')
+                    self.assertIsInstance(parsed.payload, Raw)
+                    self.assertEqual(parsed.payload.data, IPV4)
+                    self.assertIs(parsed.payload.info.protocol,
+                                  EtherType.Internet_Protocol_version_4)
+                    self.assertEqual(parsed.data, raw)
+                    self.assertEqual(version.from_data(parsed.info).data, raw)
+
+    def test_a_stop_at_another_protocol_still_dispatches_here(self) -> None:
+        """The other version's name and a protocol further in do not stop this layer."""
+        from pcapkit.const.linux.arphrd import ARPHRD
+
+        frad = CASES['FRAD'][0]
+        for version, header, alias in self._versions():
+            other = 'LinuxSLL2' if version.__name__ == 'LinuxSLL1' else 'LinuxSLL1'
+            for stop, chain in (({'protocol': other}, 'FRAD:IPv4:UDP'),
+                                ({'protocol': 'FRAD'}, 'FRAD:Raw')):
+                with self.subTest(version=version.__name__, stop=stop):
+                    raw = header(0x0800, ARPHRD.FRAD) + frad
+                    parsed = version(raw, len(raw), **stop)
+                    self.assertEqual(parsed.protochain.chain, f'{alias}:{chain}')
+                    self.assertEqual(version.from_data(parsed.info).data, raw)
 
-                    raw = header(0x0800, ARPHRD.NETLINK) + CASES['NETLINK'][0]
+                    raw = header(0x0800, 1) + IPV4
                     self.assertEqual(version(raw, len(raw), **stop).protochain.chain,
-                                     f'{alias}:Netlink:Raw')
+                                     f'{alias}:IPv4:UDP')
 
     def test_index_is_the_arphrd_type_that_dispatches_here(self) -> None:
         from pcapkit.const.linux.arphrd import ARPHRD

tests/protocols/link/test_linux_sll_selector_unit.py

@@ -21,16 +21,20 @@
         + bytes.fromhex('3039003500080000'))
 
 
-def _header_v1(protocol_type: int = 0x0800) -> bytes:
+#: A FRAD frame, Q.922 address and UI control, carrying :data:`IPV4` by NLPID ``0xCC``.
+FRAD = bytes.fromhex('0c2103cc') + IPV4
+
+
+def _header_v1(protocol_type: int = 0x0800, arphrd_type: int = 1) -> bytes:
     """Pack a minimal ``LINUX_SLL`` (v1) header."""
     link_addr = bytes.fromhex('001122334455').ljust(8, b'\x00')
-    return struct.pack('>HHH', 0, 1, 6) + link_addr + struct.pack('>H', protocol_type)
+    return struct.pack('>HHH', 0, arphrd_type, 6) + link_addr + struct.pack('>H', protocol_type)
 
 
-def _header_v2(protocol_type: int = 0x0800) -> bytes:
+def _header_v2(protocol_type: int = 0x0800, arphrd_type: int = 1) -> bytes:
     """Pack a minimal ``LINUX_SLL2`` header."""
     link_addr = bytes.fromhex('001122334455').ljust(8, b'\x00')
-    return struct.pack('>HHIHBB', protocol_type, 0, 1, 1, 0, 6) + link_addr
+    return struct.pack('>HHIHBB', protocol_type, 0, 1, arphrd_type, 0, 6) + link_addr
 
 
 def _pcap(records: 'list[bytes]', linktype: int) -> bytes:
@@ -156,6 +160,20 @@ def test_extract_version_name_stops_only_that_version(self) -> None:
         self.assertEqual(_extract(_header_v1() + IPV4, LinkType.LINUX_SLL, 'LinuxSLL2'),
                          'LINUX_SLL1:IPv4:UDP')
 
+    def test_extract_stops_a_registered_arphrd_type_alike(self) -> None:
+        """A FRAD frame stops where an EtherType one does (#2532)."""
+        from pcapkit.const.linux.arphrd import ARPHRD
+        from pcapkit.const.reg.linktype import LinkType
+
+        for linktype, header, alias, own, other in (
+                (LinkType.LINUX_SLL, _header_v1, 'LINUX_SLL1', 'LinuxSLL1', 'LinuxSLL2'),
+                (LinkType.LINUX_SLL2, _header_v2, 'LINUX_SLL2', 'LinuxSLL2', 'LinuxSLL1')):
+            record = header(arphrd_type=ARPHRD.FRAD) + FRAD
+            for protocol in ('LinuxSLL', own):
+                with self.subTest(alias=alias, protocol=protocol):
+                    self.assertEqual(_extract(record, linktype, protocol), f'{alias}:Raw')
+            self.assertEqual(_extract(record, linktype, other), f'{alias}:FRAD:IPv4:UDP')
+
 
 if __name__ == '__main__':
     unittest.main()