diff --git a/chia/util/streamable.py b/chia/util/streamable.py index 5f8d2799da..a073ddb916 100644 --- a/chia/util/streamable.py +++ b/chia/util/streamable.py @@ -5,7 +5,21 @@ import io import os import pprint from enum import Enum -from typing import Any, BinaryIO, Callable, Dict, Iterator, List, Optional, Tuple, Type, TypeVar, Union, get_type_hints +from typing import ( + Any, + BinaryIO, + Callable, + Collection, + Dict, + Iterator, + List, + Optional, + Tuple, + Type, + TypeVar, + Union, + get_type_hints, +) from blspy import G1Element, G2Element, PrivateKey from typing_extensions import Literal, get_args, get_origin @@ -52,6 +66,7 @@ ConvertFunctionType = Callable[[object], object] class Field: name: str type: Type[object] + has_default: bool # Caches to store the fields and (de)serialization methods for all available streamable classes. @@ -63,7 +78,14 @@ CONVERT_FUNCTIONS_FOR_STREAMABLE_CLASS: Dict[Type[object], List[ConvertFunctionT def create_fields_cache(cls: Type[object]) -> Tuple[Field, ...]: hints = get_type_hints(cls) - fields = tuple(Field(field.name, hints.get(field.name, None)) for field in dataclasses.fields(cls)) + fields = tuple( + Field( + name=field.name, + type=hints.get(field.name, None), + has_default=field.default is not dataclasses.MISSING or field.default_factory is not dataclasses.MISSING, + ) + for field in dataclasses.fields(cls) + ) assert all(field.type is not None for field in fields) return fields @@ -89,40 +111,61 @@ def convert_optional(convert_func: ConvertFunctionType, item: Any) -> Any: return convert_func(item) -def convert_tuple(convert_funcs: List[ConvertFunctionType], items: Tuple[Any, ...]) -> Tuple[Any, ...]: - tuple_data = [] - for i in range(len(items)): - tuple_data.append(convert_funcs[i](items[i])) - return tuple(tuple_data) +def convert_tuple(convert_funcs: List[ConvertFunctionType], items: Collection[Any]) -> Tuple[Any, ...]: + if len(items) != len(convert_funcs): + raise ValueError(f"Invalid size. Expected: {len(convert_funcs)}, got: {len(items)}") + if not isinstance(items, (list, tuple)): + raise TypeError(f"expected: tuple or list, actual: {type(items).__name__}") + return tuple(convert_func(item) for convert_func, item in zip(convert_funcs, items)) def convert_list(convert_func: ConvertFunctionType, items: List[Any]) -> List[Any]: - list_data = [] - for item in items: - list_data.append(convert_func(item)) - return list_data + if not isinstance(items, list): + raise TypeError(f"expected: list, actual: {type(items).__name__}") + return [convert_func(item) for item in items] + + +def convert_hex_string(item: str) -> bytes: + if not isinstance(item, str): + raise TypeError(f"expected: hex-string, actual: {type(item).__name__}") + try: + return hexstr_to_bytes(item) + except Exception as e: + raise TypeError(f"Can't convert the string {item!r} to bytes: {e}") from e def convert_byte_type(f_type: Type[Any], item: Any) -> Any: - if type(item) == f_type: + if isinstance(item, f_type): return item - return f_type(hexstr_to_bytes(item)) + if not isinstance(item, bytes): + item = convert_hex_string(item) + try: + return f_type(item) + except Exception as e: + raise TypeError(f"Can't convert {type(item).__name__} to {f_type.__name__}: {e}") from e def convert_unhashable_type(f_type: Type[Any], item: Any) -> Any: - if type(item) == f_type: + if isinstance(item, f_type): return item - if hasattr(f_type, "from_bytes_unchecked"): - from_bytes_method = f_type.from_bytes_unchecked - else: - from_bytes_method = f_type.from_bytes - return from_bytes_method(hexstr_to_bytes(item)) + if not isinstance(item, bytes): + item = convert_hex_string(item) + try: + if hasattr(f_type, "from_bytes_unchecked"): + return f_type.from_bytes_unchecked(item) + else: + return f_type.from_bytes(item) + except Exception as e: + raise TypeError(f"Can't convert {type(item).__name__} to {f_type.__name__}: {e}") from e def convert_primitive(f_type: Type[Any], item: Any) -> Any: - if type(item) == f_type: + if isinstance(item, f_type): return item - return f_type(item) + try: + return f_type(item) + except Exception as e: + raise TypeError(f"Can't convert type {type(item).__name__} to {f_type.__name__}: {e}") from e def dataclass_from_dict(klass: Type[Any], item: Any) -> Any: @@ -130,8 +173,10 @@ def dataclass_from_dict(klass: Type[Any], item: Any) -> Any: Converts a dictionary based on a dataclass, into an instance of that dataclass. Recursively goes through lists, optionals, and dictionaries. """ - if type(item) == klass: + if isinstance(item, klass): return item + if not isinstance(item, dict): + raise TypeError(f"expected: dict, actual: {type(item).__name__}") if klass not in CONVERT_FUNCTIONS_FOR_STREAMABLE_CLASS: # For non-streamable dataclasses we can't populate the cache on startup, so we do it here for convert @@ -144,13 +189,22 @@ def dataclass_from_dict(klass: Type[Any], item: Any) -> Any: fields = FIELDS_FOR_STREAMABLE_CLASS[klass] convert_funcs = CONVERT_FUNCTIONS_FOR_STREAMABLE_CLASS[klass] - return klass( - **{ - field.name: convert_func(item[field.name]) - for field, convert_func in zip(fields, convert_funcs) - if field.name in item - } - ) + try: + return klass( + **{ + field.name: convert_func(item[field.name]) + for field, convert_func in zip(fields, convert_funcs) + if field.name in item + } + ) + except TypeError as e: + missing_fields = [field.name for field in fields if field.name not in item and not field.has_default] + if len(missing_fields) > 0: + raise KeyError( + f"{len(missing_fields)} field{'s' if len(missing_fields) > 1 else ''} missing for {klass.__name__}: " + + ", ".join(missing_fields) + ) from e + raise def function_to_convert_one_item(f_type: Type[Any]) -> ConvertFunctionType: diff --git a/tests/core/util/test_streamable.py b/tests/core/util/test_streamable.py index dc9242ff15..4fef08b2cd 100644 --- a/tests/core/util/test_streamable.py +++ b/tests/core/util/test_streamable.py @@ -12,7 +12,7 @@ from typing_extensions import Literal, get_args from chia.protocols.wallet_protocol import RespondRemovals from chia.types.blockchain_format.coin import Coin from chia.types.blockchain_format.program import Program -from chia.types.blockchain_format.sized_bytes import bytes32 +from chia.types.blockchain_format.sized_bytes import bytes4, bytes32 from chia.types.full_block import FullBlock from chia.types.weight_proof import SubEpochChallengeSegment from chia.util.ints import uint8, uint32, uint64 @@ -125,19 +125,149 @@ def test_pure_dataclasses_in_dataclass_from_dict() -> None: assert d2.c == 1.2345 +@dataclass +class ConvertTupleFailures: + a: Tuple[int, int] + b: Tuple[int, Tuple[int, int]] + + +@pytest.mark.parametrize( + "input_dict, error", + [ + pytest.param({"a": (1,), "b": (1, (2, 2))}, ValueError, id="a: item missing"), + pytest.param({"a": (1, 1, 1), "b": (1, (2, 2))}, ValueError, id="a: item too much"), + pytest.param({"a": (1, 1), "b": (1, (2,))}, ValueError, id="b: item missing"), + pytest.param({"a": (1, 1), "b": (1, (2, 2, 2))}, ValueError, id="b: item too much"), + pytest.param({"a": "11", "b": (1, (2, 2))}, TypeError, id="a: invalid type list"), + pytest.param({"a": 1, "b": (1, (2, 2))}, TypeError, id="a: invalid type int"), + pytest.param({"a": "11", "b": (1, (2, 2))}, TypeError, id="a: invalid type str"), + pytest.param({"a": (1, 1), "b": (1, "22")}, TypeError, id="b: invalid type list"), + pytest.param({"a": (1, 1), "b": (1, 2)}, TypeError, id="b: invalid type int"), + pytest.param({"a": (1, 1), "b": (1, "22")}, TypeError, id="b: invalid type str"), + ], +) +def test_convert_tuple_failures(input_dict: Dict[str, Any], error: Any) -> None: + + with pytest.raises(error): + dataclass_from_dict(ConvertTupleFailures, input_dict) + + +@dataclass +class ConvertListFailures: + a: List[int] + b: List[List[int]] + + +@pytest.mark.parametrize( + "input_dict, error", + [ + pytest.param({"a": [1, 1], "b": [1, [2, 2]]}, TypeError, id="a: invalid type list"), + pytest.param({"a": 1, "b": [1, [2, 2]]}, TypeError, id="a: invalid type int"), + pytest.param({"a": "11", "b": [1, [2, 2]]}, TypeError, id="a: invalid type str"), + pytest.param({"a": [1, 1], "b": [1, [2, 2]]}, TypeError, id="b: invalid type list"), + pytest.param({"a": [1, 1], "b": [1, 2]}, TypeError, id="b: invalid type int"), + pytest.param({"a": [1, 1], "b": [1, "22"]}, TypeError, id="b: invalid type str"), + ], +) +def test_convert_list_failures(input_dict: Dict[str, Any], error: Any) -> None: + + with pytest.raises(error): + dataclass_from_dict(ConvertListFailures, input_dict) + + +@dataclass +class ConvertByteTypeFailures: + a: bytes4 + b: bytes + + +@pytest.mark.parametrize( + "input_dict, error", + [ + pytest.param({"a": 0, "b": bytes(0)}, TypeError, id="a: no string and no bytes"), + pytest.param({"a": [], "b": bytes(0)}, TypeError, id="a: no string and no bytes"), + pytest.param({"a": {}, "b": bytes(0)}, TypeError, id="a: no string and no bytes"), + pytest.param({"a": "invalid", "b": bytes(0)}, TypeError, id="a: invalid hex string"), + pytest.param({"a": "000000", "b": bytes(0)}, TypeError, id="a: hex string too short"), + pytest.param({"a": "0000000000", "b": bytes(0)}, TypeError, id="a: hex string too long"), + pytest.param({"a": b"\00\00\00", "b": bytes(0)}, TypeError, id="a: bytes too short"), + pytest.param({"a": b"\00\00\00\00\00", "b": bytes(0)}, TypeError, id="a: bytes too long"), + pytest.param({"a": "00000000", "b": 0}, TypeError, id="b: no string and no bytes"), + pytest.param({"a": "00000000", "b": []}, TypeError, id="b: no string and no bytes"), + pytest.param({"a": "00000000", "b": {}}, TypeError, id="b: no string and no bytes"), + pytest.param({"a": "00000000", "b": "invalid"}, TypeError, id="b: invalid hex string"), + ], +) +def test_convert_byte_type_failures(input_dict: Dict[str, Any], error: Any) -> None: + + with pytest.raises(error): + dataclass_from_dict(ConvertByteTypeFailures, input_dict) + + +@dataclass +class ConvertUnhashableTypeFailures: + a: G1Element + + +@pytest.mark.parametrize( + "input_dict, error", + [ + pytest.param({"a": 0}, TypeError, id="a: no string and no bytes"), + pytest.param({"a": []}, TypeError, id="a: no string and no bytes"), + pytest.param({"a": {}}, TypeError, id="a: no string and no bytes"), + pytest.param({"a": "invalid"}, TypeError, id="a: invalid hex string"), + pytest.param({"a": "00" * (G1Element.SIZE - 1)}, TypeError, id="a: hex string too short"), + pytest.param({"a": "00" * (G1Element.SIZE + 1)}, TypeError, id="a: hex string too long"), + pytest.param({"a": b"\00" * (G1Element.SIZE - 1)}, TypeError, id="a: bytes too short"), + pytest.param({"a": b"\00" * (G1Element.SIZE + 1)}, TypeError, id="a: bytes too long"), + pytest.param({"a": b"\00" * G1Element.SIZE}, TypeError, id="a: invalid g1 element"), + ], +) +def test_convert_unhashable_type_failures(input_dict: Dict[str, Any], error: Any) -> None: + + with pytest.raises(error): + dataclass_from_dict(ConvertUnhashableTypeFailures, input_dict) + + +class NoStrClass: + def __str__(self) -> str: + raise RuntimeError("No string") + + +@dataclass +class ConvertPrimitiveFailures: + a: int + b: uint8 + c: str + + +@pytest.mark.parametrize( + "input_dict, error", + [ + pytest.param({"a": "a", "b": uint8(1), "c": "2"}, TypeError, id="a: invalid value"), + pytest.param({"a": 0, "b": [], "c": "2"}, TypeError, id="b: invalid value"), + pytest.param({"a": 0, "b": uint8(1), "c": NoStrClass()}, TypeError, id="c: invalid value"), + ], +) +def test_convert_primitive_failures(input_dict: Dict[str, Any], error: Any) -> None: + + with pytest.raises(error): + dataclass_from_dict(ConvertPrimitiveFailures, input_dict) + + @pytest.mark.parametrize( "test_class, input_dict, error", [ - [TestDataclassFromDict1, {"a": "asdf", "b": "2", "c": G1Element()}, ValueError], - [TestDataclassFromDict1, {"a": 1, "b": "2"}, TypeError], - [TestDataclassFromDict1, {"a": 1, "b": "2", "c": "asd"}, ValueError], - [TestDataclassFromDict1, {"a": 1, "b": "2", "c": "00" * G1Element.SIZE}, ValueError], + [TestDataclassFromDict1, {"a": "asdf", "b": "2", "c": G1Element()}, TypeError], + [TestDataclassFromDict1, {"a": 1, "b": "2"}, KeyError], + [TestDataclassFromDict1, {"a": 1, "b": "2", "c": "asd"}, TypeError], + [TestDataclassFromDict1, {"a": 1, "b": "2", "c": "00" * G1Element.SIZE}, TypeError], [TestDataclassFromDict1, {"a": [], "b": "2", "c": G1Element()}, TypeError], [TestDataclassFromDict1, {"a": {}, "b": "2", "c": G1Element()}, TypeError], [TestDataclassFromDict2, {"a": "asdf", "b": 1.2345, "c": 1.2345}, TypeError], [TestDataclassFromDict2, {"a": 1.2345, "b": {"a": 1, "b": "2"}, "c": 1.2345}, TypeError], - [TestDataclassFromDict2, {"a": {"a": 1, "b": "2", "c": G1Element()}, "b": {"a": 1, "b": "2"}}, TypeError], - [TestDataclassFromDict2, {"a": {"a": 1, "b": "2"}, "b": {"a": 1, "b": "2"}, "c": 1.2345}, TypeError], + [TestDataclassFromDict2, {"a": {"a": 1, "b": "2", "c": G1Element()}, "b": {"a": 1, "b": "2"}}, KeyError], + [TestDataclassFromDict2, {"a": {"a": 1, "b": "2"}, "b": {"a": 1, "b": "2"}, "c": 1.2345}, KeyError], ], ) def test_dataclass_from_dict_failures(test_class: Type[Any], input_dict: Dict[str, Any], error: Any) -> None: