Testing a Contract

This documentation recommends the use of the pytest framework with the ethereum-tester package. 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.

Vyper Contract and Basic Fixtures

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
from eth_tester import (
    EthereumTester,
    PyEVMBackend,
)
from eth_tester.exceptions import (
    TransactionFailed,
)
from eth_utils.toolz import (
    compose,
)
import pytest
from web3 import Web3
from web3.contract import (
    Contract,
    mk_collision_prop,
)
from web3.providers.eth_tester import (
    EthereumTesterProvider,
)

from vyper import (
    compiler,
)

from .grammar.conftest import (
    get_lark_grammar,
)

LARK_GRAMMAR = get_lark_grammar()


class VyperMethod:
    ALLOWED_MODIFIERS = {'call', 'estimateGas', 'transact', 'buildTransaction'}

    def __init__(self, function, normalizers=None):
        self._function = function
        self._function._return_data_normalizers = normalizers

    def __call__(self, *args, **kwargs):
        return self.__prepared_function(*args, **kwargs)

    def __prepared_function(self, *args, **kwargs):
        if not kwargs:
            modifier, modifier_dict = 'call', {}
            fn_abi = [
                x
                for x
                in self._function.contract_abi
                if x.get('name') == self._function.function_identifier
            ].pop()
            # To make tests faster just supply some high gas value.
            modifier_dict.update({'gas': fn_abi.get('gas', 0) + 50000})
        elif len(kwargs) == 1:
            modifier, modifier_dict = kwargs.popitem()
            if modifier not in self.ALLOWED_MODIFIERS:
                raise TypeError(
                    f"The only allowed keyword arguments are: {self.ALLOWED_MODIFIERS}")
        else:
            raise TypeError(f"Use up to one keyword argument, one of: {self.ALLOWED_MODIFIERS}")
        return getattr(self._function(*args), modifier)(modifier_dict)


class VyperContract:
    """
    An alternative Contract Factory which invokes all methods as `call()`,
    unless you add a keyword argument. The keyword argument assigns the prep method.
    This call
    > contract.withdraw(amount, transact={'from': eth.accounts[1], 'gas': 100000, ...})
    is equivalent to this call in the classic contract:
    > contract.functions.withdraw(amount).transact({'from': eth.accounts[1], 'gas': 100000, ...})
    """

    def __init__(self, classic_contract, method_class=VyperMethod):
        classic_contract._return_data_normalizers += CONCISE_NORMALIZERS
        self._classic_contract = classic_contract
        self.address = self._classic_contract.address
        protected_fn_names = [fn for fn in dir(self) if not fn.endswith('__')]
        for fn_name in self._classic_contract.functions:
            # Override namespace collisions
            if fn_name in protected_fn_names:
                _concise_method = mk_collision_prop(fn_name)
            else:
                _classic_method = getattr(
                    self._classic_contract.functions,
                    fn_name)
                _concise_method = method_class(
                    _classic_method,
                    self._classic_contract._return_data_normalizers
                )
            setattr(self, fn_name, _concise_method)

    @classmethod
    def factory(cls, *args, **kwargs):
        return compose(cls, Contract.factory(*args, **kwargs))


def _none_addr(datatype, data):
    if datatype == 'address' and int(data, base=16) == 0:
        return (datatype, None)
    else:
        return (datatype, data)


CONCISE_NORMALIZERS = (_none_addr,)


@pytest.fixture
def tester():
    custom_genesis = PyEVMBackend._generate_genesis_params(overrides={'gas_limit': 4500000})
    backend = PyEVMBackend(genesis_parameters=custom_genesis)
    return EthereumTester(backend=backend)


def zero_gas_price_strategy(web3, transaction_params=None):
    return 0  # zero gas price makes testing simpler.


@pytest.fixture
def w3(tester):
    w3 = Web3(EthereumTesterProvider(tester))
    w3.eth.setGasPriceStrategy(zero_gas_price_strategy)
    return w3


def _get_contract(w3, source_code, *args, **kwargs):
    out = compiler.compile_code(
        source_code,
        ['abi', 'bytecode'],
        interface_codes=kwargs.pop('interface_codes', None),
        evm_version=kwargs.pop('evm_version', None),
    )
    LARK_GRAMMAR.parse(source_code + "\n")  # Test grammar.
    abi = out['abi']
    bytecode = out['bytecode']
    value = kwargs.pop('value_in_eth', 0) * 10 ** 18  # Handle deploying with an eth value.
    c = w3.eth.contract(abi=abi, bytecode=bytecode)
    deploy_transaction = c.constructor(*args)
    tx_info = {
        'from': w3.eth.accounts[0],
        'value': value,
        'gasPrice': 0,
    }
    tx_info.update(kwargs)
    tx_hash = deploy_transaction.transact(tx_info)
    address = w3.eth.getTransactionReceipt(tx_hash)['contractAddress']
    contract = w3.eth.contract(
        address,
        abi=abi,
        bytecode=bytecode,
        ContractFactoryClass=VyperContract,
    )
    return contract


@pytest.fixture
def get_contract(w3):
    def get_contract(source_code, *args, **kwargs):
        return _get_contract(w3, source_code, *args, **kwargs)

    return get_contract


@pytest.fixture
def get_logs(w3):
    def get_logs(tx_hash, c, event_name):
        tx_receipt = w3.eth.getTransactionReceipt(tx_hash)
        logs = c._classic_contract.events[event_name]().processReceipt(tx_receipt)
        return logs

    return get_logs


@pytest.fixture
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:
            assert exc_text in str(excinfo.value)

    return assert_tx_failed

This is the base requirement to load a vyper contract and start testing. The last 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.

Load Contract and Basic Tests

Assume the following simple contract storage.vy. It has a single integer variable and a function to set that value.

storedData: public(int128)

@public
def __init__(_x: int128):
  self.storedData = _x

@public
def set(_x: int128):
  self.storedData = _x

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

import pytest

INITIAL_VALUE = 4


@pytest.fixture
def storage_contract(w3, get_contract):
    with open('examples/storage/storage.vy') as f:
        contract_code = f.read()
        # Pass constructor variables directly to the contract
        contract = get_contract(contract_code, INITIAL_VALUE)
    return contract


def test_initial_state(storage_contract):
    # Check if the constructor of the contract is set up properly
    assert storage_contract.storedData() == INITIAL_VALUE


def test_set(w3, storage_contract):
    k0 = w3.eth.accounts[0]

    # Let k0 try to set the value to 10
    storage_contract.set(10, transact={"from": k0})
    assert storage_contract.storedData() == 10  # Directly access storedData

    # Let k0 try to set the value to -5
    storage_contract.set(-5, transact={"from": k0})
    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

DataChange: event({_setter: indexed(address), _value: int128})

storedData: public(int128)

@public
def __init__(_x: int128):
  self.storedData = _x

@public
def set(_x: int128):
  assert _x >= 0 # No negative values
  assert self.storedData < 100  # Storage will lock when 100 or more is stored
  self.storedData = _x
  log.DataChange(msg.sender, _x)

@public
def reset():
  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
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:
            assert exc_text in str(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.getTransactionReceipt(tx_hash)
        logs = c._classic_contract.events[event_name]().processReceipt(tx_receipt)
        return logs

    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.

import pytest
from web3.exceptions import (
    ValidationError,
)

INITIAL_VALUE = 4


@pytest.fixture
def adv_storage_contract(w3, get_contract):
    with open('examples/storage/advanced_storage.vy') as f:
        contract_code = f.read()
        # Pass constructor variables directly to the contract
        contract = get_contract(contract_code, INITIAL_VALUE)
    return contract


def test_initial_state(adv_storage_contract):
    # Check if the constructor of the contract is set up properly
    assert adv_storage_contract.storedData() == INITIAL_VALUE


def test_failed_transactions(w3, adv_storage_contract, assert_tx_failed):
    k1 = w3.eth.accounts[1]

    # Try to set the storage to a negative amount
    assert_tx_failed(lambda: adv_storage_contract.set(-10, transact={"from": k1}))

    # Lock the contract by storing more than 100. Then try to change the value
    adv_storage_contract.set(150, transact={"from": k1})
    assert_tx_failed(lambda: adv_storage_contract.set(10, transact={"from": k1}))

    # Reset the contract and try to change the value
    adv_storage_contract.reset(transact={"from": k1})
    adv_storage_contract.set(10, transact={"from": k1})
    assert adv_storage_contract.storedData() == 10

    # Assert a different exception (ValidationError for non matching argument type)
    assert_tx_failed(
        lambda: adv_storage_contract.set("foo", transact={"from": k1}),
        ValidationError
    )

    # Assert a different exception that contains specific text
    assert_tx_failed(
        lambda: adv_storage_contract.set(1, 2, transact={"from": k1}),
        ValidationError,
        "invocation failed due to improper number of arguments",
     )


def test_events(w3, adv_storage_contract, get_logs):
    k1, k2 = w3.eth.accounts[:2]

    tx1 = adv_storage_contract.set(10, transact={"from": k1})
    tx2 = adv_storage_contract.set(20, transact={"from": k2})
    tx3 = adv_storage_contract.reset(transact={"from": k1})

    # Save DataChange logs from all three transactions
    logs1 = get_logs(tx1, adv_storage_contract, "DataChange")
    logs2 = get_logs(tx2, adv_storage_contract, "DataChange")
    logs3 = get_logs(tx3, adv_storage_contract, "DataChange")

    # Check log contents
    assert len(logs1) == 1
    assert logs1[0].args._value == 10

    assert len(logs2) == 1
    assert logs2[0].args._setter == k2

    assert not logs3   # tx3 does not generate a log