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
10 changes: 8 additions & 2 deletions pcapkit/utilities/decorators.py
Original file line number Diff line number Diff line change
Expand Up @@ -206,8 +206,14 @@ def prepare(func: 'Callable[Concatenate[Type[R_prepare], bytes | IO[bytes], Opti
def unpack(*args: 'P.args', **kwargs: 'P.kwargs') -> 'R_prepare':
cls = cast('Type[R_prepare]', args[0])
data = cast('bytes | IO[bytes]', args[1])
length = cast('Optional[int]', args[2])
packet = cast('Optional[dict[str, Any]]', args[3])
# ``length`` and ``packet`` are optional, both in the decorated
# signature and here: a caller may omit them, pass them positionally,
# or pass them by keyword. ``args`` only has an ``[2]``/``[3]`` to
# subscript when the caller supplied that many positionals, so fall
# back to ``kwargs`` -- and to the documented default of ``None`` --
# rather than assuming the position is always filled.
length = cast('Optional[int]', args[2] if len(args) > 2 else kwargs.pop('length', None))
packet = cast('Optional[dict[str, Any]]', args[3] if len(args) > 3 else kwargs.pop('packet', None))
Comment thread
JarryShaw marked this conversation as resolved.

if isinstance(data, bytes):
length = len(data) if length is None else length
Expand Down
48 changes: 48 additions & 0 deletions tests/corekit/test_fields_misc.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,48 @@
from __future__ import annotations

import unittest

from tests._support import purge_modules


class SchemaFieldDefaultTests(unittest.TestCase):
"""Regression coverage for `#444 <https://github.com/JarryShaw/PyPCAPKit/issues/444>`__.

:class:`~pcapkit.corekit.fields.misc.SchemaField` accepts a documented,
typed ``default: bytes`` constructor argument and unpacks it eagerly, via
``schema.unpack(default)`` -- a single positional argument. That call
shape reached
:func:`~pcapkit.utilities.decorators.prepare`, which read ``length`` and
``packet`` out of ``args[2]``/``args[3]`` unconditionally, so any call
shorter than three positional arguments raised ``IndexError`` instead of
falling back to the documented ``None`` default -- making
``SchemaField(schema=..., default=b'...')`` unusable for exactly the
``bytes`` default it exists to accept.

"""

def setUp(self) -> None:
purge_modules(['pcapkit'])

from pcapkit.corekit.fields.misc import SchemaField
from pcapkit.corekit.fields.numbers import UInt8Field
from pcapkit.protocols.schema.schema import Schema, schema_final

@schema_final
class TwoField(Schema):
a: 'int' = UInt8Field()
b: 'int' = UInt8Field()

self.SchemaField = SchemaField
self.TwoField = TwoField

def test_schema_field_accepts_a_bytes_default(self) -> None:
field = self.SchemaField(schema=self.TwoField, default=b'\x01\x02')

self.assertIsInstance(field.default, self.TwoField)
self.assertEqual(field.default.a, 1)
self.assertEqual(field.default.b, 2)


if __name__ == '__main__':
unittest.main()
60 changes: 60 additions & 0 deletions tests/utilities/test_decorators.py
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,66 @@ def unpack(cls, data, length=None, packet=None):
self.assertEqual(result['__length__'], 7)
self.assertEqual(result['data'], b'payload')

def _demo_schema_for_call_shapes(self):
"""A ``DemoSchema`` whose ``unpack`` records what ``prepare`` bound.

Shared by the five call-shape tests below, one per row of `#444
<https://github.com/JarryShaw/PyPCAPKit/issues/444>`__'s reproduction
table: ``prepare`` used to subscript ``args[2]``/``args[3]``
unconditionally, so only the three-positional shape worked and every
shorter or keyword call raised ``IndexError`` instead of falling back
to the documented ``length=None, packet=None`` defaults.

"""
class DemoSchema:
@classmethod
def pre_unpack(cls, packet: dict[str, object]) -> None:
packet['prepped'] = True

def __init__(self, data: bytes) -> None:
self.data = data

def post_process(self, packet: dict[str, object]) -> dict[str, object]:
packet['data'] = self.data
return packet

@classmethod
@self.decorators.prepare
def unpack(cls, data, length=None, packet=None):
return cls(data.read())

return DemoSchema

def _assert_call_shape_result(self, packet: dict[str, object]) -> None:
self.assertEqual(packet['__length__'], 7)
self.assertTrue(packet['prepped'])
self.assertEqual(packet['data'], b'payload')

def test_prepare_accepts_data_only(self) -> None:
"""``unpack(data)`` -- length and packet both omitted."""
DemoSchema = self._demo_schema_for_call_shapes()
self._assert_call_shape_result(DemoSchema.unpack(b'payload'))

def test_prepare_accepts_data_and_positional_length(self) -> None:
"""``unpack(data, length)`` -- packet omitted."""
DemoSchema = self._demo_schema_for_call_shapes()
self._assert_call_shape_result(DemoSchema.unpack(b'payload', 7))

def test_prepare_accepts_data_length_and_packet_positionally(self) -> None:
"""``unpack(data, length, packet)`` -- the one shape that already worked."""
DemoSchema = self._demo_schema_for_call_shapes()
self._assert_call_shape_result(DemoSchema.unpack(b'payload', 7, {}))

def test_prepare_accepts_keyword_length(self) -> None:
"""``unpack(data, length=2)`` -- length by keyword, packet omitted."""
DemoSchema = self._demo_schema_for_call_shapes()
self._assert_call_shape_result(DemoSchema.unpack(b'payload', length=7))

def test_prepare_accepts_keyword_length_and_packet(self) -> None:
"""``unpack(data, length=2, packet={})`` -- both trailing args by keyword."""
DemoSchema = self._demo_schema_for_call_shapes()
self._assert_call_shape_result(DemoSchema.unpack(b'payload', length=7, packet={}))

def test_prepare_raises_eof_for_empty_payloads(self) -> None:
class DemoSchema:
@classmethod
Expand Down
Loading