Compare commits
9 commits
main
...
feature/at
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c6581db7db | ||
|
|
854d35977e | ||
|
|
4792469e39 | ||
|
|
8fe205bd29 | ||
|
|
2248e84f4a | ||
|
|
a1a583d95e | ||
|
|
c943a92fad | ||
|
|
f5b45792c2 | ||
|
|
9c7df9f8e3 |
26 changed files with 1431 additions and 1551 deletions
44
.github/workflows/main.yml
vendored
44
.github/workflows/main.yml
vendored
|
|
@ -1,10 +1,6 @@
|
|||
name: CI
|
||||
|
||||
on:
|
||||
push:
|
||||
pull_request:
|
||||
workflow_dispatch:
|
||||
workflow_call:
|
||||
on: [push, pull_request]
|
||||
|
||||
jobs:
|
||||
build:
|
||||
|
|
@ -12,7 +8,7 @@ jobs:
|
|||
strategy:
|
||||
matrix:
|
||||
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' ]
|
||||
|
||||
runs-on: ${{ matrix.os }}
|
||||
name: OS ${{ matrix.os }}, Python ${{ matrix.python-version }}
|
||||
|
|
@ -20,26 +16,25 @@ jobs:
|
|||
steps:
|
||||
# Checks out a copy of your repository on the machine
|
||||
- name: Checkout code
|
||||
uses: actions/checkout@v3
|
||||
uses: actions/checkout@v1
|
||||
|
||||
# Setup python
|
||||
- name: Setup python
|
||||
uses: actions/setup-python@v4
|
||||
uses: actions/setup-python@v1
|
||||
with:
|
||||
python-version: ${{ matrix.python-version }}
|
||||
architecture: x64
|
||||
|
||||
# Setup node.js (for pyright)
|
||||
- name: Setup node.js (for pyright)
|
||||
uses: actions/setup-node@v3
|
||||
uses: actions/setup-node@v2
|
||||
with:
|
||||
node-version: 16
|
||||
node-version: '14'
|
||||
|
||||
# Install pyright
|
||||
- name: Install pyright
|
||||
run: |
|
||||
npm install -g pyright
|
||||
pyright --version
|
||||
|
||||
# Install this package
|
||||
- name: Install this package
|
||||
|
|
@ -66,30 +61,3 @@ jobs:
|
|||
- name: Run pyright
|
||||
run: |
|
||||
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/
|
||||
37
.github/workflows/python-publish.yml
vendored
37
.github/workflows/python-publish.yml
vendored
|
|
@ -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
|
||||
|
||||
on:
|
||||
|
|
@ -5,26 +8,24 @@ on:
|
|||
types: [created]
|
||||
|
||||
jobs:
|
||||
create_wheel_and_sdist:
|
||||
name: create_wheel_and_sdist
|
||||
uses: ./.github/workflows/main.yml
|
||||
|
||||
deploy:
|
||||
needs: [ create_wheel_and_sdist ]
|
||||
|
||||
runs-on: ubuntu-latest
|
||||
|
||||
environment: pypi
|
||||
permissions:
|
||||
id-token: write # IMPORTANT: this permission is mandatory for Trusted Publishing
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@v3
|
||||
|
||||
- name: Download artifacts
|
||||
uses: actions/download-artifact@v4
|
||||
- uses: actions/checkout@v2
|
||||
- name: Set up Python
|
||||
uses: actions/setup-python@v2
|
||||
with:
|
||||
name: Package-Distributions-construct-typing
|
||||
path: ./dist
|
||||
|
||||
- name: Publish package distributions to PyPI
|
||||
uses: pypa/gh-action-pypi-publish@release/v1
|
||||
python-version: '3.x'
|
||||
- name: Install dependencies
|
||||
run: |
|
||||
python -m pip install --upgrade pip
|
||||
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
3
.gitignore
vendored
|
|
@ -129,6 +129,3 @@ dmypy.json
|
|||
example_737
|
||||
example_888
|
||||
example_ksy.ksy
|
||||
|
||||
# Test stuff
|
||||
devtest/
|
||||
3
.vscode/launch.json
vendored
3
.vscode/launch.json
vendored
|
|
@ -9,7 +9,8 @@
|
|||
"type": "python",
|
||||
"request": "launch",
|
||||
"program": "${file}",
|
||||
"console": "integratedTerminal"
|
||||
"console": "integratedTerminal",
|
||||
"justMyCode": false
|
||||
},
|
||||
{
|
||||
"name": "Debug Tests",
|
||||
|
|
|
|||
27
.vscode/settings.json
vendored
27
.vscode/settings.json
vendored
|
|
@ -1,23 +1,26 @@
|
|||
{
|
||||
// static analysis
|
||||
"python.pythonPath": "python",
|
||||
"python.languageServer": "Pylance",
|
||||
// "python.testing.unittestEnabled": false,
|
||||
// "python.testing.nosetestsEnabled": false,
|
||||
// "python.testing.pytestEnabled": true,
|
||||
"pythonTestExplorer.testFramework": "pytest",
|
||||
"python.formatting.provider": "black",
|
||||
"python.sortImports.path": "isort",
|
||||
"python.sortImports.args": [
|
||||
"--profile=black",
|
||||
],
|
||||
// "[python]": {
|
||||
// "editor.codeActionsOnSave": {
|
||||
// "source.organizeImports": true
|
||||
// }
|
||||
// }
|
||||
"python.analysis.typeCheckingMode": "strict",
|
||||
"python.analysis.autoImportCompletions": false,
|
||||
"python.analysis.diagnosticSeverityOverrides": {
|
||||
"reportPrivateUsage": "information",
|
||||
"reportUntypedNamedTuple": "information",
|
||||
},
|
||||
|
||||
// formating
|
||||
"python.formatting.provider": "black",
|
||||
|
||||
// sorting
|
||||
"python.sortImports.path": "isort",
|
||||
"python.sortImports.args": [
|
||||
"--profile=black",
|
||||
],
|
||||
|
||||
// tests
|
||||
"python.testing.pytestArgs": [
|
||||
"tests"
|
||||
],
|
||||
|
|
|
|||
32
README.md
32
README.md
|
|
@ -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
|
||||
[](https://pypi.org/project/construct-typing/)
|
||||

|
||||
|
|
@ -77,29 +68,28 @@ A short example:
|
|||
import dataclasses
|
||||
import typing as t
|
||||
from construct import Array, Byte, Const, Int8ub, this
|
||||
from construct_typed import DataclassMixin, DataclassStruct, EnumBase, TEnum, csfield
|
||||
from construct_typed import AttrsStruct, Enum, construct, attrs_field
|
||||
|
||||
class Orientation(EnumBase):
|
||||
class Orientation(Enum, constr=Int8ub): # TODO: Implement this
|
||||
HORIZONTAL = 0
|
||||
VERTICAL = 1
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Image(DataclassMixin):
|
||||
signature: bytes = csfield(Const(b"BMP"))
|
||||
orientation: Orientation = csfield(TEnum(Int8ub, Orientation))
|
||||
width: int = csfield(Int8ub)
|
||||
height: int = csfield(Int8ub)
|
||||
pixels: t.List[int] = csfield(Array(this.width * this.height, Byte))
|
||||
class Image(AttrsStruct):
|
||||
signature: bytes = attrs_field(Const(b"BMP"))
|
||||
orientation: Orientation = attrs_field(construct(Orientation)) # TODO: Implement this
|
||||
width: int = attrs_field(Int8ub)
|
||||
height: int = attrs_field(Int8ub)
|
||||
pixels: t.List[int] = attrs_field(Array(this.width * this.height, Byte))
|
||||
|
||||
format = DataclassStruct(Image)
|
||||
fmt = construct(Image)
|
||||
obj = Image(
|
||||
orientation=Orientation.VERTICAL,
|
||||
width=3,
|
||||
height=2,
|
||||
pixels=[7, 8, 9, 11, 12, 13],
|
||||
)
|
||||
print(format.build(obj))
|
||||
print(format.parse(b"BMP\x01\x03\x02\x07\x08\t\x0b\x0c\r"))
|
||||
print(fmt.build(obj))
|
||||
print(fmt.parse(b"BMP\x01\x03\x02\x07\x08\t\x0b\x0c\r"))
|
||||
```
|
||||
Output:
|
||||
```
|
||||
|
|
|
|||
|
|
@ -3,10 +3,6 @@ from construct.debug import *
|
|||
from construct.expr import *
|
||||
from construct.lib import *
|
||||
from construct.version import *
|
||||
from construct import lib
|
||||
|
||||
__author__: str
|
||||
__version__: str
|
||||
|
||||
#===============================================================================
|
||||
# exposed names
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
|
|
@ -1,3 +1,4 @@
|
|||
import operator
|
||||
import typing as t
|
||||
|
||||
from construct.core import *
|
||||
|
|
@ -469,7 +470,7 @@ class ExprMixin(t.Generic[ReturnType], object):
|
|||
@t.overload
|
||||
def __eq__(self: ExprMixin[float], other: ConstOrCallable[float]) -> BinExpr[bool]: ...
|
||||
@t.overload
|
||||
def __eq__(self, other: ConstOrCallable[t.Any]) -> BinExpr[t.Any]: ... # type: ignore
|
||||
def __eq__(self, other: t.Any) -> BinExpr[t.Any]: ...
|
||||
|
||||
# __ne__ ###########################################################################################################
|
||||
@t.overload
|
||||
|
|
@ -487,7 +488,7 @@ class ExprMixin(t.Generic[ReturnType], object):
|
|||
@t.overload
|
||||
def __ne__(self: ExprMixin[float], other: ConstOrCallable[float]) -> BinExpr[bool]: ...
|
||||
@t.overload
|
||||
def __ne__(self, other: t.Any) -> BinExpr[t.Any]: ... # type: ignore
|
||||
def __ne__(self, other: t.Any) -> BinExpr[t.Any]: ...
|
||||
|
||||
# __neg__ ##########################################################################################################
|
||||
@t.overload
|
||||
|
|
@ -497,7 +498,7 @@ class ExprMixin(t.Generic[ReturnType], object):
|
|||
@t.overload
|
||||
def __neg__(self: ExprMixin[float]) -> BinExpr[float]: ...
|
||||
@t.overload
|
||||
def __neg__(self) -> BinExpr[t.Any]: ...
|
||||
def __neg__(self) -> UniExpr[t.Any]: ...
|
||||
|
||||
# __pos__ ##########################################################################################################
|
||||
@t.overload
|
||||
|
|
@ -507,7 +508,7 @@ class ExprMixin(t.Generic[ReturnType], object):
|
|||
@t.overload
|
||||
def __pos__(self: ExprMixin[float]) -> BinExpr[float]: ...
|
||||
@t.overload
|
||||
def __pos__(self) -> BinExpr[t.Any]: ...
|
||||
def __pos__(self) -> UniExpr[t.Any]: ...
|
||||
|
||||
# __invert__ #######################################################################################################
|
||||
@t.overload
|
||||
|
|
@ -515,7 +516,7 @@ class ExprMixin(t.Generic[ReturnType], object):
|
|||
@t.overload
|
||||
def __invert__(self: ExprMixin[bool]) -> BinExpr[int]: ...
|
||||
@t.overload
|
||||
def __invert__(self) -> BinExpr[t.Any]: ...
|
||||
def __invert__(self) -> UniExpr[t.Any]: ...
|
||||
|
||||
# __inv__ ##########################################################################################################
|
||||
def __inv__(self) -> UniExpr[t.Any]: ...
|
||||
|
|
@ -542,7 +543,7 @@ class Path2(ExprMixin[ReturnType]):
|
|||
|
||||
|
||||
class FuncPath(ExprMixin[ReturnType]):
|
||||
def __init__(self, func: t.Callable[[t.Any], ReturnType], operand: t.Optional[t.Any] = ...) -> None: ...
|
||||
def __init__(self, func: t.Callable[[t.Any], t.Any], operand: t.Optional[t.Any] = ...) -> None: ...
|
||||
def __call__(self, operand: t.Any, *args: t.Any) -> ReturnType: ...
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -19,7 +19,7 @@ def recursion_lock(
|
|||
|
||||
class Container(t.Generic[ContainerType], t.Dict[str, ContainerType]):
|
||||
def __getattr__(self, name: str) -> ContainerType: ...
|
||||
def update( # type: ignore
|
||||
def update(
|
||||
self,
|
||||
seqordict: t.Union[t.Dict[str, ContainerType], t.Tuple[str, ContainerType]],
|
||||
) -> None: ...
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
import typing as t
|
||||
|
||||
|
||||
class HexDisplayedInteger(int): ...
|
||||
class HexDisplayedBytes(bytes): ...
|
||||
|
||||
|
|
@ -9,6 +10,3 @@ V = t.TypeVar("V")
|
|||
class HexDisplayedDict(t.Dict[K, V]): ...
|
||||
class HexDumpDisplayedBytes(bytes): ...
|
||||
class HexDumpDisplayedDict(t.Dict[K, V]): ...
|
||||
|
||||
def hexdump(data: bytes, linesize: int) -> str: ...
|
||||
def hexundump(data: str, linesize: int) -> bytes: ...
|
||||
|
|
|
|||
|
|
@ -1,6 +1,5 @@
|
|||
import typing as t
|
||||
|
||||
PY: t.Tuple[int, int]
|
||||
PY2: bool
|
||||
PY3: bool
|
||||
PYPY: bool
|
||||
|
|
|
|||
|
|
@ -1,55 +1,38 @@
|
|||
from .dataclass_struct import (
|
||||
DataclassBitStruct,
|
||||
DataclassMixin,
|
||||
DataclassStruct,
|
||||
TBitStruct,
|
||||
TContainerBase,
|
||||
TContainerMixin,
|
||||
TStruct,
|
||||
TStructField,
|
||||
csfield,
|
||||
sfield,
|
||||
EnhancedDataclassMixin
|
||||
)
|
||||
from .generic_wrapper import (
|
||||
from .attrs_struct import AttrsStruct, attrs_field, this_struct
|
||||
from .generics import (
|
||||
Adapter,
|
||||
ConstantOrContextLambda,
|
||||
ConstantOrContextLambda2,
|
||||
Construct,
|
||||
Context,
|
||||
ListContainer,
|
||||
PathType,
|
||||
Array,
|
||||
Subconstruct,
|
||||
Computed,
|
||||
Constructable,
|
||||
construct,
|
||||
)
|
||||
from .tenum import EnumBase, EnumValue, FlagsEnumBase, TEnum, TFlagsEnum
|
||||
from .tenum import EnumBase, FlagsEnumBase, EnumConstruct, FlagsEnumConstruct
|
||||
|
||||
__all__ = [
|
||||
"AttrsStruct",
|
||||
"attrs_field",
|
||||
"DataclassBitStruct",
|
||||
"DataclassMixin",
|
||||
"DataclassStruct",
|
||||
"TBitStruct",
|
||||
"TContainerBase",
|
||||
"TContainerMixin",
|
||||
"TStruct",
|
||||
"TStructField",
|
||||
"this_struct",
|
||||
"csfield",
|
||||
"sfield",
|
||||
"EnhancedDataclassMixin",
|
||||
"Constructable",
|
||||
"construct",
|
||||
"EnumBase",
|
||||
"EnumValue",
|
||||
"FlagsEnumBase",
|
||||
"TEnum",
|
||||
"TFlagsEnum",
|
||||
"EnumConstruct",
|
||||
"FlagsEnumConstruct",
|
||||
"Adapter",
|
||||
"ConstantOrContextLambda",
|
||||
"ConstantOrContextLambda2",
|
||||
"Construct",
|
||||
"Context",
|
||||
"ListContainer",
|
||||
"PathType",
|
||||
"Array",
|
||||
"Subconstruct",
|
||||
"Computed"
|
||||
]
|
||||
|
|
|
|||
237
construct_typed/attrs_struct.py
Normal file
237
construct_typed/attrs_struct.py
Normal file
|
|
@ -0,0 +1,237 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# pyright: strict
|
||||
import textwrap
|
||||
import typing as t
|
||||
|
||||
import attr
|
||||
import construct as cs
|
||||
from .generics import Adapter, Construct, Context, ParsedType, PathType
|
||||
|
||||
T = t.TypeVar("T")
|
||||
|
||||
# Static type inference support via __dataclass_transform__ implemented as per:
|
||||
# https://github.com/microsoft/pyright/blob/1.1.135/specs/dataclass_transforms.md
|
||||
def __dataclass_transform__(
|
||||
*,
|
||||
eq_default: bool = True,
|
||||
order_default: bool = False,
|
||||
kw_only_default: bool = False,
|
||||
field_descriptors: t.Tuple[t.Union[type, t.Callable[..., t.Any]], ...] = (()),
|
||||
) -> t.Callable[[T], T]:
|
||||
return lambda a: a
|
||||
|
||||
|
||||
ATTRS_METADATA_KEY = "__construct_typed_subcon"
|
||||
|
||||
|
||||
def attrs_field(
|
||||
subcon: Construct[ParsedType, t.Any],
|
||||
doc: t.Optional[str] = None,
|
||||
parsed: t.Optional[t.Callable[[t.Any, Context], None]] = None,
|
||||
) -> ParsedType:
|
||||
"""
|
||||
Helper method for `AttrsStruct` and `AttrsBitStruct` to create the attrs fields.
|
||||
|
||||
This method also processes `Const` and `Default`, to pass these values als default values to the dataclass.
|
||||
|
||||
# TODO: Implement `default` parameter for `attrs_field`
|
||||
"""
|
||||
orig_subcon = subcon
|
||||
|
||||
# Rename subcon, if doc or parsed are available
|
||||
if (doc is not None) or (parsed is not None):
|
||||
if doc is not None:
|
||||
doc = textwrap.dedent(doc).strip("\n")
|
||||
subcon = cs.Renamed(subcon, newdocs=doc, newparsed=parsed)
|
||||
|
||||
if orig_subcon.flagbuildnone is True:
|
||||
init = False
|
||||
default = None
|
||||
else:
|
||||
init = True
|
||||
default = attr.NOTHING
|
||||
|
||||
# Set default values in case of special sucons
|
||||
if isinstance(orig_subcon, cs.Const):
|
||||
const_subcon: "cs.Const[t.Any, t.Any, t.Any, t.Any]" = orig_subcon
|
||||
default = const_subcon.value
|
||||
elif isinstance(orig_subcon, cs.Default):
|
||||
default_subcon: "cs.Default[t.Any, t.Any, t.Any, t.Any]" = orig_subcon
|
||||
if callable(default_subcon.value):
|
||||
default = None # context lambda is only defined at parsing/building
|
||||
else:
|
||||
default = default_subcon.value
|
||||
|
||||
return t.cast(
|
||||
ParsedType,
|
||||
attr.field(
|
||||
default=default,
|
||||
init=init,
|
||||
metadata={ATTRS_METADATA_KEY: subcon},
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
class AttrsConstruct(Adapter[t.Any, t.Any, T, T]):
|
||||
if t.TYPE_CHECKING:
|
||||
|
||||
def __new__(
|
||||
cls,
|
||||
attrs_cls: t.Type[T],
|
||||
reverse_fields: bool = False,
|
||||
) -> "AttrsConstruct[T]":
|
||||
...
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
attrs_cls: t.Type[T],
|
||||
reverse_fields: bool = False,
|
||||
) -> None:
|
||||
if not attr.has(attrs_cls):
|
||||
raise TypeError(f"'{attrs_cls}' has to be a 'attrs' object")
|
||||
|
||||
self.attrs_cls = attrs_cls
|
||||
self.reverse_fields = reverse_fields
|
||||
|
||||
# get all fields from the dataclass
|
||||
fields = attr.fields(attrs_cls)
|
||||
if reverse_fields:
|
||||
fields = tuple(reversed(fields))
|
||||
|
||||
# extract the construct formats from the struct_type
|
||||
subcon_fields = {}
|
||||
for field in fields:
|
||||
subcon_fields[field.name] = field.metadata[ATTRS_METADATA_KEY]
|
||||
|
||||
# init adatper
|
||||
super().__init__(cs.Struct(**subcon_fields)) # type: ignore
|
||||
|
||||
def _decode(
|
||||
self,
|
||||
obj: "cs.Container[t.Any]",
|
||||
context: Context,
|
||||
path: PathType,
|
||||
) -> T:
|
||||
# get all fields from the dataclass
|
||||
fields = attr.fields(self.attrs_cls)
|
||||
|
||||
# extract all fields from the container, that are used for create the dataclass object
|
||||
dc_init = {}
|
||||
for field in fields:
|
||||
if field.init:
|
||||
value = obj[field.name]
|
||||
dc_init[field.name] = value
|
||||
|
||||
# create object of dataclass
|
||||
dc = self.attrs_cls(**dc_init) # type: ignore
|
||||
|
||||
# extract all other values from the container, an pass it to the dataclass
|
||||
for field in fields:
|
||||
if not field.init:
|
||||
value = obj[field.name]
|
||||
setattr(dc, field.name, value)
|
||||
|
||||
return dc
|
||||
|
||||
def _encode(self, obj: T, context: Context, path: PathType) -> t.Dict[str, t.Any]:
|
||||
if not isinstance(obj, self.attrs_cls):
|
||||
raise TypeError(f"'{repr(obj)}' has to be of type {repr(self.attrs_cls)}")
|
||||
|
||||
# get all fields from the dataclass
|
||||
fields = attr.fields(self.attrs_cls)
|
||||
|
||||
# extract all fields from the container, that are used for create the dataclass object
|
||||
ret_dict: t.Dict[str, t.Any] = {}
|
||||
for field in fields:
|
||||
value = getattr(obj, field.name)
|
||||
ret_dict[field.name] = value
|
||||
|
||||
return ret_dict
|
||||
|
||||
|
||||
# Helper object for defining the `constr` of a `struct`. Will be replaced with the proper construct, when class is created.
|
||||
this_struct: Construct[t.Any, t.Any] = Construct()
|
||||
|
||||
|
||||
def _replace_this_struct(constr: "Construct[t.Any, t.Any]", replacement: t.Any):
|
||||
"""Recursive search for `this_struct` in all SubConstructs and replace it with AttrsStruct"""
|
||||
subcon = getattr(constr, "subcon", None)
|
||||
if subcon is this_struct:
|
||||
setattr(constr, "subcon", replacement)
|
||||
elif subcon is not None:
|
||||
_replace_this_struct(subcon, replacement)
|
||||
else:
|
||||
raise ValueError(
|
||||
"Could not find `this_struct`. Only SubConstructs are supported"
|
||||
)
|
||||
|
||||
|
||||
@__dataclass_transform__(kw_only_default=True, field_descriptors=(attrs_field,))
|
||||
class AttrsStruct:
|
||||
"""
|
||||
Adapter for a attrs-class for optimised type hints / static autocompletion in comparision to the original Struct.
|
||||
|
||||
Before this construct can be created a dataclasses.dataclass type must be created, which must also derive from DataclassMixin. In this dataclass all fields must be assigned to a construct type using csfield.
|
||||
|
||||
Internally, all fields are converted to a Struct, which does the actual parsing/building.
|
||||
|
||||
Parses to a dataclasses.dataclass instance, and builds from such instance. Size is the sum of all subcon sizes, unless any subcon raises SizeofError.
|
||||
|
||||
Metaclass paramters::
|
||||
|
||||
:param constr: Create a more complex construct. `this_struct` can be used for representing this AttrsStruct object.
|
||||
:param reverse_fields: Flag if the fields should be reversed parsed/build
|
||||
|
||||
Example::
|
||||
|
||||
>>> from construct import Bytes, Int8ub, this
|
||||
>>> from construct_typed import AttrsStruct, attrs_field, construct
|
||||
>>> class Image(AttrsStruct):
|
||||
... width: int = attrs_field(Int8ub)
|
||||
... height: int = attrs_field(Int8ub)
|
||||
... pixels: bytes = attrs_field(Bytes(this.height * this.width))
|
||||
>>> d = construct(Image)
|
||||
>>> d.parse(b"\x01\x0212")
|
||||
Image(width=1, height=2, pixels=b'12')
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
def __init_subclass__(
|
||||
cls,
|
||||
constr: "cs.Construct[t.Any, t.Any]" = this_struct,
|
||||
reverse_fields: bool = False,
|
||||
):
|
||||
# validate types
|
||||
if not isinstance(constr, cs.Construct): # type: ignore
|
||||
raise ValueError("`constr` parameter has to be an `Construct` object")
|
||||
if not isinstance(reverse_fields, bool): # type: ignore
|
||||
raise ValueError("`reverse_fields` parameter has to be an `bool` object")
|
||||
|
||||
# create attrs class
|
||||
cls = attr.define(cls, kw_only=True, slots=False)
|
||||
|
||||
# create construct format
|
||||
attrs_constr = AttrsConstruct(cls, reverse_fields)
|
||||
if constr is this_struct:
|
||||
constr = attrs_constr
|
||||
else:
|
||||
_replace_this_struct(constr, attrs_constr)
|
||||
|
||||
# save construct format and make the class compatible to `Constructable` protocol
|
||||
setattr(cls, "__construct__", lambda: constr)
|
||||
|
||||
return cls
|
||||
|
||||
# the `construct` library is using the [] access internally, so struct objects
|
||||
# should also make this possible and not only via the dot access.
|
||||
def __getitem__(self, key: str) -> t.Any:
|
||||
return getattr(self, key)
|
||||
|
||||
def __setitem__(self, key: str, value: t.Any) -> None:
|
||||
setattr(self, key, value)
|
||||
|
||||
if t.TYPE_CHECKING:
|
||||
|
||||
@classmethod
|
||||
def __construct__(cls: t.Type[T]) -> "AttrsConstruct[T]":
|
||||
...
|
||||
|
|
@ -1,6 +1,5 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# pyright: strict
|
||||
# pyright: reportIncompatibleVariableOverride=false, reportAny=false
|
||||
import dataclasses
|
||||
import textwrap
|
||||
import typing as t
|
||||
|
|
@ -12,25 +11,250 @@ from construct.lib.containers import (
|
|||
recursion_lock,
|
||||
)
|
||||
from construct.lib.py3compat import bytestringtype, reprstring, unicodestringtype
|
||||
from typing_extensions import override
|
||||
|
||||
from .generic_wrapper import Adapter, Construct, Context, ParsedType, PathType
|
||||
from construct_typed.generics import Adapter, Construct, Context, ParsedType, PathType
|
||||
|
||||
T = t.TypeVar("T")
|
||||
|
||||
|
||||
class DataclassMixin:
|
||||
# Static type inference support via __dataclass_transform__ implemented as per:
|
||||
# https://github.com/microsoft/pyright/blob/1.1.135/specs/dataclass_transforms.md
|
||||
def __dataclass_transform__(
|
||||
*,
|
||||
eq_default: bool = True,
|
||||
order_default: bool = False,
|
||||
kw_only_default: bool = False,
|
||||
field_descriptors: t.Tuple[t.Union[type, t.Callable[..., t.Any]], ...] = (()),
|
||||
) -> t.Callable[[T], T]:
|
||||
return lambda a: a
|
||||
|
||||
|
||||
DATACLASS_METADATA_KEY = "__construct_typed_subcon"
|
||||
|
||||
if t.TYPE_CHECKING:
|
||||
# specialisation for constructs, that builds from none and dont have to be declared in the __init__ method
|
||||
@t.overload
|
||||
def csfield(
|
||||
subcon: cs.Construct[ParsedType, None],
|
||||
doc: t.Optional[str] = None,
|
||||
parsed: t.Optional[t.Callable[[t.Any, Context], None]] = None,
|
||||
init: t.Literal[False] = False,
|
||||
) -> ParsedType:
|
||||
...
|
||||
|
||||
@t.overload
|
||||
def csfield(
|
||||
subcon: Construct[ParsedType, t.Any],
|
||||
doc: t.Optional[str] = None,
|
||||
parsed: t.Optional[t.Callable[[t.Any, Context], None]] = None,
|
||||
init: bool = True,
|
||||
) -> ParsedType:
|
||||
...
|
||||
|
||||
|
||||
def csfield(
|
||||
subcon: Construct[ParsedType, t.Any],
|
||||
doc: t.Optional[str] = None,
|
||||
parsed: t.Optional[t.Callable[[t.Any, Context], None]] = None,
|
||||
init: bool = True,
|
||||
) -> ParsedType:
|
||||
"""
|
||||
Mixin for the dataclasses which are passed to "DataclassStruct" and "DataclassBitStruct".
|
||||
Helper method for "DataclassStruct" and "DataclassBitStruct" to create the dataclass fields.
|
||||
|
||||
Note: This implementation is different to the 'cs.Container' of the original 'construct'
|
||||
library. In the original 'cs.Container' some names like "update", "keys", "items", ... can
|
||||
only accessed via key access (square brackets) and not via attribute access (dot operator),
|
||||
because they are also method names. This implementation is based on "dataclasses.dataclass"
|
||||
which only uses modul-level instead of instance-level helper methods.So no instance-level
|
||||
methods exists and every name can be used.
|
||||
This method also processes Const and Default, to pass these values als default values to the dataclass.
|
||||
"""
|
||||
orig_subcon = subcon
|
||||
|
||||
# Rename subcon, if doc or parsed are available
|
||||
if (doc is not None) or (parsed is not None):
|
||||
if doc is not None:
|
||||
doc = textwrap.dedent(doc).strip("\n")
|
||||
subcon = cs.Renamed(subcon, newdocs=doc, newparsed=parsed)
|
||||
|
||||
if orig_subcon.flagbuildnone is True:
|
||||
init = False
|
||||
default = None
|
||||
else:
|
||||
init = True
|
||||
default = dataclasses.MISSING
|
||||
|
||||
# Set default values in case of special sucons
|
||||
if isinstance(orig_subcon, cs.Const):
|
||||
const_subcon: "cs.Const[t.Any, t.Any, t.Any, t.Any]" = orig_subcon
|
||||
default = const_subcon.value
|
||||
elif isinstance(orig_subcon, cs.Default):
|
||||
default_subcon: "cs.Default[t.Any, t.Any, t.Any, t.Any]" = orig_subcon
|
||||
if callable(default_subcon.value):
|
||||
default = None # context lambda is only defined at parsing/building
|
||||
else:
|
||||
default = default_subcon.value
|
||||
|
||||
return t.cast(
|
||||
ParsedType,
|
||||
dataclasses.field(
|
||||
default=default,
|
||||
init=init,
|
||||
metadata={DATACLASS_METADATA_KEY: subcon},
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
class DataclassConstruct(Adapter[t.Any, t.Any, T, T]):
|
||||
"""
|
||||
Adapter for a dataclasses for optimised type hints / static autocompletion in comparision to the original Struct.
|
||||
|
||||
Before this construct can be created a dataclasses.dataclass type must be created, which must also derive from DataclassMixin. In this dataclass all fields must be assigned to a construct type using csfield.
|
||||
|
||||
Internally, all fields are converted to a Struct, which does the actual parsing/building.
|
||||
|
||||
Parses to a dataclasses.dataclass instance, and builds from such instance. Size is the sum of all subcon sizes, unless any subcon raises SizeofError.
|
||||
|
||||
:param dc_type: Type of the dataclass, which also inherits from DataclassMixin
|
||||
:param reverse: Flag if the fields of the dataclass should be reversed
|
||||
|
||||
Example::
|
||||
|
||||
>>> import dataclasses
|
||||
>>> from construct import Bytes, Int8ub, this
|
||||
>>> from construct_typed import DataclassMixin, DataclassStruct, csfield, construct
|
||||
>>> @dataclasses.dataclass
|
||||
... class Image(DataclassStruct):
|
||||
... width: int = csfield(Int8ub)
|
||||
... height: int = csfield(Int8ub)
|
||||
... pixels: bytes = csfield(Bytes(this.height * this.width))
|
||||
>>> d = construct(Image)
|
||||
>>> d.parse(b"\x01\x0212")
|
||||
Image(width=1, height=2, pixels=b'12')
|
||||
"""
|
||||
|
||||
__dataclass_fields__: "t.ClassVar[dict[str, dataclasses.Field[t.Any]]]"
|
||||
subcon: "cs.Struct[t.Any, t.Any]"
|
||||
if t.TYPE_CHECKING:
|
||||
|
||||
def __new__(
|
||||
cls,
|
||||
dc_type: t.Type[T],
|
||||
reverse: bool = False,
|
||||
) -> "DataclassConstruct[T]":
|
||||
...
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
dc_type: t.Type[T],
|
||||
reverse: bool = False,
|
||||
) -> None:
|
||||
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
|
||||
fields = dataclasses.fields(self.dc_type)
|
||||
if self.reverse:
|
||||
fields = tuple(reversed(fields))
|
||||
|
||||
# extract the construct formats from the struct_type
|
||||
subcon_fields = {}
|
||||
for field in fields:
|
||||
subcon_fields[field.name] = field.metadata[DATACLASS_METADATA_KEY]
|
||||
|
||||
# init adatper
|
||||
super().__init__(cs.Struct(**subcon_fields)) # type: ignore
|
||||
|
||||
def __getattr__(self, name: str) -> t.Any:
|
||||
return getattr(self.subcon, name)
|
||||
|
||||
def _decode(
|
||||
self, obj: "cs.Container[t.Any]", context: Context, path: PathType
|
||||
) -> T:
|
||||
# get all fields from the dataclass
|
||||
fields = dataclasses.fields(self.dc_type)
|
||||
|
||||
# extract all fields from the container, that are used for create the dataclass object
|
||||
dc_init = {}
|
||||
for field in fields:
|
||||
if field.init:
|
||||
value = obj[field.name]
|
||||
dc_init[field.name] = value
|
||||
|
||||
# create object of dataclass
|
||||
dc = self.dc_type(**dc_init) # type: ignore
|
||||
|
||||
# extract all other values from the container, an pass it to the dataclass
|
||||
for field in fields:
|
||||
if not field.init:
|
||||
value = obj[field.name]
|
||||
setattr(dc, field.name, value)
|
||||
|
||||
return dc
|
||||
|
||||
def _encode(self, obj: T, context: Context, path: PathType) -> t.Dict[str, t.Any]:
|
||||
if not isinstance(obj, self.dc_type):
|
||||
raise TypeError(f"'{repr(obj)}' has to be of type {repr(self.dc_type)}")
|
||||
|
||||
# get all fields from the dataclass
|
||||
fields = dataclasses.fields(self.dc_type)
|
||||
|
||||
# extract all fields from the container, that are used for create the dataclass object
|
||||
ret_dict: t.Dict[str, t.Any] = {}
|
||||
for field in fields:
|
||||
value = getattr(obj, field.name)
|
||||
ret_dict[field.name] = value
|
||||
|
||||
return ret_dict
|
||||
|
||||
|
||||
# Helper object for defining the `constr` of a `struct`. Will be replaced with the proper construct, when class is created.
|
||||
this_struct: Construct[t.Any, t.Any] = Construct()
|
||||
|
||||
|
||||
def _replace_this_struct(constr: "Construct[t.Any, t.Any]", replacement: t.Any):
|
||||
"""Recursive search for `this_struct` in all SubConstructs and replace it with AttrsStruct"""
|
||||
subcon = getattr(constr, "subcon", None)
|
||||
if subcon is this_struct:
|
||||
setattr(constr, "subcon", replacement)
|
||||
elif subcon is not None:
|
||||
_replace_this_struct(subcon, replacement)
|
||||
else:
|
||||
raise ValueError(
|
||||
"Could not find `this_struct`. Only SubConstructs are supported"
|
||||
)
|
||||
|
||||
|
||||
@__dataclass_transform__(field_descriptors=(csfield,))
|
||||
class DataclassStruct:
|
||||
r"""
|
||||
TODO: Add Documentation
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
def __init_subclass__(
|
||||
cls,
|
||||
constr: "cs.Construct[t.Any, t.Any]" = this_struct,
|
||||
reverse_fields: bool = False,
|
||||
):
|
||||
# validate types
|
||||
if not isinstance(constr, cs.Construct): # type: ignore
|
||||
raise ValueError("`constr` parameter has to be an `Construct` object")
|
||||
if not isinstance(reverse_fields, bool): # type: ignore
|
||||
raise ValueError("`reverse_fields` parameter has to be an `bool` object")
|
||||
|
||||
# create attrs class
|
||||
cls = dataclasses.dataclass(cls)
|
||||
|
||||
# create construct format
|
||||
dc_constr = DataclassConstruct(cls, reverse_fields)
|
||||
if constr is this_struct:
|
||||
constr = dc_constr
|
||||
else:
|
||||
_replace_this_struct(constr, dc_constr)
|
||||
|
||||
# save construct format and make the class compatible to `Constructable` protocol
|
||||
setattr(cls, "__construct__", lambda: constr)
|
||||
|
||||
return cls
|
||||
|
||||
# the `construct` library is using the [] access internally, so struct objects
|
||||
# should also make this possible and not only via the dot access.
|
||||
def __getitem__(self, key: str) -> t.Any:
|
||||
return getattr(self, key)
|
||||
|
||||
|
|
@ -76,209 +300,42 @@ class DataclassMixin:
|
|||
text.append(indentation.join(str(v).split("\n")))
|
||||
return "".join(text)
|
||||
|
||||
if t.TYPE_CHECKING:
|
||||
|
||||
def csfield(
|
||||
subcon: Construct[ParsedType, t.Any],
|
||||
doc: str | None = None,
|
||||
parsed: t.Callable[[t.Any, Context], None] | None = None,
|
||||
) -> ParsedType:
|
||||
"""
|
||||
Helper method for "DataclassStruct" and "DataclassBitStruct" to create the dataclass fields.
|
||||
|
||||
This method also processes Const and Default, to pass these values als default values to the dataclass.
|
||||
"""
|
||||
orig_subcon = subcon
|
||||
|
||||
# Rename subcon, if doc or parsed are available
|
||||
if (doc is not None) or (parsed is not None):
|
||||
if doc is not None:
|
||||
doc = textwrap.dedent(doc).strip("\n")
|
||||
subcon = cs.Renamed(subcon, newdocs=doc, newparsed=parsed)
|
||||
|
||||
if orig_subcon.flagbuildnone is True:
|
||||
init = False
|
||||
default = None
|
||||
else:
|
||||
init = True
|
||||
default = dataclasses.MISSING
|
||||
|
||||
# Set default values in case of special sucons
|
||||
if isinstance(orig_subcon, cs.Const):
|
||||
const_subcon: "cs.Const[t.Any, t.Any]" = orig_subcon
|
||||
default = const_subcon.value
|
||||
elif isinstance(orig_subcon, cs.Default):
|
||||
default_subcon: "cs.Default[t.Any, t.Any]" = orig_subcon
|
||||
if callable(default_subcon.value):
|
||||
default = None # context lambda is only defined at parsing/building
|
||||
else:
|
||||
default = default_subcon.value
|
||||
|
||||
return t.cast(
|
||||
ParsedType,
|
||||
dataclasses.field(
|
||||
default=default,
|
||||
init=init,
|
||||
metadata={"subcon": subcon},
|
||||
),
|
||||
)
|
||||
@classmethod
|
||||
def __construct__(cls: t.Type[T]) -> "DataclassConstruct[T]":
|
||||
...
|
||||
|
||||
|
||||
DataclassType = t.TypeVar("DataclassType", bound=DataclassMixin)
|
||||
|
||||
|
||||
class DataclassStruct(Adapter[t.Any, t.Any, DataclassType, DataclassType]):
|
||||
"""
|
||||
Adapter for a dataclasses for optimised type hints / static autocompletion in comparision to the original Struct.
|
||||
|
||||
Before this construct can be created a dataclasses.dataclass type must be created, which must also derive from DataclassMixin. In this dataclass all fields must be assigned to a construct type using csfield.
|
||||
|
||||
Internally, all fields are converted to a Struct, which does the actual parsing/building.
|
||||
|
||||
Parses to a dataclasses.dataclass instance, and builds from such instance. Size is the sum of all subcon sizes, unless any subcon raises SizeofError.
|
||||
|
||||
:param dc_type: Type of the dataclass, which also inherits from DataclassMixin
|
||||
:param reverse: Flag if the fields of the dataclass should be reversed
|
||||
|
||||
Example::
|
||||
|
||||
>>> import dataclasses
|
||||
>>> from construct import Bytes, Int8ub, this
|
||||
>>> from construct_typed import DataclassMixin, DataclassStruct, csfield
|
||||
>>> @dataclasses.dataclass
|
||||
... class Image(DataclassMixin):
|
||||
... width: int = csfield(Int8ub)
|
||||
... height: int = csfield(Int8ub)
|
||||
... pixels: bytes = csfield(Bytes(this.height * this.width))
|
||||
>>> d = DataclassStruct(Image)
|
||||
>>> d.parse(b"\x01\x0212")
|
||||
Image(width=1, height=2, pixels=b'12')
|
||||
"""
|
||||
|
||||
subcon: "cs.Struct" # type: ignore
|
||||
def __init__(
|
||||
self,
|
||||
dc_type: type[DataclassType],
|
||||
reverse: bool = False,
|
||||
) -> None:
|
||||
self.dc_type: type[DataclassType] = dc_type
|
||||
self.reverse: bool = reverse
|
||||
|
||||
# get all fields from the dataclass
|
||||
fields = dataclasses.fields(self.dc_type)
|
||||
if self.reverse:
|
||||
fields = tuple(reversed(fields))
|
||||
|
||||
# extract the construct formats from the struct_type
|
||||
subcon_fields: dict[str, t.Any] = {}
|
||||
for field in fields:
|
||||
subcon_fields[field.name] = field.metadata["subcon"]
|
||||
|
||||
# init adatper
|
||||
super().__init__(cs.Struct(**subcon_fields)) # type: ignore
|
||||
|
||||
def __getattr__(self, name: str) -> t.Any:
|
||||
return getattr(self.subcon, name)
|
||||
|
||||
@override
|
||||
def _decode(
|
||||
self, obj: "cs.Container[t.Any]", context: Context, path: PathType
|
||||
) -> DataclassType:
|
||||
# get all fields from the dataclass
|
||||
fields = dataclasses.fields(self.dc_type)
|
||||
|
||||
# extract all fields from the container, that are used for create the dataclass object
|
||||
dc_init = {}
|
||||
for field in fields:
|
||||
if field.init:
|
||||
value = obj[field.name]
|
||||
dc_init[field.name] = value
|
||||
|
||||
# create object of dataclass
|
||||
dc = self.dc_type(**dc_init) # type: ignore
|
||||
|
||||
# extract all other values from the container, an pass it to the dataclass
|
||||
for field in fields:
|
||||
if not field.init:
|
||||
value = obj[field.name]
|
||||
setattr(dc, field.name, value)
|
||||
|
||||
return dc # type: ignore
|
||||
|
||||
@override
|
||||
def _encode(
|
||||
self, obj: DataclassType, context: Context, path: PathType
|
||||
) -> dict[str, t.Any]:
|
||||
if not isinstance(obj, self.dc_type):
|
||||
raise TypeError(f"'{repr(obj)}' has to be of type {repr(self.dc_type)}")
|
||||
|
||||
# get all fields from the dataclass
|
||||
fields = dataclasses.fields(self.dc_type)
|
||||
|
||||
# extract all fields from the container, that are used for create the dataclass object
|
||||
ret_dict: dict[str, t.Any] = {}
|
||||
for field in fields:
|
||||
value = getattr(obj, field.name)
|
||||
ret_dict[field.name] = value
|
||||
|
||||
return ret_dict
|
||||
|
||||
def DataclassBitStruct(
|
||||
dc_type: type[DataclassType], reverse: bool = False
|
||||
) -> "cs.Transformed[DataclassType, DataclassType] | cs.Restreamed[DataclassType, DataclassType]":
|
||||
class DataclassBitStruct(DataclassStruct):
|
||||
r"""
|
||||
Makes a DataclassStruct inside a Bitwise.
|
||||
|
||||
See :class:`~construct.core.Bitwise` and :class:`~construct_typed.dataclass_struct.DatclassStruct` for semantics and raisable exceptions.
|
||||
|
||||
:param dc_type: Type of the dataclass, which also inherits from DataclassMixin
|
||||
:param reverse: Flag if the fields of the dataclass should be reversed
|
||||
:param constr: TODO
|
||||
:param reverse_fields: Flag if the fields of the dataclass should be reversed
|
||||
|
||||
Example::
|
||||
|
||||
DataclassBitStruct <--> Bitwise(DataclassStruct(...))
|
||||
>>> import dataclasses
|
||||
TODO:
|
||||
>>> from construct import BitsInteger, Flag, Nibble, Padding
|
||||
>>> from construct_typed import DataclassBitStruct, DataclassMixin, csfield
|
||||
>>> @dataclasses.dataclass
|
||||
... class TestDataclass(DataclassMixin):
|
||||
>>> from construct_typed import DataclassBitStruct, csfield, construct
|
||||
... class TestDataclass(DataclassBitStruct):
|
||||
... a: int = csfield(Flag)
|
||||
... b: int = csfield(Nibble)
|
||||
... c: int = csfield(BitsInteger(10))
|
||||
... d: None = csfield(Padding(1))
|
||||
>>> d = DataclassBitStruct(TestDataclass)
|
||||
>>> d = construct(TestDataclass)
|
||||
>>> d.parse(b"\x01\x02")
|
||||
TestDataclass(a=False, b=0, c=129, d=None)
|
||||
"""
|
||||
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
|
||||
TStruct = DataclassStruct
|
||||
TBitStruct = DataclassBitStruct
|
||||
TContainerMixin = DataclassMixin
|
||||
TContainerBase = DataclassMixin
|
||||
TStructField = csfield
|
||||
sfield = csfield
|
||||
def __init_subclass__(
|
||||
cls,
|
||||
constr: "cs.Construct[t.Any, t.Any]" = this_struct,
|
||||
reverse_fields: bool = False,
|
||||
):
|
||||
cls = DataclassStruct.__init_subclass__.__func__(cls, cs.Bitwise(constr), reverse_fields) # type: ignore
|
||||
return cls
|
||||
|
|
|
|||
|
|
@ -12,14 +12,11 @@ if t.TYPE_CHECKING:
|
|||
# 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 ConstantOrContextLambda as ConstantOrContextLambda
|
||||
from construct import ConstantOrContextLambda2 as ConstantOrContextLambda2
|
||||
from construct import Construct as Construct
|
||||
from construct import Context as Context
|
||||
from construct import ListContainer as ListContainer
|
||||
from construct import PathType as PathType
|
||||
from construct import Array as Array
|
||||
from construct import Subconstruct as Subconstruct
|
||||
from construct import Computed as Computed
|
||||
|
||||
|
||||
else:
|
||||
import construct as cs
|
||||
|
|
@ -40,18 +37,22 @@ else:
|
|||
class Context:
|
||||
pass
|
||||
|
||||
class Array(
|
||||
t.Generic[SubconParsedType, SubconBuildTypes],
|
||||
cs.Array,
|
||||
):
|
||||
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]]
|
||||
ConstantOrContextLambda2 = t.Union[ValueType, t.Callable[[Context], ValueType]]
|
||||
PathType = str
|
||||
|
||||
|
||||
@t.runtime_checkable
|
||||
class Constructable(t.Protocol[ParsedType, BuildTypes]):
|
||||
def __construct__(self) -> "Construct[ParsedType, BuildTypes]":
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
def construct(
|
||||
constr: t.Union[
|
||||
Constructable[ParsedType, BuildTypes], "Construct[ParsedType, BuildTypes]"
|
||||
],
|
||||
) -> Construct[ParsedType, BuildTypes]:
|
||||
"""Get construct instance of `Constructable` or `Construct`"""
|
||||
if isinstance(constr, Constructable):
|
||||
constr = constr.__construct__()
|
||||
return constr
|
||||
|
|
@ -1,111 +1,101 @@
|
|||
# pyright: reportAny=false
|
||||
import enum
|
||||
import typing as t
|
||||
|
||||
from typing_extensions import Self, override
|
||||
import construct as cs
|
||||
|
||||
from .generic_wrapper import Construct, Adapter, Context, PathType
|
||||
|
||||
|
||||
# ## TEnum ############################################################################################################
|
||||
class EnumValue:
|
||||
"""
|
||||
This is a helper class for adding documentation to an enum value.
|
||||
"""
|
||||
|
||||
def __init__(self, value: int, doc: str | None = None) -> None:
|
||||
self.value: int = value
|
||||
self.__doc__ = doc if doc else ""
|
||||
from .generics import *
|
||||
|
||||
T = t.TypeVar("T")
|
||||
|
||||
|
||||
# ## EnumConstruct ############################################################################################################
|
||||
class EnumBase(enum.IntEnum):
|
||||
"""
|
||||
Base class for an Enum used in `construct_typed.TEnum`.
|
||||
Base class for an Enum used in `construct_typed.EnumConstruct`.
|
||||
|
||||
This class extends the standard `enum.IntEnum` by.
|
||||
- missing values are automatically generated
|
||||
- possibility to add documentation for each enum value (see `EnumValue`)
|
||||
|
||||
Example::
|
||||
|
||||
>>> class State(EnumBase):
|
||||
... Idle = 1
|
||||
... Running = EnumValue(2, "This is the running state.")
|
||||
|
||||
>>> State(1)
|
||||
<State.Idle: 1>
|
||||
|
||||
>>> State["Idle"]
|
||||
<State.Idle: 1>
|
||||
|
||||
>>> State.Idle
|
||||
<State.Idle: 1>
|
||||
|
||||
>>> State(3) # missing value
|
||||
<State.3: 3>
|
||||
|
||||
>>> State.Running.__doc__ # documentation
|
||||
'This is the running state.'
|
||||
This class extends the standard `enum.IntEnum`, so that missing values are automatically generated.
|
||||
"""
|
||||
|
||||
def __new__(cls, val: EnumValue | int) -> "Self":
|
||||
if isinstance(val, EnumValue):
|
||||
obj = int.__new__(cls, val.value)
|
||||
obj._value_ = val.value
|
||||
obj.__doc__ = val.__doc__
|
||||
else:
|
||||
obj = int.__new__(cls, val)
|
||||
obj._value_ = val
|
||||
obj.__doc__ = ""
|
||||
return obj
|
||||
@classmethod
|
||||
def __init_subclass__(
|
||||
cls,
|
||||
subcon: "cs.Construct[t.Any, t.Any]",
|
||||
**kwargs: t.Any,
|
||||
):
|
||||
super().__init_subclass__(**kwargs)
|
||||
|
||||
# Extend the enum type with _missing_ method. So if a enum value
|
||||
# validate types
|
||||
if not isinstance(subcon, cs.Construct): # type: ignore
|
||||
raise ValueError(
|
||||
f"`subcon` parameter has to be an `Construct` object but is {type(subcon)}"
|
||||
)
|
||||
|
||||
# create construct format
|
||||
enum_constr = EnumConstruct(subcon, cls)
|
||||
|
||||
# save construct format and make the class compatible to `Constructable` protocol
|
||||
setattr(cls, "__construct__", lambda: enum_constr)
|
||||
|
||||
return cls
|
||||
|
||||
# Extend the enum type with __missing__ method. So if a enum value
|
||||
# not found in the enum, a new pseudo member is created.
|
||||
# The idea is taken from: https://stackoverflow.com/a/57179436
|
||||
@classmethod
|
||||
@override
|
||||
def _missing_(cls, value: t.Any) -> enum.Enum | None:
|
||||
def _missing_(cls, value: t.Any) -> t.Optional["EnumBase"]:
|
||||
if isinstance(value, int):
|
||||
pseudo_member = cls._value2member_map_.get(value, None)
|
||||
if pseudo_member is None:
|
||||
new_member = int.__new__(cls, value)
|
||||
# I expect a name attribute to hold a string, hence str(value)
|
||||
# However, new_member._name_ = value works, too
|
||||
new_member._name_ = str(value)
|
||||
new_member._value_ = value
|
||||
new_member.__doc__ = "missing value"
|
||||
pseudo_member = cls._value2member_map_.setdefault(value, new_member)
|
||||
return pseudo_member
|
||||
return cls._create_pseudo_member_(value)
|
||||
return None # will raise the ValueError in Enum.__new__
|
||||
|
||||
@override
|
||||
def __reduce_ex__(self, proto: t.Any) -> tuple[t.Any, ...]:
|
||||
"""
|
||||
Pickle enums by value instead of name (restores pre-3.11 behavior).
|
||||
See https://github.com/python/cpython/pull/26658 for why this exists.
|
||||
"""
|
||||
return self.__class__, (self._value_,)
|
||||
@classmethod
|
||||
def _create_pseudo_member_(cls, value: int) -> "EnumBase":
|
||||
pseudo_member = cls._value2member_map_.get(value, None) # type: ignore
|
||||
if pseudo_member is None:
|
||||
new_member = int.__new__(cls, value)
|
||||
# I expect a name attribute to hold a string, hence str(value)
|
||||
# However, new_member._name_ = value works, too
|
||||
new_member._name_ = str(value)
|
||||
new_member._value_ = value
|
||||
pseudo_member = cls._value2member_map_.setdefault(value, new_member) # type: ignore
|
||||
return pseudo_member # type: ignore
|
||||
|
||||
if t.TYPE_CHECKING:
|
||||
|
||||
@classmethod
|
||||
def __construct__(cls: "t.Type[EnumType]") -> "EnumConstruct[EnumType]":
|
||||
...
|
||||
|
||||
|
||||
EnumType = t.TypeVar("EnumType", bound=EnumBase)
|
||||
|
||||
|
||||
class TEnum(Adapter[int, int, EnumType, EnumType]):
|
||||
class EnumConstruct(Adapter[int, int, EnumType, EnumType]):
|
||||
"""
|
||||
Typed enum.
|
||||
"""
|
||||
def __init__(self, subcon: Construct[int, int], enum_type: type[EnumType]):
|
||||
|
||||
if t.TYPE_CHECKING:
|
||||
|
||||
def __new__(
|
||||
cls, subcon: Construct[int, int], enum_type: t.Type[EnumType]
|
||||
) -> "EnumConstruct[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
|
||||
self.enum_type: type[EnumType] = enum_type
|
||||
self.enum_type = t.cast(t.Type[EnumType], enum_type) # type: ignore
|
||||
|
||||
# init adatper
|
||||
super(TEnum, self).__init__(subcon) # type: ignore
|
||||
super(EnumConstruct, self).__init__(subcon) # type: ignore
|
||||
|
||||
@override
|
||||
def _decode(self, obj: int, context: Context, path: PathType) -> EnumType:
|
||||
return self.enum_type(obj)
|
||||
|
||||
@override
|
||||
def _encode(
|
||||
self,
|
||||
obj: EnumType,
|
||||
|
|
@ -119,88 +109,69 @@ class TEnum(Adapter[int, int, EnumType, EnumType]):
|
|||
)
|
||||
|
||||
|
||||
# ## TFlagsEnum #######################################################################################################
|
||||
# ## FlagsEnumConstruct #######################################################################################################
|
||||
class FlagsEnumBase(enum.IntFlag):
|
||||
"""
|
||||
Base class for an Enum used in `construct_typed.TFlagsEnum`.
|
||||
|
||||
This class extends the standard `enum.IntFlag` by.
|
||||
- possibility to add documentation for each enum value (see `EnumValue`)
|
||||
|
||||
Example::
|
||||
|
||||
>>> class Option(FlagsEnumBase):
|
||||
... OptOne = 1
|
||||
... OptTwo = EnumValue(2, "This is option two.")
|
||||
|
||||
>>> Option(1)
|
||||
<Option.OptOne: 1>
|
||||
|
||||
>>> Option["OptOne"]
|
||||
<Option.OptOne: 1>
|
||||
|
||||
>>> Option.OptOne
|
||||
<Option.OptOne: 1>
|
||||
|
||||
>>> Option(3)
|
||||
<Option.OptTwo|OptOne: 3>
|
||||
|
||||
>>> Option(4)
|
||||
<Option.4: 4>
|
||||
|
||||
>>> Option.OptTwo.__doc__ # documentation
|
||||
'This is option two.'
|
||||
"""
|
||||
|
||||
def __new__(cls, val: EnumValue | int) -> "Self":
|
||||
if isinstance(val, EnumValue):
|
||||
obj = int.__new__(cls, val.value)
|
||||
obj._value_ = val.value
|
||||
obj.__doc__ = val.__doc__
|
||||
else:
|
||||
obj = int.__new__(cls, val)
|
||||
obj._value_ = val
|
||||
obj.__doc__ = ""
|
||||
return obj
|
||||
|
||||
@classmethod
|
||||
@override
|
||||
def _missing_(cls, value: t.Any) -> t.Any:
|
||||
"""
|
||||
Returns member (possibly creating it) if one can be found for value.
|
||||
"""
|
||||
new_member = super()._missing_(value)
|
||||
new_member.__doc__ = "missing value"
|
||||
return new_member
|
||||
def __init_subclass__(
|
||||
cls,
|
||||
subcon: "cs.Construct[t.Any, t.Any]",
|
||||
**kwargs: t.Any,
|
||||
):
|
||||
super().__init_subclass__(**kwargs)
|
||||
|
||||
@override
|
||||
def __reduce_ex__(self, proto: t.Any) -> tuple[t.Any, ...]:
|
||||
"""
|
||||
Pickle enums by value instead of name (restores pre-3.11 behavior).
|
||||
See https://github.com/python/cpython/pull/26658 for why this exists.
|
||||
"""
|
||||
return self.__class__, (self._value_,)
|
||||
# validate types
|
||||
if not isinstance(subcon, cs.Construct): # type: ignore
|
||||
raise ValueError(
|
||||
f"`subcon` parameter has to be an `Construct` object but is {type(subcon)}"
|
||||
)
|
||||
|
||||
# create construct format
|
||||
enum_constr = FlagsEnumConstruct(subcon, cls)
|
||||
|
||||
# save construct format and make the class compatible to `Constructable` protocol
|
||||
setattr(cls, "__construct__", lambda: enum_constr)
|
||||
|
||||
return cls
|
||||
|
||||
if t.TYPE_CHECKING:
|
||||
|
||||
@classmethod
|
||||
def __construct__(
|
||||
cls: "t.Type[FlagsEnumType]",
|
||||
) -> "FlagsEnumConstruct[FlagsEnumType]":
|
||||
...
|
||||
|
||||
|
||||
FlagsEnumType = t.TypeVar("FlagsEnumType", bound=FlagsEnumBase)
|
||||
|
||||
|
||||
class TFlagsEnum(Adapter[int, int, FlagsEnumType, FlagsEnumType]):
|
||||
class FlagsEnumConstruct(Adapter[int, int, FlagsEnumType, FlagsEnumType]):
|
||||
"""
|
||||
Typed enum.
|
||||
"""
|
||||
def __init__(self, subcon: Construct[int, int], enum_type: type[FlagsEnumType]):
|
||||
|
||||
if t.TYPE_CHECKING:
|
||||
|
||||
def __new__(
|
||||
cls, subcon: Construct[int, int], enum_type: t.Type[FlagsEnumType]
|
||||
) -> "FlagsEnumConstruct[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
|
||||
self.enum_type: type[FlagsEnumType] = enum_type
|
||||
self.enum_type = t.cast(t.Type[FlagsEnumType], enum_type) # type: ignore
|
||||
|
||||
# init adatper
|
||||
super(TFlagsEnum, self).__init__(subcon) # type: ignore
|
||||
super(FlagsEnumConstruct, self).__init__(subcon) # type: ignore
|
||||
|
||||
@override
|
||||
def _decode(self, obj: int, context: Context, path: PathType) -> FlagsEnumType:
|
||||
return self.enum_type(obj)
|
||||
|
||||
@override
|
||||
def _encode(
|
||||
self,
|
||||
obj: FlagsEnumType,
|
||||
|
|
|
|||
|
|
@ -1,2 +1,2 @@
|
|||
version = (0, 7, 0)
|
||||
version_string = "0.7.0+wrapper"
|
||||
version = (0, 5, 2)
|
||||
version_string = "0.5.2"
|
||||
3
mypy.ini
Normal file
3
mypy.ini
Normal file
|
|
@ -0,0 +1,3 @@
|
|||
[mypy]
|
||||
strict = True
|
||||
warn_unused_ignores = False
|
||||
|
|
@ -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
|
||||
|
|
@ -1,14 +1,10 @@
|
|||
construct==2.10.70
|
||||
construct==2.10.67
|
||||
pytest>=6.2.0
|
||||
numpy
|
||||
numpy==1.21.*
|
||||
arrow
|
||||
ruamel.yaml
|
||||
cloudpickle
|
||||
lz4
|
||||
black
|
||||
isort
|
||||
mypy
|
||||
cryptography
|
||||
build
|
||||
setuptools
|
||||
wheel
|
||||
mypy
|
||||
64
setup.py
Normal file
64
setup.py
Normal file
|
|
@ -0,0 +1,64 @@
|
|||
#!/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.67"],
|
||||
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 :: Implementation :: CPython",
|
||||
"Typing :: Typed",
|
||||
],
|
||||
)
|
||||
|
|
@ -1,170 +1,38 @@
|
|||
import binascii
|
||||
import io
|
||||
import typing as t
|
||||
|
||||
import pytest
|
||||
from construct import *
|
||||
from construct.lib import *
|
||||
|
||||
import construct_typed as cst
|
||||
|
||||
xfail = pytest.mark.xfail
|
||||
skip = pytest.mark.skip
|
||||
skipif = pytest.mark.skipif
|
||||
|
||||
Buffer = t.Union[bytes, memoryview, bytearray]
|
||||
ParsedType = t.TypeVar("ParsedType")
|
||||
BuildTypes = t.TypeVar("BuildTypes")
|
||||
ContainerType = t.TypeVar("ContainerType", bound=cst.TContainerMixin)
|
||||
T = t.TypeVar("T")
|
||||
import os, math, random, collections, itertools, io, hashlib, binascii
|
||||
|
||||
IdentType = t.TypeVar("IdentType")
|
||||
from construct import *
|
||||
from construct.lib import *
|
||||
|
||||
|
||||
class ZeroIO(io.BufferedIOBase):
|
||||
def read(self, __size: t.Optional[int] = None) -> bytes:
|
||||
def read(self, __size=None):
|
||||
if __size is not None:
|
||||
return bytes(__size)
|
||||
else:
|
||||
return bytes(0)
|
||||
|
||||
def read1(self, __size: int = 0) -> bytes:
|
||||
def read1(self, __size=0):
|
||||
return bytes(__size)
|
||||
|
||||
|
||||
def ident(x: IdentType) -> IdentType:
|
||||
return x
|
||||
ident = lambda x: x
|
||||
devzero = ZeroIO()
|
||||
|
||||
|
||||
devzero: t.BinaryIO = ZeroIO() # type: ignore
|
||||
|
||||
|
||||
def raises(
|
||||
func: t.Callable[..., t.Any], *args: t.Any, **kw: t.Any
|
||||
) -> t.Union[t.Any, Exception]:
|
||||
def raises(func, *args, **kw):
|
||||
try:
|
||||
return func(*args, **kw)
|
||||
except Exception as e:
|
||||
return e.__class__
|
||||
|
||||
|
||||
@t.overload
|
||||
def common(
|
||||
format: cst.TStruct[ContainerType],
|
||||
datasample: Buffer,
|
||||
objsample: t.Union[ContainerType, t.Dict[str, t.Any]],
|
||||
sizesample: t.Union[int, t.Type[Exception]] = ...,
|
||||
**kw: t.Any
|
||||
) -> None:
|
||||
...
|
||||
|
||||
|
||||
@t.overload
|
||||
def common(
|
||||
format: "Construct[ListContainer[ParsedType], t.Any]",
|
||||
datasample: Buffer,
|
||||
objsample: t.List[ParsedType],
|
||||
sizesample: t.Union[int, t.Type[Exception]] = ...,
|
||||
**kw: t.Any
|
||||
) -> None:
|
||||
...
|
||||
|
||||
|
||||
@t.overload
|
||||
def common(
|
||||
format: "Construct[Container[t.Any], t.Any]",
|
||||
datasample: Buffer,
|
||||
objsample: t.Dict[str, t.Any],
|
||||
sizesample: t.Union[int, t.Type[Exception]] = ...,
|
||||
**kw: t.Any
|
||||
) -> None:
|
||||
...
|
||||
|
||||
|
||||
@t.overload
|
||||
def common(
|
||||
format: "Construct[t.Union[EnumInteger, EnumIntegerString], t.Any]",
|
||||
datasample: Buffer,
|
||||
objsample: t.Union[int, str],
|
||||
sizesample: t.Union[int, t.Type[Exception]] = ...,
|
||||
**kw: t.Any
|
||||
) -> None:
|
||||
...
|
||||
|
||||
|
||||
@t.overload
|
||||
def common(
|
||||
format: "Construct[HexDisplayedInteger, t.Any]",
|
||||
datasample: Buffer,
|
||||
objsample: t.Union[HexDisplayedInteger, int],
|
||||
sizesample: t.Union[int, t.Type[Exception]] = ...,
|
||||
**kw: t.Any
|
||||
) -> None:
|
||||
...
|
||||
|
||||
|
||||
@t.overload
|
||||
def common(
|
||||
format: "Construct[HexDisplayedBytes, t.Any]",
|
||||
datasample: Buffer,
|
||||
objsample: t.Union[HexDisplayedBytes, bytes],
|
||||
sizesample: t.Union[int, t.Type[Exception]] = ...,
|
||||
**kw: t.Any
|
||||
) -> None:
|
||||
...
|
||||
|
||||
|
||||
@t.overload
|
||||
def common(
|
||||
format: "Construct[HexDisplayedDict[str, t.Any], t.Any]",
|
||||
datasample: Buffer,
|
||||
objsample: t.Dict[str, t.Any],
|
||||
sizesample: t.Union[int, t.Type[Exception]] = ...,
|
||||
**kw: t.Any
|
||||
) -> None:
|
||||
...
|
||||
|
||||
|
||||
@t.overload
|
||||
def common(
|
||||
format: "Construct[HexDumpDisplayedBytes, t.Any]",
|
||||
datasample: Buffer,
|
||||
objsample: t.Union[HexDumpDisplayedBytes, bytes],
|
||||
sizesample: t.Union[int, t.Type[Exception]] = ...,
|
||||
**kw: t.Any
|
||||
) -> None:
|
||||
...
|
||||
|
||||
|
||||
@t.overload
|
||||
def common(
|
||||
format: "Construct[HexDumpDisplayedDict[str, t.Any], t.Any]",
|
||||
datasample: Buffer,
|
||||
objsample: t.Dict[str, t.Any],
|
||||
sizesample: t.Union[int, t.Type[Exception]] = ...,
|
||||
**kw: t.Any
|
||||
) -> None:
|
||||
...
|
||||
|
||||
|
||||
@t.overload
|
||||
def common(
|
||||
format: "Construct[ParsedType, t.Any]",
|
||||
datasample: Buffer,
|
||||
objsample: ParsedType,
|
||||
sizesample: t.Union[int, t.Type[Exception]] = ...,
|
||||
**kw: t.Any
|
||||
) -> None:
|
||||
...
|
||||
|
||||
|
||||
def common(
|
||||
format: "Construct[t.Any, t.Any]",
|
||||
datasample: Buffer,
|
||||
objsample: t.Any,
|
||||
sizesample: t.Union[int, t.Type[Exception]] = SizeofError,
|
||||
**kw: t.Any
|
||||
) -> None:
|
||||
def common(format, datasample, objsample, sizesample=SizeofError, **kw):
|
||||
obj = format.parse(datasample, **kw)
|
||||
assert obj == objsample
|
||||
data = format.build(objsample, **kw)
|
||||
|
|
@ -176,35 +44,35 @@ def common(
|
|||
size = format.sizeof(**kw)
|
||||
assert size == sizesample
|
||||
else:
|
||||
size_ex = raises(format.sizeof, **kw)
|
||||
assert size_ex == sizesample
|
||||
size = raises(format.sizeof, **kw)
|
||||
assert size == sizesample
|
||||
|
||||
|
||||
def setattrs(obj: T, **kwargs: t.Any) -> T:
|
||||
"""Set multiple named values of an object"""
|
||||
def setattrs(obj, **kwargs):
|
||||
""" Set multiple named values of an object """
|
||||
for name, value in kwargs.items():
|
||||
setattr(obj, name, value)
|
||||
return obj
|
||||
|
||||
|
||||
def commonhex(format: "Construct[t.Any, t.Any]", hexdata: str) -> None:
|
||||
def commonhex(format, hexdata):
|
||||
commonbytes(format, binascii.unhexlify(hexdata))
|
||||
|
||||
|
||||
def commondumpdeprecated(format: "Construct[t.Any, t.Any]", filename: str) -> None:
|
||||
def commondumpdeprecated(format, filename):
|
||||
filename = "tests/deprecated_gallery/blobs/" + filename
|
||||
with open(filename, "rb") as f:
|
||||
data = f.read()
|
||||
commonbytes(format, data)
|
||||
|
||||
|
||||
def commondump(format: "Construct[t.Any, t.Any]", filename: str) -> None:
|
||||
def commondump(format, filename):
|
||||
filename = "tests/gallery/blobs/" + filename
|
||||
with open(filename, "rb") as f:
|
||||
data = f.read()
|
||||
commonbytes(format, data)
|
||||
|
||||
|
||||
def commonbytes(format: "Construct[t.Any, t.Any]", data: bytes) -> None:
|
||||
def commonbytes(format, data):
|
||||
obj = format.parse(data)
|
||||
format.build(obj)
|
||||
data2 = format.build(obj)
|
||||
|
|
|
|||
109
tests/declarativeunittest.pyi
Normal file
109
tests/declarativeunittest.pyi
Normal file
|
|
@ -0,0 +1,109 @@
|
|||
import typing as t
|
||||
from construct import *
|
||||
from construct.lib import *
|
||||
import construct_typed as cst
|
||||
|
||||
Buffer = t.Union[bytes, memoryview, bytearray]
|
||||
ParsedType = t.TypeVar("ParsedType")
|
||||
BuildTypes = t.TypeVar("BuildTypes")
|
||||
ContainerType = t.TypeVar("ContainerType", bound=cst.TContainerMixin)
|
||||
T = t.TypeVar("T")
|
||||
|
||||
IdentType = t.TypeVar("IdentType")
|
||||
|
||||
def ident(p1: IdentType) -> IdentType: ...
|
||||
|
||||
devzero: t.BinaryIO
|
||||
|
||||
def raises(
|
||||
func: t.Callable[..., t.Any], *args: t.Any, **kw: t.Any
|
||||
) -> t.Union[t.Any, Exception]: ...
|
||||
@t.overload
|
||||
def common(
|
||||
format: cst.TStruct[ContainerType],
|
||||
datasample: Buffer,
|
||||
objsample: t.Union[ContainerType, t.Dict[str, t.Any]],
|
||||
sizesample: t.Union[int, t.Type[Exception]] = ...,
|
||||
**kw: t.Any
|
||||
) -> None: ...
|
||||
@t.overload
|
||||
def common(
|
||||
format: Construct[ListContainer[ParsedType], t.Any],
|
||||
datasample: Buffer,
|
||||
objsample: t.List[ParsedType],
|
||||
sizesample: t.Union[int, t.Type[Exception]] = ...,
|
||||
**kw: t.Any
|
||||
) -> None: ...
|
||||
@t.overload
|
||||
def common(
|
||||
format: Construct[Container[t.Any], t.Any],
|
||||
datasample: Buffer,
|
||||
objsample: t.Dict[str, t.Any],
|
||||
sizesample: t.Union[int, t.Type[Exception]] = ...,
|
||||
**kw: t.Any
|
||||
) -> None: ...
|
||||
@t.overload
|
||||
def common(
|
||||
format: Construct[t.Union[EnumInteger, EnumIntegerString], t.Any],
|
||||
datasample: Buffer,
|
||||
objsample: t.Union[int, str],
|
||||
sizesample: t.Union[int, t.Type[Exception]] = ...,
|
||||
**kw: t.Any
|
||||
) -> None: ...
|
||||
@t.overload
|
||||
def common(
|
||||
format: Construct[HexDisplayedInteger, t.Any],
|
||||
datasample: Buffer,
|
||||
objsample: t.Union[HexDisplayedInteger, int],
|
||||
sizesample: t.Union[int, t.Type[Exception]] = ...,
|
||||
**kw: t.Any
|
||||
) -> None: ...
|
||||
@t.overload
|
||||
def common(
|
||||
format: Construct[HexDisplayedBytes, t.Any],
|
||||
datasample: Buffer,
|
||||
objsample: t.Union[HexDisplayedBytes, bytes],
|
||||
sizesample: t.Union[int, t.Type[Exception]] = ...,
|
||||
**kw: t.Any
|
||||
) -> None: ...
|
||||
@t.overload
|
||||
def common(
|
||||
format: Construct[HexDisplayedDict[str, t.Any], t.Any],
|
||||
datasample: Buffer,
|
||||
objsample: t.Dict[str, t.Any],
|
||||
sizesample: t.Union[int, t.Type[Exception]] = ...,
|
||||
**kw: t.Any
|
||||
) -> None: ...
|
||||
@t.overload
|
||||
def common(
|
||||
format: Construct[HexDumpDisplayedBytes, t.Any],
|
||||
datasample: Buffer,
|
||||
objsample: t.Union[HexDumpDisplayedBytes, bytes],
|
||||
sizesample: t.Union[int, t.Type[Exception]] = ...,
|
||||
**kw: t.Any
|
||||
) -> None: ...
|
||||
@t.overload
|
||||
def common(
|
||||
format: Construct[HexDumpDisplayedDict[str, t.Any], t.Any],
|
||||
datasample: Buffer,
|
||||
objsample: t.Dict[str, t.Any],
|
||||
sizesample: t.Union[int, t.Type[Exception]] = ...,
|
||||
**kw: t.Any
|
||||
) -> None: ...
|
||||
@t.overload
|
||||
def common(
|
||||
format: Construct[ParsedType, t.Any],
|
||||
datasample: Buffer,
|
||||
objsample: ParsedType,
|
||||
sizesample: t.Union[int, t.Type[Exception]] = ...,
|
||||
**kw: t.Any
|
||||
) -> None: ...
|
||||
def setattrs(obj: T, **kwargs: t.Any) -> T: ...
|
||||
def commonhex(format: Construct[t.Any, t.Any], hexdata: str) -> None: ...
|
||||
def commondumpdeprecated(
|
||||
format: Construct[t.Any, t.Any], filename: str
|
||||
) -> None: ...
|
||||
def commondump(format: Construct[t.Any, t.Any], filename: str) -> None: ...
|
||||
def commonbytes(
|
||||
format: Construct[ParsedType, t.Any], data: ParsedType
|
||||
) -> None: ...
|
||||
|
|
@ -1,6 +1,6 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# mypy: no-warn-unused-ignores
|
||||
from .declarativeunittest import raises, common, ident, devzero
|
||||
|
||||
from .declarativeunittest import raises, common, commonhex, commondumpdeprecated, commondump, commonbytes, ident, devzero
|
||||
from construct.core import *
|
||||
from construct import *
|
||||
from construct.lib import *
|
||||
|
|
@ -151,29 +151,17 @@ def test_formatfield_bool_issue_901() -> None:
|
|||
assert d.sizeof() == 1
|
||||
|
||||
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)
|
||||
common(d, b"\x01\x02\x03\x04", 0x01020304, 4)
|
||||
common(d, b"\xff\xff\xff\xff", -1, 4)
|
||||
d = BytesInteger(4, signed=False, swapped=this.swapped)
|
||||
common(d, b"\x01\x02\x03\x04", 0x01020304, 4, swapped=False)
|
||||
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(4, signed=False).build, -1) == IntegerError
|
||||
common(BytesInteger(0), b"", 0, 0)
|
||||
|
||||
def test_bitsinteger() -> None:
|
||||
d = BitsInteger(0)
|
||||
assert raises(d.parse, b"") == IntegerError
|
||||
assert raises(d.build, 0) == IntegerError
|
||||
d = BitsInteger(8)
|
||||
common(d, b"\x01\x01\x01\x01\x01\x01\x01\x01", 255, 8)
|
||||
d = BitsInteger(8, signed=True)
|
||||
|
|
@ -183,17 +171,9 @@ def test_bitsinteger() -> None:
|
|||
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"\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(8, signed=False).build, -1) == IntegerError
|
||||
common(BitsInteger(0), b"", 0, 0)
|
||||
|
||||
def test_varint() -> None:
|
||||
d = VarInt
|
||||
|
|
@ -244,8 +224,8 @@ def test_paddedstring() -> None:
|
|||
common(PaddedString(100, e), data, s, 100)
|
||||
|
||||
for e in ["ascii","utf8","utf16","utf-16-le","utf32","utf-32-le"]:
|
||||
assert PaddedString(10, e).sizeof() == 10
|
||||
assert PaddedString(this.n, e).sizeof(n=10) == 10
|
||||
PaddedString(10, e).sizeof() == 10
|
||||
PaddedString(this.n, e).sizeof(n=10) == 10
|
||||
|
||||
def test_pascalstring() -> None:
|
||||
for e,_ in [("utf8",1),("utf16",2),("utf_16_le",2),("utf32",4),("utf_32_le",4)]:
|
||||
|
|
@ -256,8 +236,8 @@ def test_pascalstring() -> None:
|
|||
common(PascalString(sc, e), sc.build(0), u"")
|
||||
|
||||
for e in ["utf8","utf16","utf-16-le","utf32","utf-32-le","ascii"]:
|
||||
assert raises(PascalString(Byte, e).sizeof) == SizeofError
|
||||
assert raises(PascalString(VarInt, e).sizeof) == SizeofError
|
||||
raises(PascalString(Byte, e).sizeof) == SizeofError
|
||||
raises(PascalString(VarInt, e).sizeof) == SizeofError
|
||||
|
||||
def test_cstring() -> None:
|
||||
s = u""
|
||||
|
|
@ -266,12 +246,12 @@ def test_cstring() -> None:
|
|||
common(CString(e), s.encode(e)+bytes(us), s)
|
||||
common(CString(e), bytes(us), u"")
|
||||
|
||||
assert CString("utf8").build(s) == b'\xd0\x90\xd1\x84\xd0\xbe\xd0\xbd'+b"\x00"
|
||||
assert CString("utf16").build(s) == b'\xff\xfe\x10\x04D\x04>\x04=\x04'+b"\x00\x00"
|
||||
assert CString("utf32").build(s) == b'\xff\xfe\x00\x00\x10\x04\x00\x00D\x04\x00\x00>\x04\x00\x00=\x04\x00\x00'+b"\x00\x00\x00\x00"
|
||||
CString("utf8").build(s) == b'\xd0\x90\xd1\x84\xd0\xbe\xd0\xbd'+b"\x00"
|
||||
CString("utf16").build(s) == b'\xff\xfe\x10\x04D\x04>\x04=\x04'+b"\x00\x00"
|
||||
CString("utf32").build(s) == b'\xff\xfe\x00\x00\x10\x04\x00\x00D\x04\x00\x00>\x04\x00\x00=\x04\x00\x00'+b"\x00\x00\x00\x00"
|
||||
|
||||
for e in ["utf8","utf16","utf-16-le","utf32","utf-32-le","ascii"]:
|
||||
assert raises(CString(e).sizeof) == SizeofError
|
||||
raises(CString(e).sizeof) == SizeofError
|
||||
|
||||
def test_greedystring() -> None:
|
||||
for e,_ in [("utf8",1),("utf16",2),("utf_16_le",2),("utf32",4),("utf_32_le",4)]:
|
||||
|
|
@ -280,7 +260,7 @@ def test_greedystring() -> None:
|
|||
common(GreedyString(e), b"", u"")
|
||||
|
||||
for e in ["utf8","utf16","utf-16-le","utf32","utf-32-le","ascii"]:
|
||||
assert raises(GreedyString(e).sizeof) == SizeofError
|
||||
raises(GreedyString(e).sizeof) == SizeofError
|
||||
|
||||
def test_string_encodings() -> None:
|
||||
# checks that "-" is replaced with "_"
|
||||
|
|
@ -291,7 +271,7 @@ def test_flag() -> None:
|
|||
d = Flag
|
||||
common(d, b"\x00", False, 1)
|
||||
common(d, b"\x01", True, 1)
|
||||
assert d.parse(b"\xff") == True
|
||||
d.parse(b"\xff") == True
|
||||
|
||||
def test_enum() -> None:
|
||||
d = Enum(Byte, one=1, two=2, four=4, eight=8)
|
||||
|
|
@ -440,11 +420,11 @@ def test_struct_proper_context() -> None:
|
|||
"x"/Byte,
|
||||
"inner"/Struct(
|
||||
"y"/Byte,
|
||||
"a"/Computed(this._.x+1), # type: ignore
|
||||
"b"/Computed(this.y+2), # type: ignore
|
||||
"a"/Computed(this._.x+1),
|
||||
"b"/Computed(this.y+2),
|
||||
),
|
||||
"c"/Computed(this.x+3), # type: ignore
|
||||
"d"/Computed(this.inner.y+4), # type: ignore
|
||||
"c"/Computed(this.x+3),
|
||||
"d"/Computed(this.inner.y+4),
|
||||
)
|
||||
assert d.parse(b"\x01\x0f") == Container(x=1, inner=Container(y=15, a=2, b=17), c=4, d=19)
|
||||
|
||||
|
|
@ -531,7 +511,7 @@ def test_const() -> None:
|
|||
|
||||
def test_computed() -> None:
|
||||
common(Computed(255), b"", 255, 0)
|
||||
common(Computed(lambda ctx: 255), b"", 255, 0) # type: ignore
|
||||
common(Computed(lambda ctx: 255), b"", 255, 0)
|
||||
assert Computed(255).build(None) == b""
|
||||
assert Struct(Computed(255)).build({}) == b""
|
||||
assert raises(Computed(this.missing).parse, b"") == KeyError
|
||||
|
|
@ -611,7 +591,7 @@ def test_rebuild_issue_664() -> None:
|
|||
def test_default() -> None:
|
||||
d = Default(Byte, 0)
|
||||
common(d, b"\xff", 255, 1)
|
||||
assert d.build(None) == b"\x00"
|
||||
d.build(None) == b"\x00"
|
||||
|
||||
def test_check() -> None:
|
||||
common(Check(True), b"", None, 0)
|
||||
|
|
@ -657,7 +637,8 @@ def test_numpy_error() -> None:
|
|||
numpy.load(io.BytesIO(b"")) # type: ignore
|
||||
|
||||
def test_namedtuple() -> None:
|
||||
coord = t.NamedTuple("coord", [("x", int), ("y", int), ("z", int)])
|
||||
import collections
|
||||
coord = collections.namedtuple("coord", "x y z")
|
||||
d1 = NamedTuple("coord", "x y z", Array(3, Byte))
|
||||
common(d1, b"123", coord(49,50,51), 3)
|
||||
d2 = NamedTuple("coord", "x y z", GreedyRange(Byte))
|
||||
|
|
@ -727,13 +708,10 @@ def test_hexdump() -> None:
|
|||
|
||||
def test_hexdump_regression_issue_188() -> None:
|
||||
# Hex HexDump were not inheriting subcon flags
|
||||
a = Hex(Const(b"MZ"))
|
||||
d = Struct(a)
|
||||
d = Struct(Hex(Const(b"MZ")))
|
||||
assert d.parse(b"MZ") == Container()
|
||||
assert d.build(dict()) == b"MZ"
|
||||
|
||||
b = HexDump(Const(b"MZ"))
|
||||
d = Struct(b)
|
||||
d = Struct(HexDump(Const(b"MZ")))
|
||||
assert d.parse(b"MZ") == Container()
|
||||
assert d.build(dict()) == b"MZ"
|
||||
|
||||
|
|
@ -830,10 +808,8 @@ def test_select_buildfromnone_issue_747() -> None:
|
|||
assert d.build(dict()) == b""
|
||||
|
||||
def test_if() -> None:
|
||||
d = If(True, Byte)
|
||||
common(d, b"\x01", 1, 1)
|
||||
d = If(False, Byte)
|
||||
common(d, b"", None, 0)
|
||||
common(If(True, Byte), b"\x01", 1, 1)
|
||||
common(If(False, Byte), b"", None, 0)
|
||||
|
||||
def test_ifthenelse() -> None:
|
||||
common(IfThenElse(True, Int8ub, Int16ub), b"\x01", 1, 1)
|
||||
|
|
@ -946,17 +922,6 @@ def test_peek() -> None:
|
|||
assert d4.build(Container(a=0x01, b=0x0102)) == b""
|
||||
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:
|
||||
d = Seek(5)
|
||||
assert d.parse(b"") == 5
|
||||
|
|
@ -1068,14 +1033,13 @@ def test_prefixed() -> None:
|
|||
common(d5, b"\x0a"+bytes(10), u"\x00"*10, SizeofError)
|
||||
|
||||
def test_prefixedarray() -> None:
|
||||
d = PrefixedArray(Byte, Byte)
|
||||
common(d, b"\x02\x0a\x0b", [10,11], SizeofError)
|
||||
assert d.parse(b"\x03\x01\x02\x03") == [1,2,3]
|
||||
assert d.parse(b"\x00") == []
|
||||
assert d.build([1,2,3]) == b"\x03\x01\x02\x03"
|
||||
assert raises(d.parse, b"") == StreamError
|
||||
assert raises(d.parse, b"\x03\x01") == StreamError
|
||||
assert raises(d.sizeof) == SizeofError
|
||||
common(PrefixedArray(Byte,Byte), b"\x02\x0a\x0b", [10,11], SizeofError)
|
||||
assert PrefixedArray(Byte, Byte).parse(b"\x03\x01\x02\x03") == [1,2,3]
|
||||
assert PrefixedArray(Byte, Byte).parse(b"\x00") == []
|
||||
assert PrefixedArray(Byte, Byte).build([1,2,3]) == b"\x03\x01\x02\x03"
|
||||
assert raises(PrefixedArray(Byte, Byte).parse, b"") == StreamError
|
||||
assert raises(PrefixedArray(Byte, Byte).parse, b"\x03\x01") == StreamError
|
||||
assert raises(PrefixedArray(Byte, Byte).sizeof) == SizeofError
|
||||
|
||||
def test_fixedsized() -> None:
|
||||
d1 = FixedSized(10, Byte)
|
||||
|
|
@ -1249,7 +1213,7 @@ def test_checksum() -> None:
|
|||
def test_checksum_nonbytes_issue_323() -> None:
|
||||
d = Struct(
|
||||
"vals" / Byte[2],
|
||||
"checksum" / Checksum(Byte, lambda vals: int(sum(vals)) & 0xFF, this.vals),
|
||||
"checksum" / Checksum(Byte, lambda vals: sum(vals) & 0xFF, this.vals),
|
||||
)
|
||||
assert d.parse(b"\x00\x00\x00") == Container(vals=[0, 0], checksum=0)
|
||||
assert raises(d.parse, b"\x00\x00\x01") == ChecksumError
|
||||
|
|
@ -1365,105 +1329,6 @@ def test_compressed_prefixed() -> None:
|
|||
assert st.parse(st.build(Container(one=zeros,two=zeros))) == Container(one=zeros,two=zeros)
|
||||
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:
|
||||
data = b"0" * 1000
|
||||
assert Rebuffered(Array(1000,Byte)).parse_stream(io.BytesIO(data)) == [48]*1000
|
||||
|
|
@ -1679,7 +1544,7 @@ def test_operators() -> None:
|
|||
assert d.docs == "description"
|
||||
d = "description" * Byte
|
||||
assert d.docs == "description"
|
||||
_ = """
|
||||
"""
|
||||
description
|
||||
""" * \
|
||||
Byte
|
||||
|
|
@ -1821,11 +1686,9 @@ def test_from_issue_244() -> None:
|
|||
assert d.parse(b"abcd") == [Container(num=97, index=0),Container(num=98, index=1),Container(num=99, index=2),Container(num=100, index=3),]
|
||||
|
||||
def test_from_issue_269() -> None:
|
||||
a = If(this.enabled, Padding(2))
|
||||
d = Struct("enabled" / Byte, a)
|
||||
d = Struct("enabled" / Byte, If(this.enabled, Padding(2)))
|
||||
assert d.build(dict(enabled=1)) == b"\x01\x00\x00"
|
||||
assert d.build(dict(enabled=0)) == b"\x00"
|
||||
|
||||
d = Struct("enabled" / Byte, "pad" / If(this.enabled, Padding(2)))
|
||||
assert d.build(dict(enabled=1)) == b"\x01\x00\x00"
|
||||
assert d.build(dict(enabled=0)) == b"\x00"
|
||||
|
|
@ -1841,7 +1704,7 @@ def test_from_issue_324() -> None:
|
|||
)),
|
||||
"checksum" / Checksum(
|
||||
Byte,
|
||||
lambda data: int(sum(data)) & 0xFF,
|
||||
lambda data: sum(data) & 0xFF,
|
||||
this.vals.data
|
||||
),
|
||||
)
|
||||
|
|
@ -1932,11 +1795,11 @@ def test_pickling_constructs() -> None:
|
|||
)
|
||||
data = bytes(100)
|
||||
|
||||
du = cloudpickle.loads(cloudpickle.dumps(d, protocol=-1)) # type: ignore
|
||||
du = cloudpickle.loads(cloudpickle.dumps(d, protocol=-1))
|
||||
assert du.parse(data) == d.parse(data)
|
||||
|
||||
def test_pickling_constructs_issue_894() -> None:
|
||||
import cloudpickle # type: ignore
|
||||
import cloudpickle
|
||||
|
||||
fundus_header = Struct(
|
||||
'width' / Int32un,
|
||||
|
|
@ -1948,7 +1811,7 @@ def test_pickling_constructs_issue_894() -> None:
|
|||
'img' / Int8un,
|
||||
)
|
||||
|
||||
cloudpickle.dumps(fundus_header) # type: ignore
|
||||
cloudpickle.dumps(fundus_header)
|
||||
|
||||
def test_exposing_members_attributes() -> None:
|
||||
d1 = Struct(
|
||||
|
|
@ -2159,7 +2022,7 @@ def test_struct_root_topmost() -> None:
|
|||
assert d.parse(b"", z=2) == Container(x=1, inner=Container(inner2=Container(x=1,z=2,zz=2)))
|
||||
|
||||
def test_parsedhook_repeatersdiscard() -> None:
|
||||
outputs: t.List[int] = []
|
||||
outputs = []
|
||||
def printobj1(obj: int, ctx: "Context") -> None:
|
||||
outputs.append(obj)
|
||||
d1 = GreedyRange(Byte * printobj1, discard=True)
|
||||
|
|
|
|||
|
|
@ -1,21 +1,22 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# pyright: strict
|
||||
import dataclasses
|
||||
import enum
|
||||
import textwrap
|
||||
import typing as t
|
||||
|
||||
import pytest
|
||||
import construct as cs
|
||||
|
||||
import construct_typed as cst
|
||||
from construct_typed import DataclassBitStruct, DataclassMixin, DataclassStruct, csfield
|
||||
from construct_typed import (
|
||||
DataclassBitStruct,
|
||||
DataclassStruct,
|
||||
csfield,
|
||||
construct,
|
||||
)
|
||||
|
||||
from .declarativeunittest import common, raises, setattrs
|
||||
|
||||
|
||||
def test_dataclass_const_default() -> None:
|
||||
@dataclasses.dataclass
|
||||
class ConstDefaultTest(DataclassMixin):
|
||||
class ConstDefaultTest(DataclassStruct):
|
||||
const_bytes: bytes = csfield(cs.Const(b"BMP"))
|
||||
const_int: int = csfield(cs.Const(5, cs.Int8ub))
|
||||
default_int: int = csfield(cs.Default(cs.Int8ub, 28))
|
||||
|
|
@ -31,8 +32,7 @@ def test_dataclass_const_default() -> None:
|
|||
|
||||
|
||||
def test_dataclass_access() -> None:
|
||||
@dataclasses.dataclass
|
||||
class TestTContainer(DataclassMixin):
|
||||
class TestTContainer(DataclassStruct):
|
||||
a: t.Optional[int] = csfield(cs.Const(1, cs.Byte))
|
||||
b: int = csfield(cs.Int8ub)
|
||||
|
||||
|
|
@ -52,17 +52,16 @@ def test_dataclass_access() -> None:
|
|||
assert tcontainer["a"] == 6
|
||||
|
||||
# wrong creation
|
||||
assert raises(lambda: TestTContainer(a=0, b=1)) == TypeError
|
||||
assert raises(lambda: TestTContainer(a=0, b=1)) == TypeError # type: ignore
|
||||
|
||||
|
||||
def test_dataclass_str_repr() -> None:
|
||||
@dataclasses.dataclass
|
||||
class Image(DataclassMixin):
|
||||
class Image(DataclassStruct):
|
||||
signature: t.Optional[bytes] = csfield(cs.Const(b"BMP"))
|
||||
width: int = csfield(cs.Int8ub)
|
||||
height: int = csfield(cs.Int8ub)
|
||||
|
||||
format = DataclassStruct(Image)
|
||||
format = construct(Image)
|
||||
obj = Image(width=3, height=2)
|
||||
assert (
|
||||
str(obj)
|
||||
|
|
@ -75,78 +74,192 @@ def test_dataclass_str_repr() -> None:
|
|||
)
|
||||
|
||||
|
||||
def test_dataclass_ifthenelse() -> None:
|
||||
@dataclasses.dataclass
|
||||
class IfThenElseTest(DataclassMixin):
|
||||
test_if: t.Optional[int] = csfield(cs.If(False, cs.Int8ub))
|
||||
test_ifthenelse: t.Optional[int] = csfield(
|
||||
cs.IfThenElse(True, cs.Int8ub, cs.Pass)
|
||||
)
|
||||
|
||||
a = IfThenElseTest(test_if=None, test_ifthenelse=None)
|
||||
assert a.test_if == None
|
||||
assert a.test_ifthenelse == None
|
||||
|
||||
|
||||
def test_dataclass_struct() -> None:
|
||||
@dataclasses.dataclass
|
||||
class Image(DataclassMixin):
|
||||
class Image(DataclassStruct):
|
||||
width: int = csfield(cs.Int8ub)
|
||||
height: int = csfield(cs.Int8ub)
|
||||
pixels: bytes = csfield(cs.Bytes(cs.this.height * cs.this.width))
|
||||
|
||||
common(
|
||||
cst.DataclassStruct(Image),
|
||||
construct(Image),
|
||||
b"\x01\x0212",
|
||||
Image(width=1, height=2, pixels=b"12"),
|
||||
)
|
||||
|
||||
# check __getattr__
|
||||
c = cst.DataclassStruct(Image)
|
||||
c = Image.__construct__() # TODO: construct(Image)
|
||||
assert c.width.name == "width"
|
||||
assert c.height.name == "height"
|
||||
assert c.width.subcon is cs.Int8ub
|
||||
assert c.height.subcon is cs.Int8ub
|
||||
|
||||
|
||||
def test_attrs() -> None:
|
||||
import attr
|
||||
|
||||
@attr.s(kw_only=True)
|
||||
class TestAttrs:
|
||||
a: int = attr.ib()
|
||||
b: int = attr.ib(default=5)
|
||||
c: int = attr.ib()
|
||||
|
||||
testattrs1 = TestAttrs(a=5, c=10)
|
||||
print(testattrs1)
|
||||
|
||||
|
||||
import construct_typed.attrs_struct as cst5
|
||||
|
||||
|
||||
def test_attrs_struct_example() -> None:
|
||||
from construct import Bytes, Int8ub, this
|
||||
from construct_typed import AttrsStruct, attrs_field, construct
|
||||
|
||||
class Image(AttrsStruct):
|
||||
width: int = attrs_field(Int8ub)
|
||||
height: int = attrs_field(Int8ub)
|
||||
pixels: bytes = attrs_field(Bytes(this.height * this.width))
|
||||
|
||||
d = construct(Image)
|
||||
obj = d.parse(b"\x01\x0212")
|
||||
assert obj.width is obj["width"]
|
||||
assert obj.height is obj["height"]
|
||||
assert obj.pixels is obj["pixels"]
|
||||
|
||||
|
||||
def test_attrs_struct() -> None:
|
||||
class Test(cst5.AttrsStruct):
|
||||
a: int = cst5.attrs_field(cs.Byte)
|
||||
b: int = cst5.attrs_field(cs.Byte)
|
||||
c: int = cst5.attrs_field(cs.Byte)
|
||||
d: int = cst5.attrs_field(cs.Byte)
|
||||
|
||||
common(
|
||||
cst.construct(Test),
|
||||
b"\x00\x01\x02\x03",
|
||||
Test(a=0, b=1, c=2, d=3),
|
||||
4,
|
||||
)
|
||||
|
||||
|
||||
def test_attrs_struct_to_str() -> None:
|
||||
class Test(cst5.AttrsStruct):
|
||||
a: int = cst5.attrs_field(cs.Byte)
|
||||
b: int = cst5.attrs_field(cs.Byte)
|
||||
c: int = cst5.attrs_field(cs.Byte)
|
||||
d: int = cst5.attrs_field(cs.Byte)
|
||||
|
||||
obj = Test(a=0, b=1, c=2, d=3)
|
||||
assert str(obj) == "Test(a=0, b=1, c=2, d=3)"
|
||||
|
||||
|
||||
def test_attrs_struct_simple_constr() -> None:
|
||||
class Test(cst5.AttrsStruct, constr=cst5.this_struct):
|
||||
a: int = cst5.attrs_field(cs.Byte)
|
||||
b: int = cst5.attrs_field(cs.Byte)
|
||||
c: int = cst5.attrs_field(cs.Byte)
|
||||
d: int = cst5.attrs_field(cs.Byte)
|
||||
|
||||
common(
|
||||
cst.construct(Test),
|
||||
b"\x00\x01\x02\x03",
|
||||
Test(a=0, b=1, c=2, d=3),
|
||||
4,
|
||||
)
|
||||
|
||||
|
||||
def test_attrs_struct_complex_constr() -> None:
|
||||
class Test(cst5.AttrsStruct, constr=cs.Bitwise(cst5.this_struct)):
|
||||
a: int = cst5.attrs_field(cs.BitsInteger(2))
|
||||
b: int = cst5.attrs_field(cs.BitsInteger(2))
|
||||
c: int = cst5.attrs_field(cs.BitsInteger(2))
|
||||
d: int = cst5.attrs_field(cs.BitsInteger(2))
|
||||
|
||||
common(
|
||||
cst.construct(Test),
|
||||
b"\x1b",
|
||||
Test(a=0, b=1, c=2, d=3),
|
||||
1,
|
||||
)
|
||||
|
||||
|
||||
def test_attrs_struct_overloaded_attributes() -> None:
|
||||
class Test(cst5.AttrsStruct):
|
||||
a: int = cst5.attrs_field(cs.Byte)
|
||||
b: int = cst5.attrs_field(cs.Byte)
|
||||
subcon: int = cst5.attrs_field(
|
||||
cs.Byte
|
||||
) # this is also an attribute from Construct
|
||||
docs: int = cst5.attrs_field(
|
||||
cs.Byte
|
||||
) # this is also an attribute from Construct
|
||||
|
||||
common(
|
||||
cst.construct(Test),
|
||||
b"\x00\x01\x02\x03",
|
||||
Test(a=0, b=1, subcon=2, docs=3),
|
||||
4,
|
||||
)
|
||||
|
||||
|
||||
def test_attrs_struct_reverse_fields() -> None:
|
||||
class Test(cst5.AttrsStruct, reverse_fields=True):
|
||||
a: int = cst5.attrs_field(cs.Byte)
|
||||
b: int = cst5.attrs_field(cs.Byte)
|
||||
c: int = cst5.attrs_field(cs.Byte)
|
||||
d: int = cst5.attrs_field(cs.Byte)
|
||||
|
||||
common(
|
||||
cst.construct(Test),
|
||||
b"\x03\x02\x01\x00",
|
||||
Test(a=0, b=1, c=2, d=3),
|
||||
4,
|
||||
)
|
||||
|
||||
|
||||
def test_attrs_struct_unsupported_param() -> None:
|
||||
with pytest.raises(TypeError):
|
||||
|
||||
class Test(cst5.AttrsStruct, strange_parameter=True): # type: ignore
|
||||
a: int = cst5.attrs_field(cs.Byte)
|
||||
|
||||
|
||||
@pytest.mark.skip
|
||||
def test_attrs_default() -> None:
|
||||
# TODO: Implement `default` parameter for `attrs_field`
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
def test_dataclass_struct_reverse() -> None:
|
||||
@dataclasses.dataclass
|
||||
class TestContainer(DataclassMixin):
|
||||
class TestContainerReverse(DataclassStruct, reverse_fields=True):
|
||||
a: int = csfield(cs.Int16ub)
|
||||
b: int = csfield(cs.Int8ub)
|
||||
|
||||
common(
|
||||
DataclassStruct(TestContainer, reverse=True),
|
||||
cst.construct(TestContainerReverse),
|
||||
b"\x02\x00\x01",
|
||||
TestContainer(a=1, b=2),
|
||||
TestContainerReverse(a=1, b=2),
|
||||
3,
|
||||
)
|
||||
normal = DataclassStruct(TestContainer)
|
||||
reverse = DataclassStruct(TestContainer, reverse=True)
|
||||
assert str(normal.parse(b"\x00\x01\x02")) == str(reverse.parse(b"\x02\x00\x01"))
|
||||
|
||||
|
||||
def test_dataclass_struct_nested() -> None:
|
||||
@dataclasses.dataclass
|
||||
class TestContainer(DataclassMixin):
|
||||
@dataclasses.dataclass
|
||||
class InnerDataclass(DataclassMixin):
|
||||
class TestContainer(DataclassStruct):
|
||||
class InnerDataclass(DataclassStruct):
|
||||
b: int = csfield(cs.Byte)
|
||||
c: bytes = csfield(cs.Bytes(cs.this._.length))
|
||||
|
||||
length: int = csfield(cs.Byte)
|
||||
a: InnerDataclass = csfield(DataclassStruct(InnerDataclass))
|
||||
a: InnerDataclass = csfield(cst.construct(InnerDataclass))
|
||||
|
||||
common(
|
||||
DataclassStruct(TestContainer),
|
||||
cst.construct(TestContainer),
|
||||
b"\x02\x01\xF1\xF2",
|
||||
TestContainer(length=2, a=TestContainer.InnerDataclass(b=1, c=b"\xF1\xF2")),
|
||||
)
|
||||
|
||||
|
||||
def test_dataclass_struct_default_field() -> None:
|
||||
@dataclasses.dataclass
|
||||
class Image(DataclassMixin):
|
||||
class Image(DataclassStruct):
|
||||
width: int = csfield(cs.Int8ub)
|
||||
height: int = csfield(cs.Int8ub)
|
||||
pixels: t.Optional[bytes] = csfield(
|
||||
|
|
@ -157,7 +270,7 @@ def test_dataclass_struct_default_field() -> None:
|
|||
)
|
||||
|
||||
common(
|
||||
DataclassStruct(Image),
|
||||
cst.construct(Image),
|
||||
b"\x02\x03\x00\x00\x00\x00\x00\x00",
|
||||
setattrs(Image(2, 3), pixels=bytes(6)),
|
||||
sample_building=Image(2, 3),
|
||||
|
|
@ -165,12 +278,11 @@ def test_dataclass_struct_default_field() -> None:
|
|||
|
||||
|
||||
def test_dataclass_struct_const_field() -> None:
|
||||
@dataclasses.dataclass
|
||||
class TestContainer(DataclassMixin):
|
||||
class TestContainer(DataclassStruct):
|
||||
const_field: t.Optional[bytes] = csfield(cs.Const(b"\x00"))
|
||||
|
||||
common(
|
||||
DataclassStruct(TestContainer),
|
||||
cst.construct(TestContainer),
|
||||
bytes(1),
|
||||
setattrs(TestContainer(), const_field=b"\x00"),
|
||||
1,
|
||||
|
|
@ -178,7 +290,7 @@ def test_dataclass_struct_const_field() -> None:
|
|||
|
||||
assert (
|
||||
raises(
|
||||
DataclassStruct(TestContainer).build,
|
||||
cst.construct(TestContainer).build,
|
||||
setattrs(TestContainer(), const_field=b"\x01"),
|
||||
)
|
||||
== cs.ConstError
|
||||
|
|
@ -186,12 +298,11 @@ def test_dataclass_struct_const_field() -> None:
|
|||
|
||||
|
||||
def test_dataclass_struct_array_field() -> None:
|
||||
@dataclasses.dataclass
|
||||
class TestContainer(DataclassMixin):
|
||||
class TestContainer(DataclassStruct):
|
||||
array_field: t.List[int] = csfield(cs.Array(5, cs.Int8ub))
|
||||
|
||||
common(
|
||||
DataclassStruct(TestContainer),
|
||||
cst.construct(TestContainer),
|
||||
bytes(5),
|
||||
TestContainer(array_field=[0, 0, 0, 0, 0]),
|
||||
5,
|
||||
|
|
@ -199,15 +310,14 @@ def test_dataclass_struct_array_field() -> None:
|
|||
|
||||
|
||||
def test_dataclass_struct_anonymus_fields_1() -> None:
|
||||
@dataclasses.dataclass
|
||||
class TestContainer(DataclassMixin):
|
||||
class TestContainer(DataclassStruct):
|
||||
_1: t.Optional[bytes] = csfield(cs.Const(b"\x00"))
|
||||
_2: None = csfield(cs.Padding(1))
|
||||
_3: None = csfield(cs.Pass)
|
||||
_4: None = csfield(cs.Terminated)
|
||||
|
||||
common(
|
||||
DataclassStruct(TestContainer),
|
||||
cst.construct(TestContainer),
|
||||
bytes(2),
|
||||
setattrs(TestContainer(), _1=b"\x00"),
|
||||
cs.SizeofError,
|
||||
|
|
@ -215,22 +325,20 @@ def test_dataclass_struct_anonymus_fields_1() -> None:
|
|||
|
||||
|
||||
def test_dataclass_struct_anonymus_fields_2() -> None:
|
||||
@dataclasses.dataclass
|
||||
class TestContainer(DataclassMixin):
|
||||
class TestContainer(DataclassStruct):
|
||||
_1: int = csfield(cs.Computed(7))
|
||||
_2: t.Optional[bytes] = csfield(cs.Const(b"JPEG"))
|
||||
_3: None = csfield(cs.Pass)
|
||||
_4: None = csfield(cs.Terminated)
|
||||
|
||||
d = DataclassStruct(TestContainer)
|
||||
d = cst.construct(TestContainer)
|
||||
assert d.build(TestContainer()) == d.build(TestContainer())
|
||||
|
||||
|
||||
def test_dataclass_struct_overloaded_method() -> None:
|
||||
# Test dot access to some names that are not accessable via dot
|
||||
# in the original 'cs.Container'.
|
||||
@dataclasses.dataclass
|
||||
class TestContainer(DataclassMixin):
|
||||
class TestContainer(DataclassStruct):
|
||||
clear: int = csfield(cs.Int8ul)
|
||||
copy: int = csfield(cs.Int8ul)
|
||||
fromkeys: int = csfield(cs.Int8ul)
|
||||
|
|
@ -246,7 +354,7 @@ def test_dataclass_struct_overloaded_method() -> None:
|
|||
update: int = csfield(cs.Int8ul)
|
||||
values: int = csfield(cs.Int8ul)
|
||||
|
||||
d = DataclassStruct(TestContainer)
|
||||
d = construct(TestContainer)
|
||||
obj = d.parse(
|
||||
d.build(
|
||||
TestContainer(
|
||||
|
|
@ -283,44 +391,22 @@ def test_dataclass_struct_overloaded_method() -> None:
|
|||
assert obj.values == 14
|
||||
|
||||
|
||||
def test_dataclass_struct_no_dataclass() -> None:
|
||||
class TestContainer(DataclassMixin):
|
||||
a: int = csfield(cs.Int16ub)
|
||||
b: int = csfield(cs.Int8ub)
|
||||
|
||||
assert raises(lambda: DataclassStruct(TestContainer)) == TypeError
|
||||
|
||||
|
||||
def test_dataclass_struct_no_DataclassMixin() -> None:
|
||||
@dataclasses.dataclass
|
||||
class TestContainer:
|
||||
a: int = csfield(cs.Int16ub)
|
||||
b: int = csfield(cs.Int8ub)
|
||||
|
||||
cls = t.cast(t.Type[DataclassMixin], TestContainer)
|
||||
assert raises(lambda: DataclassStruct(cls)) == TypeError
|
||||
|
||||
|
||||
def test_dataclass_struct_wrong_container() -> None:
|
||||
@dataclasses.dataclass
|
||||
class TestContainer1(DataclassMixin):
|
||||
class TestContainer1(DataclassStruct):
|
||||
a: int = csfield(cs.Int16ub)
|
||||
b: int = csfield(cs.Int8ub)
|
||||
|
||||
@dataclasses.dataclass
|
||||
class TestContainer2(DataclassMixin):
|
||||
class TestContainer2(DataclassStruct):
|
||||
a: int = csfield(cs.Int16ub)
|
||||
b: int = csfield(cs.Int8ub)
|
||||
|
||||
assert (
|
||||
raises(DataclassStruct(TestContainer1).build, TestContainer2(a=1, b=2))
|
||||
== TypeError
|
||||
raises(construct(TestContainer1).build, TestContainer2(a=1, b=2)) == TypeError
|
||||
)
|
||||
|
||||
|
||||
def test_dataclass_struct_doc() -> None:
|
||||
@dataclasses.dataclass
|
||||
class TestContainer(DataclassMixin):
|
||||
class TestContainer(DataclassStruct):
|
||||
a: int = csfield(cs.Int16ub, "This is the documentation of a")
|
||||
b: int = csfield(
|
||||
cs.Int8ub, doc="This is the documentation of b\nwhich is multiline"
|
||||
|
|
@ -333,7 +419,7 @@ def test_dataclass_struct_doc() -> None:
|
|||
""",
|
||||
)
|
||||
|
||||
format = DataclassStruct(TestContainer)
|
||||
format = TestContainer.__construct__() # TODO: construct(TestContainer)
|
||||
common(format, b"\x00\x01\x02\x03", TestContainer(a=1, b=2, c=3), 4)
|
||||
|
||||
assert format.subcon.a.docs == "This is the documentation of a"
|
||||
|
|
@ -345,39 +431,36 @@ def test_dataclass_struct_doc() -> None:
|
|||
|
||||
|
||||
def test_dataclass_bitstruct() -> None:
|
||||
@dataclasses.dataclass
|
||||
class TestContainer(DataclassMixin):
|
||||
class TestContainer(DataclassBitStruct):
|
||||
a: int = csfield(cs.BitsInteger(7))
|
||||
b: int = csfield(cs.Bit)
|
||||
c: int = csfield(cs.BitsInteger(8))
|
||||
|
||||
print("")
|
||||
|
||||
common(
|
||||
DataclassBitStruct(TestContainer),
|
||||
construct(TestContainer),
|
||||
b"\xFD\x12",
|
||||
TestContainer(a=0x7E, b=1, c=0x12),
|
||||
2,
|
||||
)
|
||||
|
||||
# check __getattr__
|
||||
c = DataclassStruct(TestContainer)
|
||||
assert c.a.name == "a"
|
||||
assert c.b.name == "b"
|
||||
assert c.c.name == "c"
|
||||
assert isinstance(c.a.subcon, cs.BitsInteger)
|
||||
assert c.b.subcon is cs.Bit
|
||||
assert isinstance(c.c.subcon, cs.BitsInteger)
|
||||
c = TestContainer.__construct__()
|
||||
assert c.subcon.a.name == "a"
|
||||
assert c.subcon.b.name == "b"
|
||||
assert c.subcon.c.name == "c"
|
||||
assert isinstance(c.subcon.a.subcon, cs.BitsInteger)
|
||||
assert c.subcon.b.subcon is cs.Bit
|
||||
assert isinstance(c.subcon.c.subcon, cs.BitsInteger)
|
||||
|
||||
|
||||
def test_tenum() -> None:
|
||||
class TestEnum(cst.EnumBase):
|
||||
class TestEnum(cst.EnumBase, subcon=cs.Byte):
|
||||
one = 1
|
||||
two = 2
|
||||
four = 4
|
||||
eight = 8
|
||||
|
||||
d = cst.TEnum(cs.Byte, TestEnum)
|
||||
d = cst.construct(TestEnum)
|
||||
|
||||
common(d, b"\x01", TestEnum.one, 1)
|
||||
common(d, b"\xff", TestEnum(255), 1)
|
||||
|
|
@ -396,114 +479,46 @@ def test_tenum_no_enumbase() -> None:
|
|||
b = 2
|
||||
|
||||
cls = t.cast(t.Type[cst.EnumBase], E)
|
||||
assert raises(lambda: cst.TEnum(cs.Byte, cls)) == TypeError
|
||||
assert raises(lambda: cst.EnumConstruct(cs.Byte, cls)) == TypeError
|
||||
|
||||
|
||||
def test_tenum_asdict() -> None:
|
||||
# see: https://github.com/timrid/construct-typing/issues/21
|
||||
import dataclasses
|
||||
def test_tenum_no_subcon() -> None:
|
||||
with pytest.raises(TypeError):
|
||||
|
||||
import construct_typed as cst
|
||||
|
||||
class TestEnum(cst.EnumBase):
|
||||
one = 1
|
||||
two = 2
|
||||
four = 4
|
||||
eight = 8
|
||||
|
||||
@dataclasses.dataclass
|
||||
class SomeDataclass:
|
||||
a: TestEnum
|
||||
|
||||
dc = SomeDataclass(TestEnum.one)
|
||||
dc_dict = dataclasses.asdict(dc)
|
||||
assert dc_dict["a"] == dc.a
|
||||
assert dc_dict["a"] is dc.a
|
||||
|
||||
dc = SomeDataclass(TestEnum(5))
|
||||
dc_dict = dataclasses.asdict(dc)
|
||||
assert dc_dict["a"] == dc.a
|
||||
assert dc_dict["a"] is dc.a
|
||||
|
||||
|
||||
def test_tenum_docstring() -> None:
|
||||
class TestEnum(cst.EnumBase):
|
||||
"""
|
||||
This is an test enum.
|
||||
"""
|
||||
|
||||
Value_WithDoc = cst.EnumValue(0, doc="an enum with a documentation")
|
||||
Value_WithMultilineDoc = cst.EnumValue(
|
||||
1,
|
||||
"""
|
||||
An enum with a multiline documentation...
|
||||
...next line...
|
||||
""",
|
||||
)
|
||||
Value_NoDoc = cst.EnumValue(2)
|
||||
Value_NoDoc2 = 3
|
||||
|
||||
assert TestEnum.__doc__ is not None
|
||||
assert textwrap.dedent(TestEnum.__doc__) == textwrap.dedent(
|
||||
"""
|
||||
This is an test enum.
|
||||
"""
|
||||
)
|
||||
assert TestEnum.Value_WithDoc.__doc__ == "an enum with a documentation"
|
||||
assert (
|
||||
TestEnum.Value_WithMultilineDoc.__doc__
|
||||
== """
|
||||
An enum with a multiline documentation...
|
||||
...next line...
|
||||
"""
|
||||
)
|
||||
assert TestEnum.Value_NoDoc.__doc__ == ""
|
||||
assert TestEnum.Value_NoDoc2.__doc__ == ""
|
||||
assert TestEnum(5).__doc__ == "missing value"
|
||||
|
||||
|
||||
def test_dataclass_struct_wrong_enumbase() -> None:
|
||||
class E1(cst.EnumBase):
|
||||
a = 1
|
||||
b = 2
|
||||
|
||||
class E2(cst.EnumBase):
|
||||
a = 1
|
||||
b = 2
|
||||
|
||||
assert raises(cst.TEnum(cs.Byte, E1).build, E2.a) == TypeError
|
||||
class E1(cst.EnumBase): # type: ignore
|
||||
a = 1
|
||||
b = 2
|
||||
|
||||
|
||||
def test_tenum_in_tstruct() -> None:
|
||||
class TestEnum(cst.EnumBase):
|
||||
class TestEnum(cst.EnumBase, subcon=cs.Int8ub):
|
||||
a = 1
|
||||
b = 2
|
||||
|
||||
@dataclasses.dataclass
|
||||
class TestContainer(DataclassMixin):
|
||||
a: TestEnum = csfield(cst.TEnum(cs.Int8ub, TestEnum))
|
||||
class TestContainer(DataclassStruct):
|
||||
a: TestEnum = csfield(cst.construct(TestEnum))
|
||||
b: int = csfield(cs.Int8ub)
|
||||
|
||||
common(
|
||||
DataclassStruct(TestContainer),
|
||||
construct(TestContainer),
|
||||
b"\x01\x02",
|
||||
TestContainer(a=TestEnum.a, b=2),
|
||||
2,
|
||||
)
|
||||
|
||||
assert (
|
||||
raises(cst.TEnum(cs.Byte, TestEnum).build, TestContainer(a=1, b=2)) == TypeError # type: ignore
|
||||
raises(cst.construct(TestEnum).build, TestContainer(a=1, b=2)) == TypeError # type: ignore
|
||||
)
|
||||
|
||||
|
||||
def test_tenum_flags() -> None:
|
||||
class TestEnum(cst.FlagsEnumBase):
|
||||
class TestEnum(cst.FlagsEnumBase, subcon=cs.Byte):
|
||||
one = 1
|
||||
two = 2
|
||||
four = 4
|
||||
eight = 8
|
||||
|
||||
d = cst.TFlagsEnum(cs.Byte, TestEnum)
|
||||
d = cst.construct(TestEnum)
|
||||
common(d, b"\x03", TestEnum.one | TestEnum.two, 1)
|
||||
assert d.build(TestEnum(0)) == b"\x00"
|
||||
assert d.build(TestEnum.one | TestEnum.two) == b"\x03"
|
||||
|
|
@ -512,65 +527,3 @@ def test_tenum_flags() -> None:
|
|||
assert d.build(TestEnum(255)) == b"\xff"
|
||||
assert d.build(TestEnum.eight) == b"\x08"
|
||||
assert raises(d.build, 2) == TypeError
|
||||
|
||||
|
||||
def test_tenum_flags_asdict() -> None:
|
||||
import dataclasses
|
||||
|
||||
import construct_typed as cst
|
||||
|
||||
class TestEnum(cst.FlagsEnumBase):
|
||||
one = 1
|
||||
two = 2
|
||||
four = 4
|
||||
eight = 8
|
||||
|
||||
@dataclasses.dataclass
|
||||
class SomeDataclass:
|
||||
a: TestEnum
|
||||
|
||||
dc = SomeDataclass(TestEnum.one)
|
||||
dc_dict = dataclasses.asdict(dc)
|
||||
assert dc_dict["a"] == dc.a
|
||||
assert dc_dict["a"] is dc.a
|
||||
|
||||
dc = SomeDataclass(TestEnum(5))
|
||||
dc_dict = dataclasses.asdict(dc)
|
||||
assert dc_dict["a"] == dc.a
|
||||
assert dc_dict["a"] is dc.a
|
||||
|
||||
|
||||
def test_tenum_flags_docstring() -> None:
|
||||
class TestEnum(cst.FlagsEnumBase):
|
||||
"""
|
||||
This is an test flags enum.
|
||||
"""
|
||||
|
||||
Value_WithDoc = cst.EnumValue(0, doc="an enum with a documentation")
|
||||
Value_WithMultilineDoc = cst.EnumValue(
|
||||
1,
|
||||
"""
|
||||
An enum with a multiline documentation...
|
||||
...next line...
|
||||
""",
|
||||
)
|
||||
Value_NoDoc = cst.EnumValue(2)
|
||||
Value_NoDoc2 = 4
|
||||
|
||||
assert TestEnum.__doc__ is not None
|
||||
assert textwrap.dedent(TestEnum.__doc__) == textwrap.dedent(
|
||||
"""
|
||||
This is an test flags enum.
|
||||
"""
|
||||
)
|
||||
assert TestEnum.Value_WithDoc.__doc__ == "an enum with a documentation"
|
||||
assert (
|
||||
TestEnum.Value_WithMultilineDoc.__doc__
|
||||
== """
|
||||
An enum with a multiline documentation...
|
||||
...next line...
|
||||
"""
|
||||
)
|
||||
assert TestEnum.Value_NoDoc.__doc__ == ""
|
||||
assert TestEnum.Value_NoDoc2.__doc__ == ""
|
||||
assert TestEnum(8).__doc__ == "missing value"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue