Ethereum Test Base Types package¶

class Address(FixedSizeBytes[20]):  # type: ignore
    """Class that helps represent Ethereum addresses in tests."""

    label: str | None = None

    def __new__(
        cls,
        input_bytes: "FixedSizeBytesConvertible | Address",
        *args: Any,
        label: str | None = None,
        **kwargs: Any,
    ) -> Self:
        """Create a new Address object with an optional label."""
        instance = super(Address, cls).__new__(
            cls, input_bytes, *args, **kwargs
        )
        if isinstance(input_bytes, Address) and label is None:
            instance.label = input_bytes.label
        else:
            instance.label = label
        return instance

class Bloom(FixedSizeBytes[256]):  # type: ignore
    """Class that helps represent blooms in tests."""

    pass

class BLSPublicKey(FixedSizeBytes[48]):  # type: ignore
    """Class that helps represent BLS public keys in tests."""

    pass

class BLSSignature(FixedSizeBytes[96]):  # type: ignore
    """Class that helps represent BLS signatures in tests."""

    pass

class Bytes(bytes, ToStringSchema):
    """Class that helps represent bytes of variable length in tests."""

    def __new__(cls, input_bytes: BytesConvertible = b"") -> Self:
        """Create a new Bytes object."""
        if type(input_bytes) is cls:
            return input_bytes
        return super(Bytes, cls).__new__(cls, to_bytes(input_bytes))

    __hash__ = bytes.__hash__

    def __str__(self) -> str:
        """Return the hexadecimal representation of the bytes."""
        return self.hex()

    def hex(self, *args: Any, **kwargs: Any) -> str:
        """Return the hexadecimal representation of the bytes."""
        return "0x" + super().hex(*args, **kwargs)

    def keccak256(self) -> "Hash":
        """Return the keccak256 hash of the opcode byte representation."""
        # Imported lazily so that merely importing the test framework does not
        # import the `ethereum` package: on xdist workers that import would
        # happen before pytest-cov starts the worker's coverage session,
        # making coverage report `ethereum` as "module-not-measured".
        from ethereum.crypto.hash import keccak256 as _keccak256

        return Hash(_keccak256(self))

    def sha256(self) -> "Hash":
        """Return the sha256 hash of the opcode byte representation."""
        return Hash(sha256(self).digest())

    @staticmethod
    def __get_pydantic_core_schema__(
        source_type: Any, handler: GetCoreSchemaHandler
    ) -> PlainValidatorFunctionSchema:
        """
        Call the class constructor without info and appends the serialization
        schema.
        """
        return no_info_plain_validator_function(
            source_type,
            serialization=to_string_ser_schema(),
            json_schema_input_schema=handler(
                Annotated[
                    str, StringConstraints(pattern=r"^0x([0-9a-fA-F]{2})*$")
                ]
            ),
        )

class BytesConcatenation(SupportsBytes, Sized, metaclass=ABCMeta):
    """A class that can be concatenated with bytes."""

    def __len__(self) -> int:
        """Return length of the object when converted to bytes."""
        return len(bytes(self))

    def __add__(self, other: bytes | SupportsBytes) -> bytes:
        """Concatenates the object with another bytes object."""
        return bytes(self) + bytes(other)

    def __radd__(self, other: bytes | SupportsBytes) -> bytes:
        """Concatenates the object with another bytes object."""
        return bytes(other) + bytes(self)

class CoerceBytes(Bytes):
    """
    Class that helps represent bytes of variable length in tests and
    supports removing white spaces anywhere in the string.
    """

    def __new__(cls, input_bytes: BytesConvertible = b"") -> Self:
        """Create a new CoerceBytes object."""
        if type(input_bytes) is cls:
            return input_bytes
        if isinstance(input_bytes, str):
            input_bytes = sub(r"\s+", "", input_bytes)
        return super(Bytes, cls).__new__(cls, to_bytes(input_bytes))

class FixedSizeBytes(Bytes):
    """Class that helps represent bytes of fixed length in tests."""

    byte_length: ClassVar[int]
    _sized_: ClassVar[Type["FixedSizeBytes"]]

    def __class_getitem__(cls, length: int) -> Type["FixedSizeBytes"]:
        """Create a new FixedSizeBytes class with the given length."""

        class Sized(cls):  # type: ignore
            byte_length = length

        Sized._sized_ = Sized
        return Sized

    def __new__(
        cls,
        input_bytes: FixedSizeBytesConvertible | Self,
        *,
        left_padding: bool = False,
        right_padding: bool = False,
    ) -> Self:
        """Create a new FixedSizeBytes object."""
        if type(input_bytes) is cls:
            return input_bytes
        return super(FixedSizeBytes, cls).__new__(
            cls,
            to_fixed_size_bytes(
                input_bytes,
                cls.byte_length,
                left_padding=left_padding,
                right_padding=right_padding,
            ),
        )

    # Defining `__eq__` drops the inherited hash, so restore the C one.
    __hash__ = bytes.__hash__

    def __eq__(self, other: object) -> bool:
        """Compare two FixedSizeBytes objects to be equal."""
        if other is None:
            return False
        if not isinstance(other, FixedSizeBytes):
            if not (
                isinstance(other, str)
                or isinstance(other, int)
                or isinstance(other, bytes)
                or isinstance(other, SupportsBytes)
            ):
                return False
            other = self._sized_(other)
        return super().__eq__(other)

    def __ne__(self, other: object) -> bool:
        """Compare two FixedSizeBytes objects to be not equal."""
        return not self.__eq__(other)

    @classmethod
    def __get_pydantic_core_schema__(
        cls, source_type: Any, handler: GetCoreSchemaHandler
    ) -> PlainValidatorFunctionSchema:
        """
        Call the class constructor without info and appends the serialization
        schema.
        """
        pattern = f"^0x([0-9a-fA-F]{{{cls.byte_length * 2}}})*$"
        return no_info_plain_validator_function(
            source_type,
            serialization=to_string_ser_schema(),
            json_schema_input_schema=handler(
                Annotated[str, StringConstraints(pattern=pattern)]
            ),
        )

class ForkHash(FixedSizeBytes[4]):  # type: ignore
    """
    Class that helps represent the CRC config hashes and identifiers of a fork.
    """

    pass

class Hash(FixedSizeBytes[32]):  # type: ignore
    """Class that helps represent hashes in tests."""

    pass

class HashInt(FixedSizeHexNumber[32]):  # type: ignore
    """Class that helps represent hashes in tests."""

    pass

class HeaderNonce(FixedSizeBytes[8]):  # type: ignore
    """Class that helps represent the header nonce in tests."""

    pass

class HexNumber(Number):
    """Class that helps represent hexadecimal numbers in tests."""

    def __str__(self) -> str:
        """Return the string representation of the number."""
        return self.hex()

    @staticmethod
    def __get_pydantic_core_schema__(
        source_type: Any, handler: GetCoreSchemaHandler
    ) -> PlainValidatorFunctionSchema:
        """
        Call the class constructor without info and appends the serialization
        schema.
        """
        return no_info_plain_validator_function(
            source_type,
            serialization=to_string_ser_schema(),
            json_schema_input_schema=handler(
                Annotated[str, StringConstraints(pattern=r"^0x[0-9a-fA-F]*$")]
            ),
        )

class Number(int, ToStringSchema):
    """Class that helps represent numbers in tests."""

    def __new__(cls, input_number: NumberConvertible | Self) -> Self:
        """Create a new Number object."""
        return super(Number, cls).__new__(cls, to_number(input_number))

    def __str__(self) -> str:
        """Return the string representation of the number."""
        return str(int(self))

    def hex(self) -> str:
        """Return the hexadecimal representation of the number."""
        return hex(self)

class StorageKey(FixedSizeBytes[32]):  # type: ignore
    """
    Storage key type that automatically applies left padding for values shorter
    than 32 bytes.
    """

    def __new__(
        cls,
        input_bytes: FixedSizeBytesConvertible | FixedSizeBytes,
        **kwargs: Any,
    ) -> Self:
        """Create a new StorageKey with automatic left padding."""
        # Always apply left_padding for storage keys unless explicitly set to
        # False
        if "left_padding" not in kwargs:
            kwargs["left_padding"] = True
        return super().__new__(cls, input_bytes, **kwargs)

class Wei(Number):
    """Class that helps represent wei that can be parsed from strings."""

    def __new__(cls, input_number: NumberConvertible | Self) -> Self:
        """Create a new Number object."""
        if isinstance(input_number, str):
            words = input_number.split()
            multiplier = 1
            assert len(words) <= 2
            value_str = words[0]
            if len(words) > 1:
                unit = words[1].lower()
                multiplier = cls._get_multiplier(unit)
            value: float
            if "**" in value_str:
                base, exp = value_str.split("**")
                value = float(base) ** int(exp)
            else:
                value = (
                    int(value_str)
                    if value_str.isdecimal()
                    else float(value_str)
                )
            return super(Number, cls).__new__(cls, value * multiplier)
        return super(Number, cls).__new__(cls, to_number(input_number))

    @staticmethod
    def _get_multiplier(unit: str) -> int:
        """
        Return the multiplier for the given unit of wei, handling synonyms.
        """
        match unit:
            case "wei":
                return 1
            case "kwei" | "babbage" | "femtoether":
                return 10**3
            case "mwei" | "lovelace" | "picoether":
                return 10**6
            case "gwei" | "shannon" | "nanoether" | "nano":
                return 10**9
            case "szabo" | "microether" | "micro":
                return 10**12
            case "finney" | "milliether" | "milli":
                return 10**15
            case "ether" | "eth":
                return 10**18
            case _:
                raise ValueError(f"Invalid unit {unit}")

class ZeroPaddedHexNumber(HexNumber):
    """Class that helps represent zero padded hexadecimal numbers in tests."""

    def hex(self) -> str:
        """Return the hexadecimal representation of the number."""
        if self == 0:
            return "0x00"
        hex_str = hex(self)[2:]
        if len(hex_str) % 2 == 1:
            return "0x0" + hex_str
        return "0x" + hex_str

    @staticmethod
    def __get_pydantic_core_schema__(
        source_type: Any, handler: GetCoreSchemaHandler
    ) -> PlainValidatorFunctionSchema:
        """
        Call the class constructor without info and appends the serialization
        schema.
        """
        return no_info_plain_validator_function(
            source_type,
            serialization=to_string_ser_schema(),
            json_schema_input_schema=handler(
                Annotated[
                    str, StringConstraints(pattern=r"^0x([0-9a-fA-F]{2})*$")
                ]
            ),
        )

class AccessList(CamelModel, RLPSerializable):
    """Access List for transactions."""

    address: Address
    storage_keys: List[Hash]

    rlp_fields: ClassVar[List[str]] = ["address", "storage_keys"]

class Account(CamelModel):
    """State associated with an address."""

    nonce: ZeroPaddedHexNumber = ZeroPaddedHexNumber(0)
    """
    The scalar value equal to a) the number of transactions sent by an
    Externally Owned Account, b) the amount of contracts created by a contract.
    """
    balance: ZeroPaddedHexNumber = ZeroPaddedHexNumber(0)
    """The amount of Wei (10<sup>-18</sup> Eth) the account has."""
    code: Bytes = Bytes(b"")
    """Bytecode contained by the account."""
    storage: Storage = Field(default_factory=Storage)
    """Storage within a contract."""

    model_config = {
        **CamelModel.model_config,
        "frozen": True,
    }

    @model_validator(mode="after")
    def _freeze_storage(self) -> Self:
        """
        Freeze the storage along with the account, so that accounts shared
        between allocations cannot change under any of them.
        """
        self.storage.freeze()
        return self

class Alloc(EthereumTestRootModel[Dict[Address, Account | None]]):
    """Allocation of accounts in the state, pre and post test execution."""

    root: Dict[Address, Account | None] = Field(
        default_factory=dict, validate_default=True
    )

    def get(self, address: Address) -> Account | None:
        """Get an account if it's present in the allocation, otherwise None."""
        account = self.root.get(address)
        if not account:
            return None
        return account

class BlobSchedule(EthereumTestRootModel[Dict[str, ForkBlobSchedule]]):
    """Blob schedule configuration dictionary."""

    root: Dict[str, ForkBlobSchedule] = Field(
        default_factory=dict, validate_default=True
    )

    def append(self, *, fork: str, schedule: Any) -> None:
        """Append a new fork schedule."""
        if not isinstance(schedule, ForkBlobSchedule):
            schedule = ForkBlobSchedule(**schedule)
        self.root[fork] = schedule

    def last(self) -> ForkBlobSchedule | None:
        """Return the last schedule."""
        if len(self.root) == 0:
            return None
        return list(self.root.values())[-1]

    def __getitem__(self, key: str) -> ForkBlobSchedule:
        """Return the schedule for a given fork."""
        return self.root[key]

class ForkBlobSchedule(CamelModel):
    """Representation of the blob schedule of a given fork."""

    target_blobs_per_block: HexNumber = Field(..., alias="target")
    max_blobs_per_block: HexNumber = Field(..., alias="max")
    base_fee_update_fraction: HexNumber = Field(...)

class StateCommitment(Enum):
    """
    The state-commitment scheme used to compute an allocation's state root.
    """

    MPT = auto()
    """Merkle-Patricia trie."""