chore(fw): raise an exception if an opcode recieves an invalid keyword argument

docs/CHANGELOG.md

@@ -26,6 +26,7 @@ Users can select any of the artifacts depending on their testing needs for their
 #### 🔀 Refactoring
 
 - 🔀 Move `TransactionType` enum from test file to proper module location in `ethereum_test_types.transaction_types` for better code organization and reusability.
+- ✨ Opcode classes now validate keyword arguments and raise `ValueError` with clear error messages
 
 #### `fill`
 

src/ethereum_test_vm/opcode.py

@@ -87,7 +87,7 @@ class Opcode(Bytecode):
     data_portion_length: int
     data_portion_formatter: Optional[Callable[[Any], bytes]]
     stack_properties_modifier: Optional[Callable[[Any], tuple[int, int, int, int]]]
-    kwargs: List[str] | None
+    kwargs: List[str]
     kwargs_defaults: KW_ARGS_DEFAULTS_TYPE
     unchecked_stack: bool = False
 
@@ -137,7 +137,10 @@ def __new__(
             obj.data_portion_formatter = data_portion_formatter
             obj.stack_properties_modifier = stack_properties_modifier
             obj.unchecked_stack = unchecked_stack
-            obj.kwargs = kwargs
+            if kwargs is None:
+                obj.kwargs = []
+            else:
+                obj.kwargs = kwargs
             obj.kwargs_defaults = kwargs_defaults
             return obj
         raise TypeError("Opcode constructor '__new__' didn't return an instance!")
@@ -254,6 +257,16 @@ def __call__(
 
         if self.kwargs is not None and len(kwargs) > 0:
             assert len(args) == 0, f"Cannot mix positional and keyword arguments {args} {kwargs}"
+
+            # Validate that all provided kwargs are valid
+            invalid_kwargs = set(kwargs.keys()) - set(self.kwargs)
+            if invalid_kwargs:
+                raise ValueError(
+                    f"Invalid keyword argument(s) {list(invalid_kwargs)} for opcode "
+                    f"{self._name_}. Valid arguments are: {self.kwargs}"
+                    f"Valid arguments are: {self.kwargs}"
+                )
+
             for kw in self.kwargs:
                 args.append(kwargs[kw] if kw in kwargs else self.kwargs_defaults.get(kw, 0))
 

src/ethereum_test_vm/tests/test_vm.py

@@ -431,3 +431,38 @@ def test_bytecode_concatenation_with_bytes():
     assert code.max_stack_height == base.max_stack_height
     assert code.min_stack_height == base.min_stack_height
     assert code.terminating == base.terminating
+
+
+def test_opcode_kwargs_validation():
+    """Test that invalid keyword arguments raise ValueError."""
+    # Test valid kwargs work
+    Op.MSTORE(offset=0, value=1)
+    Op.CALL(gas=1, address=2, value=3, args_offset=4, args_size=5, ret_offset=6, ret_size=7)
+
+    # Test invalid kwargs raise ValueError
+    with pytest.raises(
+        ValueError, match=r"Invalid keyword argument\(s\) \['offest'\] for opcode MSTORE"
+    ):
+        Op.MSTORE(offest=0, value=1)  # codespell:ignore offest
+
+    with pytest.raises(
+        ValueError, match=r"Invalid keyword argument\(s\) \['wrong_arg'\] for opcode MSTORE"
+    ):
+        Op.MSTORE(offset=0, value=1, wrong_arg=2)
+
+    with pytest.raises(
+        ValueError, match=r"Invalid keyword argument\(s\) \['addres'\] for opcode CALL"
+    ):
+        Op.CALL(
+            gas=1,
+            addres=2,  # codespell:ignore
+            value=3,
+            args_offset=4,
+            args_size=5,
+            ret_offset=6,
+            ret_size=7,
+        )
+
+    # Test multiple invalid kwargs
+    with pytest.raises(ValueError, match=r"Invalid keyword argument\(s\).*for opcode MSTORE"):
+        Op.MSTORE(offest=0, valu=1, extra=2)  # codespell:ignore offest,valu

tests/cancun/eip1153_tstore/test_tstorage_create_contexts.py

@@ -155,7 +155,16 @@ def creator_contract_code(  # noqa: D102
             + Op.TSTORE(1, 0x0200)
             + Op.TSTORE(2, 0x0300)
             + Op.CALLDATACOPY(0, 0, Op.CALLDATASIZE)
-            + Op.SSTORE(4, Op.CALL(address=opcode(size=Op.CALLDATASIZE, salt=create2_salt)))
+            + Op.SSTORE(
+                4,
+                Op.CALL(
+                    address=(
+                        opcode(size=Op.CALLDATASIZE, salt=create2_salt)
+                        if opcode == Op.CREATE2
+                        else opcode(size=Op.CALLDATASIZE)
+                    )
+                ),
+            )
             # Save the state of transient storage following call to storage; the transient
             # storage should not have been overwritten
             + Op.SSTORE(0, Op.TLOAD(0))

tests/cancun/eip4844_blobs/test_blobhash_opcode_contexts.py

@@ -132,7 +132,11 @@ def deploy_contract(
                 create_opcode = Op.CREATE if self == BlobhashContext.CREATE else Op.CREATE2
                 create_bytecode = Op.EXTCODECOPY(
                     address=initcode_address, dest_offset=0, offset=0, size=len(initcode)
-                ) + Op.POP(create_opcode(value=0, offset=0, size=len(initcode), salt=0))
+                ) + Op.POP(
+                    create_opcode(value=0, offset=0, size=len(initcode), salt=0)
+                    if create_opcode == Op.CREATE2
+                    else create_opcode(value=0, offset=0, size=len(initcode))
+                )
                 return pre.deploy_contract(create_bytecode)
 
             case _:

tests/cancun/eip6780_selfdestruct/test_reentrancy_selfdestruct_revert.py

@@ -53,7 +53,14 @@ def executor_contract_bytecode(
 ) -> Bytecode:
     """Contract code that performs a selfdestruct call then revert."""
     return (
-        Op.SSTORE(1, first_suicide(address=selfdestruct_contract_address, value=0))
+        Op.SSTORE(
+            1,
+            (
+                first_suicide(address=selfdestruct_contract_address, value=0)
+                if first_suicide in [Op.CALL, Op.CALLCODE]
+                else first_suicide(address=selfdestruct_contract_address)
+            ),
+        )
         + Op.SSTORE(2, Op.CALL(address=revert_contract_address))
         + Op.RETURNDATACOPY(0, 0, Op.RETURNDATASIZE())
         + Op.SSTORE(3, Op.MLOAD(0))
@@ -95,7 +102,11 @@ def revert_contract_bytecode(
     selfdestruct_contract_address: Address,
 ) -> Bytecode:
     """Contract code that performs a call and then reverts."""
-    call_op = second_suicide(address=selfdestruct_contract_address, value=100)
+    call_op = (
+        second_suicide(address=selfdestruct_contract_address, value=100)
+        if second_suicide in [Op.CALL, Op.CALLCODE]
+        else second_suicide(address=selfdestruct_contract_address)
+    )
     return Op.MSTORE(0, Op.ADD(15, call_op)) + Op.REVERT(0, 32)
 
 

tests/constantinople/eip1014_create2/test_create_returndata.py

@@ -48,7 +48,6 @@ def test_create2_return_data(
     slot_begin_memory_after_create = 8
 
     # CREATE2 Initcode
-    create2_salt = 1
     return_data_in_create = 0xFFFAFB
     initcode = Op.MSTORE(0, return_data_in_create) + return_type_in_create(0, 32)
     call_return_data_value = 0x1122334455667788991011121314151617181920212223242526272829303132
@@ -78,7 +77,7 @@ def test_create2_return_data(
         + Op.SSTORE(slot_return_data_hash_before_create, Op.SHA3(0, call_return_size))
         #
         #
-        + create_type(offset=0x100, size=Op.CALLDATASIZE(), salt=create2_salt)
+        + create_type(offset=0x100, size=Op.CALLDATASIZE())
         + Op.SSTORE(slot_returndatasize_after_create, Op.RETURNDATASIZE())
         + Op.RETURNDATACOPY(0x300, 0, Op.RETURNDATASIZE())
         + Op.SSTORE(slot_returndatacopy_after_create, Op.MLOAD(0x300))

tests/frontier/create/test_create_one_byte.py

@@ -46,7 +46,7 @@ def test_create_one_byte(
     # make a subcontract that deploys code, because deploy 0xef eats ALL gas
     create_contract = pre.deploy_contract(
         code=Op.MSTORE(0, Op.CALLDATALOAD(0))
-        + Op.MSTORE(32, create_opcode(offset=32 - initcode_length, salt=0, size=initcode_length))
+        + Op.MSTORE(32, create_opcode(offset=32 - initcode_length, size=initcode_length))
         + Op.RETURN(32, 32)
     )
     code = pre.deploy_contract(

tests/frontier/identity_precompile/test_identity_returndatasize.py

@@ -49,8 +49,8 @@ def test_identity_precompile_returndata(
                 address=Constants.IDENTITY_PRECOMPILE_ADDRESS,
                 args_offset=0,
                 args_size=args_size,
-                output_offset=0x10,
-                output_size=output_size,
+                ret_offset=0x10,
+                ret_size=output_size,
             )
         )
         + Op.SSTORE(storage.store_next(expected_returndatasize), Op.RETURNDATASIZE)

tests/frontier/precompiles/test_precompiles.py

@@ -70,15 +70,15 @@ def test_precompiles(
     env = Environment()
 
     account = pre.deploy_contract(
-        Op.MSTORE(0, 0)  # Pre-expand the memory so the gas costs are exactly the same
+        Op.MSTORE(1, 32)  # Pre-expand the memory so the gas costs are exactly the same
         + Op.GAS
         + Op.CALL(
             address=address,
             value=0,
             args_offset=0,
             args_size=32,
-            output_offset=32,
-            output_size=32,
+            ret_offset=32,
+            ret_size=32,
         )
         + Op.POP
         + Op.SUB(Op.SWAP1, Op.GAS)
@@ -88,8 +88,8 @@ def test_precompiles(
             value=0,
             args_offset=0,
             args_size=32,
-            output_offset=32,
-            output_size=32,
+            ret_offset=32,
+            ret_size=32,
         )
         + Op.POP
         + Op.SUB(Op.SWAP1, Op.GAS)
@@ -109,6 +109,6 @@ def test_precompiles(
 
     # A high gas cost will result from calling a precompile
     # Expect 0x00 when a precompile exists at the address, 0x01 otherwise
-    post = {account: Account(storage={0: "0x00" if precompile_exists else "0x01"})}
+    post = {account: Account(storage={0: 0 if precompile_exists else 1})}
 
     state_test(env=env, pre=pre, post=post, tx=tx)

tests/frontier/scenarios/scenarios/call_combinations.py

@@ -93,13 +93,23 @@ def _generate_one_call_scenario(self, first_call: Opcode) -> Scenario:
 
         scenario_contract = pre.deploy_contract(
             code=Op.MSTORE(32, 1122334455)
-            + first_call(
-                gas=Op.SUB(Op.GAS, self.keep_gas),
-                address=operation_contract,
-                args_offset=32,
-                args_size=40,
-                ret_size=32,
-                value=balance.first_call_value,
+            + (
+                first_call(
+                    gas=Op.SUB(Op.GAS, self.keep_gas),
+                    address=operation_contract,
+                    args_offset=32,
+                    args_size=40,
+                    ret_size=32,
+                    value=balance.first_call_value,
+                )
+                if first_call not in [Op.DELEGATECALL, Op.STATICCALL]
+                else first_call(
+                    gas=Op.SUB(Op.GAS, self.keep_gas),
+                    address=operation_contract,
+                    args_offset=32,
+                    args_size=40,
+                    ret_size=32,
+                )
             )
             + Op.RETURN(0, 32),
             balance=balance.scenario_contract_balance,
@@ -230,23 +240,41 @@ def _compute_callvalue() -> int:
         )
         sub_contract = pre.deploy_contract(
             code=Op.MSTORE(32, 1122334455)
-            + second_call(
-                gas=Op.SUB(Op.GAS, self.keep_gas),
-                address=operation_contract,
-                args_size=40,
-                args_offset=32,
-                ret_size=32,
-                value=second_call_value,
+            + (
+                second_call(
+                    gas=Op.SUB(Op.GAS, self.keep_gas),
+                    address=operation_contract,
+                    args_size=40,
+                    args_offset=32,
+                    ret_size=32,
+                    value=second_call_value,
+                )
+                if second_call not in [Op.DELEGATECALL, Op.STATICCALL]
+                else second_call(
+                    gas=Op.SUB(Op.GAS, self.keep_gas),
+                    address=operation_contract,
+                    args_size=40,
+                    args_offset=32,
+                    ret_size=32,
+                )
             )
             + Op.RETURN(0, 32),
             balance=balance.sub_contract_balance,
         )
         scenario_contract = pre.deploy_contract(
-            code=first_call(
-                gas=Op.SUB(Op.GAS, self.keep_gas),
-                address=sub_contract,
-                ret_size=32,
-                value=balance.first_call_value,
+            code=(
+                first_call(
+                    gas=Op.SUB(Op.GAS, self.keep_gas),
+                    address=sub_contract,
+                    ret_size=32,
+                    value=balance.first_call_value,
+                )
+                if first_call not in [Op.DELEGATECALL, Op.STATICCALL]
+                else first_call(
+                    gas=Op.SUB(Op.GAS, self.keep_gas),
+                    address=sub_contract,
+                    ret_size=32,
+                )
             )
             + Op.RETURN(0, 32),
             balance=balance.scenario_contract_balance,

tests/frontier/scenarios/scenarios/create_combinations.py

@@ -109,14 +109,25 @@ def _compute_selfbalance() -> int:
                 + Op.MSTORE(32, create(balance.create_value, 0, deploy_code_size, *salt))
                 + Op.MSTORE(0, 0)
                 + Op.MSTORE(64, 1122334455)
-                + call(
-                    gas=Op.SUB(Op.GAS, keep_gas),
-                    address=Op.MLOAD(32),
-                    args_offset=64,
-                    args_size=40,
-                    ret_offset=0,
-                    ret_size=32,
-                    value=balance.call_value,
+                + (
+                    call(
+                        gas=Op.SUB(Op.GAS, keep_gas),
+                        address=Op.MLOAD(32),
+                        args_offset=64,
+                        args_size=40,
+                        ret_offset=0,
+                        ret_size=32,
+                        value=balance.call_value,
+                    )
+                    if call not in [Op.DELEGATECALL, Op.STATICCALL]
+                    else call(
+                        gas=Op.SUB(Op.GAS, keep_gas),
+                        address=Op.MLOAD(32),
+                        args_offset=64,
+                        args_size=40,
+                        ret_offset=0,
+                        ret_size=32,
+                    )
                 )
                 + Op.RETURN(0, 32),
             )

tests/frontier/scenarios/scenarios/double_call_combinations.py

@@ -2,10 +2,9 @@
 
 from typing import List
 
-from ethereum_test_tools import Conditional
-from ethereum_test_tools.vm.opcode import Macro, Opcode
-from ethereum_test_tools.vm.opcode import Macros as Om
-from ethereum_test_tools.vm.opcode import Opcodes as Op
+from ethereum_test_tools import Bytecode, Conditional
+from ethereum_test_tools import Macros as Om
+from ethereum_test_tools import Opcodes as Op
 
 from ..common import Scenario, ScenarioEnvironment, ScenarioGeneratorInput
 
@@ -20,14 +19,14 @@ def scenarios_double_call_combinations(scenario_input: ScenarioGeneratorInput) -
     """
     scenarios_list: List[Scenario] = []
     keep_gas = 300000
-    revert_types: List[Opcode | Macro] = [Op.STOP, Om.OOG, Op.RETURN]
+    revert_types: List[Bytecode] = [Op.STOP(), Om.OOG(), Op.RETURN(offset=0, size=32)]
     if Op.REVERT in scenario_input.fork.valid_opcodes():
-        revert_types.append(Op.REVERT)
+        revert_types.append(Op.REVERT(offset=0, size=32))
     for revert in revert_types:
         operation_contract = scenario_input.pre.deploy_contract(code=scenario_input.operation_code)
         subcall_contract = scenario_input.pre.deploy_contract(
             code=Op.MSTORE(0, 0x1122334455667788991011121314151617181920212223242526272829303132)
-            + revert(offset=0, size=32)
+            + revert
         )
         scenario_contract = scenario_input.pre.deploy_contract(
             code=Op.CALL(gas=Op.SUB(Op.GAS, keep_gas), address=operation_contract, ret_size=32)