Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pcapkit/protocols/link/l2tpv2.py
Original file line number Diff line number Diff line change
Expand Up @@ -362,7 +362,7 @@ def _make_data(cls, data: 'Data_L2TP') -> 'dict[str, Any]': # type: ignore[over
"""
return {
'type': data.flags.type,
'prio': data.flags.prio,
'priority': data.flags.prio,
'version': data.version,
'length': data.length,
'tunnel_id': data.tunnelid,
Expand Down
2 changes: 1 addition & 1 deletion pcapkit/protocols/misc/pcap/frame.py
Original file line number Diff line number Diff line change
Expand Up @@ -465,7 +465,7 @@ def _make_data(cls, data: 'Data_Frame') -> 'dict[str, Any]': # type: ignore[ove

"""
return {
'ts_src': data.frame_info.ts_sec,
'ts_sec': data.frame_info.ts_sec,
'ts_usec': data.frame_info.ts_usec,
'incl_len': data.frame_info.incl_len,
'orig_len': data.frame_info.orig_len,
Expand Down
2 changes: 1 addition & 1 deletion tests/protocols/link/test_link_unit.py
Original file line number Diff line number Diff line change
Expand Up @@ -370,7 +370,7 @@ def test_l2tp_index_is_unsupported_and_make_data_preserves_flags(self) -> None:
self.assertEqual(proto.__length_hint__(), 16)
values = L2TPv2._make_data(data)
self.assertEqual(values['type'], True)
self.assertEqual(values['prio'], False)
self.assertEqual(values['priority'], False)
self.assertEqual(values['version'], 2)
self.assertEqual(values['tunnel_id'], 3)
self.assertEqual(values['session_id'], 4)
Expand Down
2 changes: 1 addition & 1 deletion tests/protocols/misc/pcap/test_header_frame_unit.py
Original file line number Diff line number Diff line change
Expand Up @@ -183,7 +183,7 @@ def test_frame_make_index_length_and_make_data(self) -> None:
self.assertEqual(explicit_schema.incl_len, 2)
self.assertEqual(explicit_schema.orig_len, 3)
values = Frame._make_data(data)
self.assertEqual(values['ts_src'], 1)
self.assertEqual(values['ts_sec'], 1)
self.assertEqual(values['ts_usec'], 250000)
self.assertEqual(values['incl_len'], 4)
self.assertEqual(values['orig_len'], 6)
Expand Down
34 changes: 17 additions & 17 deletions tests/protocols/test_construction_keyword_check_unit.py
Original file line number Diff line number Diff line change
Expand Up @@ -600,56 +600,56 @@ def test_from_data_warns_instead_of_raising(self) -> None:
<pcapkit.protocols.protocol.ProtocolBase._make_data>` returned into
``__init__``, so nobody typed those keywords: a mismatch is a defect in the
protocol's own pair of mappings, and the caller who meets it cannot fix it.
Raising there would also convert three latent defects of that shape into a
Raising there would also convert a latent defect of that shape into a
broken ``from_data`` -- see
:meth:`test_the_three_known_make_data_mismatches_are_recorded`.
:meth:`test_the_known_make_data_mismatches_are_recorded`.

``L2TPv2`` is used because it is one of the three and therefore exercises
the real path rather than a contrived one. When its ``_make_data`` is fixed
this test goes red, which is the point: the entry is then deleted.
``L2TPv2`` is subclassed with a ``_make_data`` that returns one key no
signature declares, so the case does not depend on a real mismatch that
will itself be fixed.

"""
from pcapkit.protocols.link.l2tpv2 import L2TPv2
from pcapkit.utilities.warnings import UnknownFieldWarning

class Mismatched(L2TPv2):
@classmethod
def _make_data(cls, data: 'Any') -> 'dict[str, Any]':
return {**super()._make_data(data), 'prio': True}

octets = bytes(L2TPv2(version=2, tunnel_id=1, session_id=2, payload=b'ab'))
parsed = L2TPv2(io.BytesIO(octets), len(octets))

with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter('always')
rebuilt = L2TPv2.from_data(parsed.info)
rebuilt = Mismatched.from_data(parsed.info)

self.assertIsInstance(rebuilt, L2TPv2)
self.assertIsInstance(rebuilt, Mismatched)
messages = [item for item in caught
if isinstance(item.message, UnknownFieldWarning)]
self.assertEqual(len(messages), 1, [str(item.message) for item in caught])
self.assertIn("'prio'", str(messages[0].message))
self.assertIn('_make_data', str(messages[0].message))

def test_the_three_known_make_data_mismatches_are_recorded(self) -> None:
def test_the_known_make_data_mismatches_are_recorded(self) -> None:
"""The latent defects this check made visible, named so they can be fixed.

Each of these has ``_make_data`` returning a key that no signature of the
same protocol declares, so :meth:`ProtocolBase.from_data
<pcapkit.protocols.protocol.ProtocolBase.from_data>` has been dropping that
field in silence -- a frame's timestamp, an L2TPv2 priority bit, and a
capture's byte order. They are recorded rather than fixed here because each
is a defect in its own protocol rather than in this mechanism, and because
two of them need a decision about what ``make`` should be called rather than
a rename.
<pcapkit.protocols.protocol.ProtocolBase.from_data>` drops that field --
a capture's byte order. It is recorded rather than fixed here because it
is a defect in its own protocol rather than in this mechanism.

Written as an expected-failure table for the reason the round-trip module
writes its own that way: fixing one of these turns this red and the entry
gets deleted, where a silent skip would leave the defect recorded forever.

"""
from pcapkit.protocols.link.l2tpv2 import L2TPv2
from pcapkit.protocols.misc.pcap.frame import Frame
from pcapkit.protocols.misc.pcap.header import Header
from pcapkit.protocols.protocol import _declared_keywords

# protocol -> the ``_make_data`` key it returns that nothing declares
recorded = {Frame: 'ts_src', L2TPv2: 'prio', Header: 'magic_number'}
recorded = {Header: 'magic_number'}

for protocol, key in recorded.items():
with self.subTest(protocol=protocol.__name__):
Expand Down
86 changes: 86 additions & 0 deletions tests/protocols/test_make_data_keys_unit.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,86 @@
# -*- coding: utf-8 -*-
"""``_make_data`` returns the keywords ``make()`` takes.

GitHub issue #1097: :meth:`Frame._make_data
<pcapkit.protocols.misc.pcap.frame.Frame._make_data>` returned ``ts_src`` where
:meth:`Frame.make <pcapkit.protocols.misc.pcap.frame.Frame.make>` takes
``ts_sec``, and :meth:`L2TPv2._make_data
<pcapkit.protocols.link.l2tpv2.L2TPv2._make_data>` returned ``prio`` where
:meth:`L2TPv2.make <pcapkit.protocols.link.l2tpv2.L2TPv2.make>` takes
``priority``. :meth:`~pcapkit.protocols.protocol.ProtocolBase.from_data`
discards such a key with an
:class:`~pcapkit.utilities.warnings.UnknownFieldWarning`, so the field was lost
on a round-trip. Each case here rebuilds a packet with ``from_data`` and checks
both that the field survives and that nothing was discarded.

Every case builds its own packet in memory, so this belongs to the unit tier.

"""
from __future__ import annotations

import importlib.util
import io
import unittest
import warnings
from unittest import mock

RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper')
HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS)


@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed')
class MakeDataKeysTests(unittest.TestCase):
"""A ``from_data`` round-trip keeps every field ``_make_data`` returns."""

def assertNoUnknownField(self, caught: 'list[warnings.WarningMessage]') -> None:
from pcapkit.utilities.warnings import UnknownFieldWarning

discarded = [str(item.message) for item in caught
if isinstance(item.message, UnknownFieldWarning)]
self.assertEqual(discarded, [])

def test_l2tpv2_priority_survives_from_data(self) -> None:
from pcapkit.protocols.link.l2tpv2 import L2TPv2

octets = bytes(L2TPv2(version=2, priority=True, tunnel_id=1, session_id=2))
parsed = L2TPv2(io.BytesIO(octets), len(octets))
self.assertTrue(parsed.info.flags.prio)

with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter('always')
rebuilt = L2TPv2.from_data(parsed.info)

self.assertNoUnknownField(caught)
self.assertTrue(rebuilt.info.flags.prio)
self.assertEqual(bytes(rebuilt), octets)

def test_frame_ts_sec_survives_from_data(self) -> None:
from pcapkit.protocols.misc.pcap.frame import Frame
from pcapkit.protocols.misc.pcap.header import Header

header = Header(network=1)
frame = Frame(num=1, header=header.info, ts_sec=1234, ts_usec=5,
packet=b'\x00' * 14)

# NOTE: ``Frame`` also needs ``num`` and ``header``, which are not part
# of its data, so wrap ``_make_data`` to supply them and still go
# through the real :meth:`from_data` and the real keys.
make_data = Frame._make_data

def with_context(cls: 'type', data: 'object') -> 'dict[str, object]':
return {**make_data(data), 'num': 1, 'header': header.info}

with mock.patch.object(Frame, '_make_data', classmethod(with_context)), \
warnings.catch_warnings(record=True) as caught:
warnings.simplefilter('always')
rebuilt = Frame.from_data(frame.info)

self.assertNoUnknownField(caught)
self.assertEqual(rebuilt.info.frame_info.ts_sec, 1234)
self.assertEqual(rebuilt.info.frame_info.ts_usec, 5)
self.assertEqual(rebuilt.info.frame_info.incl_len, 14)
self.assertEqual(rebuilt.info.frame_info.orig_len, 14)


if __name__ == '__main__':
unittest.main()
Loading