From 9426e78b3e58e2529e1deccf8333f76c86ecbbd6 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Thu, 8 Oct 2026 09:25:06 -0400 Subject: [PATCH] test(schema): every BitField namespace covers all of its bits Walk every schema under pcapkit.protocols.schema and assert each BitField names its full width exactly once. ForwardMatchField-wrapped peeks are exempt (they pack no octets). Discovery is cross-checked against the BitField(...) call count per module. The 27 remaining HIP and MH gaps are an explicit KNOWN_GAPS allowlist, matched exactly so a fixed site must drop its entry. Closes #1288 --- .../schema/test_bitfield_coverage_unit.py | 218 ++++++++++++++++++ 1 file changed, 218 insertions(+) create mode 100644 tests/protocols/schema/test_bitfield_coverage_unit.py diff --git a/tests/protocols/schema/test_bitfield_coverage_unit.py b/tests/protocols/schema/test_bitfield_coverage_unit.py new file mode 100644 index 0000000000..6569b0b458 --- /dev/null +++ b/tests/protocols/schema/test_bitfield_coverage_unit.py @@ -0,0 +1,218 @@ +# -*- coding: utf-8 -*- +"""Every :class:`~pcapkit.corekit.fields.strings.BitField` names all of its bits. + +GitHub issue #1288: :meth:`BitField.pre_process +` starts from an all-zero +buffer and writes only the subfields its ``namespace`` names, so a bit no +subfield covers is packed as zero whatever was parsed. The per-protocol fixes +(#1226, #1253, #1254, #1307, #1310) named a ``reserved`` subfield site by site. +This module is the invariant behind them: it walks every schema under +:mod:`pcapkit.protocols.schema` and checks that each bit field's namespace +covers its full width exactly once -- no gap and no overlap. + +A bit field wrapped in :class:`~pcapkit.corekit.fields.misc.ForwardMatchField` +is exempt by construction, not by allowlist: such a field only peeks ahead, +packs to no octets, and the bits it reads are packed again by the fields that +follow it. + +Discovery is cross-checked against the source: the number of ``BitField(...)`` +calls in each schema module must equal the number of bit fields found in it, so +one constructed where the walk cannot see it -- inside a selector, say -- fails +here rather than escaping the check. + +:mod:`pcapkit` is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`, not at module load. + +""" +from __future__ import annotations + +import ast +import importlib +import os +import pkgutil +import unittest +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class + +if TYPE_CHECKING: + from typing import Any, Iterator + +#: Bit fields known to leave bits unnamed, as ``schema.field`` (relative to +#: :mod:`pcapkit.protocols.schema`) -> the ``(start, length)`` bit ranges left +#: uncovered. An entry must match exactly: once a site is fixed, its entry has +#: to go, and a site that changes shape has to be looked at again. +#: +#: Every entry below packs its unnamed bits as zero, so a parsed packet with +#: any of them set does not rebuild byte-exactly. Naming them means carrying +#: a ``reserved`` value through the protocol's reader, data model and +#: ``make()``, which lies outside the schema files. +KNOWN_GAPS: 'dict[str, tuple[tuple[int, int], ...]]' = { + # HIP [RFC 7401]: the 3 reserved bits between Version and the fixed 1 bit + # of the version octet, and the 14 reserved bits of ROUTE_DST/ROUTE_VIA + # flags [RFC 6028]. Needs pcapkit/protocols/internet/hip.py and + # pcapkit/protocols/data/internet/hip.py. + 'internet.hip.HIP.ver': ((4, 3),), + 'internet.hip.RouteDstParameter.flags': ((2, 14),), + 'internet.hip.RouteViaParameter.flags': ((2, 14),), + # Mobility Header: reserved bits of 24 message and option flag fields. + # Needs pcapkit/protocols/internet/mh.py and + # pcapkit/protocols/data/internet/mh.py -- a family-wide change that wants + # its own review. + 'internet.mh.BindingRevocationMessage.flags': ((3, 13),), + 'internet.mh.DNSUpdateOption.flags': ((1, 7),), + 'internet.mh.DelegatedMNPOption.flags': ((1, 7),), + 'internet.mh.DynamicIPMulticastSelectorOption.flags': ((1, 7),), + 'internet.mh.FastBindingAcknowledgmentMessage.flags': ((1, 7),), + 'internet.mh.FastBindingUpdateMessage.flags': ((4, 12),), + 'internet.mh.FlowBindingMessage.flags': ((1, 7),), + 'internet.mh.HandoverAcknowledgeMessage.flags': ((3, 5),), + 'internet.mh.HandoverInitiateMessage.flags': ((4, 4),), + 'internet.mh.HeartbeatMessage.flags': ((0, 14),), + 'internet.mh.IPv4AddressAcknowledgementOption.flags': ((6, 2),), + 'internet.mh.IPv4DHCPSupportModeOption.flags': ((0, 15),), + 'internet.mh.IPv4HomeAddressReplyOption.flags': ((6, 2),), + 'internet.mh.IPv4HomeAddressRequestOption.flags': ((6, 10),), + 'internet.mh.IPv4TrafficOffloadSelectorOption.flags': ((1, 31),), + 'internet.mh.LocalizedRoutingAcknowledgmentMessage.flags': ((1, 7),), + 'internet.mh.MAGMultipathBindingOption.flags': ((2, 22),), + 'internet.mh.MultiPrefixExtension.flags': ((1, 31),), + 'internet.mh.NATDetectionOption.flags': ((1, 15),), + 'internet.mh.PerSessionBitRateAttribute.flags': ((2, 14),), + 'internet.mh.RedirectOption.flags': ((2, 14),), + 'internet.mh.SubscriptionResponseMessage.flags': ((1, 7),), + 'internet.mh.TransientBindingOption.flags': ((0, 7),), + 'internet.mh.UpdateNotificationMessage.flags': ((2, 14),), +} + + +def _gaps(width: 'int', namespace: 'dict[str, tuple[int, int]]') -> 'tuple[tuple[int, int], ...]': + """Return the ``(start, length)`` runs of bits no subfield covers.""" + covered = [False] * width + for start, size in namespace.values(): + covered[start:start + size] = [True] * size + runs = [] # type: list[tuple[int, int]] + bit = 0 + while bit < width: + if covered[bit]: + bit += 1 + continue + end = bit + while end < width and not covered[end]: + end += 1 + runs.append((bit, end - bit)) + bit = end + return tuple(runs) + + +def _overlaps(namespace: 'dict[str, tuple[int, int]]') -> 'list[int]': + """Return the bits more than one subfield claims.""" + seen = set() # type: set[int] + shared = set() # type: set[int] + for start, size in namespace.values(): + for bit in range(start, start + size): + (shared if bit in seen else seen).add(bit) + return sorted(shared) + + +class TestBitFieldCoverage(unittest.TestCase): + """Pin that no schema bit field packs an unnamed bit.""" + + maxDiff = None + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _discover(self) -> 'dict[str, tuple[Any, bool, str]]': + """Map ``schema.field`` to each bit field found, whether it sits under a + :class:`ForwardMatchField`, and the module declaring it.""" + import pcapkit.protocols.schema as package + from pcapkit.corekit.fields.misc import ForwardMatchField + from pcapkit.corekit.fields.strings import BitField + from pcapkit.protocols.schema.schema import Schema + + repo = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) + self.assertTrue(os.path.abspath(package.__file__).startswith(repo + os.sep), package.__file__) + for info in pkgutil.walk_packages(package.__path__, f'{package.__name__}.'): + importlib.import_module(info.name) + + def subclasses(cls: 'type') -> 'Iterator[type]': + for sub in cls.__subclasses__(): + yield sub + yield from subclasses(sub) + + found = {} # type: dict[int, tuple[str, Any, bool, str]] + + def walk(obj: 'Any', where: 'tuple[str, str]', forward: 'bool', seen: 'set[int]') -> 'None': + if id(obj) in seen: + return + seen.add(id(obj)) + if isinstance(obj, BitField): + found.setdefault(id(obj), (where[0], obj, forward, where[1])) + elif isinstance(obj, (list, tuple, set, frozenset)): + for item in obj: + walk(item, where, forward, seen) + elif isinstance(obj, dict): + for item in obj.values(): + walk(item, where, forward, seen) + elif type(obj).__module__.startswith('pcapkit.corekit.fields') and hasattr(obj, '__dict__'): + inner = forward or isinstance(obj, ForwardMatchField) + for item in vars(obj).values(): + walk(item, where, inner, seen) + + prefix = f'{package.__name__}.' + for cls in set(subclasses(Schema)): + if not cls.__module__.startswith(prefix): + continue + module = cls.__module__[len(prefix):] + for name, field in cls.__dict__.get('__fields__', {}).items(): + walk(field, (f'{module}.{cls.__qualname__}.{name}', module), False, set()) + return {where: (field, forward, module) for where, field, forward, module in found.values()} + + def test_discovery_matches_source(self) -> None: + import pcapkit.protocols.schema as package + + found = self._discover() + per_module = {} # type: dict[str, int] + for _, _, module in found.values(): + per_module[module] = per_module.get(module, 0) + 1 + + root = os.path.dirname(package.__file__) + for info in pkgutil.walk_packages(package.__path__, ''): + if info.ispkg: + continue + path = os.path.join(root, *info.name.split('.')) + '.py' + with open(path, encoding='utf-8') as file: + tree = ast.parse(file.read(), path) + calls = sum(1 for node in ast.walk(tree) + if isinstance(node, ast.Call) and getattr(node.func, 'id', None) == 'BitField') + with self.subTest(module=info.name): + self.assertEqual(per_module.get(info.name, 0), calls) + self.assertGreater(len(found), 0) + + def test_every_bitfield_names_all_its_bits(self) -> None: + found = self._discover() + gaps = {} # type: dict[str, tuple[tuple[int, int], ...]] + for where, (field, forward, _) in sorted(found.items()): + namespace = field._namespace # pylint: disable=protected-access + with self.subTest(field=where): + self.assertEqual(_overlaps(namespace), []) + if forward: + continue + run = _gaps(field.length * 8, namespace) + if run: + gaps[where] = run + self.assertEqual(gaps, KNOWN_GAPS) + + def test_gap_finder_catches_an_unnamed_bit(self) -> None: + from pcapkit.corekit.fields.strings import BitField + + field = BitField(length=1, namespace={'x': (0, 1), 'y': (7, 1)}) + namespace = field._namespace # pylint: disable=protected-access + self.assertEqual(_gaps(field.length * 8, namespace), ((1, 6),)) + self.assertEqual(_gaps(8, {'x': (0, 8)}), ()) + self.assertEqual(_overlaps({'x': (0, 4), 'y': (3, 5)}), [3]) + + +if __name__ == '__main__': + unittest.main()