From 0bc5b370099c12a1fc94e0fc0cbb7ff45928e0bf Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Thu, 8 Oct 2026 21:07:36 -0400 Subject: [PATCH] =?UTF-8?q?test:=20capture-driven=20parse=E2=86=92rebuild?= =?UTF-8?q?=20and=20dumper=20round=20trips=20(#1202)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - tests/protocols/test_capture_roundtrip_runtime.py: every layer of every frame in examples/captures/ (payload chain plus IPv6 extension headers) rebuilds from its info, from info.to_dict(), and Info survives to_dict/from_dict; every capture rebuilds whole, byte for byte. - tests/dumpkit/test_capture_dump_runtime.py: format='pcap' is byte-identical to its input; json/plist/tree are deterministic and hold one section per record or block; json frames carry every Info key and the protocol chain; plistlib reads the plist dump. - Known failures sit in LAYER_GAPS (#1446), DICT_GAPS (#1447) and KNOWN_FAILURES (#1448), asserted in both directions; the existing plistlib skip in tests/integration/test_output_formats.py cites #1448. Both modules are fixture tier (integration job): 4 + 6 tests, 24 s and 33 s, peak RSS 223 and 279 MiB. --- tests/dumpkit/test_capture_dump_runtime.py | 201 ++++++++++++ tests/integration/test_output_formats.py | 2 +- .../test_capture_roundtrip_runtime.py | 286 ++++++++++++++++++ 3 files changed, 488 insertions(+), 1 deletion(-) create mode 100644 tests/dumpkit/test_capture_dump_runtime.py create mode 100644 tests/protocols/test_capture_roundtrip_runtime.py diff --git a/tests/dumpkit/test_capture_dump_runtime.py b/tests/dumpkit/test_capture_dump_runtime.py new file mode 100644 index 0000000000..993df86840 --- /dev/null +++ b/tests/dumpkit/test_capture_dump_runtime.py @@ -0,0 +1,201 @@ +# -*- coding: utf-8 -*- +"""Every output format over every sample capture: exact, complete, deterministic. + +GitHub issue #1202. For each capture in :file:`examples/captures/`: + +* ``format='pcap'`` writes a file byte-identical to its input. PCAP-NG input is + left out, because PCAP output of it is refused by design. +* ``json``, ``plist`` and ``tree`` each write the same bytes on a second run. +* each of the three holds one section per record or block of the input, counted + from the file itself rather than from the extractor, and in the same order; +* a ``json`` frame section carries every top-level key of the frame's + :class:`~pcapkit.corekit.infoclass.Info` and its protocol chain. + +Deeper key-by-key completeness is not asserted: options dump as one mapping per +entry (#1263), so the rendering legitimately differs from ``Info.to_dict()``. + +A failing check is listed in :data:`KNOWN_FAILURES` with its tracking issue and +must still fail in the recorded way; every other check must pass. + +The module reads generated captures, so it belongs to the fixture-dependent tier. + +""" +from __future__ import annotations + +import importlib.util +import json +import os +import pathlib +import plistlib +import struct +import tempfile +import unittest +import warnings +import xml.etree.ElementTree as ET +from typing import TYPE_CHECKING, NamedTuple + +from tests._support import close_extractor, reimport_once_per_class, 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 sweep over every capture may take. The slowest takes ~15. +SWEEP_TIMEOUT = 300 + + +class Gap(NamedTuple): + """One check that fails today, and the defect that stops it.""" + + #: Tracking issue, or :data:`None` while the defect is not yet filed. + issue: 'Optional[int]' + #: Name of the exception the check raises. + status: 'str' + #: The defect and how to reproduce it. + defect: 'str' + + +#: Every check that fails today, by test method; each applies to every capture. +KNOWN_FAILURES = { + 'test_plist_dump_reads_back_through_plistlib': Gap( + 1448, 'AttributeError', + "dictdumper/plist.py:278 writes each timestamp as '%Y-%m-%dT%H:%M:%S.%fZ', " + 'and a property list has no fractional part, so plistlib.load ' + "raises AttributeError: 'NoneType' object has no attribute 'groupdict' on " + "every capture's first frame. Repro: extract(fin='in.pcap', fout='o', " + "format='plist'); plistlib.load(open('o.plist', 'rb')). Also skipped, " + 'citing #1448, in tests/integration/test_output_formats.py PlistRoundTripTests.', + ), +} + + +def captures(pcapng: 'bool' = True) -> 'list[str]': + """Every capture under :data:`~tests._tiers.SAMPLE_ROOT`, by file name.""" + return sorted(path.name for path in SAMPLE_ROOT.iterdir() + if path.suffix in ('.pcap', '.cap') or (pcapng and path.suffix == '.pcapng')) + + +def section_count(raw: 'bytes') -> 'int': + """Sections a dump of ``raw`` should hold, counted from the file's own framing. + + PCAP: the global header plus one per record. PCAP-NG: one per block. + + """ + count, offset = 0, 0 + if raw[:4] == b'\x0a\x0d\x0d\x0a': + endian = '<' + while offset < len(raw): + if raw[offset:offset + 4] == b'\x0a\x0d\x0d\x0a': + endian = '<' if raw[offset + 8:offset + 12] == b'\x4d\x3c\x2b\x1a' else '>' + offset += struct.unpack(f'{endian}I', raw[offset + 4:offset + 8])[0] + count += 1 + return count + endian = '<' if raw[:4] in (b'\xd4\xc3\xb2\xa1', b'\x4d\x3c\xb2\xa1') else '>' + offset, count = 24, 1 + while offset + 16 <= len(raw): + offset += 16 + struct.unpack(f'{endian}I', raw[offset + 8:offset + 12])[0] + count += 1 + return count + + +def sections(fmt: 'str', path: 'str') -> 'list[str]': + """Top-level section names of a ``fmt`` dump, read without pcapkit.""" + if fmt == 'json': + with open(path, encoding='utf-8') as file: + return list(json.load(file)) + if fmt == 'plist': + # ElementTree, not plistlib, which cannot read the dump at all; see + # :data:`KNOWN_FAILURES`. + root = ET.parse(path).getroot() + return [key.text or '' for key in root[0].findall('key')] + with open(path, encoding='utf-8') as file: + return [line.rstrip('\n') for line in file if line.strip() and not line[0].isspace()] + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class CaptureDumpTests(unittest.TestCase): + """Pin what each writer makes of every capture, by reading the file back.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + self.tmpdir = tempfile.TemporaryDirectory() # pylint: disable=consider-using-with + self.addCleanup(self.tmpdir.cleanup) + quiet = warnings.catch_warnings() + quiet.__enter__() # pylint: disable=unnecessary-dunder-call + self.addCleanup(quiet.__exit__, None, None, None) + warnings.simplefilter('ignore') + + def dump(self, name: 'str', fmt: 'str', stem: 'str', **kwargs: 'Any') -> 'tuple[Any, str]': + """Extract capture ``name`` to ``stem`` as ``fmt``; return the extractor and output path.""" + from pcapkit import extract + + out = os.path.join(self.tmpdir.name, stem) + extractor = extract(fin=str(SAMPLE_ROOT / name), fout=out, format=fmt, **kwargs) + self.addCleanup(close_extractor, extractor) + return extractor, extractor.output + + def test_pcap_output_is_byte_identical_to_its_input(self) -> None: + with time_limit(SWEEP_TIMEOUT): + for name in captures(pcapng=False): + _, path = self.dump(name, 'pcap', f'{name}.out.pcap', store=False, extension=False) + with self.subTest(capture=name), open(path, 'rb') as file: + self.assertEqual(file.read(), (SAMPLE_ROOT / name).read_bytes()) + + def assertDumpIsCompleteAndDeterministic(self, fmt: 'str') -> None: + """Two runs agree byte for byte, with one section per record or block.""" + with time_limit(SWEEP_TIMEOUT): + for name in captures(): + outputs = [self.dump(name, fmt, f'{name}.{run}', store=False)[1] for run in (1, 2)] + first, second = (pathlib.Path(path).read_bytes() for path in outputs) + names = sections(fmt, outputs[0]) + with self.subTest(capture=name, format=fmt): + self.assertEqual(first, second, 'the second run wrote different bytes') + self.assertEqual(len(names), section_count((SAMPLE_ROOT / name).read_bytes())) + self.assertEqual(len(set(names)), len(names), 'a section name repeats') + # Every format names the same sections in the same order. + if fmt != 'json': + json_path = self.dump(name, 'json', f'{name}.ref', store=False)[1] + self.assertEqual(names, sections('json', json_path)) + + def test_json_dump_is_complete_and_deterministic(self) -> None: + self.assertDumpIsCompleteAndDeterministic('json') + + def test_plist_dump_is_complete_and_deterministic(self) -> None: + self.assertDumpIsCompleteAndDeterministic('plist') + + def test_tree_dump_is_complete_and_deterministic(self) -> None: + self.assertDumpIsCompleteAndDeterministic('tree') + + def test_json_frame_sections_carry_every_info_key_and_the_chain(self) -> None: + with time_limit(SWEEP_TIMEOUT): + for name in captures(): + extractor, path = self.dump(name, 'json', f'{name}.frames', store=True) + with open(path, encoding='utf-8') as file: + report = json.load(file) + for number, frame in enumerate(extractor.frame, start=1): + section = report[f'Frame {number}'] + with self.subTest(capture=name, frame=number): + self.assertEqual(list(section), list(frame.info.to_dict())) + self.assertEqual(section['protocols'], str(frame.protochain)) + + def test_plist_dump_reads_back_through_plistlib(self) -> None: + gap = KNOWN_FAILURES.get('test_plist_dump_reads_back_through_plistlib') + with time_limit(SWEEP_TIMEOUT): + for name in captures(): + _, path = self.dump(name, 'plist', f'{name}.lib', store=False) + with self.subTest(capture=name), open(path, 'rb') as file: + try: + report = plistlib.load(file) + except Exception as exc: # pylint: disable=broad-except + got = type(exc).__name__ + else: + got = 'OK' + self.assertEqual(list(report), sections('plist', path)) + self.assertEqual(got, 'OK' if gap is None else gap.status) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/integration/test_output_formats.py b/tests/integration/test_output_formats.py index 8c5a438f0a..730c8c18a5 100644 --- a/tests/integration/test_output_formats.py +++ b/tests/integration/test_output_formats.py @@ -119,7 +119,7 @@ class PlistRoundTripTests(EndToEndTestCase): """ @unittest.skip('blocked on dictdumper writing values with fractional seconds, ' - 'which plistlib rejects (dictdumper/plist.py:278)') + 'which plistlib rejects (dictdumper/plist.py:278, #1448)') def test_plist_report_round_trips_through_plistlib(self) -> None: """A ``plist`` report should be readable by :func:`plistlib.load`. diff --git a/tests/protocols/test_capture_roundtrip_runtime.py b/tests/protocols/test_capture_roundtrip_runtime.py new file mode 100644 index 0000000000..78f8d5b10d --- /dev/null +++ b/tests/protocols/test_capture_roundtrip_runtime.py @@ -0,0 +1,286 @@ +# -*- coding: utf-8 -*- +"""Parse -> rebuild identity over every layer of every sample capture. + +GitHub issue #1202. :file:`tests/protocols/test_option_roundtrip_unit.py` runs +construct -> parse -> construct over cases built from the dispatch registries. +This module runs the other direction, over real octets: for every layer of every +frame in :file:`examples/captures/`, ``Cls.from_data(layer.info).data`` must equal +``layer.data``. A *layer* is each protocol on the payload chain, from the +:class:`~pcapkit.protocols.misc.pcap.frame.Frame` or +:class:`~pcapkit.protocols.misc.pcapng.PCAPNG` record down, plus every IPv6 +extension header, which IPv6 keeps off that chain. + +The same layers are also rebuilt from ``layer.info.to_dict()``, which +:meth:`~pcapkit.protocols.protocol.ProtocolBase.from_data` accepts too, and each +:class:`~pcapkit.corekit.infoclass.Info` is put through ``to_dict`` and +``from_dict``. Finally each capture is rebuilt whole, header and records or +blocks, and compared with the file. + +A case that does not close today is listed in :data:`LAYER_GAPS` or +:data:`DICT_GAPS` with the issue that tracks it. Both tables are asserted in both +directions: a listed case must still fail in the recorded way, a case not listed +must pass, and every entry must name a case that exists. + +The module reads generated captures, so it belongs to the fixture-dependent tier. + +""" +from __future__ import annotations + +import importlib.util +import struct +import unittest +import warnings +from typing import TYPE_CHECKING, NamedTuple + +from tests._support import close_extractor, reimport_once_per_class, time_limit +from tests._tiers import SAMPLE_ROOT + +if TYPE_CHECKING: + from typing import Any, Iterator, 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 sweep over every capture may take. A sweep takes under ten. +SWEEP_TIMEOUT = 300 + + +class Gap(NamedTuple): + """One case that does not round-trip, and the defect that stops it.""" + + #: Tracking issue, or :data:`None` while the defect is not yet filed. + issue: 'Optional[int]' + #: Expected outcome: ``'TRUNCATED'`` when the rebuild is a strict prefix of + #: the original, ``'MISMATCH'`` for any other difference, or the name of the + #: exception the rebuild raises. + status: 'str' + #: The defect and how to reproduce it. + defect: 'str' + + +_FRAG = Gap( + 1446, 'TRUNCATED', + 'An IPv6 extension header parsed with extension=True keeps the rest of the ' + 'datagram in .data, but its .info holds the header alone, so from_data ' + 'rebuilds 8 of 442 octets for ipv6.pcap frame 16. Repro: ex = ' + 'next(iter(frame[IPv6]._exthdr.values())); ' + 'type(ex).from_data(ex.info).data != ex.data. Any extension header followed ' + 'by upper-layer octets does the same (HOPOPT + UDP: 8 of 20).', +) + +#: Every (capture, frame, layer path) whose object-form rebuild does not close. +#: Frame numbers are one-based, as in the dumps. +LAYER_GAPS = { + f'ipv6.pcap#{frame}:Frame/Ethernet/IPv6/IPv6_Frag': _FRAG for frame in (13, 14, 15, 16) +} + +_DICT_DROPS_PAYLOAD = ( + 'from_data(info.to_dict()) rebuilds the header only: to_dict() leaves out the ' + '__next_type__ and __next_name__ keys _make_payload dispatches on ' + '(pcapkit/protocols/protocol.py:1926), so the payload becomes NoPayload. ' + 'IPv6 also loses its extension headers, which _make_data reads from the ' + 'equally dropped __exthdr__ key (pcapkit/protocols/internet/ipv6.py:367). ' + 'Repro: Ethernet.from_data(eth.info.to_dict()).data is 14 of 1870 octets ' + 'for ipv4.pcap frame 1.' +) + + +class DictGap(NamedTuple): + """A protocol whose dict-form rebuild does not close.""" + + #: Tracking issue, or :data:`None` while the defect is not yet filed. + issue: 'Optional[int]' + #: As :attr:`Gap.status`. + status: 'str' + #: ``'payload'`` if only a layer carrying a payload or an IPv6 extension + #: header fails, ``'always'`` if every layer of the protocol does. + when: 'str' + #: The defect and how to reproduce it. + defect: 'str' + + +#: Every protocol, by class name, whose rebuild from ``info.to_dict()`` fails. +DICT_GAPS = { + **{name: DictGap(1447, 'TRUNCATED', 'payload', _DICT_DROPS_PAYLOAD) + for name in ('Frame', 'Ethernet', 'ARP', 'IPv4', 'IPv6', 'TCP', 'UDP')}, + 'HTTP': DictGap( + 1447, 'ProtocolError', 'always', + "HTTP.from_data(info.to_dict()) raises ProtocolError('invalid HTTP data: " + "Raw'): the dict is rebuilt into a generic data model that " + 'HTTP._make_data does not accept. Repro: the first HTTP layer of ' + 'http.pcap.', + ), + # The object-form gap, which the dict form inherits unchanged. + 'IPv6_Frag': DictGap(_FRAG.issue, _FRAG.status, 'always', _FRAG.defect), +} + + +def captures() -> 'list[str]': + """Every capture under :data:`~tests._tiers.SAMPLE_ROOT`, by file name.""" + return sorted(path.name for path in SAMPLE_ROOT.iterdir() + if path.suffix in ('.pcap', '.pcapng', '.cap')) + + +def outcome(layer: 'Any', as_dict: 'bool' = False) -> 'str': + """Classify ``layer``'s rebuild from its info against ``layer.data``. + + Args: + layer: The parsed layer. + as_dict: Rebuild from ``layer.info.to_dict()`` rather than ``layer.info``. + + Returns: + ``'OK'``, or a :attr:`Gap.status`. + + """ + original = layer.data + try: + data = type(layer).from_data(layer.info.to_dict() if as_dict else layer.info, + **keywords(layer)).data + except Exception as exc: # pylint: disable=broad-except + return type(exc).__name__ + if data == original: + return 'OK' + if len(data) < len(original) and original.startswith(data): + return 'TRUNCATED' + return 'MISMATCH' + + +def walk(frame: 'Any') -> 'Iterator[tuple[str, Any]]': + """Yield ``(path, layer)`` for the payload chain and every IPv6 extension header.""" + from pcapkit.protocols.misc.null import NoPayload + + path = '' + layer = frame + while not isinstance(layer, NoPayload): + path = f'{path}/{type(layer).__name__}' if path else type(layer).__name__ + yield path, layer + for ext in getattr(layer, '_exthdr', {}).values(): + yield f'{path}/{type(ext).__name__}', ext + layer = layer.payload + + +def keywords(layer: 'Any') -> 'dict[str, Any]': + """Construction keywords ``layer.info`` does not carry, as the engines pass them.""" + from pcapkit.protocols.misc.pcap.frame import Frame + from pcapkit.protocols.misc.pcapng import PCAPNG + + if isinstance(layer, Frame): + return {'num': layer._fnum, 'header': layer._ghdr} # pylint: disable=protected-access + if isinstance(layer, PCAPNG): + return {'num': layer._fnum, 'sct': layer._sect, 'ctx': layer._ctx} # pylint: disable=protected-access + return {} + + +def pcapng_blocks(raw: 'bytes') -> 'Iterator[bytes]': + """Split a PCAP-NG file into its blocks by each block's total length.""" + offset, endian = 0, '<' + while offset < len(raw): + if raw[offset:offset + 4] == b'\x0a\x0d\x0d\x0a': + endian = '<' if raw[offset + 8:offset + 12] == b'\x4d\x3c\x2b\x1a' else '>' + (length,) = struct.unpack(f'{endian}I', raw[offset + 4:offset + 8]) + yield raw[offset:offset + length] + offset += length + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class CaptureRoundTripTests(unittest.TestCase): + """Every layer of every capture rebuilds from what was parsed.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + # The option captures carry deliberately odd values, and parsing or + # rebuilding them warns; the octets are what is asserted. + quiet = warnings.catch_warnings() + quiet.__enter__() # pylint: disable=unnecessary-dunder-call + self.addCleanup(quiet.__exit__, None, None, None) + warnings.simplefilter('ignore') + + def layers(self) -> 'Iterator[tuple[str, Any]]': + """Yield ``(case id, layer)`` for every layer of every capture.""" + from pcapkit import extract + + names = captures() + self.assertGreater(len(names), 2, f'no generated captures under {SAMPLE_ROOT}') + for name in names: + extractor = extract(fin=str(SAMPLE_ROOT / name), store=True, nofile=True) + try: + for number, frame in enumerate(extractor.frame, start=1): + for path, layer in walk(frame): + yield f'{name}#{number}:{path}', layer + finally: + close_extractor(extractor) + + def test_every_layer_rebuilds_from_its_info(self) -> None: + seen = set() + with time_limit(SWEEP_TIMEOUT): + for case, layer in self.layers(): + seen.add(case) + got = outcome(layer) + gap = LAYER_GAPS.get(case) + with self.subTest(case=case): + self.assertEqual(got, 'OK' if gap is None else gap.status) + self.assertGreater(len(seen), 5000) + self.assertEqual(set(LAYER_GAPS) - seen, set(), 'LAYER_GAPS names cases that no longer exist') + + def test_every_layer_rebuilds_from_its_info_as_a_dict(self) -> None: + from pcapkit.protocols.misc.null import NoPayload + + hit = set() + with time_limit(SWEEP_TIMEOUT): + for case, layer in self.layers(): + name = type(layer).__name__ + gap = DICT_GAPS.get(name) + if gap is not None and gap.when == 'payload': + payload = getattr(layer, '_next', None) + if (payload is None or isinstance(payload, NoPayload)) \ + and not getattr(layer, '_exthdr', None): + gap = None + got = outcome(layer, as_dict=True) + if gap is not None: + hit.add(name) + with self.subTest(case=case): + self.assertEqual(got, 'OK' if gap is None else gap.status) + self.assertEqual(set(DICT_GAPS) - hit, set(), 'DICT_GAPS names protocols no capture exercises') + + def test_every_info_survives_to_dict_and_from_dict(self) -> None: + with time_limit(SWEEP_TIMEOUT): + for case, layer in self.layers(): + with self.subTest(case=case): + self.assertEqual(type(layer.info).from_dict(layer.info.to_dict()), layer.info) + + def test_every_capture_rebuilds_byte_for_byte(self) -> None: + """Header and records (PCAP) or every block (PCAP-NG), rebuilt and joined.""" + from pcapkit import extract + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.foundation.engines.pcapng import Context + from pcapkit.protocols.misc.pcap.header import Header + from pcapkit.protocols.misc.pcapng import PCAPNG + + with time_limit(SWEEP_TIMEOUT): + for name in captures(): + raw = (SAMPLE_ROOT / name).read_bytes() + if name.endswith('.pcapng'): + parts, section, ctx = [], 0, None + for number, block in enumerate(pcapng_blocks(raw)): + if block[:4] == b'\x0a\x0d\x0d\x0a': + section += 1 + parsed = PCAPNG(block, num=0, sct=section, ctx=None) + ctx = Context(parsed.info) + else: + parsed = PCAPNG(block, num=number, sct=section, ctx=ctx) + if parsed.info.type == BlockType.Interface_Description_Block: + ctx.interfaces.append(parsed.info) + parts.append(PCAPNG.from_data(parsed.info, **keywords(parsed)).data) + else: + extractor = extract(fin=str(SAMPLE_ROOT / name), store=True, nofile=True) + self.addCleanup(close_extractor, extractor) + header = extractor._exeng._gbhdr # pylint: disable=protected-access + parts = [Header.from_data(header.info).data] + parts.extend(type(frame).from_data(frame.info, **keywords(frame)).data + for frame in extractor.frame) + with self.subTest(capture=name): + self.assertEqual(b''.join(parts), raw) + + +if __name__ == '__main__': + unittest.main()