GlcoinCore/specter-desktop
1
0
Fork 0
mirror of https://github.com/cryptoadvance/specter-desktop.git synced 2026-08-13 12:33:29 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

Chore: Refactor UTXO scan (and timeout bugfix for utxo_scan) (#1687)

Browse source 
* refactor fixtures and add testnet fixture

* make timeouts configurable

* test timeout more properly

* refactor utxo_scanning

* refactor psbt_creator and utxo_scanner into commands (pattern)

* basic testing based on testnet

* import fix

* Update tests/test_commands_utxo_scanner.py

Co-authored-by: Manolis <70536101+moneymanolis@users.noreply.github.com>

* Update tests/fix_testnet.py

* changes after the review

* black and updating black

Co-authored-by: Manolis <70536101+moneymanolis@users.noreply.github.com>
This commit is contained in:
Kim Neunert 2022-05-03 18:52:48 +02:00 • committed by GitHub
parent bb78f55e79
commit 9dbde39c2d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
18 changed files with 421 additions and 166 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/cryptoadvance/specter/api/rest/resource_psbt.py
View file

@ -6,7 +6,7 @@ from .base import (
)
from flask import current_app as app, request
from ...wallet import Wallet
from ...util.psbt_creator import PsbtCreator
from ...commands.psbt_creator import PsbtCreator
from .. import auth

2
src/cryptoadvance/specter/util/psbt_creator.py → src/cryptoadvance/specter/commands/psbt_creator.py
View file

@ -8,7 +8,7 @@ from cryptoadvance.specter.specter_error import SpecterError
from cryptoadvance.specter.util.common import str2bool
from ..helpers import is_testnet
from .descriptor import AddChecksum, Descriptor
from ..util.descriptor import AddChecksum, Descriptor
logger = logging.getLogger(__name__)

234
src/cryptoadvance/specter/commands/utxo_scanner.py Normal file
View file

@ -0,0 +1,234 @@
import threading
import logging
logger = logging.getLogger(__name__)
class UtxoScanner:
"""A Command class which simplifies the scanning of UTXOs and the import into a (core+specter) wallet"""
# 1. pass descriptor (change + recv) and range to rpc.scantxoutset
# get list of tx each having deriv-path
# 2. Update the address-indexes with the max from recv+change (to get fresh addresses and not used once)
# 3. rpc.getblockhash of height(each tx)
# 4. rpc.gettxoutproof get a proof that the tx is included in a block, needs tx_id and blockhash for each tx
# This will only work on fullnodes (!)
timeout = (
300 # 0.001 # As this is async, have longer timeout for all rpc-calls: 5 mins
)
def __init__(
self, wallet, explorer: str = None, requests_session=None, only_tor=False
):
self.wallet = wallet
self.explorer = explorer
if self.explorer:
self.explorer = self.explorer.rstrip("/")
self.requests_session = requests_session
self.only_tor = only_tor
self.error_msgs = []
def execute(self, asyncc=False):
if asyncc:
t = threading.Thread(
target=self._execute,
)
t.start()
else:
self._execute()
def _execute(self):
self.args = self.get_scantxoutset_args()
# get something like:
# [{'txid': '917a4d55...', 'vout': 1, 'scriptPubKey': '0014c313...5b19317',
# 'desc': "wpkh([e3b947d9/84'/1'/0'/1/0]0244290480...ab2f94d940a7a)#gey9w279",
# 'amount': 0.02119209, 'height': 1939628
# }]
self.unspents = self.wallet.rpc.scantxoutset(*self.args, timeout=self.timeout)[
"unspents"
]
logger.info(f"Found {len(self.unspents)} utxo")
self.adjust_keypools()
# Endgoal: calling importprunedfunds with the raw-transaction and a proof that the tx was included in a block
self.add_blockhashes_to_txs_in_unspents()
self.add_proofs_to_txs_in_unspents()
self.add_rawtxs_to_txs_in_unspents()
self.missing = [tx for tx in self.unspents if tx["raw"] is None]
self.existing = [tx for tx in self.unspents if tx["raw"] is not None]
# Now we're ready to import at least the existing txs
logger.info(f"Importing {len(self.existing)} utxos to core-wallet")
self.wallet.rpc.multi(
[("importprunedfunds", tx["raw"], tx["proof"]) for tx in self.existing],
timeout=self.timeout,
)
if len(self.missing) == 0:
logger.info(f"no more missing utxos. Rescan completed successfully!")
return self.execute_post_processing()
# let's continue with the missing ones via an explorer
if not self.check_explorer_working():
logger.error(f"Completed unsuccessfully!")
return self.execute_post_processing()
self.add_proofs_to_txs_in_missing_via_exlorer()
self.add_rawtxs_to_txs_in_missing_via_exlorer()
self.existing_via_explorer = [
tx for tx in self.missing if tx["raw"] is not None
]
self.missing_even_after_explorer = [
tx for tx in self.missing if tx["raw"] is None
]
# Now importing the missing ones
logger.info(
f"Importing {len(self.existing_via_explorer)} utxos to core-wallet (found via Explorer)"
)
self.wallet.rpc.multi(
[
("importprunedfunds", tx["raw"], tx["proof"])
for tx in self.existing_via_explorer
],
timeout=self.timeout,
)
if len(self.missing_even_after_explorer) != 0:
logger.error(
f"Even after explorer {len(self.missing_even_after_explorer)} txs could not be resolved"
)
logger.error(
f"Giving up, here is a list of the still missing TXIDs: {[ tx['txid'] for tx in self.missing_even_after_explorer]}"
)
self.error_msgs.append(
f"Some TXs could not be imported: {[ tx['txid'] for tx in self.missing_even_after_explorer]}"
)
return self.execute_post_processing()
def execute_post_processing(self):
self.wallet.fetch_transactions()
self.wallet.check_addresses()
def get_scantxoutset_args(self):
return [
"start",
[
{
"desc": self.wallet.recv_descriptor,
"range": max(self.wallet.keypool, 1000),
},
{
"desc": self.wallet.change_descriptor,
"range": max(self.wallet.change_keypool, 1000),
},
],
]
def adjust_keypools(self):
"""check the unspent for max indexes of recv+change addresses and adjusts the index of the wallet (if needed)
saves the wallet if changed.
"""
# check derivation indexes in found unspents (last 2 indexes in [brackets])
derivations = [
# purpose / account / cointype / change / index
# {... 'desc': "wpkh([e3b947d9/84'/1'/0'/1/0]02442...7a)#gey9w279"
tx["desc"].split("[")[1].split("]")[0].split("/")[-2:]
for tx in self.unspents
]
# we get a list of change / index parts of the derivations
# [['1', '0'],['1', '1'],['0', '2']] represents 2 change and one recv-address (index 2)
# get the maximum index for both address-types:
max_recv = max([-1] + [int(der[1]) for der in derivations if der[0] == "0"])
max_change = max([-1] + [int(der[1]) for der in derivations if der[0] == "1"])
updated = False
if max_recv >= self.wallet.address_index:
# skip to max_recv
self.wallet.address_index = max_recv
logger.info(f"Adjusted address_index of {self.wallet} to {max_change}")
# get next
self.wallet.getnewaddress(change=False, save=False)
updated = True
if max_change >= self.wallet.change_index:
# skip to max_change
self.wallet.change_index = max_change
logger.info(f"Adjusted change_index of {self.wallet} to {max_change}")
# get next
self.wallet.getnewaddress(change=True, save=False)
updated = True
# save only if needed
if updated:
self.wallet.save_to_file()
def add_blockhashes_to_txs_in_unspents(self):
"""includes the blockhash of the block each tx in unspents has been mined into"""
res = self.wallet.rpc.multi(
[("getblockhash", tx["height"]) for tx in self.unspents],
timeout=self.timeout,
)
block_hashes = [r["result"] for r in res]
for i, tx in enumerate(self.unspents):
# each tx in the unspents get the hash of the block it has been included in
tx["blockhash"] = block_hashes[i]
def add_proofs_to_txs_in_unspents(self):
"""includes the proof that the txs in unspents has been mined into the block"""
res = self.wallet.rpc.multi(
[("gettxoutproof", [tx["txid"]], tx["blockhash"]) for tx in self.unspents],
timeout=self.timeout,
)
proofs = [r["result"] for r in res]
for i, tx in enumerate(self.unspents):
tx["proof"] = proofs[i]
def add_rawtxs_to_txs_in_unspents(self):
"""includes the raw tx to the txs in unspents"""
res = self.wallet.rpc.multi(
[
("getrawtransaction", tx["txid"], False, tx["blockhash"])
for tx in self.unspents
],
timeout=self.timeout,
)
raws = [r["result"] for r in res]
for i, tx in enumerate(self.unspents):
tx["raw"] = raws[i]
def check_explorer_working(self):
"""returns a boolean if the explorer link is usable. Will log.error and fill self.error_msgs if not"""
if self.explorer is None:
logger.error(
f"Not all txs could be resolved and can't use explorer to get the rest"
)
self.error_msgs.append(
"Not all txs could be resolved and can't use explorer to get the rest"
)
return False
if not self.explorer.startswith("http"):
logger.error(f"explorer seem to have an invalid url: {self.explorer}")
self.error_msgs.append(
f"explorer seem to have an invalid url: {self.explorer}"
)
return False
request = self.requests_session.get(f"{self.explorer}")
if not request.ok:
logger.error(f"explorer seem to have an invalid url: {self.explorer}")
self.error_msgs.append(
f"explorer seem to have an invalid url: {self.explorer}"
)
return False
return True
def add_proofs_to_txs_in_missing_via_exlorer(self):
proofs = [
self.requests_session.get(
f"{self.explorer}/api/tx/{tx['txid']}/merkleblock-proof"
).text
for tx in self.missing
]
for i, tx in enumerate(self.missing):
tx["proof"] = proofs[i]
def add_rawtxs_to_txs_in_missing_via_exlorer(self):
raws = [
self.requests_session.get(f"{self.explorer}/api/tx/{tx['txid']}/hex").text
for tx in self.missing
]
for i, tx in enumerate(self.missing):
tx["raw"] = raws[i]

4
src/cryptoadvance/specter/config.py
View file

@ -251,8 +251,8 @@ class ProductionConfig(BaseConfig):
SECRET_KEY = secrets.token_urlsafe(16)
# There are some really slow machines out there. Creating a 2/4 multisig on an older MacBookAir
# Take already >30secs
BITCOIN_RPC_TIMEOUT = 60
LIQUID_RPC_TIMEOUT = 120
BITCOIN_RPC_TIMEOUT = float(os.getenv("BITCOIN_RPC_TIMEOUT", "60"))
LIQUID_RPC_TIMEOUT = float(os.getenv("LIQUID_RPC_TIMEOUT", "120"))
# Repeating it here as it's SECURITY CRITICAL. Check comments in BaseConfig
SERVICES_LOAD_FROM_CWD = False

5
src/cryptoadvance/specter/server_endpoints/wallets.py
View file

@ -4,7 +4,7 @@ import random
from functools import wraps
import requests
from cryptoadvance.specter.util.psbt_creator import PsbtCreator
from cryptoadvance.specter.commands.psbt_creator import PsbtCreator
from cryptoadvance.specter.util.wallet_importer import WalletImporter
from cryptoadvance.specter.wallet import Wallet
from cryptoadvance.specter.util.tx import is_hex, convert_rawtransaction_to_psbt
@ -785,6 +785,9 @@ def settings(wallet_alias):
)
app.specter.info["utxorescan"] = 1
app.specter.utxorescanwallet = wallet.alias
flash(
"Rescan started. Check the status bar on the left for progress and/or the logs for potential issues."
)
elif action == "abortrescanutxo":
app.specter.abortrescanutxo()
app.specter.info["utxorescan"] = None

3
src/cryptoadvance/specter/server_endpoints/wallets_api.py
View file

@ -9,7 +9,7 @@ from math import isnan
from numbers import Number
import requests
from cryptoadvance.specter.util.psbt_creator import PsbtCreator
from cryptoadvance.specter.commands.psbt_creator import PsbtCreator
from cryptoadvance.specter.wallet import Wallet
from flask import Blueprint, stream_with_context
from flask import current_app as app
@ -325,7 +325,6 @@ def rescan_progress(wallet_alias):
)
except SpecterError as se:
app.logger.error("SpecterError while get wallet rescan_progress: %s" % se)
return {}
@wallets_endpoint_api.route("/wallet/<wallet_alias>/get_label", methods=["POST"])

7
src/cryptoadvance/specter/services/extension_gen.py
View file

@ -80,7 +80,7 @@ class ExtGen:
def env_for_template(self, template):
"""chooses the right env for the template"""
if Path(template).name in ["conftest.py", "ghost_machine.py"]:
if Path(template).name in ["conftest.py", "fix_ghost_machine.py"]:
return self.sd_env
if Path(template).suffix.endswith("jinja"):
return self.jinja_env
@ -111,8 +111,9 @@ class ExtGen:
self.render(f"{package_path}/templates/dummy/components/dummy_tab.jinja")
self.render(f"tests/conftest.py", env=self.sd_env)
self.render(f"tests/ghost_machine.py", env=self.sd_env)
self.render(f"tests/devices_and_wallets.py", env=self.sd_env)
self.render(f"tests/fix_ghost_machine.py", env=self.sd_env)
self.render(f"tests/fix_devices_and_wallets.py", env=self.sd_env)
self.render(f"tests/fix_testnet.py", env=self.sd_env)
def create_binary_file(self, sourcepath):
"""textfiles can all be rendered. Binaries must be wgettet or copied"""

148
src/cryptoadvance/specter/wallet.py
View file

@ -15,6 +15,8 @@ from embit.transaction import Transaction
from io import StringIO
from typing import List
from cryptoadvance.specter.commands.utxo_scanner import UtxoScanner
from .addresslist import Address, AddressList
from .device import Device
from .key import Key
@ -1178,17 +1180,13 @@ class Wallet:
self.rpc.abandontransaction(txid)
def rescanutxo(self, explorer=None, requests_session=None, only_tor=False):
"""rescans the utxo via a thread. internally calls _rescan_utxo_thread
explorer: something like https://mempool.space/testnet/
"""
delete_file(self._transactions.path)
self.fetch_transactions()
t = threading.Thread(
target=self._rescan_utxo_thread,
args=(
explorer,
requests_session,
only_tor,
),
)
t.start()
command = UtxoScanner(self, explorer, requests_session, only_tor)
command.execute(asyncc=True)
def export_labels(self):
return self._addresses.get_labels()
@ -1206,132 +1204,6 @@ class Wallet:
for address in addresses:
self._addresses.set_label(address, label)
def _rescan_utxo_thread(self, explorer=None, requests_session=None, only_tor=False):
# rescan utxo is pretty fast,
# so we can check large range of addresses
# and adjust keypool accordingly
args = [
"start",
[
{"desc": self.recv_descriptor, "range": max(self.keypool, 1000)},
{
"desc": self.change_descriptor,
"range": max(self.change_keypool, 1000),
},
],
]
unspents = self.rpc.scantxoutset(*args)["unspents"]
# if keypool adjustments fails - not a big deal
try:
# check derivation indexes in found unspents (last 2 indexes in [brackets])
derivations = [
tx["desc"].split("[")[1].split("]")[0].split("/")[-2:]
for tx in unspents
]
# get max derivation for change and receive branches
max_recv = max([-1] + [int(der[1]) for der in derivations if der[0] == "0"])
max_change = max(
[-1] + [int(der[1]) for der in derivations if der[0] == "1"]
)
updated = False
if max_recv >= self.address_index:
# skip to max_recv
self.address_index = max_recv
# get next
self.getnewaddress(change=False, save=False)
updated = True
while max_change >= self.change_index:
# skip to max_change
self.change_index = max_change
# get next
self.getnewaddress(change=True, save=False)
updated = True
# save only if needed
if updated:
self.save_to_file()
except Exception as e:
logger.warning(f"Failed to get derivation path from utxo transaction: {e}")
# keep working with unspents
res = self.rpc.multi([("getblockhash", tx["height"]) for tx in unspents])
block_hashes = [r["result"] for r in res]
for i, tx in enumerate(unspents):
tx["blockhash"] = block_hashes[i]
res = self.rpc.multi(
[("gettxoutproof", [tx["txid"]], tx["blockhash"]) for tx in unspents]
)
proofs = [r["result"] for r in res]
for i, tx in enumerate(unspents):
tx["proof"] = proofs[i]
res = self.rpc.multi(
[
("getrawtransaction", tx["txid"], False, tx["blockhash"])
for tx in unspents
]
)
raws = [r["result"] for r in res]
for i, tx in enumerate(unspents):
tx["raw"] = raws[i]
missing = [tx for tx in unspents if tx["raw"] is None]
existing = [tx for tx in unspents if tx["raw"] is not None]
self.rpc.multi(
[("importprunedfunds", tx["raw"], tx["proof"]) for tx in existing]
)
# handle missing transactions now
# if Tor is running, requests will be sent over Tor
if explorer is not None:
# make sure there is no trailing /
explorer = explorer.rstrip("/")
try:
# get raw transactions
raws = [
requests_session.get(f"{explorer}/api/tx/{tx['txid']}/hex").text
for tx in missing
]
# get proofs
proofs = [
requests_session.get(
f"{explorer}/api/tx/{tx['txid']}/merkleblock-proof"
).text
for tx in missing
]
# import funds
self.rpc.multi(
[
("importprunedfunds", raws[i], proofs[i])
for i in range(len(raws))
]
)
except Exception as e:
logger.warning(f"Failed to fetch data from block explorer: {e}")
# retry if using requests_session failed
if not only_tor:
try:
# get raw transactions
raws = [
requests.get(f"{explorer}/api/tx/{tx['txid']}/hex").text
for tx in missing
]
# get proofs
proofs = [
requests.get(
f"{explorer}/api/tx/{tx['txid']}/merkleblock-proof"
).text
for tx in missing
]
# import funds
self.rpc.multi(
[
("importprunedfunds", raws[i], proofs[i])
for i in range(len(raws))
]
)
except:
logger.warning(f"Failed to fetch data from block explorer: {e}")
self.fetch_transactions()
self.check_addresses()
@property
def rescan_progress(self):
"""Returns None if rescanblockchain is not launched,
@ -1395,9 +1267,9 @@ class Wallet:
def get_address(self, index, change=False, check_keypool=True) -> str:
if check_keypool:
pool = self.change_keypool if change else self.keypool
logger.debug(
f"get_address index={index} pool={pool} gapLIMIT={self.GAP_LIMIT} change={change} will_keypoolrefill={pool < index + self.GAP_LIMIT}"
)
# logger.debug(
# f"get_address index={index} pool={pool} gapLIMIT={self.GAP_LIMIT} change={change} will_keypoolrefill={pool < index + self.GAP_LIMIT}"
# )
if pool < index + self.GAP_LIMIT:
self.keypoolrefill(pool, index + self.GAP_LIMIT, change=change)
return self.descriptor.derive(index, branch_index=int(change)).address(

2
test_requirements.txt
View file

@ -1,5 +1,5 @@
# requirements for testing
black==21.4b0
black==22.3.0
pre-commit==2.13.0
docker==4.3.1
pip-tools==5.5.0

2
tests/conftest.py
View file

@ -28,7 +28,7 @@ from cryptoadvance.specter.util.wallet_importer import WalletImporter
logger = logging.getLogger(__name__)
pytest_plugins = ["ghost_machine", "devices_and_wallets"]
pytest_plugins = ["fix_ghost_machine", "fix_devices_and_wallets", "fix_testnet"]
# This is from https://stackoverflow.com/questions/132058/showing-the-stack-trace-from-a-running-python-application
# it enables stopping a hanging test via sending the pytest-process a SIGUSR2 (12)

0
tests/devices_and_wallets.py → tests/fix_devices_and_wallets.py
View file

0
tests/ghost_machine.py → tests/fix_ghost_machine.py
View file

59
tests/fix_testnet.py Normal file
View file

@ -0,0 +1,59 @@
""" A set of fixtures which assume a testnet-node on localhost. This can be helpfull while developing.
"""
from cryptoadvance.specter.user import User, hash_password
import pytest
from cryptoadvance.specter.specter import Specter
from cryptoadvance.specter.user import User
@pytest.fixture
def specter_testnet_configured(bitcoin_regtest, devices_filled_data_folder):
"""This assumes a bitcoin-testnet-node is running on loalhost"""
config = {
"rpc": {
"autodetect": False,
"datadir": "",
"user": "bitcoin", # change this to your credential in bitcoin.conf (for testnet)
"password": "secret",
"port": 18332,
"host": "localhost",
"protocol": "http",
},
"auth": {
"method": "rpcpasswordaspin",
},
}
specter = Specter(data_folder=devices_filled_data_folder, config=config)
specter.check()
assert specter.chain == "test"
# Create a User
someuser = specter.user_manager.add_user(
User.from_json(
user_dict={
"id": "someuser",
"username": "someuser",
"password": hash_password("somepassword"),
"config": {},
"is_admin": False,
"services": None,
},
specter=specter,
)
)
specter.user_manager.save()
specter.check()
assert not specter.wallet_manager.working_folder is None
try:
yield specter
finally:
# Deleting all Wallets (this will also purge them on core)
for user in specter.user_manager.users:
for wallet in list(user.wallet_manager.wallets.values()):
user.wallet_manager.delete_wallet(
wallet, bitcoin_datadir=bitcoin_regtest.datadir, chain="regtest"
)

2
tests/test_util_psbt_creator.py → tests/test_commands_psbt_creator.py
View file

@ -1,7 +1,7 @@
import logging
from cryptoadvance.specter.key import Key
from cryptoadvance.specter.util.descriptor import Descriptor
from cryptoadvance.specter.util.psbt_creator import PsbtCreator
from cryptoadvance.specter.commands.psbt_creator import PsbtCreator
from mock import MagicMock, call, patch

87
tests/test_commands_utxo_scanner.py Normal file
View file

@ -0,0 +1,87 @@
import logging
from random import randint
import time
import pytest
from cryptoadvance.specter.commands.utxo_scanner import UtxoScanner
from cryptoadvance.specter.specter import Specter
from cryptoadvance.specter.wallet import Wallet
from fix_devices_and_wallets import create_hot_wallet_device, create_hot_wallet_with_ID
@pytest.mark.skip()
def test_rescan_utxo(specter_testnet_configured: Specter, caplog):
caplog.set_level(logging.DEBUG)
logging.getLogger("urllib3.connectionpool").setLevel(logging.INFO)
specter: Specter = specter_testnet_configured
is_pruned_node = specter.rpc.getblockchaininfo()["pruned"]
assert is_pruned_node
# You should not change the UTXO-Set of this wallet ... ever!
# Satoshi himself will punish you if you ever move these UTXOs !
hot_device = create_hot_wallet_device(
specter_testnet_configured,
"hold_accident" + str(randint(0, 100000)),
11 * "hold " + "accident",
)
assert hot_device
wallet: Wallet = create_hot_wallet_with_ID(
specter_testnet_configured, hot_device, "hold_accident"
)
if is_pruned_node:
# The pruned node on testnet might have incredible lots of TXs
mycmd = UtxoScanner(wallet)
mycmd.execute(asyncc=False)
# check_utxo is not part of the execution (but maybe should?!).
# It's usually called from the server_endpoints
wallet.check_utxo()
utxos = wallet.full_utxo
assert len(utxos) == 6
# assert_exact_utxo_set(utxos)
# When the txs are no longer in th pruned-set, this should work:
# With an explorer, it should work on a pruned_node:
mycmd = UtxoScanner(wallet, explorer="https://mempool.space/testnet/")
mycmd.execute(asyncc=False)
utxos = wallet.full_utxo
assert len(utxos) == 6
assert_exact_utxo_set(utxos)
else:
# With a full node, you don't need any explorer
mycmd = UtxoScanner(wallet)
mycmd.execute(asyncc=False)
utxos = wallet.full_utxo
assert len(utxos) == 6
assert_exact_utxo_set(utxos)
assert False
def assert_exact_utxo_set(utxos):
assert len(utxos) == 6
assert (utxos[0]["address"], utxos[0]["amount"]) == (
"tb1q2e5eev0wpz72eew7g5ypl0xeg38sm6tx2z50nm",
0.00020745,
)
assert (utxos[1]["address"], utxos[1]["amount"]) == (
"tb1qs74297wdnd0wmztekcmz3wnd6f6c3gljuhj56v",
0.00039557,
)
assert (utxos[2]["address"], utxos[2]["amount"]) == (
"tb1qk4e29xa2cxxy02yq6glpwet6hkgdcjfz8gnnwk",
0.00209453,
)
assert (utxos[3]["address"], utxos[3]["amount"]) == (
"tb1q7uqexf8yhcvcx04trf5cx0cls53jyq8lynsmns",
0.00231033,
)
assert (utxos[4]["address"], utxos[4]["amount"]) == (
"tb1q9lgm78033652xt0e0t3yqzjlu9uzf0fdwpelkq",
0.0008,
)
assert (utxos[5]["address"], utxos[5]["amount"]) == (
"tb1qhxrx4nyxfesq3vp9rrae6q8zxsayl6ndwmhmv0",
0.0006,
)

2
tests/test_rest.py
View file

@ -9,7 +9,7 @@ from cryptoadvance.specter.specter import Specter
from cryptoadvance.specter.specter_error import SpecterError
from cryptoadvance.specter.util.wallet_importer import WalletImporter
from devices_and_wallets import create_hot_wallet_with_ID
from fix_devices_and_wallets import create_hot_wallet_with_ID
def almost_equal(a: Number, b: Number, precision: float = 0.01) -> bool:

23
tests/test_rpc.py
View file

@ -28,19 +28,24 @@ def test_BitcoinRpc(bitcoin_regtest):
def test_BitcoinRpc_timeout(bitcoin_regtest, caplog):
brt = bitcoin_regtest # stupid long name
BitcoinRPC.default_timeout = 0.001
rpc = BitcoinRPC(
brt.rpcconn.rpcuser,
brt.rpcconn.rpcpassword,
host=brt.rpcconn.ipaddress,
port=brt.rpcconn.rpcport,
)
rpc.timeout = 0.0000000000001
try:
with pytest.raises(SpecterError) as se:
rpc.createwallet("some_test_wallet_name_392")
assert False, "Should raise an exception"
except SpecterError:
assert "Timeout after " in caplog.text
assert (
"while BitcoinRPC call( ) payload:[{'method': 'createwallet', 'params': ['some_test_wallet_name_392'], 'jsonrpc': '2.0', 'id': 0}]"
in caplog.text
)
assert "Timeout after 0.001" in str(se.value)
assert (
"while BitcoinRPC call( ) payload:[{'method': 'createwallet', 'params': ['some_test_wallet_name_392'], 'jsonrpc': '2.0', 'id': 0}]"
in caplog.text
)
rpc.timeout = 0.0001
with pytest.raises(SpecterError) as se:
rpc.createwallet("some_test_wallet_name_393")
assert "Timeout after 0.0001" in str(se.value)
assert "Timeout after 0.001" in caplog.text

5
tests/test_wallet.py
View file

@ -1,10 +1,5 @@
import json, logging, pytest, time, os
from cryptoadvance.specter.specter import Specter
from cryptoadvance.specter.managers.wallet_manager import WalletManager
from cryptoadvance.specter.wallet import Wallet
from conftest import instantiate_bitcoind_controller
logger = logging.getLogger(__name__)
def test_check_utxo_and_amounts(