Compare commits

..

No commits in common. "main" and "v0.6.2" have entirely different histories.
main ... v0.6.2

18 changed files with 188 additions and 483 deletions

View file

@ -1,10 +1,6 @@
name: CI name: CI
on: on: [push, pull_request]
push:
pull_request:
workflow_dispatch:
workflow_call:
jobs: jobs:
build: build:
@ -12,7 +8,7 @@ jobs:
strategy: strategy:
matrix: matrix:
os: ['ubuntu-latest', 'windows-latest'] os: ['ubuntu-latest', 'windows-latest']
python-version: [ '3.9', '3.10', '3.11', '3.12', '3.13' ] python-version: [ '3.7', '3.8', '3.9', '3.10', '3.11' ]
runs-on: ${{ matrix.os }} runs-on: ${{ matrix.os }}
name: OS ${{ matrix.os }}, Python ${{ matrix.python-version }} name: OS ${{ matrix.os }}, Python ${{ matrix.python-version }}
@ -39,7 +35,6 @@ jobs:
- name: Install pyright - name: Install pyright
run: | run: |
npm install -g pyright npm install -g pyright
pyright --version
# Install this package # Install this package
- name: Install this package - name: Install this package
@ -66,30 +61,3 @@ jobs:
- name: Run pyright - name: Run pyright
run: | run: |
pyright pyright
create_wheel_and_sdist:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.13'
architecture: x64
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install wheel build
- name: Build wheel and sdist
run: |
python -m build
- name: Upload wheel and sdist as artifact
uses: actions/upload-artifact@v4
with:
name: Package-Distributions-construct-typing
path: dist/

View file

@ -1,3 +1,6 @@
# This workflows will upload a Python Package using Twine when a release is created
# For more information see: https://help.github.com/en/actions/language-and-framework-guides/using-python-with-github-actions#publishing-to-package-registries
name: Upload Python Package name: Upload Python Package
on: on:
@ -5,26 +8,24 @@ on:
types: [created] types: [created]
jobs: jobs:
create_wheel_and_sdist:
name: create_wheel_and_sdist
uses: ./.github/workflows/main.yml
deploy: deploy:
needs: [ create_wheel_and_sdist ]
runs-on: ubuntu-latest runs-on: ubuntu-latest
environment: pypi
permissions:
id-token: write # IMPORTANT: this permission is mandatory for Trusted Publishing
steps: steps:
- uses: actions/checkout@v3 - uses: actions/checkout@v2
- name: Set up Python
- name: Download artifacts uses: actions/setup-python@v2
uses: actions/download-artifact@v4
with: with:
name: Package-Distributions-construct-typing python-version: '3.x'
path: ./dist - name: Install dependencies
run: |
- name: Publish package distributions to PyPI python -m pip install --upgrade pip
uses: pypa/gh-action-pypi-publish@release/v1 pip install setuptools wheel twine
- name: Build and publish
env:
TWINE_USERNAME: ${{ secrets.PYPI_USERNAME }}
TWINE_PASSWORD: ${{ secrets.PYPI_PASSWORD }}
run: |
python setup.py sdist bdist_wheel
twine upload dist/*

3
.gitignore vendored
View file

@ -129,6 +129,3 @@ dmypy.json
example_737 example_737
example_888 example_888
example_ksy.ksy example_ksy.ksy
# Test stuff
devtest/

View file

@ -1,12 +1,3 @@
## Modified version of "construct-typing" module used in my projects.
This modification features:
- **[EnhancedDataclassMixin](https://github.com/waszil/construct-typing/commit/479b51344bfd95149596a75ee574ac2e63c032df) with additional features**
- **ConstantOrContextLambda2 type**
- **Typing for Subconstruct**
- **Type hint for Computed**
- **Switch typing fixes**
The original README.md file was described down below:
# construct-typing # construct-typing
[![PyPI](https://img.shields.io/pypi/v/construct-typing)](https://pypi.org/project/construct-typing/) [![PyPI](https://img.shields.io/pypi/v/construct-typing)](https://pypi.org/project/construct-typing/)
![PyPI - Implementation](https://img.shields.io/pypi/implementation/construct-typing) ![PyPI - Implementation](https://img.shields.io/pypi/implementation/construct-typing)

View file

@ -17,10 +17,7 @@ from construct.lib import (
ListType, ListType,
RebufferedBytesIO, RebufferedBytesIO,
) )
from cryptography.hazmat.primitives.ciphers import Cipher from typing_extensions import Buffer
from cryptography.hazmat.primitives.ciphers.aead import AESCCM, AESGCM, ChaCha20Poly1305
from cryptography.hazmat.primitives.ciphers.modes import Mode
from typing_extensions import Buffer, TypeAlias
# unfortunately, there are a few duplications with "typing", e.g. Union and Optional, which is why the t. prefix must be used everywhere # unfortunately, there are a few duplications with "typing", e.g. Union and Optional, which is why the t. prefix must be used everywhere
@ -29,7 +26,7 @@ from typing_extensions import Buffer, TypeAlias
# - Higher Kinded Types: https://github.com/python/typing/issues/548 # - Higher Kinded Types: https://github.com/python/typing/issues/548
# - Higher Kinded Types: https://sobolevn.me/2020/10/higher-kinded-types-in-python # - Higher Kinded Types: https://sobolevn.me/2020/10/higher-kinded-types-in-python
ReadableBuffer: TypeAlias = Buffer ReadableBuffer: t.TypeAlias = Buffer
StreamType = t.IO[bytes] StreamType = t.IO[bytes]
FilenameType = t.Union[str, bytes, os.PathLike[str], os.PathLike[bytes]] FilenameType = t.Union[str, bytes, os.PathLike[str], os.PathLike[bytes]]
PathType = str PathType = str
@ -70,7 +67,6 @@ class RawCopyError(ConstructError): ...
class RotationError(ConstructError): ... class RotationError(ConstructError): ...
class ChecksumError(ConstructError): ... class ChecksumError(ConstructError): ...
class CancelParsing(ConstructError): ... class CancelParsing(ConstructError): ...
class CipherError(ConstructError): ...
# =============================================================================== # ===============================================================================
# used internally # used internally
@ -90,17 +86,6 @@ def stream_size(stream: StreamType) -> int: ...
def stream_iseof(stream: StreamType) -> bool: ... def stream_iseof(stream: StreamType) -> bool: ...
def evaluate(param: ConstantOrContextLambda2[T], context: Context) -> T: ... def evaluate(param: ConstantOrContextLambda2[T], context: Context) -> T: ...
class BytesIOWithOffsets(io.BytesIO):
@staticmethod
def from_reading(
stream: StreamType, length: int, path: PathType
) -> BytesIOWithOffsets: ...
def __init__(
self, contents: bytes, parent_stream: StreamType, offset: int
) -> None: ...
def tell(self) -> int: ...
def seek(self, offset: int, whence: int = ...) -> int: ...
# =============================================================================== # ===============================================================================
# abstract constructs # abstract constructs
# =============================================================================== # ===============================================================================
@ -150,20 +135,9 @@ class Construct(t.Generic[ParsedType, BuildTypes]):
) -> Renamed[ParsedType, BuildTypes]: ... ) -> Renamed[ParsedType, BuildTypes]: ...
def __add__(self, other: Construct[t.Any, t.Any]) -> Struct: ... def __add__(self, other: Construct[t.Any, t.Any]) -> Struct: ...
def __rshift__(self, other: Construct[t.Any, t.Any]) -> Sequence: ... def __rshift__(self, other: Construct[t.Any, t.Any]) -> Sequence: ...
def __getitem__(self, count: t.Union[int, t.Callable[[Context], int]]) -> Array[ def __getitem__(
ParsedType, self, count: t.Union[int, t.Callable[[Context], int]]
BuildTypes, ) -> Array[ParsedType, BuildTypes,]: ...
]: ...
def _parse(
self, stream: StreamType, context: Context, path: PathType
) -> ParsedType: ...
def _parsereport(
self, stream: StreamType, context: Context, path: PathType
) -> ParsedType: ...
def _build(
self, obj: BuildTypes, stream: StreamType, context: Context, path: PathType
) -> int: ...
def _sizeof(self, context: Context, path: PathType) -> int: ...
@t.type_check_only @t.type_check_only
class Context(Container[t.Any]): class Context(Container[t.Any]):
@ -195,7 +169,7 @@ class Subconstruct(
subcon: Construct[SubconParsedType, SubconBuildTypes], subcon: Construct[SubconParsedType, SubconBuildTypes],
) -> None: ... ) -> None: ...
@t.overload @t.overload
def __init__( # type: ignore def __init__(
self, self,
*args: t.Any, *args: t.Any,
**kwargs: t.Any, **kwargs: t.Any,
@ -247,20 +221,24 @@ class Compiled(Construct[t.Any, t.Any]):
# =============================================================================== # ===============================================================================
# bytes and bits # bytes and bits
# =============================================================================== # ===============================================================================
class Bytes(Construct[bytes, t.Union[bytes, bytearray, int]]): class Bytes(Construct[bytes, t.Union[bytes, int]]):
length: ConstantOrContextLambda[int] length: ConstantOrContextLambda[int]
def __init__( def __init__(
self, self,
length: ConstantOrContextLambda[int], length: ConstantOrContextLambda[int],
) -> None: ... ) -> None: ...
GreedyBytes: Construct[bytes, t.Union[bytes, bytearray]] GreedyBytes: Construct[bytes, bytes]
def Bitwise(subcon: Construct[SubconParsedType, SubconBuildTypes]) -> t.Union[ def Bitwise(
subcon: Construct[SubconParsedType, SubconBuildTypes]
) -> t.Union[
Transformed[SubconParsedType, SubconBuildTypes], Transformed[SubconParsedType, SubconBuildTypes],
Restreamed[SubconParsedType, SubconBuildTypes], Restreamed[SubconParsedType, SubconBuildTypes],
]: ... ]: ...
def Bytewise(subcon: Construct[SubconParsedType, SubconBuildTypes]) -> t.Union[ def Bytewise(
subcon: Construct[SubconParsedType, SubconBuildTypes]
) -> t.Union[
Transformed[SubconParsedType, SubconBuildTypes], Transformed[SubconParsedType, SubconBuildTypes],
Restreamed[SubconParsedType, SubconBuildTypes], Restreamed[SubconParsedType, SubconBuildTypes],
]: ... ]: ...
@ -622,7 +600,7 @@ class Check(Construct[None, None]):
func: ConstantOrContextLambda[bool], func: ConstantOrContextLambda[bool],
) -> None: ... ) -> None: ...
Error: Construct[t.NoReturn, t.NoReturn] Error: Construct[None, None]
class FocusedSeq(Construct[t.Any, t.Any]): class FocusedSeq(Construct[t.Any, t.Any]):
subcons: t.List[Construct[t.Any, t.Any]] subcons: t.List[Construct[t.Any, t.Any]]
@ -800,10 +778,6 @@ def If(
) -> IfThenElse[t.Optional[ThenParsedType], t.Optional[ThenBuildTypes]]: ... ) -> IfThenElse[t.Optional[ThenParsedType], t.Optional[ThenBuildTypes]]: ...
SwitchType = t.TypeVar("SwitchType") SwitchType = t.TypeVar("SwitchType")
SwitchParsedType = t.TypeVar("SwitchParsedType")
SwitchBuildTypes = t.TypeVar("SwitchBuildTypes")
SwitchDefaultParsedType = t.TypeVar("SwitchDefaultParsedType")
SwitchDefaultBuildTypes = t.TypeVar("SwitchDefaultBuildTypes")
class Switch(Construct[ParsedType, BuildTypes]): class Switch(Construct[ParsedType, BuildTypes]):
keyfunc: ConstantOrContextLambda[t.Any] keyfunc: ConstantOrContextLambda[t.Any]
@ -811,31 +785,17 @@ class Switch(Construct[ParsedType, BuildTypes]):
default: Construct[t.Any, t.Any] default: Construct[t.Any, t.Any]
@t.overload @t.overload
def __new__( def __new__(
cls: "type[Switch[SwitchParsedType | None, SwitchBuildTypes | None]]", cls: "type[Switch[int, t.Optional[int]]]",
keyfunc: ConstantOrContextLambda[SwitchType], keyfunc: ConstantOrContextLambda[SwitchType],
cases: dict[t.Any, Construct[SwitchParsedType, SwitchBuildTypes]], cases: t.Dict[SwitchType, Construct[int, int]],
default: None = ..., default: t.Optional[Construct[int, int]] = ...,
) -> Switch[SwitchParsedType | None, SwitchBuildTypes | None]: ... ) -> Switch[int, t.Optional[int]]: ...
@t.overload
def __new__(
cls: "type[Switch[SwitchParsedType, SwitchBuildTypes]]",
keyfunc: ConstantOrContextLambda[SwitchType],
cases: dict[t.Any, Construct[SwitchParsedType, SwitchBuildTypes]],
default: Construct[t.NoReturn, t.NoReturn],
) -> Switch[SwitchParsedType, SwitchBuildTypes]: ...
@t.overload
def __new__(
cls: "type[Switch[SwitchParsedType | SwitchDefaultParsedType, SwitchBuildTypes | SwitchDefaultBuildTypes]]",
keyfunc: ConstantOrContextLambda[SwitchType],
cases: dict[t.Any, Construct[SwitchParsedType, SwitchBuildTypes]],
default: Construct[SwitchDefaultParsedType, SwitchDefaultBuildTypes],
) -> Switch[SwitchParsedType | SwitchDefaultParsedType, SwitchBuildTypes | SwitchDefaultBuildTypes]: ...
@t.overload @t.overload
def __new__( def __new__(
cls: "type[Switch[t.Any, t.Any]]", cls: "type[Switch[t.Any, t.Any]]",
keyfunc: ConstantOrContextLambda[SwitchType], keyfunc: ConstantOrContextLambda[t.Any],
cases: dict[t.Any, Construct[t.Any, t.Any]], cases: t.Dict[t.Any, Construct[t.Any, t.Any]],
default: Construct[t.Any, t.Any] | None = ..., default: t.Optional[Construct[t.Any, t.Any]] = ...,
) -> Switch[t.Any, t.Any]: ... ) -> Switch[t.Any, t.Any]: ...
class StopIf(Construct[None, None]): class StopIf(Construct[None, None]):
@ -916,16 +876,6 @@ class Peek(
subcon: Construct[SubconParsedType, SubconBuildTypes], subcon: Construct[SubconParsedType, SubconBuildTypes],
) -> None: ... ) -> None: ...
class OffsettedEnd(
Subconstruct[SubconParsedType, SubconBuildTypes, SubconParsedType, SubconBuildTypes]
):
endoffset: ConstantOrContextLambda[int]
def __init__(
self,
endoffset: ConstantOrContextLambda[int],
subcon: Construct[SubconParsedType, SubconBuildTypes],
) -> None: ...
class Seek(Construct[int, None]): class Seek(Construct[int, None]):
at: ConstantOrContextLambda[int] at: ConstantOrContextLambda[int]
if sys.version_info >= (3, 8): if sys.version_info >= (3, 8):
@ -970,7 +920,9 @@ class RawCopy(
def ByteSwapped( def ByteSwapped(
subcon: Construct[SubconParsedType, SubconBuildTypes] subcon: Construct[SubconParsedType, SubconBuildTypes]
) -> Transformed[SubconParsedType, SubconBuildTypes]: ... ) -> Transformed[SubconParsedType, SubconBuildTypes]: ...
def BitsSwapped(subcon: Construct[SubconParsedType, SubconBuildTypes]) -> t.Union[ def BitsSwapped(
subcon: Construct[SubconParsedType, SubconBuildTypes]
) -> t.Union[
Transformed[SubconParsedType, SubconBuildTypes], Transformed[SubconParsedType, SubconBuildTypes],
Restreamed[SubconParsedType, SubconBuildTypes], Restreamed[SubconParsedType, SubconBuildTypes],
]: ... ]: ...
@ -990,10 +942,7 @@ class Prefixed(
def PrefixedArray( def PrefixedArray(
countfield: Construct[int, int], countfield: Construct[int, int],
subcon: Construct[SubconParsedType, SubconBuildTypes], subcon: Construct[SubconParsedType, SubconBuildTypes],
) -> Array[ ) -> Array[SubconParsedType, SubconBuildTypes,]: ...
SubconParsedType,
SubconBuildTypes,
]: ...
class FixedSized( class FixedSized(
Subconstruct[SubconParsedType, SubconBuildTypes, SubconParsedType, SubconBuildTypes] Subconstruct[SubconParsedType, SubconBuildTypes, SubconParsedType, SubconBuildTypes]
@ -1142,26 +1091,6 @@ class Rebuffered(
tailcutoff: t.Optional[int] = ..., tailcutoff: t.Optional[int] = ...,
) -> None: ... ) -> None: ...
class EncryptedSym(Tunnel[SubconParsedType, SubconBuildTypes]):
cipher: ConstantOrContextLambda2[Cipher[Mode]]
def __init__(
self,
subcon: Construct[SubconParsedType, SubconBuildTypes],
cipher: ConstantOrContextLambda2[Cipher[Mode]],
) -> None: ...
class EncryptedSymAead(Tunnel[SubconParsedType, SubconBuildTypes]):
cipher: ConstantOrContextLambda2[t.Union[AESGCM, AESCCM, ChaCha20Poly1305]]
nonce: ConstantOrContextLambda2[bytes]
associated_data: ConstantOrContextLambda2[bytes]
def __init__(
self,
subcon: Construct[SubconParsedType, SubconBuildTypes],
cipher: ConstantOrContextLambda2[t.Union[AESGCM, AESCCM, ChaCha20Poly1305]],
nonce: ConstantOrContextLambda2[bytes],
associated_data: ConstantOrContextLambda2[bytes] = ...,
) -> None: ...
# =============================================================================== # ===============================================================================
# lazy equivalents # lazy equivalents
# =============================================================================== # ===============================================================================
@ -1181,9 +1110,9 @@ class Lazy(
class LazyContainer(t.Generic[ContainerType], t.Dict[str, ContainerType]): class LazyContainer(t.Generic[ContainerType], t.Dict[str, ContainerType]):
def __getattr__(self, name: str) -> ContainerType: ... def __getattr__(self, name: str) -> ContainerType: ...
def __getitem__(self, index: t.Union[str, int]) -> ContainerType: ... def __getitem__(self, index: t.Union[str, int]) -> ContainerType: ...
def keys(self) -> t.Iterator[str]: ... # type: ignore def keys(self) -> t.Iterator[str]: ...
def values(self) -> t.List[ContainerType]: ... # type: ignore def values(self) -> t.List[ContainerType]: ...
def items(self) -> t.List[t.Tuple[str, ContainerType]]: ... # type: ignore def items(self) -> t.List[t.Tuple[str, ContainerType]]: ...
class LazyStruct(Construct[LazyContainer[t.Any], t.Optional[t.Dict[str, t.Any]]]): class LazyStruct(Construct[LazyContainer[t.Any], t.Optional[t.Dict[str, t.Any]]]):
subcons: t.List[Construct[t.Any, t.Any]] subcons: t.List[Construct[t.Any, t.Any]]

View file

@ -469,7 +469,7 @@ class ExprMixin(t.Generic[ReturnType], object):
@t.overload @t.overload
def __eq__(self: ExprMixin[float], other: ConstOrCallable[float]) -> BinExpr[bool]: ... def __eq__(self: ExprMixin[float], other: ConstOrCallable[float]) -> BinExpr[bool]: ...
@t.overload @t.overload
def __eq__(self, other: ConstOrCallable[t.Any]) -> BinExpr[t.Any]: ... # type: ignore def __eq__(self, other: t.Any) -> BinExpr[t.Any]: ...
# __ne__ ########################################################################################################### # __ne__ ###########################################################################################################
@t.overload @t.overload
@ -487,7 +487,7 @@ class ExprMixin(t.Generic[ReturnType], object):
@t.overload @t.overload
def __ne__(self: ExprMixin[float], other: ConstOrCallable[float]) -> BinExpr[bool]: ... def __ne__(self: ExprMixin[float], other: ConstOrCallable[float]) -> BinExpr[bool]: ...
@t.overload @t.overload
def __ne__(self, other: t.Any) -> BinExpr[t.Any]: ... # type: ignore def __ne__(self, other: t.Any) -> BinExpr[t.Any]: ...
# __neg__ ########################################################################################################## # __neg__ ##########################################################################################################
@t.overload @t.overload

View file

@ -19,7 +19,7 @@ def recursion_lock(
class Container(t.Generic[ContainerType], t.Dict[str, ContainerType]): class Container(t.Generic[ContainerType], t.Dict[str, ContainerType]):
def __getattr__(self, name: str) -> ContainerType: ... def __getattr__(self, name: str) -> ContainerType: ...
def update( # type: ignore def update(
self, self,
seqordict: t.Union[t.Dict[str, ContainerType], t.Tuple[str, ContainerType]], seqordict: t.Union[t.Dict[str, ContainerType], t.Tuple[str, ContainerType]],
) -> None: ... ) -> None: ...

View file

@ -9,19 +9,15 @@ from .dataclass_struct import (
TStructField, TStructField,
csfield, csfield,
sfield, sfield,
EnhancedDataclassMixin
) )
from .generic_wrapper import ( from .generic_wrapper import (
Adapter, Adapter,
ConstantOrContextLambda, ConstantOrContextLambda,
ConstantOrContextLambda2,
Construct, Construct,
Context, Context,
ListContainer, ListContainer,
PathType, PathType,
Array, Array
Subconstruct,
Computed,
) )
from .tenum import EnumBase, EnumValue, FlagsEnumBase, TEnum, TFlagsEnum from .tenum import EnumBase, EnumValue, FlagsEnumBase, TEnum, TFlagsEnum
@ -36,7 +32,6 @@ __all__ = [
"TStructField", "TStructField",
"csfield", "csfield",
"sfield", "sfield",
"EnhancedDataclassMixin",
"EnumBase", "EnumBase",
"EnumValue", "EnumValue",
"FlagsEnumBase", "FlagsEnumBase",
@ -44,12 +39,9 @@ __all__ = [
"TFlagsEnum", "TFlagsEnum",
"Adapter", "Adapter",
"ConstantOrContextLambda", "ConstantOrContextLambda",
"ConstantOrContextLambda2",
"Construct", "Construct",
"Context", "Context",
"ListContainer", "ListContainer",
"PathType", "PathType",
"Array", "Array"
"Subconstruct",
"Computed"
] ]

View file

@ -1,6 +1,5 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
# pyright: strict # pyright: strict
# pyright: reportIncompatibleVariableOverride=false, reportAny=false
import dataclasses import dataclasses
import textwrap import textwrap
import typing as t import typing as t
@ -12,7 +11,6 @@ from construct.lib.containers import (
recursion_lock, recursion_lock,
) )
from construct.lib.py3compat import bytestringtype, reprstring, unicodestringtype from construct.lib.py3compat import bytestringtype, reprstring, unicodestringtype
from typing_extensions import override
from .generic_wrapper import Adapter, Construct, Context, ParsedType, PathType from .generic_wrapper import Adapter, Construct, Context, ParsedType, PathType
@ -29,7 +27,7 @@ class DataclassMixin:
methods exists and every name can be used. methods exists and every name can be used.
""" """
__dataclass_fields__: "t.ClassVar[dict[str, dataclasses.Field[t.Any]]]" __dataclass_fields__: "t.ClassVar[t.Dict[str, dataclasses.Field[t.Any]]]"
def __getitem__(self, key: str) -> t.Any: def __getitem__(self, key: str) -> t.Any:
return getattr(self, key) return getattr(self, key)
@ -79,8 +77,8 @@ class DataclassMixin:
def csfield( def csfield(
subcon: Construct[ParsedType, t.Any], subcon: Construct[ParsedType, t.Any],
doc: str | None = None, doc: t.Optional[str] = None,
parsed: t.Callable[[t.Any, Context], None] | None = None, parsed: t.Optional[t.Callable[[t.Any, Context], None]] = None,
) -> ParsedType: ) -> ParsedType:
""" """
Helper method for "DataclassStruct" and "DataclassBitStruct" to create the dataclass fields. Helper method for "DataclassStruct" and "DataclassBitStruct" to create the dataclass fields.
@ -154,14 +152,18 @@ class DataclassStruct(Adapter[t.Any, t.Any, DataclassType, DataclassType]):
Image(width=1, height=2, pixels=b'12') Image(width=1, height=2, pixels=b'12')
""" """
subcon: "cs.Struct" # type: ignore subcon: "cs.Struct"
def __init__( def __init__(
self, self,
dc_type: type[DataclassType], dc_type: t.Type[DataclassType],
reverse: bool = False, reverse: bool = False,
) -> None: ) -> None:
self.dc_type: type[DataclassType] = dc_type if not issubclass(dc_type, DataclassMixin):
self.reverse: bool = reverse raise TypeError(f"'{repr(dc_type)}' has to be a '{repr(DataclassMixin)}'")
if not dataclasses.is_dataclass(dc_type):
raise TypeError(f"'{repr(dc_type)}' has to be a 'dataclasses.dataclass'")
self.dc_type = dc_type
self.reverse = reverse
# get all fields from the dataclass # get all fields from the dataclass
fields = dataclasses.fields(self.dc_type) fields = dataclasses.fields(self.dc_type)
@ -169,7 +171,7 @@ class DataclassStruct(Adapter[t.Any, t.Any, DataclassType, DataclassType]):
fields = tuple(reversed(fields)) fields = tuple(reversed(fields))
# extract the construct formats from the struct_type # extract the construct formats from the struct_type
subcon_fields: dict[str, t.Any] = {} subcon_fields = {}
for field in fields: for field in fields:
subcon_fields[field.name] = field.metadata["subcon"] subcon_fields[field.name] = field.metadata["subcon"]
@ -179,7 +181,6 @@ class DataclassStruct(Adapter[t.Any, t.Any, DataclassType, DataclassType]):
def __getattr__(self, name: str) -> t.Any: def __getattr__(self, name: str) -> t.Any:
return getattr(self.subcon, name) return getattr(self.subcon, name)
@override
def _decode( def _decode(
self, obj: "cs.Container[t.Any]", context: Context, path: PathType self, obj: "cs.Container[t.Any]", context: Context, path: PathType
) -> DataclassType: ) -> DataclassType:
@ -204,10 +205,9 @@ class DataclassStruct(Adapter[t.Any, t.Any, DataclassType, DataclassType]):
return dc # type: ignore return dc # type: ignore
@override
def _encode( def _encode(
self, obj: DataclassType, context: Context, path: PathType self, obj: DataclassType, context: Context, path: PathType
) -> dict[str, t.Any]: ) -> t.Dict[str, t.Any]:
if not isinstance(obj, self.dc_type): if not isinstance(obj, self.dc_type):
raise TypeError(f"'{repr(obj)}' has to be of type {repr(self.dc_type)}") raise TypeError(f"'{repr(obj)}' has to be of type {repr(self.dc_type)}")
@ -215,16 +215,20 @@ class DataclassStruct(Adapter[t.Any, t.Any, DataclassType, DataclassType]):
fields = dataclasses.fields(self.dc_type) fields = dataclasses.fields(self.dc_type)
# extract all fields from the container, that are used for create the dataclass object # extract all fields from the container, that are used for create the dataclass object
ret_dict: dict[str, t.Any] = {} ret_dict: t.Dict[str, t.Any] = {}
for field in fields: for field in fields:
value = getattr(obj, field.name) value = getattr(obj, field.name)
ret_dict[field.name] = value ret_dict[field.name] = value
return ret_dict return ret_dict
def DataclassBitStruct( def DataclassBitStruct(
dc_type: type[DataclassType], reverse: bool = False dc_type: t.Type[DataclassType], reverse: bool = False
) -> "cs.Transformed[DataclassType, DataclassType] | cs.Restreamed[DataclassType, DataclassType]": ) -> t.Union[
"cs.Transformed[DataclassType, DataclassType]",
"cs.Restreamed[DataclassType, DataclassType]",
]:
r""" r"""
Makes a DataclassStruct inside a Bitwise. Makes a DataclassStruct inside a Bitwise.
@ -251,29 +255,6 @@ def DataclassBitStruct(
""" """
return cs.Bitwise(DataclassStruct(dc_type, reverse)) return cs.Bitwise(DataclassStruct(dc_type, reverse))
class EnhancedDataclassMixin(DataclassMixin):
@classmethod
def format(cls):
return DataclassStruct(cls)
@classmethod
def build(cls, obj: t.Self, **kw: dict[str, t.Any]):
return cls.format().build(obj, **kw)
@classmethod
def parse(cls, data: bytes | bytearray, **kw: dict[str, t.Any]):
return cls.format().parse(data, **kw)
@classmethod
def parse_file(cls, file: str, **kw: dict[str, t.Any]):
return cls.format().parse_file(file, **kw)
@classmethod
def parse_stream(cls, stream: t.IO[bytes], **kw: dict[str, t.Any]):
return cls.format().parse_stream(stream, **kw)
def build_self(self) -> bytes:
return self.build(self)
# support legacy names # support legacy names
TStruct = DataclassStruct TStruct = DataclassStruct

View file

@ -12,14 +12,12 @@ if t.TYPE_CHECKING:
# while type checking, the original classes are already generics, because they are defined like this in the stubs. # while type checking, the original classes are already generics, because they are defined like this in the stubs.
from construct import Adapter as Adapter from construct import Adapter as Adapter
from construct import ConstantOrContextLambda as ConstantOrContextLambda from construct import ConstantOrContextLambda as ConstantOrContextLambda
from construct import ConstantOrContextLambda2 as ConstantOrContextLambda2
from construct import Construct as Construct from construct import Construct as Construct
from construct import Context as Context from construct import Context as Context
from construct import ListContainer as ListContainer from construct import ListContainer as ListContainer
from construct import PathType as PathType from construct import PathType as PathType
from construct import Array as Array from construct import Array as Array
from construct import Subconstruct as Subconstruct
from construct import Computed as Computed
else: else:
import construct as cs import construct as cs
@ -46,12 +44,5 @@ else:
): ):
pass pass
class Subconstruct(t.Generic[SubconParsedType, SubconBuildTypes, ParsedType, BuildTypes], cs.Subconstruct):
pass
class Computed(t.Generic[ParsedType], cs.Computed):
pass
ConstantOrContextLambda = t.Union[ValueType, t.Callable[[Context], t.Any]] ConstantOrContextLambda = t.Union[ValueType, t.Callable[[Context], t.Any]]
ConstantOrContextLambda2 = t.Union[ValueType, t.Callable[[Context], ValueType]]
PathType = str PathType = str

View file

@ -1,10 +1,9 @@
# pyright: reportAny=false
import enum import enum
import typing as t import typing as t
from typing_extensions import Self, override from typing_extensions import Self
from .generic_wrapper import Construct, Adapter, Context, PathType from .generic_wrapper import *
# ## TEnum ############################################################################################################ # ## TEnum ############################################################################################################
@ -13,8 +12,8 @@ class EnumValue:
This is a helper class for adding documentation to an enum value. This is a helper class for adding documentation to an enum value.
""" """
def __init__(self, value: int, doc: str | None = None) -> None: def __init__(self, value: int, doc: t.Optional[str] = None) -> None:
self.value: int = value self.value = value
self.__doc__ = doc if doc else "" self.__doc__ = doc if doc else ""
@ -48,7 +47,7 @@ class EnumBase(enum.IntEnum):
'This is the running state.' 'This is the running state.'
""" """
def __new__(cls, val: EnumValue | int) -> "Self": def __new__(cls, val: t.Union[EnumValue, int]) -> "Self":
if isinstance(val, EnumValue): if isinstance(val, EnumValue):
obj = int.__new__(cls, val.value) obj = int.__new__(cls, val.value)
obj._value_ = val.value obj._value_ = val.value
@ -63,8 +62,7 @@ class EnumBase(enum.IntEnum):
# not found in the enum, a new pseudo member is created. # not found in the enum, a new pseudo member is created.
# The idea is taken from: https://stackoverflow.com/a/57179436 # The idea is taken from: https://stackoverflow.com/a/57179436
@classmethod @classmethod
@override def _missing_(cls, value: t.Any) -> t.Optional[enum.Enum]:
def _missing_(cls, value: t.Any) -> enum.Enum | None:
if isinstance(value, int): if isinstance(value, int):
pseudo_member = cls._value2member_map_.get(value, None) pseudo_member = cls._value2member_map_.get(value, None)
if pseudo_member is None: if pseudo_member is None:
@ -78,8 +76,7 @@ class EnumBase(enum.IntEnum):
return pseudo_member return pseudo_member
return None # will raise the ValueError in Enum.__new__ return None # will raise the ValueError in Enum.__new__
@override def __reduce_ex__(self, proto: t.Any) -> t.Tuple[t.Any, ...]:
def __reduce_ex__(self, proto: t.Any) -> tuple[t.Any, ...]:
""" """
Pickle enums by value instead of name (restores pre-3.11 behavior). Pickle enums by value instead of name (restores pre-3.11 behavior).
See https://github.com/python/cpython/pull/26658 for why this exists. See https://github.com/python/cpython/pull/26658 for why this exists.
@ -94,18 +91,21 @@ class TEnum(Adapter[int, int, EnumType, EnumType]):
""" """
Typed enum. Typed enum.
""" """
def __init__(self, subcon: Construct[int, int], enum_type: type[EnumType]): def __init__(self, subcon: Construct[int, int], enum_type: t.Type[EnumType]):
if not issubclass(enum_type, EnumBase):
raise TypeError(
"'{}' has to be a '{}'".format(repr(enum_type), repr(EnumBase))
)
# save enum type # save enum type
self.enum_type: type[EnumType] = enum_type self.enum_type = t.cast(t.Type[EnumType], enum_type) # type: ignore
# init adatper # init adatper
super(TEnum, self).__init__(subcon) # type: ignore super(TEnum, self).__init__(subcon) # type: ignore
@override
def _decode(self, obj: int, context: Context, path: PathType) -> EnumType: def _decode(self, obj: int, context: Context, path: PathType) -> EnumType:
return self.enum_type(obj) return self.enum_type(obj)
@override
def _encode( def _encode(
self, self,
obj: EnumType, obj: EnumType,
@ -152,7 +152,7 @@ class FlagsEnumBase(enum.IntFlag):
'This is option two.' 'This is option two.'
""" """
def __new__(cls, val: EnumValue | int) -> "Self": def __new__(cls, val: t.Union[EnumValue, int]) -> "Self":
if isinstance(val, EnumValue): if isinstance(val, EnumValue):
obj = int.__new__(cls, val.value) obj = int.__new__(cls, val.value)
obj._value_ = val.value obj._value_ = val.value
@ -164,7 +164,6 @@ class FlagsEnumBase(enum.IntFlag):
return obj return obj
@classmethod @classmethod
@override
def _missing_(cls, value: t.Any) -> t.Any: def _missing_(cls, value: t.Any) -> t.Any:
""" """
Returns member (possibly creating it) if one can be found for value. Returns member (possibly creating it) if one can be found for value.
@ -173,8 +172,7 @@ class FlagsEnumBase(enum.IntFlag):
new_member.__doc__ = "missing value" new_member.__doc__ = "missing value"
return new_member return new_member
@override def __reduce_ex__(self, proto: t.Any) -> t.Tuple[t.Any, ...]:
def __reduce_ex__(self, proto: t.Any) -> tuple[t.Any, ...]:
""" """
Pickle enums by value instead of name (restores pre-3.11 behavior). Pickle enums by value instead of name (restores pre-3.11 behavior).
See https://github.com/python/cpython/pull/26658 for why this exists. See https://github.com/python/cpython/pull/26658 for why this exists.
@ -189,18 +187,21 @@ class TFlagsEnum(Adapter[int, int, FlagsEnumType, FlagsEnumType]):
""" """
Typed enum. Typed enum.
""" """
def __init__(self, subcon: Construct[int, int], enum_type: type[FlagsEnumType]): def __init__(self, subcon: Construct[int, int], enum_type: t.Type[FlagsEnumType]):
if not issubclass(enum_type, FlagsEnumBase):
raise TypeError(
"'{}' has to be a '{}'".format(repr(enum_type), repr(FlagsEnumBase))
)
# save enum type # save enum type
self.enum_type: type[FlagsEnumType] = enum_type self.enum_type = t.cast(t.Type[FlagsEnumType], enum_type) # type: ignore
# init adatper # init adatper
super(TFlagsEnum, self).__init__(subcon) # type: ignore super(TFlagsEnum, self).__init__(subcon) # type: ignore
@override
def _decode(self, obj: int, context: Context, path: PathType) -> FlagsEnumType: def _decode(self, obj: int, context: Context, path: PathType) -> FlagsEnumType:
return self.enum_type(obj) return self.enum_type(obj)
@override
def _encode( def _encode(
self, self,
obj: FlagsEnumType, obj: FlagsEnumType,

View file

@ -1,2 +1,2 @@
version = (0, 7, 0) version = (0, 6, 2)
version_string = "0.7.0+wrapper" version_string = "0.6.2"

3
mypy.ini Normal file
View file

@ -0,0 +1,3 @@
[mypy]
strict = True
warn_unused_ignores = False

View file

@ -1,76 +0,0 @@
[build-system]
requires = ["setuptools >= 75.8.0"]
build-backend = "setuptools.build_meta"
[project]
name="construct-typing"
dynamic = ["version"]
license = { file = "LICENSE" }
description="Extension for the python package 'construct' that adds typing features"
readme = "README.md"
authors=[{ name = "Tim Riddermann" }]
requires-python = ">=3.9"
dependencies = [
"construct==2.10.70",
"typing_extensions>=4.6.0"
]
keywords = [
"construct",
"kaitai",
"declarative",
"data structure",
"struct",
"binary",
"symmetric",
"parser",
"builder",
"parsing",
"building",
"pack",
"unpack",
"packer",
"unpacker",
"bitstring",
"bytestring",
"annotation",
"type hint",
"typing",
"typed",
"bitstruct",
"PEP 561",
]
classifiers = [
"Development Status :: 3 - Alpha",
"License :: OSI Approved :: MIT License",
"Intended Audience :: Developers",
"Topic :: Software Development :: Libraries :: Python Modules",
"Topic :: Software Development :: Build Tools",
"Topic :: Software Development :: Code Generators",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Programming Language :: Python :: Implementation :: CPython",
"Typing :: Typed",
]
[project.urls]
"Homepage" = "https://github.com/timrid/construct-typing"
"Bug Reports" = "https://github.com/timrid/construct-typing/issues"
[tool.setuptools]
packages=[
"construct-stubs",
"construct-stubs.lib",
"construct_typed"
]
[tool.setuptools.dynamic]
version = {attr = "construct_typed.version.version_string"}
[tool.mypy]
strict = true
warn_unused_ignores = false

View file

@ -1,4 +1,4 @@
construct==2.10.70 construct==2.10.68
pytest>=6.2.0 pytest>=6.2.0
numpy numpy
arrow arrow
@ -7,8 +7,4 @@ cloudpickle
lz4 lz4
black black
isort isort
mypy mypy
cryptography
build
setuptools
wheel

69
setup.py Normal file
View file

@ -0,0 +1,69 @@
#!/usr/bin/env python
from setuptools import setup
version_string = "?.?.?"
exec(open("./construct_typed/version.py").read())
setup(
name="construct-typing",
version=version_string,
packages=["construct-stubs", "construct_typed"],
package_data={
"construct-stubs": ["*.pyi", "lib/*.pyi"],
"construct_typed": ["py.typed"],
},
license="MIT",
license_files=("LICENSE",),
description="Extension for the python package 'construct' that adds typing features",
long_description=open("README.md").read(),
long_description_content_type="text/markdown",
platforms=["POSIX", "Windows"],
url="https://github.com/timrid/construct-typing",
author="Tim Riddermann",
python_requires=">=3.7",
install_requires=[
"construct==2.10.68",
"typing_extensions>=4.6.0"
],
keywords=[
"construct",
"kaitai",
"declarative",
"data structure",
"struct",
"binary",
"symmetric",
"parser",
"builder",
"parsing",
"building",
"pack",
"unpack",
"packer",
"unpacker",
"bitstring",
"bytestring",
"annotation",
"type hint",
"typing",
"typed",
"bitstruct",
"PEP 561",
],
classifiers=[
"Development Status :: 3 - Alpha",
"License :: OSI Approved :: MIT License",
"Intended Audience :: Developers",
"Topic :: Software Development :: Libraries :: Python Modules",
"Topic :: Software Development :: Build Tools",
"Topic :: Software Development :: Code Generators",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.7",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: Implementation :: CPython",
"Typing :: Typed",
],
)

View file

@ -151,29 +151,17 @@ def test_formatfield_bool_issue_901() -> None:
assert d.sizeof() == 1 assert d.sizeof() == 1
def test_bytesinteger() -> None: def test_bytesinteger() -> None:
d = BytesInteger(0)
assert raises(d.parse, b"") == IntegerError
assert raises(d.build, 0) == IntegerError
d = BytesInteger(4, signed=True, swapped=False) d = BytesInteger(4, signed=True, swapped=False)
common(d, b"\x01\x02\x03\x04", 0x01020304, 4) common(d, b"\x01\x02\x03\x04", 0x01020304, 4)
common(d, b"\xff\xff\xff\xff", -1, 4) common(d, b"\xff\xff\xff\xff", -1, 4)
d = BytesInteger(4, signed=False, swapped=this.swapped) d = BytesInteger(4, signed=False, swapped=this.swapped)
common(d, b"\x01\x02\x03\x04", 0x01020304, 4, swapped=False) common(d, b"\x01\x02\x03\x04", 0x01020304, 4, swapped=False)
common(d, b"\x04\x03\x02\x01", 0x01020304, 4, swapped=True) common(d, b"\x04\x03\x02\x01", 0x01020304, 4, swapped=True)
assert raises(BytesInteger(-1).parse, b"") == IntegerError
assert raises(BytesInteger(-1).build, 0) == IntegerError
assert raises(BytesInteger(8).build, None) == IntegerError
assert raises(BytesInteger(8, signed=False).build, -1) == IntegerError
assert raises(BytesInteger(8, True).build, -2**64) == IntegerError
assert raises(BytesInteger(8, True).build, 2**64) == IntegerError
assert raises(BytesInteger(8, False).build, -2**64) == IntegerError
assert raises(BytesInteger(8, False).build, 2**64) == IntegerError
assert raises(BytesInteger(this.missing).sizeof) == SizeofError assert raises(BytesInteger(this.missing).sizeof) == SizeofError
assert raises(BytesInteger(4, signed=False).build, -1) == IntegerError
common(BytesInteger(0), b"", 0, 0)
def test_bitsinteger() -> None: def test_bitsinteger() -> None:
d = BitsInteger(0)
assert raises(d.parse, b"") == IntegerError
assert raises(d.build, 0) == IntegerError
d = BitsInteger(8) d = BitsInteger(8)
common(d, b"\x01\x01\x01\x01\x01\x01\x01\x01", 255, 8) common(d, b"\x01\x01\x01\x01\x01\x01\x01\x01", 255, 8)
d = BitsInteger(8, signed=True) d = BitsInteger(8, signed=True)
@ -183,17 +171,9 @@ def test_bitsinteger() -> None:
d = BitsInteger(16, swapped=this.swapped) d = BitsInteger(16, swapped=this.swapped)
common(d, b"\x01\x01\x01\x01\x01\x01\x01\x01\x00\x00\x00\x00\x00\x00\x00\x00", 0xff00, 16, swapped=False) common(d, b"\x01\x01\x01\x01\x01\x01\x01\x01\x00\x00\x00\x00\x00\x00\x00\x00", 0xff00, 16, swapped=False)
common(d, b"\x00\x00\x00\x00\x00\x00\x00\x00\x01\x01\x01\x01\x01\x01\x01\x01", 0xff00, 16, swapped=True) common(d, b"\x00\x00\x00\x00\x00\x00\x00\x00\x01\x01\x01\x01\x01\x01\x01\x01", 0xff00, 16, swapped=True)
assert raises(BitsInteger(-1).parse, b"") == IntegerError
assert raises(BitsInteger(-1).build, 0) == IntegerError
assert raises(BitsInteger(5, swapped=True).parse, bytes(5)) == IntegerError
assert raises(BitsInteger(5, swapped=True).build, 0) == IntegerError
assert raises(BitsInteger(8).build, None) == IntegerError
assert raises(BitsInteger(8, signed=False).build, -1) == IntegerError
assert raises(BitsInteger(8, True).build, -2**64) == IntegerError
assert raises(BitsInteger(8, True).build, 2**64) == IntegerError
assert raises(BitsInteger(8, False).build, -2**64) == IntegerError
assert raises(BitsInteger(8, False).build, 2**64) == IntegerError
assert raises(BitsInteger(this.missing).sizeof) == SizeofError assert raises(BitsInteger(this.missing).sizeof) == SizeofError
assert raises(BitsInteger(8, signed=False).build, -1) == IntegerError
common(BitsInteger(0), b"", 0, 0)
def test_varint() -> None: def test_varint() -> None:
d = VarInt d = VarInt
@ -946,17 +926,6 @@ def test_peek() -> None:
assert d4.build(Container(a=0x01, b=0x0102)) == b"" assert d4.build(Container(a=0x01, b=0x0102)) == b""
assert d4.sizeof() == 0 assert d4.sizeof() == 0
def test_offsettedend() -> None:
d1 = Struct(
"header" / Bytes(2),
"data" / OffsettedEnd(-2, GreedyBytes),
"footer" / Bytes(2),
)
common(d1, b"\x01\x02\x03\x04\x05\x06\x07", Container(header=b'\x01\x02', data=b'\x03\x04\x05', footer=b'\x06\x07'))
d2 = OffsettedEnd(0, Byte)
assert raises(d2.sizeof) == SizeofError
def test_seek() -> None: def test_seek() -> None:
d = Seek(5) d = Seek(5)
assert d.parse(b"") == 5 assert d.parse(b"") == 5
@ -1365,105 +1334,6 @@ def test_compressed_prefixed() -> None:
assert st.parse(st.build(Container(one=zeros,two=zeros))) == Container(one=zeros,two=zeros) assert st.parse(st.build(Container(one=zeros,two=zeros))) == Container(one=zeros,two=zeros)
assert raises(d.sizeof) == SizeofError assert raises(d.sizeof) == SizeofError
@pytest.mark.xfail(ONWINDOWS and PYPY, reason="no wheel for 'cryptography' is currently available for pypy on windows")
def test_encryptedsym() -> None:
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
key128 = b"\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19\x1a\x1b\x1c\x1d\x1e\x1f"
key256 = b"\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19\x1a\x1b\x1c\x1d\x1e\x1f\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19\x1a\x1b\x1c\x1d\x1e\x1f"
iv = b"\x20\x21\x22\x23\x24\x25\x26\x27\x28\x29\x2a\x2b\x2c\x2d\x2e\x2f"
nonce = iv
# AES 128/256 bit - ECB
d = EncryptedSym(GreedyBytes, lambda ctx: Cipher(algorithms.AES(ctx.key), modes.ECB()))
common(d, b"\xf4\x0f\x54\xb7\x6a\x7a\xf1\xdb\x92\x73\x14\xde\x2f\xa0\x3e\x2d", b'Secret Message..', key=key128, iv=iv)
common(d, b"\x82\x6b\x01\x82\x90\x02\xa1\x9e\x35\x0a\xe2\xc3\xee\x1a\x42\xf5", b'Secret Message..', key=key256, iv=iv)
# AES 128/256 bit - CBC
d = EncryptedSym(GreedyBytes, lambda ctx: Cipher(algorithms.AES(ctx.key), modes.CBC(ctx.iv)))
common(d, b"\xba\x79\xc2\x62\x22\x08\x29\xb9\xfb\xd3\x90\xc4\x04\xb7\x55\x87", b'Secret Message..', key=key128, iv=iv)
common(d, b"\x60\xc2\x45\x0d\x7e\x41\xd4\xf8\x85\xd4\x8a\x64\xd1\x45\x49\xe3", b'Secret Message..', key=key256, iv=iv)
# AES 128/256 bit - CTR
d = EncryptedSym(GreedyBytes, lambda ctx: Cipher(algorithms.AES(ctx.key), modes.CTR(ctx.nonce)))
common(d, b"\x80\x78\xb6\x0c\x07\xf5\x0c\x90\xce\xa2\xbf\xcb\x5b\x22\xb9\xb5", b'Secret Message..', key=key128, nonce=nonce)
common(d, b"\x6a\xae\x7b\x86\x1a\xa6\xe0\x6a\x49\x02\x02\x1b\xf2\x3c\xd8\x0d", b'Secret Message..', key=key256, nonce=nonce)
assert raises(EncryptedSym(GreedyBytes, "AES").build, b"") == CipherError # type: ignore
assert raises(EncryptedSym(GreedyBytes, "AES").parse, b"") == CipherError # type: ignore
@pytest.mark.xfail(ONWINDOWS and PYPY, reason="no wheel for 'cryptography' is currently available for pypy on windows")
def test_encryptedsym_cbc_example() -> None:
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
d = Struct(
"iv" / Default(Bytes(16), os.urandom(16)),
"enc_data" / EncryptedSym(
Aligned(16,
Struct(
"width" / Int16ul,
"height" / Int16ul
)
),
lambda ctx: Cipher(algorithms.AES(ctx._.key), modes.CBC(ctx.iv))
)
)
key128 = b"\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19\x1a\x1b\x1c\x1d\x1e\x1f"
byts = d.build({"enc_data": {"width": 5, "height": 4}}, key=key128)
obj = d.parse(byts, key=key128)
assert obj.enc_data == Container(width=5, height=4)
@pytest.mark.xfail(ONWINDOWS and PYPY, reason="no wheel for 'cryptography' is currently available for pypy on windows")
def test_encryptedsymaead() -> None:
from cryptography.hazmat.primitives.ciphers import aead
key128 = b"\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19\x1a\x1b\x1c\x1d\x1e\x1f"
key256 = b"\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19\x1a\x1b\x1c\x1d\x1e\x1f\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19\x1a\x1b\x1c\x1d\x1e\x1f"
nonce = b"\x20\x21\x22\x23\x24\x25\x26\x27\x28\x29\x2a\x2b\x2c\x2d\x2e\x2f"
# AES 128/256 bit - GCM
d = Struct(
"associated_data" / Bytes(21),
"data" / EncryptedSymAead(
GreedyBytes,
lambda ctx: aead.AESGCM(ctx._.key),
this._.nonce,
this.associated_data
)
)
common(
d,
b"This is authenticated\xb6\xd3\x64\x0c\x7a\x31\xaa\x16\xa3\x58\xec\x17\x39\x99\x2e\xf8\x4e\x41\x17\x76\x3f\xd1\x06\x47\x04\x9f\x42\x1c\xf4\xa9\xfd\x99\x9c\xe9",
Container(associated_data=b"This is authenticated", data=b"The secret message"),
key=key128,
nonce=nonce
)
common(
d,
b"This is authenticated\xde\xb4\x41\x79\xc8\x7f\xea\x8d\x0e\x41\xf6\x44\x2f\x93\x21\xe6\x37\xd1\xd3\x29\xa4\x97\xc3\xb5\xf4\x81\x72\xa1\x7f\x3b\x9b\x53\x24\xe4",
Container(associated_data=b"This is authenticated", data=b"The secret message"),
key=key256,
nonce=nonce
)
assert raises(EncryptedSymAead(GreedyBytes, "AESGCM", bytes(16)).build, b"") == CipherError # type: ignore
assert raises(EncryptedSymAead(GreedyBytes, "AESGCM", bytes(16)).parse, b"") == CipherError # type: ignore
@pytest.mark.xfail(ONWINDOWS and PYPY, reason="no wheel for 'cryptography' is currently available for pypy on windows")
def test_encryptedsymaead_gcm_example() -> None:
from cryptography.hazmat.primitives.ciphers import aead
d = Struct(
"nonce" / Default(Bytes(16), os.urandom(16)),
"associated_data" / Bytes(21),
"enc_data" / EncryptedSymAead(
GreedyBytes,
lambda ctx: aead.AESGCM(ctx._.key),
this.nonce,
this.associated_data
)
)
key128 = b"\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19\x1a\x1b\x1c\x1d\x1e\x1f"
byts = d.build({"associated_data": b"This is authenticated", "enc_data": b"The secret message"}, key=key128)
obj = d.parse(byts, key=key128)
assert obj.enc_data == b"The secret message"
assert obj.associated_data == b"This is authenticated"
def test_rebuffered() -> None: def test_rebuffered() -> None:
data = b"0" * 1000 data = b"0" * 1000
assert Rebuffered(Array(1000,Byte)).parse_stream(io.BytesIO(data)) == [48]*1000 assert Rebuffered(Array(1000,Byte)).parse_stream(io.BytesIO(data)) == [48]*1000

View file

@ -2,11 +2,9 @@
# pyright: strict # pyright: strict
import dataclasses import dataclasses
import enum import enum
import textwrap
import typing as t import typing as t
import construct as cs import construct as cs
import construct_typed as cst import construct_typed as cst
from construct_typed import DataclassBitStruct, DataclassMixin, DataclassStruct, csfield from construct_typed import DataclassBitStruct, DataclassMixin, DataclassStruct, csfield
@ -74,20 +72,16 @@ def test_dataclass_str_repr() -> None:
== "Image: \n signature = b'BMP' (total 3)\n width = 3\n height = 2" == "Image: \n signature = b'BMP' (total 3)\n width = 3\n height = 2"
) )
def test_dataclass_ifthenelse() -> None: def test_dataclass_ifthenelse() -> None:
@dataclasses.dataclass @dataclasses.dataclass
class IfThenElseTest(DataclassMixin): class IfThenElseTest(DataclassMixin):
test_if: t.Optional[int] = csfield(cs.If(False, cs.Int8ub)) test_if: t.Optional[int] = csfield(cs.If(False, cs.Int8ub))
test_ifthenelse: t.Optional[int] = csfield( test_ifthenelse: t.Optional[int] = csfield(cs.IfThenElse(True, cs.Int8ub, cs.Pass))
cs.IfThenElse(True, cs.Int8ub, cs.Pass)
)
a = IfThenElseTest(test_if=None, test_ifthenelse=None) a = IfThenElseTest(test_if=None, test_ifthenelse=None)
assert a.test_if == None assert a.test_if == None
assert a.test_ifthenelse == None assert a.test_ifthenelse == None
def test_dataclass_struct() -> None: def test_dataclass_struct() -> None:
@dataclasses.dataclass @dataclasses.dataclass
class Image(DataclassMixin): class Image(DataclassMixin):
@ -401,9 +395,8 @@ def test_tenum_no_enumbase() -> None:
def test_tenum_asdict() -> None: def test_tenum_asdict() -> None:
# see: https://github.com/timrid/construct-typing/issues/21 # see: https://github.com/timrid/construct-typing/issues/21
import dataclasses
import construct_typed as cst import construct_typed as cst
import dataclasses
class TestEnum(cst.EnumBase): class TestEnum(cst.EnumBase):
one = 1 one = 1
@ -443,9 +436,9 @@ def test_tenum_docstring() -> None:
Value_NoDoc = cst.EnumValue(2) Value_NoDoc = cst.EnumValue(2)
Value_NoDoc2 = 3 Value_NoDoc2 = 3
assert TestEnum.__doc__ is not None assert (
assert textwrap.dedent(TestEnum.__doc__) == textwrap.dedent( TestEnum.__doc__
""" == """
This is an test enum. This is an test enum.
""" """
) )
@ -515,9 +508,8 @@ def test_tenum_flags() -> None:
def test_tenum_flags_asdict() -> None: def test_tenum_flags_asdict() -> None:
import dataclasses
import construct_typed as cst import construct_typed as cst
import dataclasses
class TestEnum(cst.FlagsEnumBase): class TestEnum(cst.FlagsEnumBase):
one = 1 one = 1
@ -557,9 +549,9 @@ def test_tenum_flags_docstring() -> None:
Value_NoDoc = cst.EnumValue(2) Value_NoDoc = cst.EnumValue(2)
Value_NoDoc2 = 4 Value_NoDoc2 = 4
assert TestEnum.__doc__ is not None assert (
assert textwrap.dedent(TestEnum.__doc__) == textwrap.dedent( TestEnum.__doc__
""" == """
This is an test flags enum. This is an test flags enum.
""" """
) )