From efce64592032c812ce3d8d7d275a5ef932c2fc36 Mon Sep 17 00:00:00 2001 From: timrid <6593626+timrid@users.noreply.github.com> Date: Wed, 22 Jul 2026 12:40:39 +0200 Subject: [PATCH] typed: Added `Subconstruct`, `SymmetricAdapter`, `Tunnel` and `Validator` to `construct_typed` as subscriptable types. --- CHANGELOG.md | 4 + construct_typed/__init__.py | 8 ++ construct_typed/generic_wrapper.py | 42 +++++++++-- tests/test_typed.py | 116 ++++++++++++++++++++++++++++- 4 files changed, 160 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 63236bc..0c67fa0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,9 @@ # Changelog +## [UNRELEASED] +**New features:** +- Added `Subconstruct`, `SymmetricAdapter`, `Tunnel` and `Validator` to `construct_typed` as subscriptable types. + ## [0.8.0] - 2026-07-20 **Breaking changes:** - `csfield` should only be used for constructs that cannot build from `None`. Every other construct should use the new `csfield_noinit`, `csfield_const` or `csfield_default`. diff --git a/construct_typed/__init__.py b/construct_typed/__init__.py index 4fd89e3..8c46aa7 100644 --- a/construct_typed/__init__.py +++ b/construct_typed/__init__.py @@ -21,6 +21,10 @@ Context, ListContainer, PathType, + Subconstruct, + SymmetricAdapter, + Tunnel, + Validator, ) from .tenum import EnumBase, EnumValue, FlagsEnumBase, TEnum, TFlagsEnum @@ -43,6 +47,7 @@ "FlagsEnumBase", "TEnum", "TFlagsEnum", + "Subconstruct", "Adapter", "ConstantOrContextLambda", "Construct", @@ -50,4 +55,7 @@ "ListContainer", "PathType", "Array", + "SymmetricAdapter", + "Tunnel", + "Validator", ] diff --git a/construct_typed/generic_wrapper.py b/construct_typed/generic_wrapper.py index c6f1084..7a1a0d4 100644 --- a/construct_typed/generic_wrapper.py +++ b/construct_typed/generic_wrapper.py @@ -14,6 +14,10 @@ from construct import Array as Array from construct import Construct as Construct from construct import ListContainer as ListContainer + from construct import Subconstruct as Subconstruct + from construct import SymmetricAdapter as SymmetricAdapter + from construct import Tunnel as Tunnel + from construct import Validator as Validator from construct.core import ConstantOrContextLambda as ConstantOrContextLambda from construct.core import Context as Context from construct.core import PathType as PathType @@ -22,26 +26,50 @@ else: import construct as cs - # at runtime, the original classes are no generics, so whe have to make new classes with generics support + # at runtime, the original classes are not generic, so we have to make new classes with generics support class Construct(t.Generic[ParsedType, BuildTypes], cs.Construct): - pass + """Subscriptable version of `construct.Construct` that can be used as a generic type.""" + + class Subconstruct( + t.Generic[SubconParsedType, SubconBuildTypes, ParsedType, BuildTypes], + cs.Subconstruct, + ): + """Subscriptable version of `construct.Subconstruct` that can be used as a generic type.""" class Adapter( t.Generic[SubconParsedType, SubconBuildTypes, ParsedType, BuildTypes], cs.Adapter, ): - pass + """Subscriptable version of `construct.Adapter` that can be used as a generic type.""" - class ListContainer(t.Generic[ListType], cs.ListContainer): - pass + class Tunnel( + t.Generic[SubconParsedType, SubconBuildTypes], + cs.Tunnel, + ): + """Subscriptable version of `construct.Tunnel` that can be used as a generic type.""" - class Context: - pass + class SymmetricAdapter( + t.Generic[SubconParsedType, SubconBuildTypes, ParsedType, BuildTypes], + cs.SymmetricAdapter, + ): + """Subscriptable version of `construct.SymmetricAdapter` that can be used as a generic type.""" + + class Validator( + t.Generic[SubconParsedType, SubconBuildTypes], + cs.Validator, + ): + """Subscriptable version of `construct.Validator` that can be used as a generic type.""" class Array( t.Generic[SubconParsedType, SubconBuildTypes], cs.Array, ): + """Subscriptable version of `construct.Array` that can be used as a generic type.""" + + class ListContainer(t.Generic[ListType], cs.ListContainer): + pass + + class Context: pass ConstantOrContextLambda = ValueType | t.Callable[[Context], t.Any] diff --git a/tests/test_typed.py b/tests/test_typed.py index ca537f0..060c63a 100644 --- a/tests/test_typed.py +++ b/tests/test_typed.py @@ -203,9 +203,7 @@ def test_dataclass_ifthenelse() -> None: @dataclasses.dataclass class IfThenElseTest(DataclassMixin): test_if: int | None = csfield(cs.If(False, cs.Int8ub)) - test_ifthenelse: int | None = csfield( - cs.IfThenElse(True, cs.Int8ub, cs.Pass) - ) + test_ifthenelse: int | None = csfield(cs.IfThenElse(True, cs.Int8ub, cs.Pass)) a = IfThenElseTest(test_if=None, test_ifthenelse=None) assert a.test_if is None @@ -745,3 +743,115 @@ class TestEnum(cst.FlagsEnumBase): assert TestEnum.Value_NoDoc.__doc__ == "" assert TestEnum.Value_NoDoc2.__doc__ == "" assert TestEnum(8).__doc__ == "missing value" + + +def test_construct_subscriptable() -> None: + """Construct has to be subscriptable at runtime.""" + + class MyByte(cst.Construct[int, int]): + def _parse( + self, stream: t.IO[bytes], context: cst.Context, path: cst.PathType + ) -> int: + return cs.stream_read(stream, 1, path)[0] + + def _build( + self, + obj: int, + stream: t.IO[bytes], + context: cst.Context, + path: cst.PathType, + ) -> int: + cs.stream_write(stream, bytes([obj]), 1, path) + return obj + + def _sizeof(self, context: cst.Context, path: cst.PathType) -> int: + return 1 + + fmt = MyByte() + assert fmt.parse(b"\x05") == 5 + assert fmt.build(5) == b"\x05" + assert fmt.sizeof() == 1 + + +def test_subconstruct_subscriptable() -> None: + """Subconstruct has to be subscriptable at runtime.""" + + class MyPaddedBytes(cst.Subconstruct[bytes, bytes, bytes, bytes]): + def __init__(self) -> None: + pass + + fmt = MyPaddedBytes() + assert fmt is not None + + +def test_adapter_subscriptable() -> None: + """Adapter has to be subscriptable at runtime.""" + + class IpAddressAdapter(cst.Adapter[cst.ListContainer[int], list[int], str, str]): + def _encode( + self, obj: str, context: cst.Context, path: cst.PathType + ) -> list[int]: + return list(map(int, obj.split("."))) + + def _decode( + self, obj: list[int], context: cst.Context, path: cst.PathType + ) -> str: + return "{0}.{1}.{2}.{3}".format(*obj) + + IpAddress2 = IpAddressAdapter(cs.Byte[4]) + + assert IpAddress2.parse(b"\x7f\x80\x81\x82") == "127.128.129.130" + assert IpAddress2.build("127.1.2.3") == b"\x7f\x01\x02\x03" + assert IpAddress2.sizeof() == 4 + + +def test_symmetric_adapter_subscriptable() -> None: + """SymmetricAdapter has to be subscriptable at runtime.""" + + T = t.TypeVar("T") + + class ReversedList(cst.SymmetricAdapter[list[T], list[T], list[T], list[T]]): + def _decode( + self, obj: list[T], context: cst.Context, path: cst.PathType + ) -> list[T]: + return list(reversed(obj)) + + assert ReversedList(cs.Array(4, cs.Byte)).build([1, 2, 3, 4]) == b"\x04\x03\x02\x01" + assert ReversedList(cs.Array(4, cs.Byte)).parse(b"\x01\x02\x03\x04") == [4, 3, 2, 1] + + +def test_validator_subscriptable() -> None: + """Validator has to be subscriptable at runtime.""" + + class OneOf(cst.Validator[int, int]): + def _validate( + self, obj: int, context: cst.Context, path: cst.PathType + ) -> bool: + return obj in (1, 2, 3) + + fmt = OneOf(cs.Byte) + + assert fmt.parse(b"\x01") == 1 + assert fmt.build(2) == b"\x02" + assert raises(fmt.build, 4) is cs.ValidationError + assert raises(fmt.parse, b"\x04") is cs.ValidationError + + +def test_tunnel_subscriptable() -> None: + """Tunnel has to be subscriptable at runtime.""" + + class ROT1(cst.Tunnel[bytes, bytes]): + def _decode( + self, data: bytes, context: cst.Context, path: cst.PathType + ) -> bytes: + return bytes((b - 1) % 256 for b in data) + + def _encode( + self, data: bytes, context: cst.Context, path: cst.PathType + ) -> bytes: + return bytes((b + 1) % 256 for b in data) + + fmt = ROT1(cs.GreedyBytes) + + assert fmt.parse(b"\x02\x03\x04") == b"\x01\x02\x03" + assert fmt.build(b"\x01\x02\x03") == b"\x02\x03\x04"