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

Expand coverage and allow calling using class or instance

Browse Source
This commit is contained in:
Piper Merriam 2016-09-04 17:17:36 -06:00
parent 0f25b90960
commit 74dd65902f
4 changed files with 123 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

67
tests/filtering/test_contract_on_event_filtering.py
View File

@ -4,16 +4,47 @@ from flaky import flaky
@flaky(max_runs=3)
@pytest.mark.parametrize('call_as_instance', (True, False))
def test_on_filter_using_get_interface(web3_empty,
emitter,
Emitter,
wait_for_transaction,
emitter_log_topics,
emitter_event_ids,
call_as_instance):
web3 = web3_empty
if call_as_instance:
filter = emitter.on('LogNoArguments', {})
else:
filter = Emitter.on('LogNoArguments', {})
txn_hash = emitter.transact().logNoArgs(emitter_event_ids.LogNoArguments)
txn_receipt = wait_for_transaction(web3, txn_hash)
log_entries = filter.get()
assert len(log_entries) == 1
assert log_entries[0]['transactionHash'] == txn_hash
@flaky(max_runs=3)
@pytest.mark.parametrize('call_as_instance', (True, False))
def test_on_filter_with_only_event_name(web3_empty,
emitter,
Emitter,
wait_for_transaction,
emitter_log_topics,
emitter_event_ids):
emitter_event_ids,
call_as_instance):
web3 = web3_empty
seen_logs = []
filter = emitter.on('LogNoArguments', {}, seen_logs.append)
if call_as_instance:
filter = emitter.on('LogNoArguments', {}, seen_logs.append)
else:
filter = Emitter.on('LogNoArguments', {}, seen_logs.append)
txn_hash = emitter.transact().logNoArgs(emitter_event_ids.LogNoArguments)
txn_receipt = wait_for_transaction(web3, txn_hash)
@ -29,18 +60,26 @@ def test_on_filter_with_only_event_name(web3_empty,
@flaky(max_runs=3)
@pytest.mark.parametrize('call_as_instance', (True, False))
def test_on_filter_with_event_name_and_single_argument(web3_empty,
emitter,
Emitter,
wait_for_transaction,
emitter_log_topics,
emitter_event_ids):
emitter_event_ids,
call_as_instance):
web3 = web3_empty
seen_logs = []
filter = emitter.on('LogTripleWithIndex', {'filter': {
'arg1': 2,
}}, seen_logs.append)
if call_as_instance:
filter = emitter.on('LogTripleWithIndex', {'filter': {
'arg1': 2,
}}, seen_logs.append)
else:
filter = Emitter.on('LogTripleWithIndex', {'filter': {
'arg1': 2,
}}, seen_logs.append)
txn_hashes = []
txn_hashes.append(
@ -66,18 +105,26 @@ def test_on_filter_with_event_name_and_single_argument(web3_empty,
@flaky(max_runs=3)
@pytest.mark.parametrize('call_as_instance', (True, False))
def test_on_filter_with_event_name_and_non_indexed_argument(web3_empty,
emitter,
Emitter,
wait_for_transaction,
emitter_log_topics,
emitter_event_ids):
emitter_event_ids,
call_as_instance):
web3 = web3_empty
seen_logs = []
filter = emitter.on('LogTripleWithIndex', {'filter': {
'arg0': 1, 'arg1': 2,
}}, seen_logs.append)
if call_as_instance:
filter = emitter.on('LogTripleWithIndex', {'filter': {
'arg0': 1, 'arg1': 2,
}}, seen_logs.append)
else:
filter = Emitter.on('LogTripleWithIndex', {'filter': {
'arg0': 1, 'arg1': 2,
}}, seen_logs.append)
txn_hashes = []
txn_hashes.append(

53
tests/filtering/test_contract_past_event_filtering.py
View File

@ -1,15 +1,19 @@
import pytest
import random
import gevent
from flaky import flaky
@flaky(max_runs=3)
def test_past_events_filter_with_only_event_name(web3_empty,
emitter,
wait_for_transaction,
emitter_log_topics,
emitter_event_ids,
LogTopics):
@pytest.mark.parametrize('call_as_instance', (True, False))
def test_past_events_filter_with_callback(web3_empty,
emitter,
Emitter,
wait_for_transaction,
emitter_log_topics,
emitter_event_ids,
LogTopics,
call_as_instance):
web3 = web3_empty
txn_hash = emitter.transact().logNoArgs(emitter_event_ids.LogNoArguments)
@ -17,7 +21,10 @@ def test_past_events_filter_with_only_event_name(web3_empty,
seen_logs = []
filter = emitter.pastEvents('LogNoArguments', {}, seen_logs.append)
if call_as_instance:
filter = emitter.pastEvents('LogNoArguments', {}, seen_logs.append)
else:
filter = Emitter.pastEvents('LogNoArguments', {}, seen_logs.append)
with gevent.Timeout(5):
while not seen_logs:
@ -33,3 +40,35 @@ def test_past_events_filter_with_only_event_name(web3_empty,
assert event_data['transactionIndex'] == txn_receipt['transactionIndex']
assert event_data['address'] == emitter.address
assert event_data['event'] == 'LogNoArguments'
@flaky(max_runs=3)
@pytest.mark.parametrize('call_as_instance', (True, False))
def test_past_events_filter_using_get_api(web3_empty,
emitter,
Emitter,
wait_for_transaction,
emitter_log_topics,
emitter_event_ids,
LogTopics,
call_as_instance):
web3 = web3_empty
txn_hash = emitter.transact().logNoArgs(emitter_event_ids.LogNoArguments)
txn_receipt = wait_for_transaction(web3, txn_hash)
if call_as_instance:
filter = emitter.pastEvents('LogNoArguments')
else:
filter = Emitter.pastEvents('LogNoArguments')
log_entries = filter.get()
assert len(log_entries) == 1
event_data = log_entries[0]
assert event_data['args'] == {}
assert event_data['blockHash'] == txn_receipt['blockHash']
assert event_data['blockNumber'] == txn_receipt['blockNumber']
assert event_data['transactionIndex'] == txn_receipt['transactionIndex']
assert event_data['address'] == emitter.address
assert event_data['event'] == 'LogNoArguments'

5
web3/contract.py
View File

@ -36,6 +36,9 @@ from web3.utils.abi import (
check_if_arguments_can_be_encoded,
function_abi_to_4byte_selector,
)
from web3.utils.decorators import (
combomethod,
)
from web3.utils.transactions import (
get_block_gas_limit,
)
@ -379,6 +382,7 @@ class Contract(object):
return deploy_data
@combomethod
def on(self, event_name, default_filter_params=None, *callbacks):
"""
register a callback to be triggered on the appropriate events.
@ -410,6 +414,7 @@ class Contract(object):
return log_filter
@combomethod
def pastEvents(self, event_name, default_filter_params=None, *callbacks):
"""
register a callback to be triggered on all past events.

15
web3/utils/decorators.py Normal file
View File

@ -0,0 +1,15 @@
import functools
class combomethod(object):
def __init__(self, method):
self.method = method
def __get__(self, obj=None, objtype=None):
@functools.wraps(self.method)
def _wrapper(*args, **kwargs):
if obj is not None:
return self.method(obj, *args, **kwargs)
else:
return self.method(objtype, *args, **kwargs)
return _wrapper