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
39 changes: 22 additions & 17 deletions pcapkit/protocols/schema/misc/pcapng.py
Original file line number Diff line number Diff line change
Expand Up @@ -399,14 +399,19 @@ class Option(EnumSchema[Enum_OptionType]):
)

def __init_subclass__(cls, /, code: 'Optional[Enum_OptionType | Iterable[Enum_OptionType]]' = None,
namespace: 'Optional[str]' = None, *args: 'Any', **kwargs: 'Any') -> 'None':
ns: 'Optional[str]' = None, *args: 'Any', **kwargs: 'Any') -> 'None':
"""Register option type to :attr:`__enum__` mapping.

Args:
code: Option type code. It can be either a single option type enumeration
or a list of option type enumerations.
namespace: Namespace of option type enumeration. If not given, the value
will be inferred from the option type code.
ns: Namespace of option type enumeration. If not given, the value
will be inferred from the option type code. Spelled ``ns`` rather
than ``namespace`` because :meth:`abc.ABCMeta.__new__` names its
own fourth parameter ``namespace``, and before Python 3.11 that
parameter is positional-or-keyword rather than positional-only --
so a class keyword literally called ``namespace`` bound it twice.
See GitHub issue #439.
*args: Arbitrary positional arguments.
**kwargs: Arbitrary keyword arguments.

Expand All @@ -421,25 +426,25 @@ def __init_subclass__(cls, /, code: 'Optional[Enum_OptionType | Iterable[Enum_Op
from pcapkit.const.pcapng.option_type import OptionType as Enum_OptionType
from pcapkit.protocols.schema.misc.pcapng improt Option

class NewOption(Option, namespace='opt', code=Enum_OptionType.opt_new):
class NewOption(Option, ns='opt', code=Enum_OptionType.opt_new):
...

See Also:
- :class:`pcapkit.const.pcapng.option_type.OptionType`

"""
if namespace is not None:
cls.__namespace__ = namespace
if ns is not None:
cls.__namespace__ = ns

if code is not None:
if namespace is None:
namespace = cast('Optional[str]', cls.__namespace__)
if ns is None:
ns = cast('Optional[str]', cls.__namespace__)

if not isinstance(code, Enum_OptionType):
for _code in code:
Option.register(_code, cls, namespace)
Option.register(_code, cls, ns)
else:
Option.register(code, cls, namespace)
Option.register(code, cls, ns)
super().__init_subclass__()

@staticmethod
Expand Down Expand Up @@ -472,7 +477,7 @@ def register(code: 'Enum_OptionType', cls: 'Type[Option]', ns: 'Optional[str]' =
length: 'int'


class _OPT_Option(Option, namespace='opt'):
class _OPT_Option(Option, ns='opt'):
"""Header schema for ``opt_*`` options."""

#: Option type.
Expand Down Expand Up @@ -623,7 +628,7 @@ def __init__(self, length: 'int', magic: 'Literal[0x1A2B3C4D]', major: 'int',
length2: 'int') -> 'None': ...


class _IF_Option(Option, namespace='if'):
class _IF_Option(Option, ns='if'):
"""Header schema for ``if_*`` options."""

#: Option type.
Expand Down Expand Up @@ -892,7 +897,7 @@ def __init__(self, length: 'int', linktype: 'int', snaplen: 'int',
options: 'list[Option | bytes] | bytes', length2: 'int') -> 'None': ...


class _EPB_Option(Option, namespace='epb'):
class _EPB_Option(Option, ns='epb'):
"""Header schema for ``epb_*`` options."""

#: Option type.
Expand Down Expand Up @@ -1159,7 +1164,7 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema':
def __init__(self, type: 'Enum_RecordType', length: 'int', ip: 'IPv6Address | str | bytes | int', resol: 'str') -> 'None': ...


class _NS_Option(Option, namespace='ns'):
class _NS_Option(Option, ns='ns'):
"""Header schema for ``ns_*`` options."""

#: Option type.
Expand Down Expand Up @@ -1266,7 +1271,7 @@ def __init__(self, length: 'int',
options: 'list[Option | bytes] | bytes', length2: 'int') -> 'None': ...


class _ISB_Option(Option, namespace='isb'):
class _ISB_Option(Option, ns='isb'):
"""Header schema for ``isb_*`` options."""

#: Option type.
Expand Down Expand Up @@ -1570,7 +1575,7 @@ class ZigBeeAPSKey(DSBSecrets, code=Enum_SecretsType.ZigBee_APS_Key):
def __init__(self, key: 'bytes', panid: 'int', addr_low: 'int', addr_high: 'int') -> 'None': ...


class _DSB_Option(Option, namespace='dsb'):
class _DSB_Option(Option, ns='dsb'):
"""Header schema for ``dsb_*`` options."""

#: Option type.
Expand Down Expand Up @@ -1643,7 +1648,7 @@ class CustomBlock(BlockType, code=[Enum_BlockType.Custom_Block_that_rewriters_ca
def __init__(self, length: 'int', pen: 'int', data: 'bytes', length2: 'int') -> 'None': ...


class _PACK_Option(Option, namespace='pack'):
class _PACK_Option(Option, ns='pack'):
"""Header schema for ``pack_*`` options."""

#: Option type.
Expand Down
83 changes: 76 additions & 7 deletions pcapkit/protocols/schema/schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@
import collections.abc
import io
import itertools
import sys
from typing import TYPE_CHECKING, Any, Generic, TypeVar, cast, final

from pcapkit.corekit.fields.collections import ListField, OptionField
Expand All @@ -16,7 +15,7 @@
from pcapkit.corekit.infoclass import FinalisedState
from pcapkit.utilities.compat import Mapping
from pcapkit.utilities.decorators import prepare
from pcapkit.utilities.exceptions import NoDefaultValue, ProtocolUnbound, stacklevel
from pcapkit.utilities.exceptions import NoDefaultValue, ProtocolUnbound, SchemaError, stacklevel
from pcapkit.utilities.warnings import SchemaWarning, UnknownFieldWarning, warn

if TYPE_CHECKING:
Expand Down Expand Up @@ -210,7 +209,55 @@ def __prepare__(cls, name: 'str', bases: 'tuple[type, ...]', /, **kwds: 'Any') -
fields.update(base.__fields__)
return collections.OrderedDict(__fields__=fields)

def __new__(cls, name: 'str', bases: 'tuple[type, ...]', attrs: 'dict[str, Any]', **kwargs: 'Any') -> 'Type[Schema]':
#: Class keywords that collide with a parameter this class does not
#: control, so a schema class declared with one fails with an opaque
#: ``TypeError`` from several frames away instead of a clear message here.
#: Two different collisions, both reserved:
#:
#: ``mcls``, ``name``, ``bases``, ``namespace`` collide with
#: :meth:`abc.ABCMeta.__new__` -- a *different* function from this one, one
#: level up the ``super().__new__(...)`` call below. Before Python 3.11
#: those four are positional-or-keyword there (from 3.11 they are
#: positional-only, ``def __new__(mcls, name, bases, namespace, /,
#: **kwargs)``), so a class keyword spelled the same as any of them binds
#: that parameter twice: ``TypeError: ABCMeta.__new__() got multiple
#: values for argument '...'``. That is GitHub issue #439's root cause --
#: ``namespace`` collided this way, which is why
#: :mod:`pcapkit.protocols.schema.misc.pcapng`'s ``Option`` subclasses
#: spell it ``ns=`` instead.
#:
#: ``cls`` collides one level *later*: every ``__init_subclass__`` is an
#: implicit classmethod, so ``cls`` is always bound as its first argument,
#: and a class keyword also spelled ``cls`` binds it twice there --
#: ``TypeError: Generic.__init_subclass__() got multiple values for
#: argument 'cls'`` for a :class:`Schema` subclass (:class:`Schema`
#: inherits :class:`typing.Generic`), or the equivalent from whichever
#: class in the MRO defines ``__init_subclass__`` first. This one is not
#: specific to ``abc.ABCMeta`` or to this metaclass at all -- it holds for
#: *any* Python class with any ``__init_subclass__`` in its MRO -- but it
#: is reserved here anyway since it is exactly the class of mistake this
#: guard exists to catch early and legibly.
#:
#: Listed explicitly rather than derived from
#: :func:`inspect.signature(abc.ABCMeta.__new__) <inspect.signature>` at
#: import time: the signature's positional-only marker differs across
#: Python versions, introspecting a CPython internal's exact shape to
#: guard against a CPython internal's exact shape is circular, and five
#: names that will not change are cheaper to read than the machinery to
#: recompute them.
#:
#: ``name``, ``bases`` and ``attrs`` -- *this* method's own parameters --
#: collide the same way one level earlier than ``ABCMeta.__new__``, on
#: every Python version, and are not in this set for that reason alone:
#: they are made positional-only below instead (the same fix CPython gave
#: ``ABCMeta.__new__`` in 3.11), which removes the collision rather than
#: merely naming it. ``name`` and ``bases`` stay in this set regardless,
#: because they still collide with ``ABCMeta.__new__``'s parameters of the
#: same name one level up; ``attrs`` does not appear anywhere downstream
#: and so needs no entry once it is positional-only here.
_RESERVED_CLASS_KWARGS = frozenset({'mcls', 'name', 'bases', 'namespace', 'cls'})

def __new__(cls, name: 'str', bases: 'tuple[type, ...]', attrs: 'dict[str, Any]', /, **kwargs: 'Any') -> 'Type[Schema]':
"""Create the schema class.

Args:
Expand All @@ -223,7 +270,22 @@ def __new__(cls, name: 'str', bases: 'tuple[type, ...]', attrs: 'dict[str, Any]'
:attr:`~Schema.__excluded__` fields from the base classes, as well as
populating both fields from the subclass attributes.

Raises:
SchemaError: If a class keyword in ``**kwargs`` collides with a
parameter of :meth:`abc.ABCMeta.__new__` or of
``__init_subclass__`` -- see :attr:`_RESERVED_CLASS_KWARGS`.

"""
if clash := cls._RESERVED_CLASS_KWARGS.intersection(kwargs):
raise SchemaError(
f'{name}: class keyword(s) {sorted(clash)!r} are reserved -- '
f'each collides with a same-named parameter of either '
f'abc.ABCMeta.__new__ or the implicit __init_subclass__ '
f'classmethod binding, and cannot be used as a class keyword '
f'on a Schema subclass; rename to a different spelling (see '
f'GitHub issue #439)'
)

if '__additional__' not in attrs:
attrs['__additional__'] = []
if '__excluded__' not in attrs:
Expand All @@ -235,10 +297,17 @@ def __new__(cls, name: 'str', bases: 'tuple[type, ...]', attrs: 'dict[str, Any]'
if hasattr(base, '__excluded__'):
attrs['__excluded__'].extend(name for name in base.__excluded__ if name not in attrs['__excluded__'])

# NOTE: for unknown reason, the following code will cause an error
# for duplicated keyword arguments in class definition.
if sys.version_info < (3, 11):
return type.__new__(cls, name, bases, attrs, **kwargs)
# See #439: this used to branch on ``sys.version_info < (3, 11)`` and
# call ``type.__new__`` directly below that, to dodge the ``namespace``
# collision described above. That branch skipped ``ABCMeta.__new__``'s
# call to ``abc._abc_init(cls)``, so no :class:`Schema` subclass ever
# got its own ``_abc_impl``, and every one of them fell through the
# MRO to :class:`collections.abc.Mapping`'s -- corrupting
# ``isinstance`` against *any* of them for as long as the process ran.
# The actual fix was renaming the one colliding class keyword that was
# actually in use, which means this can now call ``ABCMeta.__new__``
# unconditionally, on every supported version, like any other
# metaclass would.
return super().__new__(cls, name, bases, attrs, **kwargs) # type: ignore[return-value]


Expand Down
2 changes: 1 addition & 1 deletion tests/protocols/misc/test_pcapng_unit.py
Original file line number Diff line number Diff line change
Expand Up @@ -1261,7 +1261,7 @@ class LocalDefaultOption(schema_pcapng.Option, code=OptionType.opt_comment):

class LocalIterableOption(schema_pcapng.Option,
code=[OptionType.if_name],
namespace='localtest'):
ns='localtest'):
pass

self.assertIs(registry['opt'][OptionType.opt_comment], LocalDefaultOption)
Expand Down
Loading
Loading