10527345804

Committed 23 Aug 2024 02:20PM UTC coverage: 93.892% (-0.01%) from 93.903%

Build # 10527345804

Build Type

push

github

Committed by

web-flow

Commit Message

Add addresses 1.4.1 for chain BEVM_TESTNET (#1297)

* Add new master copy address 0x41675C099F32341bf84BFc5382aF534df5C7461a

* Add new master copy address 0x29fcB43b46531BcA003ddC8FCB67FFE91900C762

* Add new proxy address 0x4e1DCf7AD4e460CfD30791CCC4F9c8a4f820ec67

* Apply linter fixes

---------

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>

Run Details

8670 of 9234 relevant lines covered (93.89%)

3.76 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

67.07

/gnosis/eth/tests/utils.py

import functools
import json
import os
from copy import deepcopy
from typing import Any

import pytest
import requests
from eth_account.signers.local import LocalAccount
from eth_typing import ChecksumAddress
from hexbytes import HexBytes
from web3 import Web3
from web3.contract import Contract
from web3.types import TxParams

from ..contracts import get_example_erc20_contract


def just_test_if_node_available(node_url_variable_name: str) -> str:
    """
    Just run the test if ``node url`` is defined on the ``node_url_variable_name`` environment variable
    and it's accessible. Node JSON RPC url will only be tested the first call to this function.

    :param node_url_variable_name: Environment variable name for ``node url``
    :return: ``node url``
    """
    node_url = getattr(just_test_if_node_available, node_url_variable_name, None)
    if node_url:  # Just check node first time
        return node_url

    node_url = os.environ.get(node_url_variable_name)
    if not node_url:
        pytest.skip(
            f"{node_url_variable_name} not defined, skipping test",
            allow_module_level=True,
        )
    else:
        try:
            response = requests.post(
                node_url,
                timeout=5,
                json={
                    "jsonrpc": "2.0",
                    "method": "eth_blockNumber",
                    "params": [],
                    "id": 1,
                },
            )
            if not response.ok:
                pytest.fail(
                    f"Problem connecting to node {node_url}: {response.status_code} - {response.content}"
                )
        except IOError:
            pytest.fail(f"Problem connecting to {node_url}")
    setattr(just_test_if_node_available, node_url_variable_name, node_url)
    return node_url


def just_test_if_mainnet_node() -> str:
    return just_test_if_node_available("ETHEREUM_MAINNET_NODE")


def just_test_if_polygon_node() -> str:
    return just_test_if_node_available("ETHEREUM_POLYGON_NODE")


def skip_on(exception, reason="Test skipped due to a controlled exception"):
    """
    Decorator to skip a test if an exception is raised instead of failing it

    :param exception:
    :param reason:
    :return:
    """

    def decorator_func(f):
        @functools.wraps(f)
        def wrapper(*args, **kwargs):
            try:
                # Run the test
                return f(*args, **kwargs)
            except exception:
                pytest.skip(reason)

        return wrapper

    return decorator_func


# TODO Move this to EthereumClient
def send_tx(w3: Web3, tx: TxParams, account: LocalAccount) -> bytes:
    tx["from"] = account.address
    if "nonce" not in tx:
        tx["nonce"] = w3.eth.get_transaction_count(
            account.address, block_identifier="pending"
        )

    if "gasPrice" not in tx and "maxFeePerGas" not in tx:
        tx["gasPrice"] = w3.eth.gas_price

    if "gas" not in tx:
        tx["gas"] = w3.eth.estimate_gas(tx)

    signed_tx = account.sign_transaction(tx)
    tx_hash = w3.eth.send_raw_transaction(bytes(signed_tx.rawTransaction))
    tx_receipt = w3.eth.wait_for_transaction_receipt(tx_hash)
    assert tx_receipt["status"] == 1, "Error with tx %s - %s" % (tx_hash.hex(), tx)
    return tx_hash


def deploy_erc20(
    w3: Web3,
    account: LocalAccount,
    name: str,
    symbol: str,
    owner: ChecksumAddress,
    amount: int,
    decimals: int = 18,
) -> Contract:
    erc20_contract = get_example_erc20_contract(w3)
    tx = erc20_contract.constructor(
        name, symbol, decimals, owner, amount
    ).build_transaction(
        {
            "nonce": w3.eth.get_transaction_count(
                account.address, block_identifier="pending"
            )
        }
    )
    signed_tx = account.sign_transaction(tx)
    tx_hash = w3.eth.send_raw_transaction(signed_tx.rawTransaction)

    tx_receipt = w3.eth.wait_for_transaction_receipt(tx_hash)
    erc20_address = tx_receipt["contractAddress"]
    deployed_erc20 = get_example_erc20_contract(w3, erc20_address)
    assert deployed_erc20.functions.balanceOf(owner).call() == amount
    return deployed_erc20


def bytes_to_str(o: Any) -> Any:
    """
    Converts bytes (and hexbytes) fields to `str` in nested data types

    :param o:
    :return:
    """
    if isinstance(o, bytes):
        return HexBytes(o).hex()
    if isinstance(o, dict):
        o = dict(o)  # Remove AttributeDict
        for k in o.keys():
            o[k] = bytes_to_str(o[k])
    elif isinstance(o, (list, tuple)):
        o = deepcopy(o)
        for i, v in enumerate(o):
            o[i] = bytes_to_str(o[i])
    elif isinstance(o, set):
        o = {bytes_to_str(element) for element in o}
    return o


def to_json_with_hexbytes(o: Any) -> str:
    """
    Convert RPC calls with nested bytes/Hexbytes to json and compare. Useful for RPC calls

    :param o:
    :return: Object as JSON with Hexbytes/bytes parsed correctly as a hex string
    """
    return json.dumps(bytes_to_str(o), indent=4, sort_keys=True)

1	import functools	4✔
2	import json	4✔
3	import os	4✔
4	from copy import deepcopy	4✔
5	from typing import Any	4✔
6
7	import pytest	4✔
8	import requests	4✔
9	from eth_account.signers.local import LocalAccount	4✔
10	from eth_typing import ChecksumAddress	4✔
11	from hexbytes import HexBytes	4✔
12	from web3 import Web3	4✔
13	from web3.contract import Contract	4✔
14	from web3.types import TxParams	4✔
15
16	from ..contracts import get_example_erc20_contract	4✔
17
18
19	def just_test_if_node_available(node_url_variable_name: str) -> str:	4✔
20	"""
21	Just run the test if ``node url`` is defined on the ``node_url_variable_name`` environment variable
22	and it's accessible. Node JSON RPC url will only be tested the first call to this function.
23
24	:param node_url_variable_name: Environment variable name for ``node url``
25	:return: ``node url``
26	"""
27	node_url = getattr(just_test_if_node_available, node_url_variable_name, None)	4✔
28	if node_url: # Just check node first time	4✔
29	return node_url	4✔
30
31	node_url = os.environ.get(node_url_variable_name)	4✔
32	if not node_url:	4✔
33	pytest.skip(	×
34	f"{node_url_variable_name} not defined, skipping test",
35	allow_module_level=True,
36	)
37	else:
38	try:	4✔
39	response = requests.post(	4✔
40	node_url,
41	timeout=5,
42	json={
43	"jsonrpc": "2.0",
44	"method": "eth_blockNumber",
45	"params": [],
46	"id": 1,
47	},
48	)
49	if not response.ok:	4✔
50	pytest.fail(	×
51	f"Problem connecting to node {node_url}: {response.status_code} - {response.content}"
52	)
53	except IOError:	×
54	pytest.fail(f"Problem connecting to {node_url}")	×
55	setattr(just_test_if_node_available, node_url_variable_name, node_url)	4✔
56	return node_url	4✔
57
58
59	def just_test_if_mainnet_node() -> str:	4✔
60	return just_test_if_node_available("ETHEREUM_MAINNET_NODE")	4✔
61
62
63	def just_test_if_polygon_node() -> str:	4✔
64	return just_test_if_node_available("ETHEREUM_POLYGON_NODE")	4✔
65
66
67	def skip_on(exception, reason="Test skipped due to a controlled exception"):	4✔
68	"""
69	Decorator to skip a test if an exception is raised instead of failing it
70
71	:param exception:
72	:param reason:
73	:return:
74	"""
75
76	def decorator_func(f):	×
77	@functools.wraps(f)	×
78	def wrapper(args, *kwargs):	×
79	try:	×
80	# Run the test
81	return f(args, *kwargs)	×
82	except exception:	×
83	pytest.skip(reason)	×
84
85	return wrapper	×
86
87	return decorator_func	×
88
89
90	# TODO Move this to EthereumClient
91	def send_tx(w3: Web3, tx: TxParams, account: LocalAccount) -> bytes:	4✔
92	tx["from"] = account.address	4✔
93	if "nonce" not in tx:	4✔
94	tx["nonce"] = w3.eth.get_transaction_count(	4✔
95	account.address, block_identifier="pending"
96	)
97
98	if "gasPrice" not in tx and "maxFeePerGas" not in tx:	4✔
99	tx["gasPrice"] = w3.eth.gas_price	4✔
100
101	if "gas" not in tx:	4✔
102	tx["gas"] = w3.eth.estimate_gas(tx)	4✔
103
104	signed_tx = account.sign_transaction(tx)	4✔
105	tx_hash = w3.eth.send_raw_transaction(bytes(signed_tx.rawTransaction))	4✔
106	tx_receipt = w3.eth.wait_for_transaction_receipt(tx_hash)	4✔
107	assert tx_receipt["status"] == 1, "Error with tx %s - %s" % (tx_hash.hex(), tx)	4✔
108	return tx_hash	4✔
109
110
111	def deploy_erc20(	4✔
112	w3: Web3,
113	account: LocalAccount,
114	name: str,
115	symbol: str,
116	owner: ChecksumAddress,
117	amount: int,
118	decimals: int = 18,
119	) -> Contract:
120	erc20_contract = get_example_erc20_contract(w3)	4✔
121	tx = erc20_contract.constructor(	4✔
122	name, symbol, decimals, owner, amount
123	).build_transaction(
124	{
125	"nonce": w3.eth.get_transaction_count(
126	account.address, block_identifier="pending"
127	)
128	}
129	)
130	signed_tx = account.sign_transaction(tx)	4✔
131	tx_hash = w3.eth.send_raw_transaction(signed_tx.rawTransaction)	4✔
132
133	tx_receipt = w3.eth.wait_for_transaction_receipt(tx_hash)	4✔
134	erc20_address = tx_receipt["contractAddress"]	4✔
135	deployed_erc20 = get_example_erc20_contract(w3, erc20_address)	4✔
136	assert deployed_erc20.functions.balanceOf(owner).call() == amount	4✔
137	return deployed_erc20	4✔
138
139
140	def bytes_to_str(o: Any) -> Any:	4✔
141	"""
142	Converts bytes (and hexbytes) fields to `str` in nested data types
143
144	:param o:
145	:return:
146	"""
147	if isinstance(o, bytes):	×
148	return HexBytes(o).hex()	×
149	if isinstance(o, dict):	×
150	o = dict(o) # Remove AttributeDict	×
151	for k in o.keys():	×
152	o[k] = bytes_to_str(o[k])	×
153	elif isinstance(o, (list, tuple)):	×
154	o = deepcopy(o)	×
155	for i, v in enumerate(o):	×
156	o[i] = bytes_to_str(o[i])	×
157	elif isinstance(o, set):	×
158	o = {bytes_to_str(element) for element in o}	×
159	return o	×
160
161
162	def to_json_with_hexbytes(o: Any) -> str:	4✔
163	"""
164	Convert RPC calls with nested bytes/Hexbytes to json and compare. Useful for RPC calls
165
166	:param o:
167	:return: Object as JSON with Hexbytes/bytes parsed correctly as a hex string
168	"""
169	return json.dumps(bytes_to_str(o), indent=4, sort_keys=True)	×

safe-global / safe-eth-py / 10527345804

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous