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: 9 additions & 1 deletion pcapkit/corekit/fields/field.py
Original file line number Diff line number Diff line change
Expand Up @@ -61,10 +61,18 @@ class FieldBase(Generic[_T], metaclass=FieldMeta):

if TYPE_CHECKING:
_name: 'str'
_default: '_T | NoValueType'
_template: 'str'
_callback: 'Callable[[Self, dict[str, Any]], None]'

# NOTE: Declared on the class, not only assigned in :meth:`__init__`, so that
# :attr:`default` is answerable for every field. A field class is free to
# replace :meth:`__init__` without chaining to this one -- as
# :class:`~pcapkit.corekit.fields.collections.ListField` does, since a list of
# fields takes no default value of its own -- and reading :attr:`default` off
# one of those raised :exc:`AttributeError` for a private attribute rather
# than reporting that the field declares no default. See #422.
_default: '_T | NoValueType' = NoValue

@property
def name(self) -> 'str':
"""Field name."""
Expand Down
110 changes: 98 additions & 12 deletions pcapkit/protocols/schema/schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -72,18 +72,41 @@ def schema_final(cls: '_ST', *, _finalised: 'bool' = True) -> '_ST':
args_ = [f'{key}=NoValue' for key in cls.__fields__]
dict_ = [f'{key}={key}' for key in cls.__fields__]

# NOTE: We shall only attempt to generate ``__init__`` method
# if the class does not define such method.
if not hasattr(cls, '__init__'):
# NOTE: We shall only attempt to generate ``__init__`` method if the class
# does not define such method -- which is a test on ``cls.__dict__``, not on
# ``hasattr``: every class inherits ``__init__`` from :obj:`object`, so
# ``hasattr(cls, '__init__')`` is unconditionally true and the generated
# method was never installed. ``Schema(...)`` therefore ran
# :meth:`Schema.__update__` alone and never reached
# :meth:`Schema.__post_init__`, leaving a schema built from a subset of its
# fields holding :class:`~pcapkit.corekit.fields.field.FieldBase` objects in
# place of the omitted values, so that it could not be packed at all and
# failed with an error naming a field class rather than a field. See #422.
#
# :class:`~pcapkit.protocols.schema.misc.null.NoPayload` is what the test
# protects: it declares an argument-less ``__init__`` of its own so that no
# generated one displaces it.
if '__init__' not in cls.__dict__:
# NOTE: We only generate typed ``__init__`` method if only the class
# has field definition from any of itself and its base classes.
if args_:
# NOTE: The following code is to make the ``__init__`` method work.
# It is inspired from the :func:`dataclasses._create_fn` function.
#
# ``**kwargs`` is forwarded rather than rejected, so that a keyword
# naming something other than a field keeps reaching
# :meth:`Schema.__update__` and drawing its
# :class:`~pcapkit.utilities.warnings.UnknownFieldWarning`, as it did
# while ``__init__`` *was* ``__update__``. Several schemas are
# constructed that way on purpose -- the Multipath TCP options take a
# ``kind`` and a ``length`` that the enclosing option owns and that
# ``MPTCP`` declares only for the type checker -- so a strict
# signature here would turn a warning into a :exc:`TypeError` on a
# path that has nothing to do with the missing ``__post_init__``.
init_ = (
f'def __create_fn__():\n'
f' def __init__(self, {", ".join(args_)}, *, __packet__=None):\n'
f' self.__update__({", ".join(dict_)})\n'
f' def __init__(self, {", ".join(args_)}, *, __packet__=None, **kwargs):\n'
f' self.__update__({", ".join(dict_)}, **kwargs)\n'
f' self.__post_init__(__packet__)\n'
f' return __init__\n'
)
Expand Down Expand Up @@ -278,10 +301,64 @@ def __new__(cls, *args: '_VT', **kwargs: '_VT') -> 'Self': # pylint: disable=un
return self

def __post_init__(self, packet: 'Optional[dict[str, Any]]' = None) -> 'None':
"""Fill in the fields the caller left unset.

Args:
packet: Packet data, as forwarded from the ``__packet__`` keyword
argument of the generated ``__init__``. The schema is packed
here only when one is given; see the note below.

"""
for name, field in self.__fields__.items():
if self.__dict__[name] in (NoValue, None):
self.__dict__[name] = field.default
self.pack(packet)
# NOTE: Read with a fallback rather than by subscript, since the
# generated ``__init__`` is not the only caller: :meth:`from_dict`
# seeds only the keys its argument carries, so a field the caller left
# out is missing from ``__dict__`` entirely rather than holding
# ``NoValue``, and subscripting it raised :exc:`KeyError` naming the
# field.
#
# What is tested is ``NoValue`` alone, not ``NoValue`` or ``None``.
# This method fills in what the caller did not say, and a ``None`` the
# caller passed *is* something said: on an optional field it is the
# chosen value, meaning this packet does not carry the field. It is
# also what :meth:`unpack` stores for a
# :class:`~pcapkit.corekit.fields.misc.ConditionalField` whose test
# fails -- including one that declares a default of its own -- so
# substituting the default here would leave a constructed schema
# disagreeing with a parsed one about the same packet, and
# ``from_dict(parsed.to_dict())`` no longer reproducing what it was
# given. Telling the two apart is what ``NoValue`` is for.
value = self.__dict__.get(name, NoValue)
if value is not NoValue:
continue
Comment thread
JarryShaw marked this conversation as resolved.

default = field.default
if default is not NoValue:
self.__dict__[name] = default
else:
# NOTE: Nothing to fill an unset field with, so the ``NoValue``
# the generated ``__init__`` seeded it with is dropped rather than
# kept: it is a *field* sentinel, not a value a schema may hold.
# Dropping it rather than storing ``None`` also keeps the name out
# of the context :meth:`pack` builds from ``__dict__``, which a
# schema may be relying on to seed for itself -- the PCAP-NG
# section header block reads its Byte-Order Magic from a ``match``
# its own :meth:`pre_pack` supplies, and only when the context
# does not name one already. :meth:`pack` reads an absent field as
# ``None`` regardless.
self.__dict__.pop(name, None)

# NOTE: Packed here only when a packet context was actually handed over.
# A schema is not in general packable from its own fields alone: a field
# callback may read a key that the *enclosing* layer owns, as the
# Multipath TCP options do with the ``length`` of the TCP option that
# carries them, and packing without it raises rather than producing
# octets. ``__updated__`` is still set, so a schema left unpacked here is
# packed by :meth:`__bytes__` on first use -- by which time the enclosing
# layer has supplied the context, which is where the octets were produced
# before this method ran on construction at all.
if packet is not None:
self.pack(packet)

def __update__(self, dict_: 'Optional[Mapping[str, _VT] | Iterable[tuple[str, _VT]]]' = None,
**kwargs: '_VT') -> 'None':
Expand Down Expand Up @@ -508,11 +585,22 @@ class will consider negative value as a placeholder.
for field in self.__fields__.values():
field = field(packet)

# NOTE: Read from the instance rather than with :func:`getattr`, which
# finds the *class* attribute when the instance has none -- and a
# schema's class attribute for a field is the
# :class:`~pcapkit.corekit.fields.field.FieldBase` object itself. A
# field the caller never set therefore arrived below as the field
# rather than as a value: ``getattr(self, name, None)`` could not
# return its ``None`` for one, so the absent-value branches never
# fired, and what surfaced instead was a failure from inside the
# packing of a field object -- naming a field *class*, and so saying
# nothing about which field had been left out. See #422.
data = self.__dict__.get(self.__map__.get(field.name, field.name))

if isinstance(field, PayloadField):
from pcapkit.protocols.protocol import \
Protocol # pylint: disable=import-outside-toplevel

data = getattr(self, field.name, None)
if data is None:
self.__buffer__[field.name] = b''
elif isinstance(data, Protocol):
Expand All @@ -526,7 +614,6 @@ class will consider negative value as a placeholder.
continue

if isinstance(field, ListField):
data = getattr(self, field.name, None)
if data is None:
self.__buffer__[field.name] = b''
elif isinstance(data, bytes):
Expand All @@ -551,9 +638,8 @@ class will consider negative value as a placeholder.
self.__buffer__[field.name] = b''
continue

value = getattr(self, field.name)
try:
temp = field.pack(value, packet)
temp = field.pack(data, packet)
except NoDefaultValue:
temp = bytes(field.length)
self.__buffer__[field.name] = temp
Expand Down
99 changes: 87 additions & 12 deletions tests/protocols/schema/test_schema_unit.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
from __future__ import annotations

import builtins
import collections
import enum
import importlib.util
Expand Down Expand Up @@ -73,6 +72,9 @@ def test_schema_pack_unpack_and_mapping_methods(self) -> None:
self.assertEqual(schema['kind'], 9)
self.assertIn('kind=9', str(schema))
self.assertIn('NestedSchema(...)', repr(schema))
# ``pad`` is absent: the generated ``__init__`` seeds every field, but
# ``__post_init__`` keeps only the ones it can fill, and a padding field
# declaring no default has nothing to be filled with
self.assertEqual(list(schema), ['kind', 'maybe', 'peek', 'repeated', 'nested', 'payload'])

as_dict = schema.to_dict()
Expand All @@ -92,6 +94,9 @@ def test_schema_pack_unpack_and_mapping_methods(self) -> None:
self.assertEqual(unpacked.repeated, [8, 9])
self.assertEqual(unpacked.nested.marker, 0x33)
self.assertEqual(unpacked.payload, b'zz')
# unpacking reads every field off the wire, so unlike the construction
# above it leaves none of them absent
self.assertEqual(list(unpacked), list(FeatureSchema.__fields__))

def test_schema_update_unknown_fields_and_builtin_field_mapping(self) -> None:
_, _, _, _, BuiltinNameSchema = self._make_schema_classes()
Expand Down Expand Up @@ -225,23 +230,14 @@ def test_schema_final_generated_init_and_legacy_version_branch(self) -> None:
from pcapkit.corekit.fields.numbers import UInt8Field
from pcapkit.protocols.schema.schema import Schema, schema_final

@schema_final
class GeneratedInitSchema(Schema):
value: int = UInt8Field(default=1)

@schema_final
class EmptyGeneratedSchema(Schema):
pass

original_hasattr = builtins.hasattr

def fake_hasattr(obj: object, name: str) -> bool:
if obj in (GeneratedInitSchema, EmptyGeneratedSchema) and name == '__init__':
return False
return original_hasattr(obj, name)

with mock.patch('builtins.hasattr', side_effect=fake_hasattr):
GeneratedInitSchema = schema_final(GeneratedInitSchema)
EmptyGeneratedSchema = schema_final(EmptyGeneratedSchema)

self.assertEqual(bytes(GeneratedInitSchema(value=2)), b'\x02')
self.assertEqual(bytes(GeneratedInitSchema()), b'\x01')
self.assertEqual(bytes(EmptyGeneratedSchema()), b'')
Expand All @@ -252,6 +248,85 @@ class LegacyVersionSchema(Schema):

self.assertIn('value', LegacyVersionSchema.__fields__)

def test_generated_init_is_installed_and_runs_post_init(self) -> None:
from pcapkit.corekit.fields.collections import ListField
from pcapkit.corekit.fields.numbers import UInt8Field, UInt16Field
from pcapkit.protocols.schema.misc.null import NoPayload
from pcapkit.protocols.schema.schema import Schema, schema_final

@schema_final
class HeaderSchema(Schema):
kind: int = UInt8Field(default=3)
size: int = UInt16Field(default=0x0102)
spare: int = UInt8Field()
trailer: list[int] = ListField(length=2, item_type=UInt8Field())

# the guard read ``hasattr(cls, '__init__')``, which every class satisfies
# through :obj:`object`, so the generated method was never installed and
# ``__init__`` stayed bound to ``Schema.__update__``
self.assertIsNot(HeaderSchema.__init__, Schema.__update__)
self.assertEqual(HeaderSchema.__init__.__qualname__, 'HeaderSchema.__init__')

schema = HeaderSchema(kind=9)

# ``__post_init__`` ran: ``size`` carries its declared default, and the
# two that declare none are left absent rather than holding the
# ``NoValue`` the generated ``__init__`` seeded them with
self.assertEqual(schema.to_dict(), {'kind': 9, 'size': 0x0102})

# so the schema packs, where before the fix the fields left out reached
# the packing as the field objects themselves
self.assertEqual(bytes(schema), b'\x09\x01\x02\x00')

# and the two construction paths now agree
self.assertEqual(bytes(HeaderSchema.from_dict({'kind': 9})), bytes(schema))

# a packet context is what makes packing at construction possible, so it
# is what asks for it
eager = HeaderSchema(kind=9, __packet__={})
self.assertFalse(eager.__updated__)
self.assertEqual(eager.__buffer__['size'], b'\x01\x02')

# a schema declaring an ``__init__`` of its own keeps it
self.assertEqual(NoPayload.__init__.__qualname__, 'NoPayload.__init__')
self.assertEqual(bytes(NoPayload()), b'')

def test_post_init_fills_the_unset_and_keeps_an_explicit_none(self) -> None:
from pcapkit.corekit.fields.misc import ConditionalField
from pcapkit.corekit.fields.numbers import UInt8Field
from pcapkit.protocols.schema.schema import Schema, schema_final

@schema_final
class OptionalSchema(Schema):
kind: int = UInt8Field(default=1)
#: declares a default of its own, and is off the wire unless kind is 9
maybe: int = ConditionalField(UInt8Field(default=0xCC),
lambda packet: packet['kind'] == 9)
#: declares no default
spare: int = UInt8Field()

# a field the caller left out is filled from its declared default, and one
# declaring none is left absent rather than holding ``NoValue``
self.assertEqual(OptionalSchema(kind=9).to_dict(), {'kind': 9, 'maybe': 0xCC})

# a ``None`` the caller passed is a value they chose, not a field they
# omitted, so it survives even where the field declares a default
self.assertEqual(OptionalSchema(kind=9, maybe=None, spare=None).to_dict(),
{'kind': 9, 'maybe': None, 'spare': None})

# keeping it costs nothing on the wire, since ``FieldBase.pack`` resolves a
# ``None`` from the field's own default anyway
self.assertEqual(bytes(OptionalSchema(kind=9, maybe=None)),
bytes(OptionalSchema(kind=9)))

# and it is what keeps a constructed schema agreeing with a parsed one:
# ``unpack`` stores ``None`` for a conditional field whose test fails, so
# substituting the default would stop ``to_dict`` surviving ``from_dict``
parsed = OptionalSchema.unpack(b'\x01\x07', 2, None)
self.assertIsNone(parsed.maybe)
self.assertEqual(OptionalSchema.from_dict(parsed.to_dict()).to_dict(),
parsed.to_dict())

def test_schema_mapping_payload_list_and_default_edge_branches(self) -> None:
NestedSchema, FeatureSchema, PayloadOnlySchema, _, _ = self._make_schema_classes()
from pcapkit.corekit.fields.field import NoValue
Expand Down