Testing with Ethereum Tester¶
Ethereum Tester is a tool suite for testing Ethereum based applications.
This section provides a quick overview of testing with eth-tester
. To learn more, you can view the documentation at the Github repo or join the Gitter channel.
Getting Started¶
Prior to testing, the Vyper specific contract conversion and the blockchain related fixtures need to be set up. These fixtures will be used in every test file and should therefore be defined in conftest.py.
Note
Since the testing is done in the pytest framework, you can make use of pytest.ini, tox.ini and setup.cfg and you can use most IDEs’ pytest plugins.
1import json
2
3import pytest
4import web3.exceptions
5from eth_tester import EthereumTester, PyEVMBackend
6from eth_tester.exceptions import TransactionFailed
7from eth_utils.toolz import compose
8from hexbytes import HexBytes
9from web3 import Web3
10from web3.contract import Contract
11from web3.providers.eth_tester import EthereumTesterProvider
12
13from vyper import compiler
14from vyper.ast.grammar import parse_vyper_source
15
16
17class VyperMethod:
18 ALLOWED_MODIFIERS = {"call", "estimateGas", "transact", "buildTransaction"}
19
20 def __init__(self, function, normalizers=None):
21 self._function = function
22 self._function._return_data_normalizers = normalizers
23
24 def __call__(self, *args, **kwargs):
25 return self.__prepared_function(*args, **kwargs)
26
27 def __prepared_function(self, *args, **kwargs):
28 if not kwargs:
29 modifier, modifier_dict = "call", {}
30 fn_abi = [
31 x
32 for x in self._function.contract_abi
33 if x.get("name") == self._function.function_identifier
34 ].pop()
35 # To make tests faster just supply some high gas value.
36 modifier_dict.update({"gas": fn_abi.get("gas", 0) + 500000})
37 elif len(kwargs) == 1:
38 modifier, modifier_dict = kwargs.popitem()
39 if modifier not in self.ALLOWED_MODIFIERS:
40 raise TypeError(f"The only allowed keyword arguments are: {self.ALLOWED_MODIFIERS}")
41 else:
42 raise TypeError(f"Use up to one keyword argument, one of: {self.ALLOWED_MODIFIERS}")
43 return getattr(self._function(*args), modifier)(modifier_dict)
44
45
46class VyperContract:
47 """
48 An alternative Contract Factory which invokes all methods as `call()`,
49 unless you add a keyword argument. The keyword argument assigns the prep method.
50 This call
51 > contract.withdraw(amount, transact={'from': eth.accounts[1], 'gas': 100000, ...})
52 is equivalent to this call in the classic contract:
53 > contract.functions.withdraw(amount).transact({'from': eth.accounts[1], 'gas': 100000, ...})
54 """
55
56 def __init__(self, classic_contract, method_class=VyperMethod):
57 classic_contract._return_data_normalizers += CONCISE_NORMALIZERS
58 self._classic_contract = classic_contract
59 self.address = self._classic_contract.address
60 protected_fn_names = [fn for fn in dir(self) if not fn.endswith("__")]
61
62 try:
63 fn_names = [fn["name"] for fn in self._classic_contract.functions._functions]
64 except web3.exceptions.NoABIFunctionsFound:
65 fn_names = []
66
67 for fn_name in fn_names:
68 # Override namespace collisions
69 if fn_name in protected_fn_names:
70 raise AttributeError(f"{fn_name} is protected!")
71 else:
72 _classic_method = getattr(self._classic_contract.functions, fn_name)
73 _concise_method = method_class(
74 _classic_method, self._classic_contract._return_data_normalizers
75 )
76 setattr(self, fn_name, _concise_method)
77
78 @classmethod
79 def factory(cls, *args, **kwargs):
80 return compose(cls, Contract.factory(*args, **kwargs))
81
82
83def _none_addr(datatype, data):
84 if datatype == "address" and int(data, base=16) == 0:
85 return (datatype, None)
86 else:
87 return (datatype, data)
88
89
90CONCISE_NORMALIZERS = (_none_addr,)
91
92
93@pytest.fixture(scope="module")
94def tester():
95 # set absurdly high gas limit so that london basefee never adjusts
96 # (note: 2**63 - 1 is max that evm allows)
97 custom_genesis = PyEVMBackend._generate_genesis_params(overrides={"gas_limit": 10**10})
98 custom_genesis["base_fee_per_gas"] = 0
99 backend = PyEVMBackend(genesis_parameters=custom_genesis)
100 return EthereumTester(backend=backend)
101
102
103def zero_gas_price_strategy(web3, transaction_params=None):
104 return 0 # zero gas price makes testing simpler.
105
106
107@pytest.fixture(scope="module")
108def w3(tester):
109 w3 = Web3(EthereumTesterProvider(tester))
110 w3.eth.set_gas_price_strategy(zero_gas_price_strategy)
111 return w3
112
113
114def _get_contract(w3, source_code, no_optimize, *args, **kwargs):
115 out = compiler.compile_code(
116 source_code,
117 # test that metadata gets generated
118 ["abi", "bytecode", "metadata"],
119 interface_codes=kwargs.pop("interface_codes", None),
120 no_optimize=no_optimize,
121 evm_version=kwargs.pop("evm_version", None),
122 show_gas_estimates=True, # Enable gas estimates for testing
123 )
124 parse_vyper_source(source_code) # Test grammar.
125 json.dumps(out["metadata"]) # test metadata is json serializable
126 abi = out["abi"]
127 bytecode = out["bytecode"]
128 value = kwargs.pop("value_in_eth", 0) * 10**18 # Handle deploying with an eth value.
129 c = w3.eth.contract(abi=abi, bytecode=bytecode)
130 deploy_transaction = c.constructor(*args)
131 tx_info = {"from": w3.eth.accounts[0], "value": value, "gasPrice": 0}
132 tx_info.update(kwargs)
133 tx_hash = deploy_transaction.transact(tx_info)
134 address = w3.eth.get_transaction_receipt(tx_hash)["contractAddress"]
135 return w3.eth.contract(address, abi=abi, bytecode=bytecode, ContractFactoryClass=VyperContract)
136
137
138def _deploy_blueprint_for(w3, source_code, no_optimize, initcode_prefix=b"", **kwargs):
139 out = compiler.compile_code(
140 source_code,
141 ["abi", "bytecode"],
142 interface_codes=kwargs.pop("interface_codes", None),
143 no_optimize=no_optimize,
144 evm_version=kwargs.pop("evm_version", None),
145 show_gas_estimates=True, # Enable gas estimates for testing
146 )
147 parse_vyper_source(source_code) # Test grammar.
148 abi = out["abi"]
149 bytecode = HexBytes(initcode_prefix) + HexBytes(out["bytecode"])
150 bytecode_len = len(bytecode)
151 bytecode_len_hex = hex(bytecode_len)[2:].rjust(4, "0")
152 # prepend a quick deploy preamble
153 deploy_preamble = HexBytes("61" + bytecode_len_hex + "3d81600a3d39f3")
154 deploy_bytecode = HexBytes(deploy_preamble) + bytecode
155
156 deployer_abi = [] # just a constructor
157 c = w3.eth.contract(abi=deployer_abi, bytecode=deploy_bytecode)
158 deploy_transaction = c.constructor()
159 tx_info = {"from": w3.eth.accounts[0], "value": 0, "gasPrice": 0}
160
161 tx_hash = deploy_transaction.transact(tx_info)
162 address = w3.eth.get_transaction_receipt(tx_hash)["contractAddress"]
163
164 # sanity check
165 assert w3.eth.get_code(address) == bytecode, (w3.eth.get_code(address), bytecode)
166
167 def factory(address):
168 return w3.eth.contract(
169 address, abi=abi, bytecode=bytecode, ContractFactoryClass=VyperContract
170 )
171
172 return w3.eth.contract(address, bytecode=deploy_bytecode), factory
173
174
175@pytest.fixture(scope="module")
176def deploy_blueprint_for(w3, no_optimize):
177 def deploy_blueprint_for(source_code, *args, **kwargs):
178 return _deploy_blueprint_for(w3, source_code, no_optimize, *args, **kwargs)
179
180 return deploy_blueprint_for
181
182
183@pytest.fixture(scope="module")
184def get_contract(w3, no_optimize):
185 def get_contract(source_code, *args, **kwargs):
186 return _get_contract(w3, source_code, no_optimize, *args, **kwargs)
187
188 return get_contract
189
190
191@pytest.fixture
192def get_logs(w3):
193 def get_logs(tx_hash, c, event_name):
194 tx_receipt = w3.eth.get_transaction_receipt(tx_hash)
195 return c._classic_contract.events[event_name]().process_receipt(tx_receipt)
196
197 return get_logs
198
199
200@pytest.fixture(scope="module")
201def assert_tx_failed(tester):
202 def assert_tx_failed(function_to_test, exception=TransactionFailed, exc_text=None):
203 snapshot_id = tester.take_snapshot()
204 with pytest.raises(exception) as excinfo:
205 function_to_test()
206 tester.revert_to_snapshot(snapshot_id)
207 if exc_text:
208 # TODO test equality
209 assert exc_text in str(excinfo.value), (exc_text, excinfo.value)
210
211 return assert_tx_failed
The final two fixtures are optional and will be discussed later. The rest of this chapter assumes that you have this code set up in your conftest.py
file.
Alternatively, you can import the fixtures to conftest.py
or use pytest plugins.
Writing a Basic Test¶
Assume the following simple contract storage.vy
. It has a single integer variable and a function to set that value.
1storedData: public(int128)
2
3@external
4def __init__(_x: int128):
5 self.storedData = _x
6
7@external
8def set(_x: int128):
9 self.storedData = _x
We create a test file test_storage.py
where we write our tests in pytest style.
1import pytest
2
3INITIAL_VALUE = 4
4
5
6@pytest.fixture
7def storage_contract(w3, get_contract):
8 with open("examples/storage/storage.vy") as f:
9 contract_code = f.read()
10 # Pass constructor variables directly to the contract
11 contract = get_contract(contract_code, INITIAL_VALUE)
12 return contract
13
14
15def test_initial_state(storage_contract):
16 # Check if the constructor of the contract is set up properly
17 assert storage_contract.storedData() == INITIAL_VALUE
18
19
20def test_set(w3, storage_contract):
21 k0 = w3.eth.accounts[0]
22
23 # Let k0 try to set the value to 10
24 storage_contract.set(10, transact={"from": k0})
25 assert storage_contract.storedData() == 10 # Directly access storedData
26
27 # Let k0 try to set the value to -5
28 storage_contract.set(-5, transact={"from": k0})
29 assert storage_contract.storedData() == -5
First we create a fixture for the contract which will compile our contract and set up a Web3 contract object. We then use this fixture for our test functions to interact with the contract.
Note
To run the tests, call pytest
or python -m pytest
from your project directory.
Events and Failed Transactions¶
To test events and failed transactions we expand our simple storage contract to include an event and two conditions for a failed transaction: advanced_storage.vy
1event DataChange:
2 setter: indexed(address)
3 value: int128
4
5storedData: public(int128)
6
7@external
8def __init__(_x: int128):
9 self.storedData = _x
10
11@external
12def set(_x: int128):
13 assert _x >= 0, "No negative values"
14 assert self.storedData < 100, "Storage is locked when 100 or more is stored"
15 self.storedData = _x
16 log DataChange(msg.sender, _x)
17
18@external
19def reset():
20 self.storedData = 0
Next, we take a look at the two fixtures that will allow us to read the event logs and to check for failed transactions.
@pytest.fixture(scope="module")
def assert_tx_failed(tester):
def assert_tx_failed(function_to_test, exception=TransactionFailed, exc_text=None):
snapshot_id = tester.take_snapshot()
with pytest.raises(exception) as excinfo:
function_to_test()
tester.revert_to_snapshot(snapshot_id)
if exc_text:
# TODO test equality
assert exc_text in str(excinfo.value), (exc_text, excinfo.value)
return assert_tx_failed
The fixture to assert failed transactions defaults to check for a TransactionFailed
exception, but can be used to check for different exceptions too, as shown below. Also note that the chain gets reverted to the state before the failed transaction.
@pytest.fixture
def get_logs(w3):
def get_logs(tx_hash, c, event_name):
tx_receipt = w3.eth.get_transaction_receipt(tx_hash)
return c._classic_contract.events[event_name]().process_receipt(tx_receipt)
return get_logs
This fixture will return a tuple with all the logs for a certain event and transaction. The length of the tuple equals the number of events (of the specified type) logged and should be checked first.
Finally, we create a new file test_advanced_storage.py
where we use the new fixtures to test failed transactions and events.
1import pytest
2from web3.exceptions import ValidationError
3
4INITIAL_VALUE = 4
5
6
7@pytest.fixture
8def adv_storage_contract(w3, get_contract):
9 with open("examples/storage/advanced_storage.vy") as f:
10 contract_code = f.read()
11 # Pass constructor variables directly to the contract
12 contract = get_contract(contract_code, INITIAL_VALUE)
13 return contract
14
15
16def test_initial_state(adv_storage_contract):
17 # Check if the constructor of the contract is set up properly
18 assert adv_storage_contract.storedData() == INITIAL_VALUE
19
20
21def test_failed_transactions(w3, adv_storage_contract, assert_tx_failed):
22 k1 = w3.eth.accounts[1]
23
24 # Try to set the storage to a negative amount
25 assert_tx_failed(lambda: adv_storage_contract.set(-10, transact={"from": k1}))
26
27 # Lock the contract by storing more than 100. Then try to change the value
28 adv_storage_contract.set(150, transact={"from": k1})
29 assert_tx_failed(lambda: adv_storage_contract.set(10, transact={"from": k1}))
30
31 # Reset the contract and try to change the value
32 adv_storage_contract.reset(transact={"from": k1})
33 adv_storage_contract.set(10, transact={"from": k1})
34 assert adv_storage_contract.storedData() == 10
35
36 # Assert a different exception (ValidationError for non matching argument type)
37 assert_tx_failed(
38 lambda: adv_storage_contract.set("foo", transact={"from": k1}), ValidationError
39 )
40
41 # Assert a different exception that contains specific text
42 assert_tx_failed(
43 lambda: adv_storage_contract.set(1, 2, transact={"from": k1}),
44 ValidationError,
45 "invocation failed due to improper number of arguments",
46 )
47
48
49def test_events(w3, adv_storage_contract, get_logs):
50 k1, k2 = w3.eth.accounts[:2]
51
52 tx1 = adv_storage_contract.set(10, transact={"from": k1})
53 tx2 = adv_storage_contract.set(20, transact={"from": k2})
54 tx3 = adv_storage_contract.reset(transact={"from": k1})
55
56 # Save DataChange logs from all three transactions
57 logs1 = get_logs(tx1, adv_storage_contract, "DataChange")
58 logs2 = get_logs(tx2, adv_storage_contract, "DataChange")
59 logs3 = get_logs(tx3, adv_storage_contract, "DataChange")
60
61 # Check log contents
62 assert len(logs1) == 1
63 assert logs1[0].args.value == 10
64
65 assert len(logs2) == 1
66 assert logs2[0].args.setter == k2
67
68 assert not logs3 # tx3 does not generate a log