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

lots of mucking

Browse Source
This commit is contained in:
Piper Merriam 2016-08-05 12:55:02 -06:00
parent c0537f8df9
commit 4495f4f67f
25 changed files with 200 additions and 162 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

161
conftest.py
View File

@ -4,6 +4,7 @@ import tempfile
import shutil
import random
import requests
import gevent
# This has to go here so that the `gevent.monkey.patch_all()` happens in the
@ -28,10 +29,7 @@ def get_open_port():
return port
def wait_for_http_connection(port, timeout=30):
import gevent
import requests
def wait_for_http_connection(port, timeout=60):
with gevent.Timeout(timeout):
while True:
try:
@ -45,6 +43,16 @@ def wait_for_http_connection(port, timeout=30):
raise ValueError("Unable to establish HTTP connection")
@pytest.fixture()
def skip_if_testrpc():
from web3.providers.rpc import TestRPCProvider
def _skip_if_testrpc(web3):
if isinstance(web3.currentProvider, TestRPCProvider):
pytest.skip()
return _skip_if_testrpc
@pytest.fixture()
def wait_for_miner_start():
def _wait_for_miner_start(web3, timeout=60):
@ -56,7 +64,6 @@ def wait_for_miner_start():
@pytest.fixture()
def wait_for_block():
import gevent
from web3.providers.rpc import TestRPCProvider
def _wait_for_block(web3, block_number=1, timeout=60 * 10):
@ -72,8 +79,6 @@ def wait_for_block():
@pytest.fixture()
def wait_for_transaction():
import gevent
def _wait_for_transaction(web3, txn_hash, timeout=120):
with gevent.Timeout(timeout):
while True:
@ -120,7 +125,7 @@ def web3_tester_provider():
@pytest.fixture()
def web3_tester(request, web3_tester_provider):
def web3_tester_empty(request, web3_tester_provider):
from web3 import Web3
if getattr(request, 'reset_chain', True):
@ -130,58 +135,89 @@ def web3_tester(request, web3_tester_provider):
return web3
def _web3_rpc():
from web3 import (
Web3,
RPCProvider,
)
@pytest.fixture()
def web3_tester_persistent(request, web3_tester_provider):
from web3 import Web3
with tempdir() as base_dir:
with GethProcess('testing', base_dir=base_dir) as geth:
geth.wait_for_rpc(30)
geth.wait_for_dag(600)
provider = RPCProvider(port=geth.rpc_port)
provider._geth = geth
web3 = Web3(provider)
yield web3
web3_rpc_empty = pytest.yield_fixture()(_web3_rpc)
web3_rpc_persistent = pytest.yield_fixture(scope="session")(_web3_rpc)
web3 = Web3(web3_tester_provider)
return web3
@pytest.fixture()
def web3_rpc(request):
if getattr(request, 'reset_chain', False):
return request.getfuncargvalue('web3_rpc_empty')
else:
return request.getfuncargvalue('web3_rpc_persistent')
def web3_tester(web3_tester_persistent):
# alias
return web3_tester_persistent
def _web3_ipc():
@contextlib.contextmanager
def setup_testing_geth():
with tempdir() as base_dir:
geth_process = GethProcess(
'testing',
base_dir=base_dir,
overrides={'verbosity': '3'},
)
with geth_process as running_geth_process:
running_geth_process.wait_for_ipc(60)
running_geth_process.wait_for_rpc(60)
running_geth_process.wait_for_dag(600)
yield running_geth_process
@pytest.yield_fixture(scope="session")
def geth_persistent():
with setup_testing_geth() as geth:
yield geth
@pytest.fixture(scope="session")
def web3_rpc_persistent(geth_persistent):
from web3 import (
Web3,
IPCProvider,
Web3, RPCProvider,
)
with tempdir() as base_dir:
with GethProcess('testing', base_dir=base_dir) as geth:
geth.wait_for_ipc(30)
geth.wait_for_dag(600)
provider = IPCProvider(geth.ipc_path)
provider._geth = geth
web3 = Web3(provider)
yield web3
web3_ipc_empty = pytest.yield_fixture()(_web3_ipc)
web3_ipc_persistent = pytest.yield_fixture(scope="session")(_web3_ipc)
provider = RPCProvider(port=geth_persistent.rpc_port)
provider._geth = geth_persistent
web3 = Web3(provider)
return web3
@pytest.fixture()
def web3_ipc(request):
if getattr(request, 'reset_chain', False):
return request.getfuncargvalue('web3_ipc_empty')
else:
return request.getfuncargvalue('web3_ipc_persistent')
@pytest.yield_fixture()
def web3_rpc_empty():
from web3 import (
Web3, RPCProvider,
)
with setup_testing_geth() as geth:
provider = RPCProvider(port=geth.rpc_port)
provider._geth = geth
web3 = Web3(provider)
yield web3
@pytest.fixture(scope="session")
def web3_ipc_persistent(geth_persistent):
from web3 import (
Web3, IPCProvider,
)
provider = IPCProvider(ipc_path=geth_persistent.ipc_path)
provider._geth = geth_persistent
web3 = Web3(provider)
return web3
@pytest.yield_fixture()
def web3_ipc_empty():
from web3 import (
Web3, IPCProvider,
)
with setup_testing_geth() as geth:
provider = IPCProvider(ipc_path=geth.ipc_path)
provider._geth = geth
web3 = Web3(provider)
yield web3
@pytest.fixture(params=[
@ -191,19 +227,26 @@ def web3_ipc(request):
])
def web3(request):
if request.param == "tester":
return request.getfuncargvalue('web3_tester')
return request.getfuncargvalue('web3_tester_persistent')
elif request.param == "rpc":
return request.getfuncargvalue('web3_rpc')
return request.getfuncargvalue('web3_rpc_persistent')
elif request.param == "ipc":
return request.getfuncargvalue('web3_ipc')
return request.getfuncargvalue('web3_ipc_persistent')
else:
raise ValueError("Unknown param")
@pytest.fixture()
def empty_account(web3):
from eth_tester_client.utils import mk_random_privkey
address = web3.personal.importRawKey(mk_random_privkey(), "a-password")
assert web3.eth.getBalance(address) == 0
return address
@pytest.fixture(params=[
'tester',
pytest.mark.slow('rpc'),
pytest.mark.slow('ipc'),
])
def web3_empty(request):
if request.param == "tester":
return request.getfuncargvalue('web3_tester_empty')
elif request.param == "rpc":
return request.getfuncargvalue('web3_rpc_empty')
elif request.param == "ipc":
return request.getfuncargvalue('web3_ipc_empty')
else:
raise ValueError("Unknown param")

8
tests/eth-module/test_eth_getBlockTransactionCount.py
View File

@ -2,18 +2,14 @@ import collections
import itertools
import pytest
from web3.providers.rpc import TestRPCProvider
@pytest.fixture(autouse=True)
def wait_for_first_block(web3, wait_for_block):
def wait_for_first_block(web3, wait_for_block, skip_if_testrpc):
skip_if_testrpc(web3)
wait_for_block(web3)
def test_eth_getBlockTransactionCount(web3, extra_accounts, wait_for_transaction):
if isinstance(web3.currentProvider, TestRPCProvider):
pytest.skip("testrpc doesn't implement `getBlockTransactionCount`")
transaction_hashes = []
# send some transaction

8
tests/eth-module/test_eth_getTransactionFromBlock.py
View File

@ -2,18 +2,14 @@ import collections
import itertools
import pytest
from web3.providers.rpc import TestRPCProvider
@pytest.fixture(autouse=True)
def wait_for_first_block(web3, wait_for_block):
def wait_for_first_block(web3, wait_for_block, skip_if_testrpc):
skip_if_testrpc(web3)
wait_for_block(web3)
def test_eth_getTransactionFromBlock(web3, extra_accounts, wait_for_transaction):
if isinstance(web3.currentProvider, TestRPCProvider):
pytest.skip("testrpc doesn't implement `eth_getTransactionByBlockNumberAndIndex`")
current_block_number = web3.eth.blockNumber
transaction_hashes = []

14
tests/eth-module/test_eth_mining.py
View File

@ -1,21 +1,11 @@
reset_chain = True
def test_mining_property(web3_tester):
web3 = web3_tester
assert web3.eth.mining is False
def test_mining_property_ipc(web3_ipc_empty, wait_for_miner_start):
web3 = web3_ipc_empty
wait_for_miner_start(web3)
assert web3.eth.mining is True
def test_mining_property_rpc(web3_rpc_empty, wait_for_miner_start):
web3 = web3_rpc_empty
def test_mining_property_ipc(web3_empty, wait_for_miner_start):
web3 = web3_empty
wait_for_miner_start(web3)
assert web3.eth.mining is True

5
tests/eth-module/test_eth_sign.py
View File

@ -57,9 +57,8 @@ def extract_ecdsa_signer(msg_hash, signature):
return address
def test_eth_sign(web3):
if isinstance(web3.currentProvider, TestRPCProvider):
pytest.skip("testrpc doesn't implement `getBlockTransactionCount`")
def test_eth_sign(web3, skip_if_testrpc):
skip_if_testrpc(web3)
private_key_hex = '0x5e95384d8050109aab08c1922d3c230739bc16976553c317e5d0b87b59371f2a'
private_key = decode_hex(private_key_hex)

19
tests/filtering/conftest.py
View File

@ -3,18 +3,14 @@ import json
import textwrap
from sha3 import sha3_256
from web3.providers.rpc import TestRPCProvider
assert sha3_256(b'').hexdigest() == 'c5d2460186f7233c927e7db2dcc703c0e500b653ca82273b7bfad8045d85a470'
@pytest.fixture(autouse=True)
def skip_testrpc_and_wait_for_mining_start(web3, wait_for_block):
if isinstance(web3.currentProvider, TestRPCProvider):
pytest.skip("No miner interface on eth-testrpc")
wait_for_block(web3)
def skip_testrpc_and_wait_for_mining_start(web3_empty, wait_for_block,
skip_if_testrpc):
skip_if_testrpc(web3_empty) # TODO: enable testrpc
wait_for_block(web3_empty)
CONTRACT_EMITTER_SOURCE = textwrap.dedent(("""
@ -118,12 +114,15 @@ def EMITTER(EMITTER_CODE,
@pytest.fixture()
def Emitter(web3, EMITTER):
def Emitter(web3_empty, EMITTER):
web3 = web3_empty
return web3.eth.contract(**EMITTER)
@pytest.fixture()
def emitter(web3, Emitter, wait_for_transaction, wait_for_block):
def emitter(web3_empty, Emitter, wait_for_transaction, wait_for_block):
web3 = web3_empty
wait_for_block(web3)
deploy_txn_hash = Emitter.deploy({'from': web3.eth.coinbase, 'gas': 1000000})
deploy_receipt = wait_for_transaction(web3, deploy_txn_hash)

12
tests/filtering/test_contract_on_event_filtering.py
View File

@ -3,15 +3,13 @@ import gevent
from flaky import flaky
reset_chain = True
@flaky(max_runs=3)
def test_on_filter_with_only_event_name(web3,
def test_on_filter_with_only_event_name(web3_empty,
emitter,
wait_for_transaction,
emitter_log_topics,
emitter_event_ids):
web3 = web3_empty
seen_logs = []
@ -31,11 +29,12 @@ def test_on_filter_with_only_event_name(web3,
@flaky(max_runs=3)
def test_on_filter_with_event_name_and_single_argument(web3,
def test_on_filter_with_event_name_and_single_argument(web3_empty,
emitter,
wait_for_transaction,
emitter_log_topics,
emitter_event_ids):
web3 = web3_empty
seen_logs = []
@ -67,11 +66,12 @@ def test_on_filter_with_event_name_and_single_argument(web3,
@flaky(max_runs=3)
def test_on_filter_with_event_name_and_non_indexed_argument(web3,
def test_on_filter_with_event_name_and_non_indexed_argument(web3_empty,
emitter,
wait_for_transaction,
emitter_log_topics,
emitter_event_ids):
web3 = web3_empty
seen_logs = []

7
tests/filtering/test_contract_past_event_filtering.py
View File

@ -3,15 +3,14 @@ import gevent
from flaky import flaky
reset_chain = True
@flaky(max_runs=3)
def test_past_events_filter_with_only_event_name(web3,
def test_past_events_filter_with_only_event_name(web3_empty,
emitter,
wait_for_transaction,
emitter_log_topics,
emitter_event_ids):
web3 = web3_empty
txn_hash = emitter.transact().logNoArgs(emitter_event_ids.LogNoArguments)
txn_receipt = wait_for_transaction(web3, txn_hash)

3
tests/filtering/test_filter_against_latest_blocks.py
View File

@ -3,9 +3,6 @@ import gevent
from flaky import flaky
reset_chain = True
@flaky(max_runs=3)
def test_filter_against_latest_blocks(web3, wait_for_block):
seen_blocks = []

3
tests/filtering/test_filter_against_pending_transactions.py
View File

@ -3,9 +3,6 @@ import gevent
from flaky import flaky
reset_chain = True
@flaky(max_runs=3)
def test_filter_against_pending_transactions(web3, wait_for_transaction):
seen_txns = []

6
tests/filtering/test_filter_against_transaction_logs.py
View File

@ -3,15 +3,13 @@ import gevent
from flaky import flaky
reset_chain = True
@flaky(max_runs=3)
def test_filter_against_log_events(web3,
def test_filter_against_log_events(web3_empty,
emitter,
wait_for_transaction,
emitter_log_topics,
emitter_event_ids):
web3 = web3_empty
seen_logs = []
txn_filter = web3.eth.filter({})

7
tests/mining/conftest.py
View File

@ -2,8 +2,11 @@ import pytest
@pytest.fixture(autouse=True)
def always_wait_for_mining_start(web3_ipc_empty, wait_for_miner_start):
web3 = web3_ipc_empty
def always_wait_for_mining_start(web3_empty, wait_for_miner_start,
skip_if_testrpc):
web3 = web3_empty
skip_if_testrpc(web3)
wait_for_miner_start(web3)

7
tests/mining/test_miner_hashrate.py
View File

@ -1,8 +1,5 @@
from web3.providers.rpc import TestRPCProvider
def test_miner_hashrate(web3_ipc_empty, wait_for_miner_start):
web3 = web3_ipc_empty
def test_miner_hashrate(web3_empty, wait_for_miner_start):
web3 = web3_empty
hashrate = web3.miner.hashrate
assert hashrate > 0

6
tests/mining/test_miner_setExtra.py
View File

@ -5,8 +5,8 @@ import gevent
from web3.utils.encoding import decode_hex
def test_miner_setExtra(web3_ipc_empty, wait_for_block):
web3 = web3_ipc_empty
def test_miner_setExtra(web3_empty, wait_for_block):
web3 = web3_empty
initial_extra = decode_hex(web3.eth.getBlock(web3.eth.blockNumber)['extraData'])
@ -17,7 +17,7 @@ def test_miner_setExtra(web3_ipc_empty, wait_for_block):
web3.miner.setExtra(new_extra_data)
with gevent.Timeout(30):
with gevent.Timeout(60):
while True:
extra_data = decode_hex(web3.eth.getBlock(web3.eth.blockNumber)['extraData'])
if extra_data == new_extra_data:

6
tests/mining/test_miner_setGasPrice.py
View File

@ -2,8 +2,8 @@ import random
import gevent
def test_miner_setGasPrice(web3_ipc_empty, wait_for_block):
web3 = web3_ipc_empty
def test_miner_setGasPrice(web3_empty, wait_for_block):
web3 = web3_empty
initial_gas_price = web3.eth.gasPrice
@ -12,7 +12,7 @@ def test_miner_setGasPrice(web3_ipc_empty, wait_for_block):
web3.miner.setGasPrice(initial_gas_price // 2)
with gevent.Timeout(30):
with gevent.Timeout(60):
while web3.eth.gasPrice == initial_gas_price:
gevent.sleep(random.random())

6
tests/mining/test_miner_start.py
View File

@ -3,8 +3,8 @@ import random
import gevent
def test_miner_start(web3_ipc_empty, wait_for_miner_start):
web3 = web3_ipc_empty
def test_miner_start(web3_empty, wait_for_miner_start):
web3 = web3_empty
# sanity
assert web3.eth.mining
@ -12,7 +12,7 @@ def test_miner_start(web3_ipc_empty, wait_for_miner_start):
web3.miner.stop()
with gevent.Timeout(30):
with gevent.Timeout(60):
while web3.eth.mining or web3.eth.hashrate:
gevent.sleep(random.random())

6
tests/mining/test_miner_stop.py
View File

@ -6,15 +6,15 @@ from flaky import flaky
@flaky(max_runs=3)
def test_miner_stop(web3_ipc_empty):
web3 = web3_ipc_empty
def test_miner_stop(web3_empty):
web3 = web3_empty
assert web3.eth.mining
assert web3.miner.hashrate
web3.miner.stop()
with gevent.Timeout(30):
with gevent.Timeout(60):
while web3.eth.mining or web3.eth.hashrate:
gevent.sleep(random.random())

20
tests/personal-module/conftest.py
View File

@ -20,10 +20,15 @@ def account_public_key(account_private_key):
@pytest.fixture()
def password_account(web3, account_password,
account_private_key, account_public_key,
def password_account(web3_empty,
account_password,
account_private_key,
account_public_key,
wait_for_transaction):
from eth_tester_client.utils import normalize_address
web3 = web3_empty
address = web3.personal.importRawKey(account_private_key, account_password)
# sanity check
@ -40,3 +45,14 @@ def password_account(web3, account_password,
assert web3.eth.getBalance(address) == initial_balance
return address
@pytest.fixture()
def empty_account(web3_empty):
web3 = web3_empty
from eth_tester_client.utils import mk_random_privkey
address = web3.personal.importRawKey(mk_random_privkey(), "a-password")
assert web3.eth.getBalance(address) == 0
return address

16
tests/personal-module/test_personal_importRawKey.py
View File

@ -5,11 +5,10 @@ from eth_tester_client.utils import (
)
reset_chain = True
def test_personal_importRawKey_as_bytes(web3, account_private_key,
def test_personal_importRawKey_as_bytes(web3_empty, account_private_key,
account_password, account_public_key):
web3 = web3_empty
address = web3.personal.importRawKey(account_private_key, account_password)
# sanity check
@ -18,9 +17,11 @@ def test_personal_importRawKey_as_bytes(web3, account_private_key,
assert web3.personal.unlockAccount(address, account_password) is True
def test_personal_importRawKey_as_hex_with_0x(web3, account_private_key,
def test_personal_importRawKey_as_hex_with_0x(web3_empty, account_private_key,
account_password,
account_public_key):
web3 = web3_empty
address = web3.personal.importRawKey(encode_32bytes(account_private_key), account_password)
# sanity check
@ -29,9 +30,12 @@ def test_personal_importRawKey_as_hex_with_0x(web3, account_private_key,
assert web3.personal.unlockAccount(address, account_password) is True
def test_personal_importRawKey_as_hex_without_0x(web3, account_private_key,
def test_personal_importRawKey_as_hex_without_0x(web3_empty,
account_private_key,
account_password,
account_public_key):
web3 = web3_empty
address = web3.personal.importRawKey(strip_0x(encode_32bytes(account_private_key)), account_password)
# sanity check

7
tests/personal-module/test_personal_lockAccount.py
View File

@ -1,11 +1,10 @@
import pytest
reset_chain = True
def test_personal_lockAccount(web3, password_account, account_password,
def test_personal_lockAccount(web3_empty, password_account, account_password,
wait_for_transaction, empty_account):
web3 = web3_empty
initial_balancee = web3.eth.getBalance(empty_account)
assert web3.personal.unlockAccount(password_account, account_password) is True

4
tests/personal-module/test_personal_signAndSendTransaction.py
View File

@ -1,7 +1,9 @@
def test_personal_signAndSendTransaction(web3, password_account,
def test_personal_signAndSendTransaction(web3_empty, password_account,
account_password,
wait_for_transaction,
empty_account):
web3 = web3_empty
txn_hash = web3.personal.signAndSendTransaction({
'from': password_account,
'to': empty_account,

11
tests/personal-module/test_personal_unlockAccount.py
View File

@ -1,16 +1,15 @@
import pytest
reset_chain = True
@pytest.fixture(autouse=True)
def wait_for_first_block(web3, wait_for_block):
wait_for_block(web3)
def wait_for_first_block(web3_empty, wait_for_block):
wait_for_block(web3_empty)
def test_personal_unlockAccount(web3, password_account, account_password,
def test_personal_unlockAccount(web3_empty, password_account, account_password,
wait_for_transaction, empty_account):
web3 = web3_empty
initial_balancee = web3.eth.getBalance(empty_account)
with pytest.raises(ValueError):

8
tests/txpool-module/conftest.py
View File

@ -2,8 +2,12 @@ import pytest
@pytest.fixture(autouse=True)
def skip_testrpc_and_wait_for_mining_start(web3_ipc_empty, wait_for_miner_start):
web3 = web3_ipc_empty
def skip_testrpc_and_wait_for_mining_start(web3_empty,
wait_for_miner_start,
skip_if_testrpc):
web3 = web3_empty
skip_if_testrpc(web3)
wait_for_miner_start(web3)

6
tests/txpool-module/test_txpool_content.py
View File

@ -2,12 +2,12 @@ import random
import gevent
def test_txpool_content(web3_ipc_empty):
web3 = web3_ipc_empty
def test_txpool_content(web3_empty):
web3 = web3_empty
web3.miner.stop()
with gevent.Timeout(30):
with gevent.Timeout(60):
while web3.miner.hashrate or web3.eth.mining:
gevent.sleep(random.random())

6
tests/txpool-module/test_txpool_inspect.py
View File

@ -2,12 +2,12 @@ import random
import gevent
def test_txpool_inspect(web3_ipc_empty):
web3 = web3_ipc_empty
def test_txpool_inspect(web3_empty):
web3 = web3_empty
web3.miner.stop()
with gevent.Timeout(30):
with gevent.Timeout(60):
while web3.miner.hashrate or web3.eth.mining:
gevent.sleep(random.random())