flip/web3.py
1
0
Fork 0
mirror of https://github.com/FlipsideCrypto/web3.py.git synced 2026-02-06 19:06:52 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

cleanup

Browse Source
This commit is contained in:
Piper Merriam 2016-09-09 13:09:09 -06:00
parent 2f240970ff
commit 4e525e5dd8
7 changed files with 183 additions and 176 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
docs/contracts.rst
View File

@ -100,7 +100,7 @@ Each Contract Factory exposes the following methods.
"0x4e3a3754410177e6937ef1f84bba68ea139e8d1a2258c5f85db9f1cd715a1bdd"
.. py:method:: Contract.call(transaction).myMethod(*args)
.. py:method:: Contract.call(transaction).myMethod(*args, **kwargs)
Call a contract function, executing the transaction locally using the
``eth_call`` API. This will not create a new public transaction.
@ -121,7 +121,7 @@ Each Contract Factory exposes the following methods.
54321 # the token balance for the account `web3.eth.accounts[1]`
.. py:method:: Contract.estimateGas(transaction).myMethod(*args)
.. py:method:: Contract.estimateGas(transaction).myMethod(*args, **kwargs)
Call a contract function, executing the transaction locally using the
``eth_call`` API. This will not create a new public transaction.

8
tests/contracts/test_contract_constructor_encoding.py
View File

@ -9,13 +9,13 @@ from web3.utils.formatting import remove_0x_prefix
def test_contract_constructor_abi_encoding_with_no_constructor_fn(MathContract, MATH_CODE):
deploy_data = MathContract._encodeConstructorData()
deploy_data = MathContract._encode_constructor_data()
assert deploy_data == MATH_CODE
def test_contract_constructor_abi_encoding_with_constructor_with_no_args(SimpleConstructorContract,
SIMPLE_CONSTRUCTOR_CODE):
deploy_data = SimpleConstructorContract._encodeConstructorData()
deploy_data = SimpleConstructorContract._encode_constructor_data()
assert deploy_data == SIMPLE_CONSTRUCTOR_CODE
@ -32,11 +32,11 @@ def test_contract_constructor_abi_encoding_with_constructor_with_no_args(SimpleC
)
def test_error_if_invalid_arguments_supplied(WithConstructorArgumentsContract, arguments):
with pytest.raises(TypeError):
WithConstructorArgumentsContract._encodeConstructorData(arguments)
WithConstructorArgumentsContract._encode_constructor_data(arguments)
def test_contract_constructor_encoding_encoding(WithConstructorArgumentsContract):
deploy_data = WithConstructorArgumentsContract._encodeConstructorData([1234, 'abcd'])
deploy_data = WithConstructorArgumentsContract._encode_constructor_data([1234, 'abcd'])
encoded_args = '0x00000000000000000000000000000000000000000000000000000000000004d26162636400000000000000000000000000000000000000000000000000000000'
expected_ending = encode_hex(encode_abi(['uint256', 'bytes32'], [1234, b'abcd']))
assert expected_ending == encoded_args

2
tests/contracts/test_contract_estimateGas.py
View File

@ -31,7 +31,7 @@ def math_contract(web3, MATH_ABI, MATH_CODE, MATH_RUNTIME, MATH_SOURCE,
def test_contract_estimateGas(web3, math_contract):
increment_abi = math_contract.find_matching_fn_abi('increment', [])
increment_abi = math_contract._find_matching_fn_abi('increment', [])
call_data = function_abi_to_4byte_selector(increment_abi)
gas_estimate = math_contract.estimateGas().increment()

10
tests/contracts/test_contract_method_to_argument_matching.py
View File

@ -14,7 +14,7 @@ MULTIPLE_FUNCTIONS = json.loads('[{"constant":false,"inputs":[],"name":"a","outp
def test_finds_single_function_without_args(web3_tester):
Contract = web3_tester.eth.contract(SINGLE_FN_NO_ARGS)
abi = Contract.find_matching_fn_abi('a', [])
abi = Contract._find_matching_fn_abi('a', [])
assert abi['name'] == 'a'
assert abi['inputs'] == []
@ -22,7 +22,7 @@ def test_finds_single_function_without_args(web3_tester):
def test_finds_single_function_with_args(web3_tester):
Contract = web3_tester.eth.contract(SINGLE_FN_ONE_ARG)
abi = Contract.find_matching_fn_abi('a', [1234])
abi = Contract._find_matching_fn_abi('a', [1234])
assert abi['name'] == 'a'
assert len(abi['inputs']) == 1
assert abi['inputs'][0]['type'] == 'uint256'
@ -32,7 +32,7 @@ def test_error_when_no_function_name_match(web3_tester):
Contract = web3_tester.eth.contract(SINGLE_FN_NO_ARGS)
with pytest.raises(ValueError):
Contract.find_matching_fn_abi('no_function_name', [1234])
Contract._find_matching_fn_abi('no_function_name', [1234])
@pytest.mark.parametrize(
@ -48,7 +48,7 @@ def test_error_when_no_function_name_match(web3_tester):
def test_finds_function_with_matching_args(web3_tester, arguments, expected_types):
Contract = web3_tester.eth.contract(MULTIPLE_FUNCTIONS)
abi = Contract.find_matching_fn_abi('a', arguments)
abi = Contract._find_matching_fn_abi('a', arguments)
assert abi['name'] == 'a'
assert len(abi['inputs']) == len(expected_types)
assert set(get_abi_input_types(abi)) == set(expected_types)
@ -58,4 +58,4 @@ def test_error_when_duplicate_match(web3_tester):
Contract = web3_tester.eth.contract(MULTIPLE_FUNCTIONS)
with pytest.raises(ValueError):
abi = Contract.find_matching_fn_abi('a', [100])
abi = Contract._find_matching_fn_abi('a', [100])

2
tests/contracts/test_extracting_event_data.py
View File

@ -61,7 +61,7 @@ def test_event_data_extraction(web3_tester_persistent,
assert len(txn_receipt['logs']) == 1
log_entry = txn_receipt['logs'][0]
event_abi = emitter.find_matching_event_abi(event_name)
event_abi = emitter._find_matching_event_abi(event_name)
event_topic = getattr(emitter_log_topics, event_name)
is_anonymous = event_abi['anonymous']

2
tests/eth-module/test_eth_estimateGas.py
View File

@ -31,7 +31,7 @@ def math_contract(web3, MATH_ABI, MATH_CODE, MATH_RUNTIME, MATH_SOURCE,
def test_eth_estimateGas(web3, math_contract):
increment_abi = math_contract.find_matching_fn_abi('increment', [])
increment_abi = math_contract._find_matching_fn_abi('increment', [])
call_data = function_abi_to_4byte_selector(increment_abi)
gas_estimate = web3.eth.estimateGas({
'to': math_contract.address,

331
web3/contract.py
View File

@ -188,119 +188,15 @@ class Contract(object):
"Cannot specify `to` for contract deployment"
)
deploy_transaction['data'] = cls._encodeConstructorData(args, kwargs)
deploy_transaction['data'] = cls._encode_constructor_data(args, kwargs)
# TODO: handle asynchronous contract creation
txn_hash = cls.web3.eth.sendTransaction(deploy_transaction)
return txn_hash
#
# ABI Helpers
# Public API
#
@classmethod
def find_matching_fn_abi(cls, fn_name=None, args=None, kwargs=None):
filters = []
if fn_name:
filters.append(functools.partial(filter_by_name, fn_name))
if args is not None or kwargs is not None:
if args is None:
args = tuple()
if kwargs is None:
kwargs = {}
num_arguments = len(args) + len(kwargs)
filters.extend([
functools.partial(filter_by_argument_count, num_arguments),
functools.partial(filter_by_encodability, args, kwargs),
])
function_candidates = filter_by_type('function', cls.abi)
for filter_fn in filters:
function_candidates = filter_fn(function_candidates)
if len(function_candidates) == 1:
return function_candidates[0]
elif not function_candidates:
break
if not function_candidates:
raise ValueError("No matching functions found")
else:
raise ValueError("Multiple functions found")
@classmethod
def find_matching_event_abi(cls, event_name=None, argument_names=None):
filters = [
functools.partial(filter_by_type, 'event'),
]
if event_name is not None:
filters.append(functools.partial(filter_by_name, event_name))
if argument_names is not None:
filters.append(
functools.partial(filter_by_argument_name, argument_names)
)
filter_fn = compose(*filters)
event_abi_candidates = filter_fn(cls.abi)
if len(event_abi_candidates) == 1:
return event_abi_candidates[0]
elif not event_abi_candidates:
raise ValueError("No matching functions found")
else:
raise ValueError("Multiple functions found")
@classmethod
def get_function_info(cls, fn_name, args=None, kwargs=None):
if args is None:
args = tuple()
if kwargs is None:
kwargs = {}
fn_abi = cls.find_matching_fn_abi(fn_name, args, kwargs)
fn_selector = function_abi_to_4byte_selector(fn_abi)
fn_arguments = merge_args_and_kwargs(fn_abi, args, kwargs)
return fn_abi, fn_selector, fn_arguments
@combomethod
def _prepare_transaction(cls,
fn_name,
fn_args=None,
fn_kwargs=None,
transaction=None):
"""
Returns a dictionary of the transaction that could be used to call this
"""
if transaction is None:
prepared_transaction = {}
else:
prepared_transaction = dict(**transaction)
if 'data' in prepared_transaction:
raise ValueError("Transaction parameter may not contain a 'data' key")
fn_abi, fn_selector, fn_arguments = cls.get_function_info(
fn_name, fn_args, fn_kwargs,
)
if cls.address:
prepared_transaction.setdefault('to', cls.address)
prepared_transaction['data'] = cls._encodeABI(
fn_abi,
fn_arguments,
data=fn_selector,
)
return prepared_transaction
@classmethod
@coerce_return_to_text
def encodeABI(cls, fn_name, args=None, kwargs=None, data=None):
@ -308,62 +204,10 @@ class Contract(object):
encodes the arguments using the Ethereum ABI for the contract function
that matches the given name and arguments..
"""
fn_abi, _, fn_arguments = cls.get_function_info(
fn_abi, _, fn_arguments = cls._get_function_info(
fn_name, args, kwargs,
)
return cls._encodeABI(fn_abi, fn_arguments, data)
@classmethod
def _encodeABI(cls, abi, arguments, data=None):
argument_types = get_abi_input_types(abi)
if not check_if_arguments_can_be_encoded(abi, arguments, {}):
raise TypeError(
"One or more arguments could not be encoded to the necessary "
"ABI type. Expected types are: {0}".format(
', '.join(argument_types),
)
)
try:
encoded_arguments = encode_abi(
argument_types,
force_obj_to_bytes(arguments),
)
except EncodingError as e:
raise TypeError(
"One or more arguments could not be encoded to the necessary "
"ABI type: {0}".format(str(e))
)
if data:
return add_0x_prefix(
force_bytes(remove_0x_prefix(data)) +
force_bytes(remove_0x_prefix(encode_hex(encoded_arguments)))
)
else:
return encode_hex(encoded_arguments)
@classmethod
@coerce_return_to_text
def _encodeConstructorData(cls, args=None, kwargs=None):
constructor_abi = get_constructor_abi(cls.abi)
if constructor_abi:
if args is None:
args = tuple()
if kwargs is None:
kwargs = {}
arguments = merge_args_and_kwargs(constructor_abi, args, kwargs)
deploy_data = add_0x_prefix(
cls._encodeABI(constructor_abi, arguments, data=cls.code)
)
else:
deploy_data = add_0x_prefix(cls.code)
return deploy_data
return cls._encode_abi(fn_abi, fn_arguments, data)
@combomethod
def on(self, event_name, filter_params=None, *callbacks):
@ -375,7 +219,10 @@ class Contract(object):
argument_filters = filter_params.pop('filter', {})
argument_filter_names = list(argument_filters.keys())
event_abi = self.find_matching_event_abi(event_name, argument_filter_names)
event_abi = self._find_matching_event_abi(
event_name,
argument_filter_names,
)
data_filter_set, event_filter_params = construct_event_filter_params(
event_abi,
@ -613,6 +460,166 @@ class Contract(object):
return Transactor()
#
# Private Helpers
#
@classmethod
def _find_matching_fn_abi(cls, fn_name=None, args=None, kwargs=None):
filters = []
if fn_name:
filters.append(functools.partial(filter_by_name, fn_name))
if args is not None or kwargs is not None:
if args is None:
args = tuple()
if kwargs is None:
kwargs = {}
num_arguments = len(args) + len(kwargs)
filters.extend([
functools.partial(filter_by_argument_count, num_arguments),
functools.partial(filter_by_encodability, args, kwargs),
])
function_candidates = filter_by_type('function', cls.abi)
for filter_fn in filters:
function_candidates = filter_fn(function_candidates)
if len(function_candidates) == 1:
return function_candidates[0]
elif not function_candidates:
break
if not function_candidates:
raise ValueError("No matching functions found")
else:
raise ValueError("Multiple functions found")
@classmethod
def _find_matching_event_abi(cls, event_name=None, argument_names=None):
filters = [
functools.partial(filter_by_type, 'event'),
]
if event_name is not None:
filters.append(functools.partial(filter_by_name, event_name))
if argument_names is not None:
filters.append(
functools.partial(filter_by_argument_name, argument_names)
)
filter_fn = compose(*filters)
event_abi_candidates = filter_fn(cls.abi)
if len(event_abi_candidates) == 1:
return event_abi_candidates[0]
elif not event_abi_candidates:
raise ValueError("No matching functions found")
else:
raise ValueError("Multiple functions found")
@classmethod
def _get_function_info(cls, fn_name, args=None, kwargs=None):
if args is None:
args = tuple()
if kwargs is None:
kwargs = {}
fn_abi = cls._find_matching_fn_abi(fn_name, args, kwargs)
fn_selector = function_abi_to_4byte_selector(fn_abi)
fn_arguments = merge_args_and_kwargs(fn_abi, args, kwargs)
return fn_abi, fn_selector, fn_arguments
@combomethod
def _prepare_transaction(cls,
fn_name,
fn_args=None,
fn_kwargs=None,
transaction=None):
"""
Returns a dictionary of the transaction that could be used to call this
"""
if transaction is None:
prepared_transaction = {}
else:
prepared_transaction = dict(**transaction)
if 'data' in prepared_transaction:
raise ValueError("Transaction parameter may not contain a 'data' key")
fn_abi, fn_selector, fn_arguments = cls._get_function_info(
fn_name, fn_args, fn_kwargs,
)
if cls.address:
prepared_transaction.setdefault('to', cls.address)
prepared_transaction['data'] = cls._encode_abi(
fn_abi,
fn_arguments,
data=fn_selector,
)
return prepared_transaction
@classmethod
def _encode_abi(cls, abi, arguments, data=None):
argument_types = get_abi_input_types(abi)
if not check_if_arguments_can_be_encoded(abi, arguments, {}):
raise TypeError(
"One or more arguments could not be encoded to the necessary "
"ABI type. Expected types are: {0}".format(
', '.join(argument_types),
)
)
try:
encoded_arguments = encode_abi(
argument_types,
force_obj_to_bytes(arguments),
)
except EncodingError as e:
raise TypeError(
"One or more arguments could not be encoded to the necessary "
"ABI type: {0}".format(str(e))
)
if data:
return add_0x_prefix(
force_bytes(remove_0x_prefix(data)) +
force_bytes(remove_0x_prefix(encode_hex(encoded_arguments)))
)
else:
return encode_hex(encoded_arguments)
@classmethod
@coerce_return_to_text
def _encode_constructor_data(cls, args=None, kwargs=None):
constructor_abi = get_constructor_abi(cls.abi)
if constructor_abi:
if args is None:
args = tuple()
if kwargs is None:
kwargs = {}
arguments = merge_args_and_kwargs(constructor_abi, args, kwargs)
deploy_data = add_0x_prefix(
cls._encode_abi(constructor_abi, arguments, data=cls.code)
)
else:
deploy_data = add_0x_prefix(cls.code)
return deploy_data
@coerce_return_to_text
def call_contract_function(contract,
@ -633,7 +640,7 @@ def call_contract_function(contract,
return_data = contract.web3.eth.call(call_transaction)
function_abi = contract.find_matching_fn_abi(function_name, args, kwargs)
function_abi = contract._find_matching_fn_abi(function_name, args, kwargs)
output_types = get_abi_output_types(function_abi)
output_data = decode_abi(output_types, return_data)