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.

conftest.py
  1import json
  2import logging
  3from contextlib import contextmanager
  4from functools import wraps
  5
  6import hypothesis
  7import pytest
  8import web3.exceptions
  9from eth_tester import EthereumTester, PyEVMBackend
 10from eth_tester.exceptions import TransactionFailed
 11from eth_utils import setup_DEBUG2_logging
 12from eth_utils.toolz import compose
 13from hexbytes import HexBytes
 14from web3 import Web3
 15from web3.contract import Contract
 16from web3.providers.eth_tester import EthereumTesterProvider
 17
 18from tests.utils import working_directory
 19from vyper import compiler
 20from vyper.ast.grammar import parse_vyper_source
 21from vyper.codegen.ir_node import IRnode
 22from vyper.compiler.input_bundle import FilesystemInputBundle, InputBundle
 23from vyper.compiler.settings import OptimizationLevel, Settings, _set_debug_mode
 24from vyper.ir import compile_ir, optimizer
 25from vyper.utils import ERC5202_PREFIX
 26
 27# Import the base fixtures
 28pytest_plugins = ["tests.fixtures.memorymock"]
 29
 30############
 31# PATCHING #
 32############
 33
 34
 35# disable hypothesis deadline globally
 36hypothesis.settings.register_profile("ci", deadline=None)
 37hypothesis.settings.load_profile("ci")
 38
 39
 40def set_evm_verbose_logging():
 41    logger = logging.getLogger("eth.vm.computation.Computation")
 42    setup_DEBUG2_logging()
 43    logger.setLevel("DEBUG2")
 44
 45
 46# Useful options to comment out whilst working:
 47# set_evm_verbose_logging()
 48#
 49# from vdb import vdb
 50# vdb.set_evm_opcode_debugger()
 51
 52
 53def pytest_addoption(parser):
 54    parser.addoption(
 55        "--optimize",
 56        choices=["codesize", "gas", "none"],
 57        default="gas",
 58        help="change optimization mode",
 59    )
 60    parser.addoption("--enable-compiler-debug-mode", action="store_true")
 61
 62
 63@pytest.fixture(scope="module")
 64def output_formats():
 65    output_formats = compiler.OUTPUT_FORMATS.copy()
 66    del output_formats["bb"]
 67    del output_formats["bb_runtime"]
 68    return output_formats
 69
 70
 71@pytest.fixture(scope="module")
 72def optimize(pytestconfig):
 73    flag = pytestconfig.getoption("optimize")
 74    return OptimizationLevel.from_string(flag)
 75
 76
 77@pytest.fixture(scope="session", autouse=True)
 78def debug(pytestconfig):
 79    debug = pytestconfig.getoption("enable_compiler_debug_mode")
 80    assert isinstance(debug, bool)
 81    _set_debug_mode(debug)
 82
 83
 84@pytest.fixture
 85def chdir_tmp_path(tmp_path):
 86    # this is useful for when you want imports to have relpaths
 87    with working_directory(tmp_path):
 88        yield
 89
 90
 91@pytest.fixture
 92def keccak():
 93    return Web3.keccak
 94
 95
 96@pytest.fixture
 97def make_file(tmp_path):
 98    # writes file_contents to file_name, creating it in the
 99    # tmp_path directory. returns final path.
100    def fn(file_name, file_contents):
101        path = tmp_path / file_name
102        path.parent.mkdir(parents=True, exist_ok=True)
103        with path.open("w") as f:
104            f.write(file_contents)
105
106        return path
107
108    return fn
109
110
111# this can either be used for its side effects (to prepare a call
112# to get_contract), or the result can be provided directly to
113# compile_code / CompilerData.
114@pytest.fixture
115def make_input_bundle(tmp_path, make_file):
116    def fn(sources_dict):
117        for file_name, file_contents in sources_dict.items():
118            make_file(file_name, file_contents)
119        return FilesystemInputBundle([tmp_path])
120
121    return fn
122
123
124# for tests which just need an input bundle, doesn't matter what it is
125@pytest.fixture
126def dummy_input_bundle():
127    return InputBundle([])
128
129
130# TODO: remove me, this is just string.encode("utf-8").ljust()
131# only used in test_logging.py.
132@pytest.fixture
133def bytes_helper():
134    def bytes_helper(str, length):
135        return bytes(str, "utf-8") + bytearray(length - len(str))
136
137    return bytes_helper
138
139
140def _none_addr(datatype, data):
141    if datatype == "address" and int(data, base=16) == 0:
142        return (datatype, None)
143    else:
144        return (datatype, data)
145
146
147CONCISE_NORMALIZERS = (_none_addr,)
148
149
150@pytest.fixture(scope="module")
151def tester():
152    # set absurdly high gas limit so that london basefee never adjusts
153    # (note: 2**63 - 1 is max that evm allows)
154    custom_genesis = PyEVMBackend._generate_genesis_params(overrides={"gas_limit": 10**10})
155    custom_genesis["base_fee_per_gas"] = 0
156    backend = PyEVMBackend(genesis_parameters=custom_genesis)
157    return EthereumTester(backend=backend)
158
159
160def zero_gas_price_strategy(web3, transaction_params=None):
161    return 0  # zero gas price makes testing simpler.
162
163
164@pytest.fixture(scope="module")
165def w3(tester):
166    w3 = Web3(EthereumTesterProvider(tester))
167    w3.eth.set_gas_price_strategy(zero_gas_price_strategy)
168    return w3
169
170
171def get_compiler_gas_estimate(code, func):
172    sigs = compiler.phases.CompilerData(code).function_signatures
173    if func:
174        return compiler.utils.build_gas_estimates(sigs)[func] + 22000
175    else:
176        return sum(compiler.utils.build_gas_estimates(sigs).values()) + 22000
177
178
179def check_gas_on_chain(w3, tester, code, func=None, res=None):
180    gas_estimate = get_compiler_gas_estimate(code, func)
181    gas_actual = tester.get_block_by_number("latest")["gas_used"]
182    # Computed upper bound on the gas consumption should
183    # be greater than or equal to the amount of gas used
184    if gas_estimate < gas_actual:
185        raise Exception(f"Gas upper bound fail: bound {gas_estimate} actual {gas_actual}")
186
187    print(f"Function name: {func} - Gas estimate {gas_estimate}, Actual: {gas_actual}")
188
189
190def gas_estimation_decorator(w3, tester, fn, source_code, func):
191    def decorator(*args, **kwargs):
192        @wraps(fn)
193        def decorated_function(*args, **kwargs):
194            result = fn(*args, **kwargs)
195            if "transact" in kwargs:
196                check_gas_on_chain(w3, tester, source_code, func, res=result)
197            return result
198
199        return decorated_function(*args, **kwargs)
200
201    return decorator
202
203
204def set_decorator_to_contract_function(w3, tester, contract, source_code, func):
205    func_definition = getattr(contract, func)
206    func_with_decorator = gas_estimation_decorator(w3, tester, func_definition, source_code, func)
207    setattr(contract, func, func_with_decorator)
208
209
210class VyperMethod:
211    ALLOWED_MODIFIERS = {"call", "estimateGas", "transact", "buildTransaction"}
212
213    def __init__(self, function, normalizers=None):
214        self._function = function
215        self._function._return_data_normalizers = normalizers
216
217    def __call__(self, *args, **kwargs):
218        return self.__prepared_function(*args, **kwargs)
219
220    def __prepared_function(self, *args, **kwargs):
221        if not kwargs:
222            modifier, modifier_dict = "call", {}
223            fn_abi = [
224                x
225                for x in self._function.contract_abi
226                if x.get("name") == self._function.function_identifier
227            ].pop()
228            # To make tests faster just supply some high gas value.
229            modifier_dict.update({"gas": fn_abi.get("gas", 0) + 500000})
230        elif len(kwargs) == 1:
231            modifier, modifier_dict = kwargs.popitem()
232            if modifier not in self.ALLOWED_MODIFIERS:
233                raise TypeError(f"The only allowed keyword arguments are: {self.ALLOWED_MODIFIERS}")
234        else:
235            raise TypeError(f"Use up to one keyword argument, one of: {self.ALLOWED_MODIFIERS}")
236        return getattr(self._function(*args), modifier)(modifier_dict)
237
238
239class VyperContract:
240    """
241    An alternative Contract Factory which invokes all methods as `call()`,
242    unless you add a keyword argument. The keyword argument assigns the prep method.
243    This call
244    > contract.withdraw(amount, transact={'from': eth.accounts[1], 'gas': 100000, ...})
245    is equivalent to this call in the classic contract:
246    > contract.functions.withdraw(amount).transact({'from': eth.accounts[1], 'gas': 100000, ...})
247    """
248
249    def __init__(self, classic_contract, method_class=VyperMethod):
250        classic_contract._return_data_normalizers += CONCISE_NORMALIZERS
251        self._classic_contract = classic_contract
252        self.address = self._classic_contract.address
253        protected_fn_names = [fn for fn in dir(self) if not fn.endswith("__")]
254
255        try:
256            fn_names = [fn["name"] for fn in self._classic_contract.functions._functions]
257        except web3.exceptions.NoABIFunctionsFound:
258            fn_names = []
259
260        for fn_name in fn_names:
261            # Override namespace collisions
262            if fn_name in protected_fn_names:
263                raise AttributeError(f"{fn_name} is protected!")
264            else:
265                _classic_method = getattr(self._classic_contract.functions, fn_name)
266                _concise_method = method_class(
267                    _classic_method, self._classic_contract._return_data_normalizers
268                )
269            setattr(self, fn_name, _concise_method)
270
271    @classmethod
272    def factory(cls, *args, **kwargs):
273        return compose(cls, Contract.factory(*args, **kwargs))
274
275
276@pytest.fixture
277def get_contract_from_ir(w3, optimize):
278    def ir_compiler(ir, *args, **kwargs):
279        ir = IRnode.from_list(ir)
280        if optimize != OptimizationLevel.NONE:
281            ir = optimizer.optimize(ir)
282
283        bytecode, _ = compile_ir.assembly_to_evm(
284            compile_ir.compile_to_assembly(ir, optimize=optimize)
285        )
286
287        abi = kwargs.get("abi") or []
288        c = w3.eth.contract(abi=abi, bytecode=bytecode)
289        deploy_transaction = c.constructor()
290        tx_hash = deploy_transaction.transact()
291        address = w3.eth.get_transaction_receipt(tx_hash)["contractAddress"]
292        contract = w3.eth.contract(
293            address, abi=abi, bytecode=bytecode, ContractFactoryClass=VyperContract
294        )
295        return contract
296
297    return ir_compiler
298
299
300def _get_contract(
301    w3,
302    source_code,
303    optimize,
304    output_formats,
305    *args,
306    override_opt_level=None,
307    input_bundle=None,
308    **kwargs,
309):
310    settings = Settings()
311    settings.evm_version = kwargs.pop("evm_version", None)
312    settings.optimize = override_opt_level or optimize
313    out = compiler.compile_code(
314        source_code,
315        # test that all output formats can get generated
316        output_formats=output_formats,
317        settings=settings,
318        input_bundle=input_bundle,
319        show_gas_estimates=True,  # Enable gas estimates for testing
320    )
321    parse_vyper_source(source_code)  # Test grammar.
322    json.dumps(out["metadata"])  # test metadata is json serializable
323    abi = out["abi"]
324    bytecode = out["bytecode"]
325    value = kwargs.pop("value_in_eth", 0) * 10**18  # Handle deploying with an eth value.
326    c = w3.eth.contract(abi=abi, bytecode=bytecode)
327    deploy_transaction = c.constructor(*args)
328    tx_info = {"from": w3.eth.accounts[0], "value": value, "gasPrice": 0}
329    tx_info.update(kwargs)
330    tx_hash = deploy_transaction.transact(tx_info)
331    address = w3.eth.get_transaction_receipt(tx_hash)["contractAddress"]
332    return w3.eth.contract(address, abi=abi, bytecode=bytecode, ContractFactoryClass=VyperContract)
333
334
335@pytest.fixture(scope="module")
336def get_contract(w3, optimize, output_formats):
337    def fn(source_code, *args, **kwargs):
338        return _get_contract(w3, source_code, optimize, output_formats, *args, **kwargs)
339
340    return fn
341
342
343@pytest.fixture
344def get_contract_with_gas_estimation(tester, w3, optimize, output_formats):
345    def get_contract_with_gas_estimation(source_code, *args, **kwargs):
346        contract = _get_contract(w3, source_code, optimize, output_formats, *args, **kwargs)
347        for abi_ in contract._classic_contract.functions.abi:
348            if abi_["type"] == "function":
349                set_decorator_to_contract_function(w3, tester, contract, source_code, abi_["name"])
350        return contract
351
352    return get_contract_with_gas_estimation
353
354
355@pytest.fixture
356def get_contract_with_gas_estimation_for_constants(w3, optimize, output_formats):
357    def get_contract_with_gas_estimation_for_constants(source_code, *args, **kwargs):
358        return _get_contract(w3, source_code, optimize, output_formats, *args, **kwargs)
359
360    return get_contract_with_gas_estimation_for_constants
361
362
363@pytest.fixture(scope="module")
364def get_contract_module(optimize, output_formats):
365    """
366    This fixture is used for Hypothesis tests to ensure that
367    the same contract is called over multiple runs of the test.
368    """
369    custom_genesis = PyEVMBackend._generate_genesis_params(overrides={"gas_limit": 4500000})
370    custom_genesis["base_fee_per_gas"] = 0
371    backend = PyEVMBackend(genesis_parameters=custom_genesis)
372    tester = EthereumTester(backend=backend)
373    w3 = Web3(EthereumTesterProvider(tester))
374    w3.eth.set_gas_price_strategy(zero_gas_price_strategy)
375
376    def get_contract_module(source_code, *args, **kwargs):
377        return _get_contract(w3, source_code, optimize, output_formats, *args, **kwargs)
378
379    return get_contract_module
380
381
382def _deploy_blueprint_for(
383    w3, source_code, optimize, output_formats, initcode_prefix=ERC5202_PREFIX, **kwargs
384):
385    settings = Settings()
386    settings.evm_version = kwargs.pop("evm_version", None)
387    settings.optimize = optimize
388    out = compiler.compile_code(
389        source_code,
390        output_formats=output_formats,
391        settings=settings,
392        show_gas_estimates=True,  # Enable gas estimates for testing
393    )
394    parse_vyper_source(source_code)  # Test grammar.
395    abi = out["abi"]
396    bytecode = HexBytes(initcode_prefix) + HexBytes(out["bytecode"])
397    bytecode_len = len(bytecode)
398    bytecode_len_hex = hex(bytecode_len)[2:].rjust(4, "0")
399    # prepend a quick deploy preamble
400    deploy_preamble = HexBytes("61" + bytecode_len_hex + "3d81600a3d39f3")
401    deploy_bytecode = HexBytes(deploy_preamble) + bytecode
402
403    deployer_abi = []  # just a constructor
404    c = w3.eth.contract(abi=deployer_abi, bytecode=deploy_bytecode)
405    deploy_transaction = c.constructor()
406    tx_info = {"from": w3.eth.accounts[0], "value": 0, "gasPrice": 0}
407
408    tx_hash = deploy_transaction.transact(tx_info)
409    address = w3.eth.get_transaction_receipt(tx_hash)["contractAddress"]
410
411    # sanity check
412    assert w3.eth.get_code(address) == bytecode, (w3.eth.get_code(address), bytecode)
413
414    def factory(address):
415        return w3.eth.contract(
416            address, abi=abi, bytecode=bytecode, ContractFactoryClass=VyperContract
417        )
418
419    return w3.eth.contract(address, bytecode=deploy_bytecode), factory
420
421
422@pytest.fixture(scope="module")
423def deploy_blueprint_for(w3, optimize, output_formats):
424    def deploy_blueprint_for(source_code, *args, **kwargs):
425        return _deploy_blueprint_for(w3, source_code, optimize, output_formats, *args, **kwargs)
426
427    return deploy_blueprint_for
428
429
430# TODO: this should not be a fixture.
431# remove me and replace all uses with `with pytest.raises`.
432@pytest.fixture
433def assert_compile_failed():
434    def assert_compile_failed(function_to_test, exception=Exception):
435        with pytest.raises(exception):
436            function_to_test()
437
438    return assert_compile_failed
439
440
441@pytest.fixture
442def create2_address_of(keccak):
443    def _f(_addr, _salt, _initcode):
444        prefix = HexBytes("0xff")
445        addr = HexBytes(_addr)
446        salt = HexBytes(_salt)
447        initcode = HexBytes(_initcode)
448        return keccak(prefix + addr + salt + keccak(initcode))[12:]
449
450    return _f
451
452
453@pytest.fixture
454def side_effects_contract(get_contract):
455    def generate(ret_type):
456        """
457        Generates a Vyper contract with an external `foo()` function, which
458        returns the specified return value of the specified return type, for
459        testing side effects using the `assert_side_effects_invoked` fixture.
460        """
461        code = f"""
462counter: public(uint256)
463
464@external
465def foo(s: {ret_type}) -> {ret_type}:
466    self.counter += 1
467    return s
468    """
469        contract = get_contract(code)
470        return contract
471
472    return generate
473
474
475@pytest.fixture
476def assert_side_effects_invoked():
477    def assert_side_effects_invoked(side_effects_contract, side_effects_trigger, n=1):
478        start_value = side_effects_contract.counter()
479
480        side_effects_trigger()
481
482        end_value = side_effects_contract.counter()
483        assert end_value == start_value + n
484
485    return assert_side_effects_invoked
486
487
488@pytest.fixture
489def get_logs(w3):
490    def get_logs(tx_hash, c, event_name):
491        tx_receipt = w3.eth.get_transaction_receipt(tx_hash)
492        return c._classic_contract.events[event_name]().process_receipt(tx_receipt)
493
494    return get_logs
495
496
497@pytest.fixture(scope="module")
498def tx_failed(tester):
499    @contextmanager
500    def fn(exception=TransactionFailed, exc_text=None):
501        snapshot_id = tester.take_snapshot()
502        with pytest.raises(exception) as excinfo:
503            yield excinfo
504        tester.revert_to_snapshot(snapshot_id)
505        if exc_text:
506            # TODO test equality
507            assert exc_text in str(excinfo.value), (exc_text, excinfo.value)
508
509    return fn

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.

storage.vy
 1#pragma version >0.3.10
 2
 3storedData: public(int128)
 4
 5@deploy
 6def __init__(_x: int128):
 7  self.storedData = _x
 8
 9@external
10def set(_x: int128):
11  self.storedData = _x

We create a test file test_storage.py where we write our tests in pytest style.

test_storage.py
 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

advanced_storage.vy
 1#pragma version >0.3.10
 2
 3event DataChange:
 4    setter: indexed(address)
 5    value: int128
 6
 7storedData: public(int128)
 8
 9@deploy
10def __init__(_x: int128):
11  self.storedData = _x
12
13@external
14def set(_x: int128):
15  assert _x >= 0, "No negative values"
16  assert self.storedData < 100, "Storage is locked when 100 or more is stored"
17  self.storedData = _x
18  log DataChange(msg.sender, _x)
19
20@external
21def reset():
22  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.

conftest.py
@pytest.fixture(scope="module")
def tx_failed(tester):
    @contextmanager
    def fn(exception=TransactionFailed, exc_text=None):
        snapshot_id = tester.take_snapshot()
        with pytest.raises(exception) as excinfo:
            yield excinfo
        tester.revert_to_snapshot(snapshot_id)
        if exc_text:
            # TODO test equality
            assert exc_text in str(excinfo.value), (exc_text, excinfo.value)

    return fn

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.

conftest.py
@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.

test_advanced_storage.py
 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, tx_failed):
22    k1 = w3.eth.accounts[1]
23
24    # Try to set the storage to a negative amount
25    with tx_failed():
26        adv_storage_contract.set(-10, transact={"from": k1})
27
28    # Lock the contract by storing more than 100. Then try to change the value
29    adv_storage_contract.set(150, transact={"from": k1})
30    with tx_failed():
31        adv_storage_contract.set(10, transact={"from": k1})
32
33    # Reset the contract and try to change the value
34    adv_storage_contract.reset(transact={"from": k1})
35    adv_storage_contract.set(10, transact={"from": k1})
36    assert adv_storage_contract.storedData() == 10
37
38    # Assert a different exception (ValidationError for non-matching argument type)
39    with tx_failed(ValidationError):
40        adv_storage_contract.set("foo", transact={"from": k1})
41
42    # Assert a different exception that contains specific text
43    with tx_failed(ValidationError, "invocation failed due to improper number of arguments"):
44        adv_storage_contract.set(1, 2, transact={"from": k1})
45
46
47def test_events(w3, adv_storage_contract, get_logs):
48    k1, k2 = w3.eth.accounts[:2]
49
50    tx1 = adv_storage_contract.set(10, transact={"from": k1})
51    tx2 = adv_storage_contract.set(20, transact={"from": k2})
52    tx3 = adv_storage_contract.reset(transact={"from": k1})
53
54    # Save DataChange logs from all three transactions
55    logs1 = get_logs(tx1, adv_storage_contract, "DataChange")
56    logs2 = get_logs(tx2, adv_storage_contract, "DataChange")
57    logs3 = get_logs(tx3, adv_storage_contract, "DataChange")
58
59    # Check log contents
60    assert len(logs1) == 1
61    assert logs1[0].args.value == 10
62
63    assert len(logs2) == 1
64    assert logs2[0].args.setter == k2
65
66    assert not logs3  # tx3 does not generate a log