diff --git a/python/x402/tests/unit/mechanisms/evm/test_signer.py b/python/x402/tests/unit/mechanisms/evm/test_signer.py index 3d6ba19d0f..ef0c18b649 100644 --- a/python/x402/tests/unit/mechanisms/evm/test_signer.py +++ b/python/x402/tests/unit/mechanisms/evm/test_signer.py @@ -1,5 +1,7 @@ """Tests for EVM signer implementations.""" +from unittest.mock import MagicMock, patch + import pytest try: @@ -7,7 +9,11 @@ except ImportError: pytest.skip("EVM signers require eth_account", allow_module_level=True) -from x402.mechanisms.evm.signers import EthAccountSigner, FacilitatorWeb3Signer +from x402.mechanisms.evm.signers import ( + EthAccountSigner, + EthAccountSignerWithRPC, + FacilitatorWeb3Signer, +) class TestEthAccountSigner: @@ -184,3 +190,185 @@ def test_facilitator_signer_implements_facilitator_protocol(self): assert hasattr(signer, "get_balance") assert hasattr(signer, "get_chain_id") assert hasattr(signer, "get_code") + + +class TestEthAccountSignerWithRPC: + """Test EthAccountSignerWithRPC client-side signer with RPC capabilities.""" + + def test_should_create_signer_and_inherit_address(self): + """Should create signer and inherit address from EthAccountSigner.""" + account = Account.create() + signer = EthAccountSignerWithRPC(account, rpc_url="https://sepolia.base.org") + + assert signer.address == account.address + assert signer.address.startswith("0x") + assert len(signer.address) == 42 + + def test_should_be_subclass_of_eth_account_signer(self): + """EthAccountSignerWithRPC should inherit from EthAccountSigner.""" + account = Account.create() + signer = EthAccountSignerWithRPC(account, rpc_url="https://sepolia.base.org") + + assert isinstance(signer, EthAccountSigner) + + def test_should_construct_web3_with_rpc_url(self): + """Constructor should wire a Web3 client at the given RPC URL.""" + account = Account.create() + with patch("x402.mechanisms.evm.signers.Web3") as mock_web3_cls: + mock_provider = MagicMock() + mock_web3_cls.HTTPProvider.return_value = mock_provider + + EthAccountSignerWithRPC(account, rpc_url="https://example.test/rpc") + + mock_web3_cls.HTTPProvider.assert_called_once_with("https://example.test/rpc") + mock_web3_cls.assert_called_once_with(mock_provider) + + def test_should_inherit_sign_typed_data(self): + """EthAccountSignerWithRPC should inherit working sign_typed_data.""" + account = Account.create() + signer = EthAccountSignerWithRPC(account, rpc_url="https://sepolia.base.org") + + from x402.mechanisms.evm.types import TypedDataDomain, TypedDataField + + domain = TypedDataDomain( + name="USD Coin", + version="2", + chain_id=8453, + verifying_contract="0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913", + ) + types = { + "TransferWithAuthorization": [ + TypedDataField(name="from", type="address"), + TypedDataField(name="to", type="address"), + TypedDataField(name="value", type="uint256"), + TypedDataField(name="validAfter", type="uint256"), + TypedDataField(name="validBefore", type="uint256"), + TypedDataField(name="nonce", type="bytes32"), + ] + } + message = { + "from": account.address, + "to": "0x1234567890123456789012345678901234567890", + "value": "1000000", + "validAfter": "1000000000", + "validBefore": "1000003600", + "nonce": "0x" + "00" * 32, + } + + signature = signer.sign_typed_data(domain, types, "TransferWithAuthorization", message) + + assert isinstance(signature, bytes) + assert len(signature) >= 65 + + def test_read_contract_should_call_function_and_return_value(self): + """read_contract should checksum the address and return the call() value.""" + account = Account.create() + signer = EthAccountSignerWithRPC(account, rpc_url="https://sepolia.base.org") + + mock_call = MagicMock(return_value=42) + mock_function = MagicMock(return_value=MagicMock(call=mock_call)) + mock_contract = MagicMock() + mock_contract.functions.balanceOf = mock_function + signer._w3 = MagicMock() + signer._w3.eth.contract.return_value = mock_contract + + # Lowercase address — read_contract must checksum it before contract(). + result = signer.read_contract( + "0x833589fcd6edb6e08f4c7c32d4f71b54bda02913", + [{"name": "balanceOf"}], + "balanceOf", + "0xabcdefabcdefabcdefabcdefabcdefabcdefabcd", + ) + + assert result == 42 + contract_call_kwargs = signer._w3.eth.contract.call_args.kwargs + assert contract_call_kwargs["address"] == "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913" + assert contract_call_kwargs["abi"] == [{"name": "balanceOf"}] + mock_function.assert_called_once_with("0xabcdefabcdefabcdefabcdefabcdefabcdefabcd") + mock_call.assert_called_once_with() + + def test_sign_transaction_should_return_hex_prefixed_raw_tx(self): + """sign_transaction should return '0x'-prefixed hex of raw_transaction bytes.""" + account = Account.create() + signer = EthAccountSignerWithRPC(account, rpc_url="https://sepolia.base.org") + + raw_bytes = b"\x02\xf8\x6c\x01\x80" + signed_tx = MagicMock() + signed_tx.raw_transaction = raw_bytes + signer._w3 = MagicMock() + signer._w3.eth.account.sign_transaction.return_value = signed_tx + + tx = { + "to": "0x1234567890123456789012345678901234567890", + "data": "0x", + "nonce": 0, + "gas": 21000, + "maxFeePerGas": 2_000_000_000, + "maxPriorityFeePerGas": 1_000_000_000, + "chainId": 84532, + } + + result = signer.sign_transaction(tx) + + assert result == "0x" + raw_bytes.hex() + assert result.startswith("0x") + signer._w3.eth.account.sign_transaction.assert_called_once_with(tx, account.key) + + def test_get_transaction_count_should_checksum_address(self): + """get_transaction_count should pass a checksummed address to web3.""" + account = Account.create() + signer = EthAccountSignerWithRPC(account, rpc_url="https://sepolia.base.org") + + signer._w3 = MagicMock() + signer._w3.eth.get_transaction_count.return_value = 7 + + result = signer.get_transaction_count("0x833589fcd6edb6e08f4c7c32d4f71b54bda02913") + + assert result == 7 + signer._w3.eth.get_transaction_count.assert_called_once_with( + "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913" + ) + + def test_estimate_fees_per_gas_should_compute_eip1559_fees(self): + """estimate_fees_per_gas should return (base*2 + priority, priority).""" + account = Account.create() + signer = EthAccountSignerWithRPC(account, rpc_url="https://sepolia.base.org") + + signer._w3 = MagicMock() + signer._w3.eth.get_block.return_value = {"baseFeePerGas": 5_000_000_000} + signer._w3.eth.max_priority_fee = 1_000_000_000 + + max_fee, max_priority = signer.estimate_fees_per_gas() + + assert max_priority == 1_000_000_000 + assert max_fee == 5_000_000_000 * 2 + 1_000_000_000 + signer._w3.eth.get_block.assert_called_once_with("latest") + + def test_estimate_fees_per_gas_should_default_base_fee_when_missing(self): + """estimate_fees_per_gas should fall back to 1 gwei base when block omits it.""" + account = Account.create() + signer = EthAccountSignerWithRPC(account, rpc_url="https://sepolia.base.org") + + signer._w3 = MagicMock() + # Block without baseFeePerGas (e.g., legacy / non-EIP-1559 chain). + signer._w3.eth.get_block.return_value = {} + signer._w3.eth.max_priority_fee = 2_500_000_000 + + max_fee, max_priority = signer.estimate_fees_per_gas() + + assert max_priority == 2_500_000_000 + assert max_fee == 1_000_000_000 * 2 + 2_500_000_000 + + def test_implements_client_evm_signer_protocol(self): + """EthAccountSignerWithRPC should expose RPC-extended client signer surface.""" + account = Account.create() + signer = EthAccountSignerWithRPC(account, rpc_url="https://sepolia.base.org") + + # Inherited from EthAccountSigner: + assert hasattr(signer, "address") + assert hasattr(signer, "sign_typed_data") + # Added by EthAccountSignerWithRPC for gas-sponsoring extensions: + assert callable(signer.read_contract) + assert callable(signer.sign_transaction) + assert callable(signer.get_transaction_count) + assert callable(signer.estimate_fees_per_gas)