diff --git a/pcapkit/protocols/data/__init__.py b/pcapkit/protocols/data/__init__.py index 00c027cc7b..f818d81927 100644 --- a/pcapkit/protocols/data/__init__.py +++ b/pcapkit/protocols/data/__init__.py @@ -94,8 +94,9 @@ 'HIP_RouteDstParameter', 'HIP_HIPTransportModeParameter', 'HIP_HIPMACParameter', 'HIP_HIPMAC2Parameter', 'HIP_HIPSignature2Parameter', 'HIP_HIPSignatureParameter', 'HIP_EchoRequestUnsignedParameter', 'HIP_EchoResponseUnsignedParameter', 'HIP_RelayFromParameter', - 'HIP_RelayToParameter', 'HIP_RouteViaParameter', 'HIP_FromParameter', - 'HIP_RVSHMACParameter', 'HIP_RelayHMACParameter', + 'HIP_RelayToParameter', 'HIP_OverlayTTLParameter', 'HIP_RouteViaParameter', + 'HIP_FromParameter', 'HIP_RVSHMACParameter', 'HIP_ViaRVSParameter', + 'HIP_RelayHMACParameter', # Hop-by-Hop Options 'HOPOPT', diff --git a/pcapkit/protocols/data/internet/__init__.py b/pcapkit/protocols/data/internet/__init__.py index 1f89dc2f17..421c83561f 100644 --- a/pcapkit/protocols/data/internet/__init__.py +++ b/pcapkit/protocols/data/internet/__init__.py @@ -47,6 +47,7 @@ NATTraversalModeParameter as HIP_NATTraversalModeParameter from pcapkit.protocols.data.internet.hip import NotificationParameter as HIP_NotificationParameter from pcapkit.protocols.data.internet.hip import OverlayIDParameter as HIP_OverlayIDParameter +from pcapkit.protocols.data.internet.hip import OverlayTTLParameter as HIP_OverlayTTLParameter from pcapkit.protocols.data.internet.hip import PayloadMICParameter as HIP_PayloadMICParameter from pcapkit.protocols.data.internet.hip import PuzzleParameter as HIP_PuzzleParameter from pcapkit.protocols.data.internet.hip import R1CounterParameter as HIP_R1CounterParameter @@ -70,6 +71,7 @@ from pcapkit.protocols.data.internet.hip import \ TransportFormatListParameter as HIP_TransportFormatListParameter from pcapkit.protocols.data.internet.hip import UnassignedParameter as HIP_UnassignedParameter +from pcapkit.protocols.data.internet.hip import ViaRVSParameter as HIP_ViaRVSParameter # Hop-by-Hop Options from pcapkit.protocols.data.internet.hopopt import HOPOPT @@ -396,8 +398,9 @@ 'HIP_RouteDstParameter', 'HIP_HIPTransportModeParameter', 'HIP_HIPMACParameter', 'HIP_HIPMAC2Parameter', 'HIP_HIPSignature2Parameter', 'HIP_HIPSignatureParameter', 'HIP_EchoRequestUnsignedParameter', 'HIP_EchoResponseUnsignedParameter', 'HIP_RelayFromParameter', - 'HIP_RelayToParameter', 'HIP_RouteViaParameter', 'HIP_FromParameter', - 'HIP_RVSHMACParameter', 'HIP_RelayHMACParameter', + 'HIP_RelayToParameter', 'HIP_OverlayTTLParameter', 'HIP_RouteViaParameter', + 'HIP_FromParameter', 'HIP_RVSHMACParameter', 'HIP_ViaRVSParameter', + 'HIP_RelayHMACParameter', # Hop-by-Hop Options 'HOPOPT', diff --git a/pcapkit/protocols/data/internet/hip.py b/pcapkit/protocols/data/internet/hip.py index 1d621a2d4d..081733912b 100644 --- a/pcapkit/protocols/data/internet/hip.py +++ b/pcapkit/protocols/data/internet/hip.py @@ -55,8 +55,9 @@ 'RouteDstParameter', 'HIPTransportModeParameter', 'HIPMACParameter', 'HIPMAC2Parameter', 'HIPSignature2Parameter', 'HIPSignatureParameter', 'EchoRequestUnsignedParameter', 'EchoResponseUnsignedParameter', 'RelayFromParameter', - 'RelayToParameter', 'RouteViaParameter', 'FromParameter', - 'RVSHMACParameter', 'RelayHMACParameter', + 'RelayToParameter', 'OverlayTTLParameter', 'RouteViaParameter', + 'FromParameter', 'RVSHMACParameter', 'ViaRVSParameter', + 'RelayHMACParameter', ] diff --git a/pcapkit/protocols/data/internet/hopopt.py b/pcapkit/protocols/data/internet/hopopt.py index d5fb1afebb..4c569f874b 100644 --- a/pcapkit/protocols/data/internet/hopopt.py +++ b/pcapkit/protocols/data/internet/hopopt.py @@ -149,7 +149,7 @@ class CALIPSOOption(Option): if TYPE_CHECKING: #: Compartment bitmap. - cmpt_bitmap: 'bytes' + cmpt_bitmap: 'tuple[int, ...]' def __init__(self, type: 'Enum_Option', action: 'int', change: 'bool', length: 'int', domain: 'int', cmpt_len: 'int', level: 'int', checksum: 'bytes', pad: 'bytes') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,redefined-builtin,multiple-statements,line-too-long diff --git a/pcapkit/protocols/data/internet/ipv4.py b/pcapkit/protocols/data/internet/ipv4.py index f10d1782e6..c7cf2f64f2 100644 --- a/pcapkit/protocols/data/internet/ipv4.py +++ b/pcapkit/protocols/data/internet/ipv4.py @@ -12,7 +12,7 @@ if TYPE_CHECKING: from datetime import timedelta from ipaddress import IPv4Address - from typing import Any, Optional + from typing import Any from typing_extensions import Literal @@ -305,12 +305,12 @@ class RROption(Option): #: Pointer. pointer: 'int' #: Route, i.e. the address slots before the pointer. - route: 'Optional[tuple[IPv4Address, ...]]' + route: 'tuple[IPv4Address, ...]' #: Address slots at or beyond the pointer. remaining: 'tuple[IPv4Address, ...]' if TYPE_CHECKING: - def __init__(self, code: 'OptionNumber', length: 'int', type: 'OptionType', pointer: 'int', route: 'Optional[tuple[IPv4Address, ...]]', remaining: 'tuple[IPv4Address, ...]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, code: 'OptionNumber', length: 'int', type: 'OptionType', pointer: 'int', route: 'tuple[IPv4Address, ...]', remaining: 'tuple[IPv4Address, ...]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final @@ -337,12 +337,12 @@ class SSROption(Option): #: Pointer. pointer: 'int' #: Route, i.e. the address slots before the pointer. - route: 'Optional[tuple[IPv4Address, ...]]' + route: 'tuple[IPv4Address, ...]' #: Address slots at or beyond the pointer. remaining: 'tuple[IPv4Address, ...]' if TYPE_CHECKING: - def __init__(self, code: 'OptionNumber', length: 'int', type: 'OptionType', pointer: 'int', route: 'Optional[tuple[IPv4Address, ...]]', remaining: 'tuple[IPv4Address, ...]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, code: 'OptionNumber', length: 'int', type: 'OptionType', pointer: 'int', route: 'tuple[IPv4Address, ...]', remaining: 'tuple[IPv4Address, ...]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final diff --git a/pcapkit/protocols/data/internet/ipv6_opts.py b/pcapkit/protocols/data/internet/ipv6_opts.py index 7c3ef8b002..0060af563a 100644 --- a/pcapkit/protocols/data/internet/ipv6_opts.py +++ b/pcapkit/protocols/data/internet/ipv6_opts.py @@ -154,7 +154,7 @@ class CALIPSOOption(Option): if TYPE_CHECKING: #: Compartment bitmap. - cmpt_bitmap: 'bytes' + cmpt_bitmap: 'tuple[int, ...]' def __init__(self, type: 'Enum_Option', action: 'int', change: 'bool', length: 'int', domain: 'int', cmpt_len: 'int', level: 'int', checksum: 'bytes', pad: 'bytes') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,redefined-builtin,multiple-statements,line-too-long diff --git a/pcapkit/protocols/data/misc/pcapng.py b/pcapkit/protocols/data/misc/pcapng.py index 983798f847..6e72bd220b 100644 --- a/pcapkit/protocols/data/misc/pcapng.py +++ b/pcapkit/protocols/data/misc/pcapng.py @@ -807,6 +807,9 @@ class SystemdJournalExportBlock(PCAPNG): data: 'tuple[OrderedMultiInfo[str, str | bytes], ...]' if TYPE_CHECKING: + #: Malformed entry kept as captured, present only for such a block. + entry_raw: 'bytes' + def __init__(self, type: 'Enum_BlockType', length: 'int', data: 'tuple[OrderedMultiDict[str, str | bytes], ...]') -> 'None': ... diff --git a/pcapkit/protocols/internet/hip.py b/pcapkit/protocols/internet/hip.py index a32f909306..509fc40f76 100644 --- a/pcapkit/protocols/internet/hip.py +++ b/pcapkit/protocols/internet/hip.py @@ -3021,7 +3021,7 @@ def _read_param_via_rvs(self, schema: 'Schema_ViaRVSParameter', *, version: 'int ) return via_rvs - def _read_param_relay_hmac(self, schema: 'Schema_RelayHMACParameter', version: 'int', # pylint: disable=unused-argument + def _read_param_relay_hmac(self, schema: 'Schema_RelayHMACParameter', *, version: 'int', # pylint: disable=unused-argument options: 'Parameter') -> 'Data_RelayHMACParameter': # pylint: disable=unused-argument """Read HIP ``RELAY_HMAC`` parameter. diff --git a/pcapkit/protocols/link/arp.py b/pcapkit/protocols/link/arp.py index 479f3225ab..4d67155c6d 100644 --- a/pcapkit/protocols/link/arp.py +++ b/pcapkit/protocols/link/arp.py @@ -394,7 +394,7 @@ def _read_proto_resolve(self, addr: 'bytes', ptype: 'int') -> 'str | IPv4Address return ipaddress.IPv6Address(addr) return addr.hex() - def _make_addr_resolve(self, addr: 'str | bytes', htype: 'int') -> 'bytes': + def _make_addr_resolve(self, addr: 'str | bytes | bytearray', htype: 'int') -> 'bytes': """Resolve hardware address according to protocol. Arguments: diff --git a/pcapkit/protocols/misc/null.py b/pcapkit/protocols/misc/null.py index 440b1c24c3..fb689b4ca5 100644 --- a/pcapkit/protocols/misc/null.py +++ b/pcapkit/protocols/misc/null.py @@ -64,6 +64,13 @@ def protocol(self) -> 'NoReturn': """ raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute 'protocol'") + # protocol chain of current instance + @property + def protochain(self) -> 'None': # type: ignore[override] + """Protocol chain of current instance, which an empty payload does not + have.""" + return None + ########################################################################## # Methods. ########################################################################## diff --git a/pcapkit/protocols/schema/__init__.py b/pcapkit/protocols/schema/__init__.py index a2e85a6dc0..601d84cc57 100644 --- a/pcapkit/protocols/schema/__init__.py +++ b/pcapkit/protocols/schema/__init__.py @@ -51,8 +51,9 @@ 'HIP_RouteDstParameter', 'HIP_HIPTransportModeParameter', 'HIP_HIPMACParameter', 'HIP_HIPMAC2Parameter', 'HIP_HIPSignature2Parameter', 'HIP_HIPSignatureParameter', 'HIP_EchoRequestUnsignedParameter', 'HIP_EchoResponseUnsignedParameter', 'HIP_RelayFromParameter', - 'HIP_RelayToParameter', 'HIP_RouteViaParameter', 'HIP_FromParameter', - 'HIP_RVSHMACParameter', 'HIP_RelayHMACParameter', + 'HIP_RelayToParameter', 'HIP_OverlayTTLParameter', 'HIP_RouteViaParameter', + 'HIP_FromParameter', 'HIP_RVSHMACParameter', 'HIP_ViaRVSParameter', + 'HIP_RelayHMACParameter', 'HOPOPT', 'HOPOPT_SMFDPDOption', 'HOPOPT_QuickStartOption', 'HOPOPT_UnassignedOption', 'HOPOPT_PadOption', 'HOPOPT_TunnelEncapsulationLimitOption', diff --git a/pcapkit/protocols/schema/internet/__init__.py b/pcapkit/protocols/schema/internet/__init__.py index a5a8e08013..4a3dc3fe12 100644 --- a/pcapkit/protocols/schema/internet/__init__.py +++ b/pcapkit/protocols/schema/internet/__init__.py @@ -52,6 +52,7 @@ NATTraversalModeParameter as HIP_NATTraversalModeParameter from pcapkit.protocols.schema.internet.hip import NotificationParameter as HIP_NotificationParameter from pcapkit.protocols.schema.internet.hip import OverlayIDParameter as HIP_OverlayIDParameter +from pcapkit.protocols.schema.internet.hip import OverlayTTLParameter as HIP_OverlayTTLParameter from pcapkit.protocols.schema.internet.hip import PayloadMICParameter as HIP_PayloadMICParameter from pcapkit.protocols.schema.internet.hip import PuzzleParameter as HIP_PuzzleParameter from pcapkit.protocols.schema.internet.hip import R1CounterParameter as HIP_R1CounterParameter @@ -76,6 +77,7 @@ from pcapkit.protocols.schema.internet.hip import \ TransportFormatListParameter as HIP_TransportFormatListParameter from pcapkit.protocols.schema.internet.hip import UnassignedParameter as HIP_UnassignedParameter +from pcapkit.protocols.schema.internet.hip import ViaRVSParameter as HIP_ViaRVSParameter # Hop-by-Hop Options from pcapkit.protocols.schema.internet.hopopt import HOPOPT @@ -405,8 +407,9 @@ 'HIP_RouteDstParameter', 'HIP_HIPTransportModeParameter', 'HIP_HIPMACParameter', 'HIP_HIPMAC2Parameter', 'HIP_HIPSignature2Parameter', 'HIP_HIPSignatureParameter', 'HIP_EchoRequestUnsignedParameter', 'HIP_EchoResponseUnsignedParameter', 'HIP_RelayFromParameter', - 'HIP_RelayToParameter', 'HIP_RouteViaParameter', 'HIP_FromParameter', - 'HIP_RVSHMACParameter', 'HIP_RelayHMACParameter', + 'HIP_RelayToParameter', 'HIP_OverlayTTLParameter', 'HIP_RouteViaParameter', + 'HIP_FromParameter', 'HIP_RVSHMACParameter', 'HIP_ViaRVSParameter', + 'HIP_RelayHMACParameter', # Hop-by-Hop Options 'HOPOPT', diff --git a/pcapkit/protocols/schema/internet/hip.py b/pcapkit/protocols/schema/internet/hip.py index b39455261f..a4f8332267 100644 --- a/pcapkit/protocols/schema/internet/hip.py +++ b/pcapkit/protocols/schema/internet/hip.py @@ -53,8 +53,9 @@ 'RouteDstParameter', 'HIPTransportModeParameter', 'HIPMACParameter', 'HIPMAC2Parameter', 'HIPSignature2Parameter', 'HIPSignatureParameter', 'EchoRequestUnsignedParameter', 'EchoResponseUnsignedParameter', 'RelayFromParameter', - 'RelayToParameter', 'RouteViaParameter', 'FromParameter', - 'RVSHMACParameter', 'RelayHMACParameter', + 'RelayToParameter', 'OverlayTTLParameter', 'RouteViaParameter', + 'FromParameter', 'RVSHMACParameter', 'ViaRVSParameter', + 'RelayHMACParameter', ] if TYPE_CHECKING: diff --git a/pcapkit/protocols/schema/internet/hopopt.py b/pcapkit/protocols/schema/internet/hopopt.py index b25a6cb7ca..50eed88986 100644 --- a/pcapkit/protocols/schema/internet/hopopt.py +++ b/pcapkit/protocols/schema/internet/hopopt.py @@ -413,7 +413,7 @@ class SMFHashBasedDPDOption(SMFDPDOption, code=Enum_SMFDPDMode.H_DPD): #: Hash assist value (HAV). hav: 'bytes' = BytesField(length=lambda pkt: pkt['len']) - def post_process(self, packet: 'dict[str, Any]') -> 'SMFIdentificationBasedDPDOption': + def post_process(self, packet: 'dict[str, Any]') -> 'SMFHashBasedDPDOption': """Revise ``schema`` data after unpacking process. Args: @@ -423,7 +423,7 @@ def post_process(self, packet: 'dict[str, Any]') -> 'SMFIdentificationBasedDPDOp Revised schema. """ - ret = super().post_process(packet) # type: SMFIdentificationBasedDPDOption + ret = super().post_process(packet) # type: SMFHashBasedDPDOption ret.mode = Enum_SMFDPDMode.H_DPD return ret diff --git a/pcapkit/protocols/schema/internet/ipv6_opts.py b/pcapkit/protocols/schema/internet/ipv6_opts.py index a4664aec42..4c0c6f30c9 100644 --- a/pcapkit/protocols/schema/internet/ipv6_opts.py +++ b/pcapkit/protocols/schema/internet/ipv6_opts.py @@ -420,7 +420,7 @@ class SMFHashBasedDPDOption(SMFDPDOption, code=Enum_SMFDPDMode.H_DPD): #: Hash assist value (HAV). hav: 'bytes' = BytesField(length=lambda pkt: pkt['len']) - def post_process(self, packet: 'dict[str, Any]') -> 'SMFIdentificationBasedDPDOption': + def post_process(self, packet: 'dict[str, Any]') -> 'SMFHashBasedDPDOption': """Revise ``schema`` data after unpacking process. Args: @@ -430,7 +430,7 @@ def post_process(self, packet: 'dict[str, Any]') -> 'SMFIdentificationBasedDPDOp Revised schema. """ - ret = super().post_process(packet) # type: SMFIdentificationBasedDPDOption + ret = super().post_process(packet) # type: SMFHashBasedDPDOption ret.mode = Enum_SMFDPDMode.H_DPD return ret diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index 278132e034..17ecd33b60 100644 --- a/pcapkit/protocols/schema/misc/pcapng.py +++ b/pcapkit/protocols/schema/misc/pcapng.py @@ -2735,7 +2735,7 @@ class PACK_FlagsOption(_PACK_Option, code=Enum_OptionType.pack_flags): padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) if TYPE_CHECKING: - def __init__(self, type: 'Enum_OptionType', length: 'int', flags: 'EPBFlags') -> 'None': ... + def __init__(self, type: 'Enum_OptionType', length: 'int', flags: 'PACKFlags') -> 'None': ... @schema_final diff --git a/tests/protocols/test_declarations_match_values_2317_unit.py b/tests/protocols/test_declarations_match_values_2317_unit.py new file mode 100644 index 0000000000..cc79ab6ad2 --- /dev/null +++ b/tests/protocols/test_declarations_match_values_2317_unit.py @@ -0,0 +1,294 @@ +# -*- coding: utf-8 -*- +"""Declarations agree with the values the code stores. C.f. #2317. + +Each annotation and ``__all__`` below disagreed with what the code actually +stores or accepts, so :program:`mypy` and the rendered docs described a value +that never occurs: ``cmpt_bitmap`` annotated :obj:`bytes` while +``_read_opt_calipso`` stores a :obj:`tuple` of :obj:`int`, +``SMFHashBasedDPDOption.post_process`` declaring its sibling's class, +``Optional`` on an ``RR``/``SSR`` ``route`` that is always built, +:attr:`~pcapkit.protocols.protocol.ProtocolBase.protochain` declaring a +:class:`~pcapkit.corekit.protochain.ProtoChain` that +:class:`~pcapkit.protocols.misc.null.NoPayload` does not have, a ``pack_flags`` +constructor stub typing the EPB flags, ``ARP._make_addr_resolve`` leaving out +the :obj:`bytearray` its body handles, an undeclared ``entry_raw``, two HIP +parameters missing from both ``__all__`` lists, and a ``version`` that one +``_read_param_*`` took positionally. + +Each case pins the declaration against the value, rather than the declaration +alone: a parse shows what is stored, and the declaration is read back from the +source where it lives only under :data:`~typing.TYPE_CHECKING`. + +:mod:`pcapkit` is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import ast +import inspect +import struct +import unittest +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class + +if TYPE_CHECKING: + from typing import Optional + +#: Test tier (#1958): pins declarations and in-memory parses, neither of which +#: differs by interpreter. +__tier__ = 'LATEST' + +#: Three address slots, the first of which is before a pointer of 8. +SLOTS = bytes([10, 0, 0, 1]) + bytes([10, 0, 0, 2]) + bytes(4) + + +def hopopt_bytes(option: 'bytes') -> 'bytes': + """A HOPOPT header carrying ``option``, padded to a multiple of 8 octets.""" + pad = -(len(option) + 2) % 8 + if pad: + option += bytes([1, pad - 2]) + bytes(pad - 2) if pad > 1 else bytes([0]) + return bytes([59, (len(option) + 2) // 8 - 1]) + option + + +def ipv4_bytes(options: 'bytes') -> 'bytes': + """An IPv4 header carrying ``options``, padded to a multiple of 4 octets.""" + options += bytes(-len(options) % 4) + length = 20 + len(options) + return (bytes([0x40 | length // 4, 0]) + length.to_bytes(2, 'big') + bytes(4) + + bytes([64, 59, 0, 0]) + bytes([10, 0, 0, 1]) + bytes([10, 0, 0, 2]) + options) + + +def body(module: 'object', name: 'str') -> 'ast.ClassDef': + """The class definition of ``name`` in ``module``'s source.""" + tree = ast.parse(inspect.getsource(module)) # type: ignore[arg-type] + return next(node for node in ast.walk(tree) + if isinstance(node, ast.ClassDef) and node.name == name) + + +def declared(module: 'object', name: 'str', field: 'str') -> 'Optional[str]': + """The annotation of ``field`` in ``name``'s class body, as written. + + :data:`None` where the class body does not declare ``field`` at all, so that + an undeclared field reads as a mismatch rather than as a + :exc:`StopIteration`. + + """ + node = next((item for item in ast.walk(body(module, name)) + if isinstance(item, ast.AnnAssign) and isinstance(item.target, ast.Name) + and item.target.id == field), None) + return None if node is None else ast.literal_eval(node.annotation) + + +def stubbed(module: 'object', name: 'str', arg: 'str') -> 'str': + """The annotation of ``arg`` in ``name``'s ``__init__`` stub, as written.""" + stub = next(item for item in ast.walk(body(module, name)) + if isinstance(item, ast.FunctionDef) and item.name == '__init__') + return ast.literal_eval(next(item for item in stub.args.args if item.arg == arg).annotation) + + +class CALIPSOBitmapTests(unittest.TestCase): + """``CALIPSOOption.cmpt_bitmap`` is the tuple of octets that is stored.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + #: A CALIPSO option with a one-word compartment bitmap. + OPTION = (bytes([7, 12]) + (0x12345678).to_bytes(4, 'big') + bytes([1, 5]) + + b'\xab\xcd' + bytes([0xde, 0xad, 0xbe, 0xef])) + + def test_parsed_bitmap_is_a_tuple_of_ints(self) -> None: + from pcapkit.const.ipv6.option import Option + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + + for protocol in (HOPOPT, IPv6_Opts): + with self.subTest(protocol=protocol.__name__): + bitmap = protocol(hopopt_bytes(self.OPTION)).info.options[Option.CALIPSO].cmpt_bitmap + self.assertEqual(bitmap, (0xde, 0xad, 0xbe, 0xef)) + self.assertIsInstance(bitmap, tuple) + for octet in bitmap: + self.assertIsInstance(octet, int) + + def test_annotation_says_tuple(self) -> None: + from pcapkit.protocols.data.internet import hopopt, ipv6_opts + + for module in (hopopt, ipv6_opts): + with self.subTest(module=module.__name__): + self.assertEqual(declared(module, 'CALIPSOOption', 'cmpt_bitmap'), + 'tuple[int, ...]') + + +class SMFHashBasedDPDTests(unittest.TestCase): + """``SMFHashBasedDPDOption.post_process`` declares the schema it returns.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + #: An ``SMF_DPD`` option in H-DPD mode, i.e. with the mode bit set. + OPTION = bytes([8, 5]) + bytes([0x80, 0x11, 0x22, 0x33, 0x44]) + + def test_declaration_names_the_schema_returned(self) -> None: + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + + for protocol in (HOPOPT, IPv6_Opts): + with self.subTest(protocol=protocol.__name__): + schema = protocol(hopopt_bytes(self.OPTION)).schema.options[0] + self.assertEqual(type(schema).__name__, 'SMFHashBasedDPDOption') + self.assertEqual(type(schema).post_process.__annotations__['return'], + type(schema).__name__) + + +class RecordRouteTests(unittest.TestCase): + """``RROption``/``SSROption`` always hold a ``route``, as ``LSROption`` does.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_parsed_route_is_a_tuple(self) -> None: + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.protocols.internet.ipv4 import IPv4 + + # One header does not hold all three: ``IHL`` caps the options at 40 + # octets, and a route option with three slots is 15. + for code, kind in ((OptionNumber.RR, 7), (OptionNumber.SSR, 137), (OptionNumber.LSR, 131)): + with self.subTest(code=code): + parsed = IPv4(ipv4_bytes(bytes([kind, 15, 8]) + SLOTS)).info.options + self.assertIsInstance(parsed[code].route, tuple) + + def test_the_three_annotations_agree(self) -> None: + from pcapkit.protocols.data.internet import ipv4 + + for name in ('RROption', 'SSROption', 'LSROption'): + with self.subTest(name=name): + self.assertEqual(declared(ipv4, name, 'route'), 'tuple[IPv4Address, ...]') + self.assertEqual(stubbed(ipv4, name, 'route'), 'tuple[IPv4Address, ...]') + + +class NoPayloadProtochainTests(unittest.TestCase): + """``NoPayload`` narrows ``protochain``, as it narrows ``name``.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_protochain_is_none_and_says_so(self) -> None: + from pcapkit.protocols.misc.null import NoPayload + from pcapkit.protocols.protocol import ProtocolBase + + self.assertIsNone(NoPayload().protochain) + self.assertEqual(NoPayload.protochain.fget.__annotations__['return'], 'None') + self.assertEqual(ProtocolBase.protochain.fget.__annotations__['return'], 'ProtoChain') + + +class PACKFlagsTests(unittest.TestCase): + """The ``pack_flags`` constructor stub types the flags its field holds.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_stub_matches_the_field(self) -> None: + from pcapkit.protocols.schema.misc import pcapng + + for name, flags in (('PACK_FlagsOption', 'PACKFlags'), ('EPB_FlagsOption', 'EPBFlags')): + with self.subTest(name=name): + self.assertEqual(getattr(pcapng, name).__annotations__['flags'], flags) + self.assertEqual(stubbed(pcapng, name, 'flags'), flags) + + +class ARPHardwareAddressTests(unittest.TestCase): + """``ARP._make_addr_resolve`` declares the ``bytearray`` its body accepts.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + #: A well-formed ARP request, as a source of an instance to call. + PACKET = (bytes([0, 1, 8, 0, 6, 4, 0, 1]) + bytes([1, 2, 3, 4, 5, 6]) + + bytes([10, 0, 0, 1]) + bytes(6) + bytes([10, 0, 0, 2])) + + def test_a_bytearray_resolves(self) -> None: + from pcapkit.const.arp.hardware import Hardware + from pcapkit.protocols.link.arp import ARP + + arp = ARP(self.PACKET) + self.assertEqual(arp._make_addr_resolve(bytearray(b'aa:bb:cc:dd:ee:ff'), + Hardware.Ethernet), b'\xaa\xbb\xcc\xdd\xee\xff') + self.assertEqual(arp._make_addr_resolve(bytearray(b'\x01\x02'), Hardware.Frame_Relay), + b'\x01\x02') + + def test_annotation_names_bytearray(self) -> None: + from pcapkit.protocols.link.arp import ARP + + self.assertEqual(ARP._make_addr_resolve.__annotations__['addr'], + 'str | bytes | bytearray') + + +class JournalEntryRawTests(unittest.TestCase): + """``SystemdJournalExportBlock`` declares the ``entry_raw`` it is given. + + The parse that adds it is pinned by + :mod:`tests.protocols.misc.test_pcapng_journal_binary_overrun_1406_unit`; + what was missing is the declaration. + + """ + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_entry_raw_is_declared(self) -> None: + from pcapkit.protocols.data.misc import pcapng + + self.assertEqual(declared(pcapng, 'SystemdJournalExportBlock', 'entry_raw'), 'bytes') + + def test_the_reader_still_adds_it(self) -> None: + import warnings + + from pcapkit.protocols.misc.pcapng import PCAPNG + + entry = b'BIN\n' + struct.pack(' None: + reimport_once_per_class(self) + + def test_all_lists_every_parameter(self) -> None: + from pcapkit.protocols.data.internet import hip as data_hip + from pcapkit.protocols.schema.internet import hip as schema_hip + + for module in (schema_hip, data_hip): + with self.subTest(module=module.__name__): + parameters = {name for name, item in vars(module).items() + if isinstance(item, type) and item.__module__ == module.__name__ + and issubclass(item, module.Parameter) and item is not module.Parameter} + self.assertEqual(parameters - set(module.__all__), set()) + self.assertLessEqual({'OverlayTTLParameter', 'ViaRVSParameter'}, parameters) + + +class HIPReadParameterSignatureTests(unittest.TestCase): + """Every ``HIP._read_param_*`` takes ``version`` and ``options`` by keyword.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_version_and_options_are_keyword_only(self) -> None: + from pcapkit.protocols.internet.hip import HIP + + readers = [name for name in dir(HIP) if name.startswith('_read_param_')] + self.assertIn('_read_param_relay_hmac', readers) + for name in readers: + with self.subTest(name=name): + parameters = inspect.signature(getattr(HIP, name)).parameters + for argument in ('version', 'options'): + self.assertEqual(parameters[argument].kind, + inspect.Parameter.KEYWORD_ONLY) + + +if __name__ == '__main__': + unittest.main()