Skip to content

Instantly share code, notes, and snippets.

@k9ert

k9ert/failing_persistence_test

Created July 22, 2021 14:02
Show Gist options
  • Save k9ert/4a8efeca81dc88468f66b37a891e58a6 to your computer and use it in GitHub Desktop.
Save k9ert/4a8efeca81dc88468f66b37a891e58a6 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
failing_persistence_test
============================= test session starts ==============================
platform linux -- Python 3.8.10, pytest-6.1.2, py-1.10.0, pluggy-0.13.1
rootdir: /home/kim/src/specter-desktop, configfile: pytest.ini
plugins: cov-2.10.1
collected 104 items
tests/test_checker.py . [ 0%]
tests/test_cli_bitcoind.py .. [ 2%]
tests/test_cli_server.py .... [ 6%]
tests/test_config.py . [ 7%]
tests/test_controller.py ... [ 10%]
tests/test_device_manager.py . [ 11%]
tests/test_hwibridge.py ........s [ 20%]
tests/test_managers_node.py . [ 21%]
tests/test_node.py . [ 22%]
tests/test_node_controller.py ss [ 24%]
tests/test_persistence.py EEE [ 26%]
tests/test_rest.py EE [ 28%]
tests/test_rpc.py F [ 29%]
tests/test_specter.py EF [ 31%]
tests/test_util_wallet_importer.py E [ 32%]
tests/test_wallet_manager.py FFFFFEE [ 39%]
tests/test_decodetx.py . [ 40%]
tests/test_device_manager.py . [ 41%]
tests/test_helpers.py ... [ 44%]
tests/test_internal_node.py s [ 45%]
tests/test_key.py ..... [ 50%]
tests/test_liquid_rpc.py F [ 50%]
tests/test_managers_config.py . [ 51%]
tests/test_managers_otp.py . [ 52%]
tests/test_merkleblock.py .............. [ 66%]
tests/test_node.py F [ 67%]
tests/test_node_controller.py . [ 68%]
tests/test_specter.py . [ 69%]
tests/test_tor_daemon.py s [ 70%]
tests/test_util_descriptor.py .................. [ 87%]
tests/test_util_psbt_creator.py ... [ 90%]
tests/test_util_shell.py .. [ 92%]
tests/test_util_wallet_importer.py . [ 93%]
tests/test_util_xpub.py ....... [100%]
==================================== ERRORS ====================================
_________________ ERROR at setup of test_write_devices[False] __________________
self = <cryptoadvance.specter.util.wallet_importer.WalletImporter object at 0x7f83086afd30>
wallet_manager = <cryptoadvance.specter.managers.wallet_manager.WalletManager object at 0x7f83085b4790>
def create_wallet(self, wallet_manager):
"""creates the wallet. Assumes all devices are there (create with create_nonexisting_signers)
will also keypoolrefill and import_labels
"""
try:
> self.wallet = wallet_manager.create_wallet(
name=self.wallet_name,
sigs_required=self.descriptor.multisig_M,
key_type=self.descriptor.address_type,
keys=self.keys,
devices=self.cosigners,
)
src/cryptoadvance/specter/util/wallet_importer.py:90:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <cryptoadvance.specter.managers.wallet_manager.WalletManager object at 0x7f83085b4790>
name = 'a_simple_wallet', sigs_required = 1, key_type = 'wpkh'
keys = [<cryptoadvance.specter.key.Key object at 0x7f8308790d30>]
devices = [<cryptoadvance.specter.devices.trezor.Trezor object at 0x7f8308790c40>]
def create_wallet(self, name, sigs_required, key_type, keys, devices):
try:
walletsindir = [
wallet["name"] for wallet in self.rpc.listwalletdir()["wallets"]
]
except:
walletsindir = []
wallet_alias = alias(name)
i = 2
while (
os.path.isfile(os.path.join(self.working_folder, "%s.json" % wallet_alias))
or os.path.join(self.rpc_path, wallet_alias) in walletsindir
):
wallet_alias = alias("%s %d" % (name, i))
i += 1
> w = self.WalletClass.create(
self.rpc,
self.rpc_path,
self.working_folder,
self.device_manager,
self,
name,
wallet_alias,
sigs_required,
key_type,
keys,
devices,
self.bitcoin_core_version_raw,
)
src/cryptoadvance/specter/managers/wallet_manager.py:304:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
cls = <class 'cryptoadvance.specter.wallet.Wallet'>
rpc = <BitcoinRpc http://localhost:18543>, rpc_path = 'specter'
working_folder = '/tmp/specter_home_tmp_q9xx8fo9/wallets_someuser/regtest'
device_manager = <cryptoadvance.specter.managers.device_manager.DeviceManager object at 0x7f83087a0640>
wallet_manager = <cryptoadvance.specter.managers.wallet_manager.WalletManager object at 0x7f83085b4790>
name = 'a_simple_wallet', alias = 'a_simple_wallet', sigs_required = 1
key_type = 'wpkh'
keys = [<cryptoadvance.specter.key.Key object at 0x7f8308790d30>]
devices = [<cryptoadvance.specter.devices.trezor.Trezor object at 0x7f8308790c40>]
core_version = 210100
@classmethod
def create(
cls,
rpc,
rpc_path,
working_folder,
device_manager,
wallet_manager,
name,
alias,
sigs_required,
key_type,
keys,
devices,
core_version=None,
):
"""Creates a wallet. If core_version is not specified - get it from rpc"""
# get xpubs in a form [fgp/der]xpub from all keys
xpubs = [key.metadata["combined"] for key in keys]
recv_keys = ["%s/0/*" % xpub for xpub in xpubs]
change_keys = ["%s/1/*" % xpub for xpub in xpubs]
is_multisig = len(keys) > 1
# we start by constructing an argument for descriptor wrappers
if is_multisig:
recv_descriptor = "sortedmulti({},{})".format(
sigs_required, ",".join(recv_keys)
)
change_descriptor = "sortedmulti({},{})".format(
sigs_required, ",".join(change_keys)
)
else:
recv_descriptor = recv_keys[0]
change_descriptor = change_keys[0]
# now we iterate over script-type in reverse order
# to get sh(wpkh(xpub)) from sh-wpkh and xpub
arr = key_type.split("-")
for el in arr[::-1]:
recv_descriptor = "%s(%s)" % (el, recv_descriptor)
change_descriptor = "%s(%s)" % (el, change_descriptor)
recv_descriptor = AddChecksum(recv_descriptor)
change_descriptor = AddChecksum(change_descriptor)
if not recv_descriptor != change_descriptor:
raise SpecterError(
f"The recv_descriptor ({recv_descriptor}) is the same than the change_descriptor ({change_descriptor})"
)
# get Core version if we don't know it
if core_version is None:
core_version = rpc.getnetworkinfo().get("version", 0)
use_descriptors = core_version >= 210000
if use_descriptors:
# Use descriptor wallet
rpc.createwallet(os.path.join(rpc_path, alias), True, True, "", False, True)
else:
rpc.createwallet(os.path.join(rpc_path, alias), True)
wallet_rpc = rpc.wallet(os.path.join(rpc_path, alias))
# import descriptors
args = [
{
"desc": desc,
"internal": change,
"timestamp": "now",
"watchonly": True,
}
for (change, desc) in [(False, recv_descriptor), (True, change_descriptor)]
]
for arg in args:
if use_descriptors:
arg["active"] = True
else:
arg["keypool"] = True
arg["range"] = [0, cls.GAP_LIMIT]
if not args[0] != args[1]:
raise SpecterError(f"{args[0]} is equal {args[1]}")
# Descriptor wallets were introduced in v0.21.0, but upgraded nodes may
# still have legacy wallets. Use getwalletinfo to check the wallet type.
# The "keypool" for descriptor wallets is automatically refilled
if use_descriptors:
res = wallet_rpc.importdescriptors(args)
else:
res = wallet_rpc.importmulti(args, {"rescan": False})
if not all([r["success"] for r in res]):
all_issues = " and ".join(
r["error"]["message"] for r in res if r["success"] == False
)
raise SpecterError(all_issues)
> return cls(
name,
alias,
"{} of {} {}".format(sigs_required, len(keys), purposes[key_type])
if len(keys) > 1
else purposes[key_type],
addrtypes[key_type],
"",
-1,
"",
-1,
0,
0,
recv_descriptor,
change_descriptor,
keys,
devices,
sigs_required,
{},
[],
os.path.join(working_folder, "%s.json" % alias),
device_manager,
wallet_manager,
)
src/cryptoadvance/specter/wallet.py:234:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <Wallet name=a_simple_wallet alias=a_simple_wallet>
name = 'a_simple_wallet', alias = 'a_simple_wallet'
description = 'Single (Segwit)', address_type = 'bech32', address = ''
address_index = -1, change_address = '', change_index = -1, keypool = 0
change_keypool = 0
recv_descriptor = 'wpkh([1ef4e492/84h/1h/0h]tpubDC5EUwdy9WWpzqMWKNhVmXdMgMbi4ywxkdysRdNr1MdM4SCfVLbNtsFvzY6WKSuzsaVAitj6FmP6TugPuNT6yKZDLsHrSwMd816TnqX7kuc/0/*)#xp8lv5nr'
change_descriptor = 'wpkh([1ef4e492/84h/1h/0h]tpubDC5EUwdy9WWpzqMWKNhVmXdMgMbi4ywxkdysRdNr1MdM4SCfVLbNtsFvzY6WKSuzsaVAitj6FmP6TugPuNT6yKZDLsHrSwMd816TnqX7kuc/1/*)#h4z73prm'
keys = [<cryptoadvance.specter.key.Key object at 0x7f8308790d30>]
devices = [<cryptoadvance.specter.devices.trezor.Trezor object at 0x7f8308790c40>]
sigs_required = 1, pending_psbts = {}, frozen_utxo = []
fullpath = '/tmp/specter_home_tmp_q9xx8fo9/wallets_someuser/regtest/a_simple_wallet.json'
device_manager = <cryptoadvance.specter.managers.device_manager.DeviceManager object at 0x7f83087a0640>
manager = <cryptoadvance.specter.managers.wallet_manager.WalletManager object at 0x7f83085b4790>
old_format_detected = False, last_block = None
def __init__(
self,
name,
alias,
description,
address_type,
address,
address_index,
change_address,
change_index,
keypool,
change_keypool,
recv_descriptor,
change_descriptor,
keys,
devices,
sigs_required,
pending_psbts,
frozen_utxo,
fullpath,
device_manager,
manager,
old_format_detected=False,
last_block=None,
):
self.name = name
self.alias = alias
self.description = description
self.address_type = address_type
self.address = address
self.address_index = address_index
self.change_address = change_address
self.change_index = change_index
self.keypool = keypool
self.change_keypool = change_keypool
self.recv_descriptor = recv_descriptor
self.change_descriptor = change_descriptor
self.keys = keys
self.devices = [
(
device
if isinstance(device, Device)
else device_manager.get_by_alias(device)
)
for device in devices
]
if None in self.devices:
raise Exception("A device used by this wallet could not have been found!")
self.sigs_required = int(sigs_required)
self.pending_psbts = pending_psbts
self.frozen_utxo = frozen_utxo
self.fullpath = fullpath
self.manager = manager
self.rpc = self.manager.rpc.wallet(
os.path.join(self.manager.rpc_path, self.alias)
)
self.last_block = last_block
addr_path = self.fullpath.replace(".json", "_addr.csv")
self._addresses = self.AddressListCls(addr_path, self.rpc)
if not self._addresses.file_exists:
self.fetch_labels()
txs_path = self.fullpath.replace(".json", "_txs.csv")
self._transactions = self.TxListCls(
txs_path, self.rpc, self._addresses, self.manager.chain
)
if address == "":
> self.getnewaddress()
src/cryptoadvance/specter/wallet.py:133:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <Wallet name=a_simple_wallet alias=a_simple_wallet>, change = False
save = True
def getnewaddress(self, change=False, save=True):
if change:
self.change_index += 1
index = self.change_index
else:
self.address_index += 1
index = self.address_index
> address = self.get_address(index, change=change)
src/cryptoadvance/specter/wallet.py:1128:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <Wallet name=a_simple_wallet alias=a_simple_wallet>, index = 0
change = False, check_keypool = True
def get_address(self, index, change=False, check_keypool=True):
if check_keypool:
pool = self.change_keypool if change else self.keypool
if pool < index + self.GAP_LIMIT:
> self.keypoolrefill(pool, index + self.GAP_LIMIT, change=change)
src/cryptoadvance/specter/wallet.py:1141:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <Wallet name=a_simple_wallet alias=a_simple_wallet>, start = 0, end = 20
change = False
def keypoolrefill(self, start, end=None, change=False):
if end is None:
# end is ignored for descriptor wallets
end = start + self.GAP_LIMIT
desc = self.recv_descriptor if not change else self.change_descriptor
args = [
{
"desc": desc,
"internal": change,
"timestamp": "now",
"watchonly": True,
}
]
if self.use_descriptors:
args[0]["active"] = True
else:
args[0]["keypool"] = True
args[0]["range"] = [start, end]
try:
addresses = [
dict(
address=self.get_address(idx, change=change, check_keypool=False),
index=idx,
change=change,
)
for idx in range(start, end)
]
self._addresses.add(addresses, check_rpc=False)
except Exception as e:
logger.warn(f"Error while calculating addresses: {e}")
# Descriptor wallets were introduced in v0.21.0, but upgraded nodes may
# still have legacy wallets. Use getwalletinfo to check the wallet type.
# The "keypool" for descriptor wallets is automatically refilled
if not self.use_descriptors:
r = self.rpc.importmulti(args, {"rescan": False})
if change:
self.change_keypool = end
else:
self.keypool = end
> self.save_to_file()
src/cryptoadvance/specter/wallet.py:1327:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <Wallet name=a_simple_wallet alias=a_simple_wallet>
def save_to_file(self):
> write_json_file(self.to_json(), self.fullpath)
src/cryptoadvance/specter/wallet.py:675:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <Wallet name=a_simple_wallet alias=a_simple_wallet>, for_export = False
def to_json(self, for_export=False):
o = {
"name": self.name,
"alias": self.alias,
"description": self.description,
"address_type": self.address_type,
"address": self.address,
"address_index": self.address_index,
"change_address": self.change_address,
"change_index": self.change_index,
"keypool": self.keypool,
"change_keypool": self.change_keypool,
"recv_descriptor": self.recv_descriptor,
"change_descriptor": self.change_descriptor,
"keys": [key.json for key in self.keys],
"devices": [device.alias for device in self.devices],
"sigs_required": self.sigs_required,
> "blockheight": self.blockheight,
}
src/cryptoadvance/specter/wallet.py:664:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <Wallet name=a_simple_wallet alias=a_simple_wallet>
@property
def blockheight(self):
> self.fetch_transactions()
src/cryptoadvance/specter/wallet.py:1093:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <Wallet name=a_simple_wallet alias=a_simple_wallet>
def fetch_transactions(self):
"""Load transactions from Bitcoin Core"""
arr = []
idx = 0
# unconfirmed_selftransfers needed since Bitcoin Core does not properly list `selftransfer` txs in `listtransactions` command
# Until v0.21, it listed there consolidations to a receive address, but not change address
# Since v0.21, it does not list there consolidations at all
# Therefore we need to check here if a transaction might got confirmed
# NOTE: This might be a problem in case of re-org...
# More details: https://github.com/cryptoadvance/specter-desktop/issues/996
unconfirmed_selftransfers = [
txid
for txid in self._transactions
if self._transactions[txid].get("category", "") == "selftransfer"
and not self._transactions[txid].get("blockhash", None)
]
unconfirmed_selftransfers_txs = []
if unconfirmed_selftransfers:
unconfirmed_selftransfers_txs = self.rpc.multi(
[("gettransaction", txid) for txid in unconfirmed_selftransfers]
)
while True:
txlist = (
self.rpc.listtransactions(
"*",
LISTTRANSACTIONS_BATCH_SIZE,
LISTTRANSACTIONS_BATCH_SIZE * idx,
True,
)
+ [tx["result"] for tx in unconfirmed_selftransfers_txs]
)
# list of transactions that we don't know about,
# or that it has a different blockhash (reorg / confirmed)
# or doesn't have an address(?)
# or has wallet conflicts
res = [
tx
for tx in txlist
if tx["txid"] not in self._transactions
or not self._transactions[tx["txid"]].get("address", None)
or self._transactions[tx["txid"]].get("blockhash", None)
!= tx.get("blockhash", None)
or (
self._transactions[tx["txid"]].get("blockhash", None)
and not self._transactions[tx["txid"]].get("blockheight", None)
) # Fix for Core v19 with Specter v1
or self._transactions[tx["txid"]].get("conflicts", [])
!= tx.get("walletconflicts", [])
]
# TODO: Looks like Core ignore a consolidation (self-transfer) going into the change address (in listtransactions)
# This means it'll show unconfirmed for us forever...
arr.extend(res)
idx += 1
# not sure if Core <20 returns last batch or empty array at the end
if (
len(res) < LISTTRANSACTIONS_BATCH_SIZE
or len(arr) < LISTTRANSACTIONS_BATCH_SIZE * idx
):
break
txs = dict.fromkeys([a["txid"] for a in arr])
txids = list(txs.keys())
# get all raw transactions
res = self.rpc.multi([("gettransaction", txid) for txid in txids])
for i, r in enumerate(res):
txid = txids[i]
# check if we already added it
if txs.get(txid, None) is not None:
continue
txs[txid] = r["result"]
# This is a fix for Bitcoin Core versions < v0.20
# These do not return the blockheight as part of the `gettransaction` command
# So here we check if this property is lacking and if so
# query the current block height and manually calculate it.
##################### Remove from here after dropping Core v0.19 support #####################
check_blockheight = False
for tx in txs.values():
if tx and tx.get("confirmations", 0) > 0 and "blockheight" not in tx:
check_blockheight = True
break
if check_blockheight:
current_blockheight = self.rpc.getblockcount()
for tx in txs.values():
if tx.get("confirmations", 0) > 0:
tx["blockheight"] = current_blockheight - tx["confirmations"] + 1
##################### Remove until here after dropping Core v0.19 support #####################
self._transactions.add(txs)
if self.use_descriptors:
while (
len(
[
tx
for tx in self._transactions
if self._transactions[tx]["category"] != "send"
and not self._transactions[tx]["address"]
]
)
!= 0
):
addresses = [
dict(
address=self.get_address(
idx, change=False, check_keypool=False
),
index=idx,
change=False,
)
for idx in range(
self._addresses.max_index(change=False),
self._addresses.max_index(change=False) + self.GAP_LIMIT,
)
]
change_addresses = [
dict(
address=self.get_address(idx, change=True, check_keypool=False),
index=idx,
change=True,
)
for idx in range(
self._addresses.max_index(change=True),
self._addresses.max_index(change=True) + self.GAP_LIMIT,
)
]
> self._addresses.add(addresses, check_rpc=False)
src/cryptoadvance/specter/wallet.py:402:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = {'bcrt1qcatuhg0gll3h7py4cmn53rjjn9xlsqfwj3zcej': Address(Address #0, bcrt1qcatuhg0gll3h7py4cmn53rjjn9xlsqfwj3zcej), 'b... 'bcrt1qx7sxzqutlu8uawl6lujyxe7xw7lluunz7m659y': Address(Address #11001, bcrt1qx7sxzqutlu8uawl6lujyxe7xw7lluunz7m659y)}
arr = [{'address': 'bcrt1q4h4nzmj5jxan2w7am9rfcrtekym2rfpg2q84m2', 'change': False, 'index': 10982}, {'address': 'bcrt1q8j0m...se, 'index': 10986}, {'address': 'bcrt1q6e634uw39zdgh50cjfrkkwc0carhcnhu9pcdv9', 'change': False, 'index': 10987}, ...]
check_rpc = False
def add(self, arr, check_rpc=False):
"""arr should be a list of dicts"""
labeled_addresses = {}
if check_rpc:
# get all available labels
labels = self.rpc.listlabels()
if "" in labels:
labels.remove("")
# get addresses for all labels
res = self.rpc.multi([("getaddressesbylabel", label) for label in labels])
for i, result in enumerate(res):
label = labels[i]
# go through all addresses and assign labels
for k in result["result"].keys():
labeled_addresses[k] = label
# go through all addresses and assign
for addr in arr:
if addr["address"] in self:
continue
# if we found a label for it - import
if addr["address"] in labeled_addresses:
addr["label"] = labeled_addresses[addr["address"]]
self[addr["address"]] = self.AddressCls(self.rpc, **addr)
# add all labeled addresses but not from the array (destination)
for addr in labeled_addresses:
if addr not in self:
self[addr] = self.AddressCls(
self.rpc,
address=addr,
label=labeled_addresses[addr],
change=None,
index=None,
)
> self.save()
src/cryptoadvance/specter/addresslist.py:151:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = {'bcrt1qcatuhg0gll3h7py4cmn53rjjn9xlsqfwj3zcej': Address(Address #0, bcrt1qcatuhg0gll3h7py4cmn53rjjn9xlsqfwj3zcej), 'b... 'bcrt1qx7sxzqutlu8uawl6lujyxe7xw7lluunz7m659y': Address(Address #11001, bcrt1qx7sxzqutlu8uawl6lujyxe7xw7lluunz7m659y)}
def save(self):
if len(list(self.keys())) > 0:
> write_csv(self.path, list(self.values()), self.AddressCls)
src/cryptoadvance/specter/addresslist.py:115:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
fname = '/tmp/specter_home_tmp_q9xx8fo9/wallets_someuser/regtest/a_simple_wallet_addr.csv'
objs = [Address(Address #0, bcrt1qcatuhg0gll3h7py4cmn53rjjn9xlsqfwj3zcej), Address(Address #1, bcrt1qsqnuk9hulcfta7kj7687favj... bcrt1q2zv9963acq3g7a62mdjgj60rr3hgmyykaccca7), Address(Address #5, bcrt1qpnem6p9vr8rmjsf7k49p9sleu0h020g34ggn6k), ...]
cls = <class 'cryptoadvance.specter.addresslist.Address'>
def write_csv(fname, objs, cls=dict):
> _write_csv(fname, objs, cls)
src/cryptoadvance/specter/persistence.py:164:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
fname = '/tmp/specter_home_tmp_q9xx8fo9/wallets_someuser/regtest/a_simple_wallet_addr.csv'
objs = [Address(Address #0, bcrt1qcatuhg0gll3h7py4cmn53rjjn9xlsqfwj3zcej), Address(Address #1, bcrt1qsqnuk9hulcfta7kj7687favj... bcrt1q2zv9963acq3g7a62mdjgj60rr3hgmyykaccca7), Address(Address #5, bcrt1qpnem6p9vr8rmjsf7k49p9sleu0h020g34ggn6k), ...]
cls = <class 'cryptoadvance.specter.addresslist.Address'>
def _write_csv(fname, objs, cls=dict):
columns = []
# if it's a custom class
if hasattr(cls, "columns"):
columns = cls.columns
# if it's just a dict
elif len(objs) > 0:
columns = objs[0].keys()
with fslock:
with open(fname, mode="w") as csv_file:
writer = csv.DictWriter(csv_file, fieldnames=columns, extrasaction="ignore")
writer.writeheader()
for obj in objs:
> writer.writerow(obj)
src/cryptoadvance/specter/persistence.py:160:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment