使用以太坊测试器进行测试

以太坊测试器 是一款用于测试基于以太坊的应用程序的工具套件。

本节简要概述了使用 eth-tester 进行测试的方法。如需了解更多信息，您可以查看 Github 仓库 中的文档或加入 Gitter 频道。

入门

在测试之前，需要设置 Vyper 特定的合约转换和区块链相关的固定装置。这些固定装置将在每个测试文件中使用，因此应该在 conftest.py 中定义。

注意

由于测试是在 pytest 框架中完成的，因此您可以使用 pytest.ini、tox.ini 和 setup.cfg，并且可以使用大多数 IDE 的 pytest 插件。

  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
 15from vyper.compiler.settings import Settings
 16
 17
 18class VyperMethod:
 19    ALLOWED_MODIFIERS = {"call", "estimateGas", "transact", "buildTransaction"}
 20
 21    def __init__(self, function, normalizers=None):
 22        self._function = function
 23        self._function._return_data_normalizers = normalizers
 24
 25    def __call__(self, *args, **kwargs):
 26        return self.__prepared_function(*args, **kwargs)
 27
 28    def __prepared_function(self, *args, **kwargs):
 29        if not kwargs:
 30            modifier, modifier_dict = "call", {}
 31            fn_abi = [
 32                x
 33                for x in self._function.contract_abi
 34                if x.get("name") == self._function.function_identifier
 35            ].pop()
 36            # To make tests faster just supply some high gas value.
 37            modifier_dict.update({"gas": fn_abi.get("gas", 0) + 500000})
 38        elif len(kwargs) == 1:
 39            modifier, modifier_dict = kwargs.popitem()
 40            if modifier not in self.ALLOWED_MODIFIERS:
 41                raise TypeError(f"The only allowed keyword arguments are: {self.ALLOWED_MODIFIERS}")
 42        else:
 43            raise TypeError(f"Use up to one keyword argument, one of: {self.ALLOWED_MODIFIERS}")
 44        return getattr(self._function(*args), modifier)(modifier_dict)
 45
 46
 47class VyperContract:
 48    """
 49    An alternative Contract Factory which invokes all methods as `call()`,
 50    unless you add a keyword argument. The keyword argument assigns the prep method.
 51    This call
 52    > contract.withdraw(amount, transact={'from': eth.accounts[1], 'gas': 100000, ...})
 53    is equivalent to this call in the classic contract:
 54    > contract.functions.withdraw(amount).transact({'from': eth.accounts[1], 'gas': 100000, ...})
 55    """
 56
 57    def __init__(self, classic_contract, method_class=VyperMethod):
 58        classic_contract._return_data_normalizers += CONCISE_NORMALIZERS
 59        self._classic_contract = classic_contract
 60        self.address = self._classic_contract.address
 61        protected_fn_names = [fn for fn in dir(self) if not fn.endswith("__")]
 62
 63        try:
 64            fn_names = [fn["name"] for fn in self._classic_contract.functions._functions]
 65        except web3.exceptions.NoABIFunctionsFound:
 66            fn_names = []
 67
 68        for fn_name in fn_names:
 69            # Override namespace collisions
 70            if fn_name in protected_fn_names:
 71                raise AttributeError(f"{fn_name} is protected!")
 72            else:
 73                _classic_method = getattr(self._classic_contract.functions, fn_name)
 74                _concise_method = method_class(
 75                    _classic_method, self._classic_contract._return_data_normalizers
 76                )
 77            setattr(self, fn_name, _concise_method)
 78
 79    @classmethod
 80    def factory(cls, *args, **kwargs):
 81        return compose(cls, Contract.factory(*args, **kwargs))
 82
 83
 84def _none_addr(datatype, data):
 85    if datatype == "address" and int(data, base=16) == 0:
 86        return (datatype, None)
 87    else:
 88        return (datatype, data)
 89
 90
 91CONCISE_NORMALIZERS = (_none_addr,)
 92
 93
 94@pytest.fixture(scope="module")
 95def tester():
 96    # set absurdly high gas limit so that london basefee never adjusts
 97    # (note: 2**63 - 1 is max that evm allows)
 98    custom_genesis = PyEVMBackend._generate_genesis_params(overrides={"gas_limit": 10**10})
 99    custom_genesis["base_fee_per_gas"] = 0
100    backend = PyEVMBackend(genesis_parameters=custom_genesis)
101    return EthereumTester(backend=backend)
102
103
104def zero_gas_price_strategy(web3, transaction_params=None):
105    return 0  # zero gas price makes testing simpler.
106
107
108@pytest.fixture(scope="module")
109def w3(tester):
110    w3 = Web3(EthereumTesterProvider(tester))
111    w3.eth.set_gas_price_strategy(zero_gas_price_strategy)
112    return w3
113
114
115def _get_contract(w3, source_code, optimize, *args, override_opt_level=None, **kwargs):
116    settings = Settings()
117    settings.evm_version = kwargs.pop("evm_version", None)
118    settings.optimize = override_opt_level or optimize
119    out = compiler.compile_code(
120        source_code,
121        # test that metadata and natspecs get generated
122        ["abi", "bytecode", "metadata", "userdoc", "devdoc"],
123        settings=settings,
124        interface_codes=kwargs.pop("interface_codes", None),
125        show_gas_estimates=True,  # Enable gas estimates for testing
126    )
127    parse_vyper_source(source_code)  # Test grammar.
128    json.dumps(out["metadata"])  # test metadata is json serializable
129    abi = out["abi"]
130    bytecode = out["bytecode"]
131    value = kwargs.pop("value_in_eth", 0) * 10**18  # Handle deploying with an eth value.
132    c = w3.eth.contract(abi=abi, bytecode=bytecode)
133    deploy_transaction = c.constructor(*args)
134    tx_info = {"from": w3.eth.accounts[0], "value": value, "gasPrice": 0}
135    tx_info.update(kwargs)
136    tx_hash = deploy_transaction.transact(tx_info)
137    address = w3.eth.get_transaction_receipt(tx_hash)["contractAddress"]
138    return w3.eth.contract(address, abi=abi, bytecode=bytecode, ContractFactoryClass=VyperContract)
139
140
141def _deploy_blueprint_for(w3, source_code, optimize, initcode_prefix=b"", **kwargs):
142    settings = Settings()
143    settings.evm_version = kwargs.pop("evm_version", None)
144    settings.optimize = optimize
145    out = compiler.compile_code(
146        source_code,
147        ["abi", "bytecode"],
148        interface_codes=kwargs.pop("interface_codes", None),
149        settings=settings,
150        show_gas_estimates=True,  # Enable gas estimates for testing
151    )
152    parse_vyper_source(source_code)  # Test grammar.
153    abi = out["abi"]
154    bytecode = HexBytes(initcode_prefix) + HexBytes(out["bytecode"])
155    bytecode_len = len(bytecode)
156    bytecode_len_hex = hex(bytecode_len)[2:].rjust(4, "0")
157    # prepend a quick deploy preamble
158    deploy_preamble = HexBytes("61" + bytecode_len_hex + "3d81600a3d39f3")
159    deploy_bytecode = HexBytes(deploy_preamble) + bytecode
160
161    deployer_abi = []  # just a constructor
162    c = w3.eth.contract(abi=deployer_abi, bytecode=deploy_bytecode)
163    deploy_transaction = c.constructor()
164    tx_info = {"from": w3.eth.accounts[0], "value": 0, "gasPrice": 0}
165
166    tx_hash = deploy_transaction.transact(tx_info)
167    address = w3.eth.get_transaction_receipt(tx_hash)["contractAddress"]
168
169    # sanity check
170    assert w3.eth.get_code(address) == bytecode, (w3.eth.get_code(address), bytecode)
171
172    def factory(address):
173        return w3.eth.contract(
174            address, abi=abi, bytecode=bytecode, ContractFactoryClass=VyperContract
175        )
176
177    return w3.eth.contract(address, bytecode=deploy_bytecode), factory
178
179
180@pytest.fixture(scope="module")
181def deploy_blueprint_for(w3, optimize):
182    def deploy_blueprint_for(source_code, *args, **kwargs):
183        return _deploy_blueprint_for(w3, source_code, optimize, *args, **kwargs)
184
185    return deploy_blueprint_for
186
187
188@pytest.fixture(scope="module")
189def get_contract(w3, optimize):
190    def get_contract(source_code, *args, **kwargs):
191        return _get_contract(w3, source_code, optimize, *args, **kwargs)
192
193    return get_contract
194
195
196@pytest.fixture
197def get_logs(w3):
198    def get_logs(tx_hash, c, event_name):
199        tx_receipt = w3.eth.get_transaction_receipt(tx_hash)
200        return c._classic_contract.events[event_name]().process_receipt(tx_receipt)
201
202    return get_logs
203
204
205@pytest.fixture(scope="module")
206def assert_tx_failed(tester):
207    def assert_tx_failed(function_to_test, exception=TransactionFailed, exc_text=None):
208        snapshot_id = tester.take_snapshot()
209        with pytest.raises(exception) as excinfo:
210            function_to_test()
211        tester.revert_to_snapshot(snapshot_id)
212        if exc_text:
213            # TODO test equality
214            assert exc_text in str(excinfo.value), (exc_text, excinfo.value)
215
216    return assert_tx_failed

最后两个固定装置是可选的，将在后面讨论。本章的其余部分假设您已在 conftest.py 文件中设置了此代码。

或者，您可以将固定装置导入到 conftest.py 中，或使用 pytest 插件。

编写基本测试

假设以下简单的合约 storage.vy。它有一个整数变量和一个用于设置该值的函数。

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

我们创建一个测试文件 test_storage.py，在其中使用 pytest 样式编写测试。

 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

首先，我们为合约创建一个固定装置，它将编译我们的合约并设置 Web3 合约对象。然后，我们将使用此固定装置让我们的测试函数与合约进行交互。

注意

要运行测试，请从项目目录调用 pytest 或 python -m pytest。

事件和失败的交易

为了测试事件和失败的交易，我们将扩展我们简单的存储合约，以包含一个事件和两个导致交易失败的条件：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

接下来，我们将看看两个固定装置，它们将允许我们读取事件日志并检查失败的交易。

@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

断言失败的交易的固定装置默认情况下会检查 TransactionFailed 异常，但也可以用于检查其他异常，如下所示。还要注意，链将恢复到失败交易之前的状态。

@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

此固定装置将返回一个包含特定事件和交易的所有日志的元组。元组的长度等于记录的事件（指定类型）的数量，应该首先进行检查。

最后，我们创建一个新文件 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, 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