You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
176 lines
6.9 KiB
176 lines
6.9 KiB
|
|
"""Blockchaininterface functionality tests.""" |
|
|
|
import binascii |
|
from commontest import create_wallet_for_sync |
|
|
|
import pytest |
|
from jmbase import get_log |
|
from jmclient import load_test_config, jm_single, BaseWallet |
|
|
|
log = get_log() |
|
|
|
pytestmark = pytest.mark.usefixtures("setup_regtest_bitcoind") |
|
|
|
|
|
def sync_test_wallet(fast, wallet_service): |
|
sync_count = 0 |
|
wallet_service.synced = False |
|
while not wallet_service.synced: |
|
wallet_service.sync_wallet(fast=fast) |
|
sync_count += 1 |
|
# avoid infinite loop |
|
assert sync_count < 10 |
|
log.debug("Tried " + str(sync_count) + " times") |
|
|
|
|
|
@pytest.mark.parametrize('fast', (False, True)) |
|
def test_empty_wallet_sync(setup_wallets, fast): |
|
wallet_service = create_wallet_for_sync([0, 0, 0, 0, 0], ['test_empty_wallet_sync']) |
|
|
|
sync_test_wallet(fast, wallet_service) |
|
|
|
broken = True |
|
for md in range(wallet_service.max_mixdepth + 1): |
|
for internal in (BaseWallet.ADDRESS_TYPE_INTERNAL, |
|
BaseWallet.ADDRESS_TYPE_EXTERNAL): |
|
broken = False |
|
assert 0 == wallet_service.get_next_unused_index(md, internal) |
|
assert not broken |
|
|
|
|
|
@pytest.mark.parametrize('fast,internal', ( |
|
(False, BaseWallet.ADDRESS_TYPE_EXTERNAL), |
|
(False, BaseWallet.ADDRESS_TYPE_INTERNAL), |
|
(True, BaseWallet.ADDRESS_TYPE_EXTERNAL), |
|
(True, BaseWallet.ADDRESS_TYPE_INTERNAL))) |
|
def test_sequentially_used_wallet_sync(setup_wallets, fast, internal): |
|
used_count = [1, 3, 6, 2, 23] |
|
wallet_service = create_wallet_for_sync( |
|
used_count, ['test_sequentially_used_wallet_sync'], |
|
populate_internal=internal) |
|
|
|
sync_test_wallet(fast, wallet_service) |
|
|
|
broken = True |
|
for md in range(len(used_count)): |
|
broken = False |
|
assert used_count[md] == wallet_service.get_next_unused_index(md, internal) |
|
assert not broken |
|
|
|
|
|
@pytest.mark.parametrize('fast', (False,)) |
|
def test_gap_used_wallet_sync(setup_wallets, fast): |
|
""" After careful examination this test now only includes the Recovery sync. |
|
Note: pre-Aug 2019, because of a bug, this code was not in fact testing both |
|
Fast and Recovery sync, but only Recovery (twice). Also, the scenario set |
|
out in this test (where coins are funded to a wallet which has no index-cache, |
|
and initially no imports) is only appropriate for recovery-mode sync, not for |
|
fast-mode (the now default). |
|
""" |
|
used_count = [1, 3, 6, 2, 23] |
|
wallet_service = create_wallet_for_sync(used_count, ['test_gap_used_wallet_sync']) |
|
wallet_service.gap_limit = 20 |
|
|
|
for md in range(len(used_count)): |
|
x = -1 |
|
for x in range(md): |
|
assert x <= wallet_service.gap_limit, "test broken" |
|
# create some unused addresses |
|
wallet_service.get_new_script(md, BaseWallet.ADDRESS_TYPE_INTERNAL) |
|
wallet_service.get_new_script(md, BaseWallet.ADDRESS_TYPE_EXTERNAL) |
|
used_count[md] += x + 2 |
|
jm_single().bc_interface.grab_coins(wallet_service.get_new_addr(md, |
|
BaseWallet.ADDRESS_TYPE_INTERNAL), 1) |
|
jm_single().bc_interface.grab_coins(wallet_service.get_new_addr(md, |
|
BaseWallet.ADDRESS_TYPE_EXTERNAL), 1) |
|
|
|
# reset indices to simulate completely unsynced wallet |
|
for md in range(wallet_service.max_mixdepth + 1): |
|
wallet_service.set_next_index(md, BaseWallet.ADDRESS_TYPE_INTERNAL, 0) |
|
wallet_service.set_next_index(md, BaseWallet.ADDRESS_TYPE_EXTERNAL, 0) |
|
sync_test_wallet(fast, wallet_service) |
|
|
|
broken = True |
|
for md in range(len(used_count)): |
|
broken = False |
|
assert md + 1 == wallet_service.get_next_unused_index(md, |
|
BaseWallet.ADDRESS_TYPE_INTERNAL) |
|
assert used_count[md] == wallet_service.get_next_unused_index(md, |
|
BaseWallet.ADDRESS_TYPE_EXTERNAL) |
|
assert not broken |
|
|
|
|
|
@pytest.mark.parametrize('fast', (False,)) |
|
def test_multigap_used_wallet_sync(setup_wallets, fast): |
|
""" See docstring for test_gap_used_wallet_sync; exactly the |
|
same applies here. |
|
""" |
|
start_index = 5 |
|
used_count = [start_index, 0, 0, 0, 0] |
|
wallet_service = create_wallet_for_sync(used_count, ['test_multigap_used_wallet_sync']) |
|
wallet_service.gap_limit = 5 |
|
|
|
mixdepth = 0 |
|
for w in range(5): |
|
for x in range(int(wallet_service.gap_limit * 0.6)): |
|
assert x <= wallet_service.gap_limit, "test broken" |
|
# create some unused addresses |
|
wallet_service.get_new_script(mixdepth, |
|
BaseWallet.ADDRESS_TYPE_INTERNAL) |
|
wallet_service.get_new_script(mixdepth, |
|
BaseWallet.ADDRESS_TYPE_EXTERNAL) |
|
used_count[mixdepth] += x + 2 |
|
jm_single().bc_interface.grab_coins(wallet_service.get_new_addr( |
|
mixdepth, BaseWallet.ADDRESS_TYPE_INTERNAL), 1) |
|
jm_single().bc_interface.grab_coins(wallet_service.get_new_addr( |
|
mixdepth, BaseWallet.ADDRESS_TYPE_EXTERNAL), 1) |
|
|
|
# reset indices to simulate completely unsynced wallet |
|
for md in range(wallet_service.max_mixdepth + 1): |
|
wallet_service.set_next_index(md, BaseWallet.ADDRESS_TYPE_INTERNAL, 0) |
|
wallet_service.set_next_index(md, BaseWallet.ADDRESS_TYPE_EXTERNAL, 0) |
|
|
|
sync_test_wallet(fast, wallet_service) |
|
|
|
assert used_count[mixdepth] - start_index == \ |
|
wallet_service.get_next_unused_index(mixdepth, |
|
BaseWallet.ADDRESS_TYPE_INTERNAL) |
|
assert used_count[mixdepth] == wallet_service.get_next_unused_index( |
|
mixdepth, BaseWallet.ADDRESS_TYPE_EXTERNAL) |
|
|
|
|
|
@pytest.mark.parametrize('fast', (False, True)) |
|
def test_retain_unused_indices_wallet_sync(setup_wallets, fast): |
|
used_count = [0, 0, 0, 0, 0] |
|
wallet_service = create_wallet_for_sync(used_count, |
|
['test_retain_unused_indices_wallet_sync']) |
|
|
|
for x in range(9): |
|
wallet_service.get_new_script(0, BaseWallet.ADDRESS_TYPE_INTERNAL) |
|
|
|
sync_test_wallet(fast, wallet_service) |
|
|
|
assert wallet_service.get_next_unused_index(0, |
|
BaseWallet.ADDRESS_TYPE_INTERNAL) == 9 |
|
|
|
|
|
@pytest.mark.parametrize('fast', (False, True)) |
|
def test_imported_wallet_sync(setup_wallets, fast): |
|
used_count = [0, 0, 0, 0, 0] |
|
wallet_service = create_wallet_for_sync(used_count, ['test_imported_wallet_sync']) |
|
source_wallet_service = create_wallet_for_sync(used_count, ['test_imported_wallet_sync_origin']) |
|
|
|
address = source_wallet_service.get_internal_addr(0) |
|
wallet_service.import_private_key(0, source_wallet_service.get_wif(0, 1, 0)) |
|
txid = binascii.unhexlify(jm_single().bc_interface.grab_coins(address, 1)) |
|
|
|
sync_test_wallet(fast, wallet_service) |
|
|
|
assert wallet_service._utxos.have_utxo(txid, 0) == 0 |
|
|
|
|
|
@pytest.fixture(scope='module') |
|
def setup_wallets(): |
|
load_test_config() |
|
jm_single().bc_interface.tick_forward_chain_interval = 1
|
|
|