2023-07-18 15:38:40 +02:00
|
|
|
import binascii
|
|
|
|
|
import io
|
|
|
|
|
import typing as t
|
|
|
|
|
|
2020-12-10 20:26:37 +01:00
|
|
|
import pytest
|
2023-07-18 15:38:40 +02:00
|
|
|
from construct import *
|
|
|
|
|
from construct.lib import *
|
|
|
|
|
|
|
|
|
|
import construct_typed as cst
|
2021-01-01 17:26:40 +01:00
|
|
|
|
2020-12-10 20:26:37 +01:00
|
|
|
xfail = pytest.mark.xfail
|
|
|
|
|
skip = pytest.mark.skip
|
|
|
|
|
skipif = pytest.mark.skipif
|
|
|
|
|
|
2023-07-18 15:38:40 +02:00
|
|
|
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")
|
2020-12-10 20:26:37 +01:00
|
|
|
|
2023-07-18 15:38:40 +02:00
|
|
|
IdentType = t.TypeVar("IdentType")
|
2020-12-10 20:26:37 +01:00
|
|
|
|
2021-01-01 17:26:40 +01:00
|
|
|
|
2020-12-10 20:26:37 +01:00
|
|
|
class ZeroIO(io.BufferedIOBase):
|
2023-07-18 17:37:29 +02:00
|
|
|
def read(self, __size: t.Optional[int] = None) -> bytes:
|
2020-12-10 20:26:37 +01:00
|
|
|
if __size is not None:
|
|
|
|
|
return bytes(__size)
|
|
|
|
|
else:
|
|
|
|
|
return bytes(0)
|
|
|
|
|
|
2023-07-18 17:37:29 +02:00
|
|
|
def read1(self, __size: int = 0) -> bytes:
|
2020-12-10 20:26:37 +01:00
|
|
|
return bytes(__size)
|
|
|
|
|
|
2021-01-01 17:26:40 +01:00
|
|
|
|
2023-07-18 15:38:40 +02:00
|
|
|
def ident(x: IdentType) -> IdentType:
|
|
|
|
|
return x
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
devzero: t.BinaryIO = ZeroIO() # type: ignore
|
2020-12-10 20:26:37 +01:00
|
|
|
|
|
|
|
|
|
2023-07-18 15:38:40 +02:00
|
|
|
def raises(
|
|
|
|
|
func: t.Callable[..., t.Any], *args: t.Any, **kw: t.Any
|
|
|
|
|
) -> t.Union[t.Any, Exception]:
|
2020-12-10 20:26:37 +01:00
|
|
|
try:
|
|
|
|
|
return func(*args, **kw)
|
|
|
|
|
except Exception as e:
|
|
|
|
|
return e.__class__
|
|
|
|
|
|
|
|
|
|
|
2023-07-18 15:38:40 +02:00
|
|
|
@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:
|
2020-12-10 20:26:37 +01:00
|
|
|
obj = format.parse(datasample, **kw)
|
|
|
|
|
assert obj == objsample
|
|
|
|
|
data = format.build(objsample, **kw)
|
|
|
|
|
assert data == datasample
|
|
|
|
|
# following are implied by above (re-parse and re-build)
|
|
|
|
|
# assert format.parse(format.build(obj)) == obj
|
|
|
|
|
# assert format.build(format.parse(data)) == data
|
|
|
|
|
if isinstance(sizesample, int):
|
|
|
|
|
size = format.sizeof(**kw)
|
|
|
|
|
assert size == sizesample
|
|
|
|
|
else:
|
2023-07-18 17:37:29 +02:00
|
|
|
size_ex = raises(format.sizeof, **kw)
|
|
|
|
|
assert size_ex == sizesample
|
2020-12-10 20:26:37 +01:00
|
|
|
|
|
|
|
|
|
2023-07-18 15:38:40 +02:00
|
|
|
def setattrs(obj: T, **kwargs: t.Any) -> T:
|
|
|
|
|
"""Set multiple named values of an object"""
|
2021-01-01 17:26:40 +01:00
|
|
|
for name, value in kwargs.items():
|
|
|
|
|
setattr(obj, name, value)
|
|
|
|
|
return obj
|
|
|
|
|
|
|
|
|
|
|
2023-07-18 17:37:29 +02:00
|
|
|
def commonhex(format: "Construct[t.Any, t.Any]", hexdata: str) -> None:
|
2020-12-10 20:26:37 +01:00
|
|
|
commonbytes(format, binascii.unhexlify(hexdata))
|
|
|
|
|
|
|
|
|
|
|
2023-07-18 17:37:29 +02:00
|
|
|
def commondumpdeprecated(format: "Construct[t.Any, t.Any]", filename: str) -> None:
|
2020-12-10 20:26:37 +01:00
|
|
|
filename = "tests/deprecated_gallery/blobs/" + filename
|
2021-01-01 17:26:40 +01:00
|
|
|
with open(filename, "rb") as f:
|
2020-12-10 20:26:37 +01:00
|
|
|
data = f.read()
|
|
|
|
|
commonbytes(format, data)
|
|
|
|
|
|
|
|
|
|
|
2023-07-18 17:37:29 +02:00
|
|
|
def commondump(format: "Construct[t.Any, t.Any]", filename: str) -> None:
|
2020-12-10 20:26:37 +01:00
|
|
|
filename = "tests/gallery/blobs/" + filename
|
2021-01-01 17:26:40 +01:00
|
|
|
with open(filename, "rb") as f:
|
2020-12-10 20:26:37 +01:00
|
|
|
data = f.read()
|
|
|
|
|
commonbytes(format, data)
|
|
|
|
|
|
|
|
|
|
|
2023-07-18 17:37:29 +02:00
|
|
|
def commonbytes(format: "Construct[t.Any, t.Any]", data: bytes) -> None:
|
2020-12-10 20:26:37 +01:00
|
|
|
obj = format.parse(data)
|
2023-07-18 15:38:40 +02:00
|
|
|
format.build(obj)
|