from pyln.testing.utils import TEST_NETWORK, TIMEOUT, VALGRIND, DEVELOPER, DEPRECATED_APIS # noqa: F401 from pyln.testing.utils import env, only_one, wait_for, write_config, TailableProc, sync_blockheight, wait_channel_quiescent, get_tx_p2wsh_outnum # noqa: F401 import bitstring from pyln.client import Millisatoshi from pyln.testing.utils import EXPERIMENTAL_DUAL_FUND EXPERIMENTAL_FEATURES = env("EXPERIMENTAL_FEATURES", "0") == "1" COMPAT = env("COMPAT", "1") == "1" def hex_bits(features): # We always to full bytes flen = (max(features + [0]) + 7) // 8 * 8 res = bitstring.BitArray(length=flen) # Big endian sucketh. for f in features: res[flen - 1 - f] = 1 return res.hex def expected_peer_features(wumbo_channels=False, extra=[]): """Return the expected peer features hexstring for this configuration""" features = [1, 5, 7, 9, 11, 13, 14, 17, 27] if EXPERIMENTAL_FEATURES: # OPT_ONION_MESSAGES features += [39] # option_anchor_outputs features += [21] if wumbo_channels: features += [19] if EXPERIMENTAL_DUAL_FUND: # option_anchor_outputs features += [21] # option_dual_fund features += [29] return hex_bits(features + extra) # With the addition of the keysend plugin, we now send a different set of # features for the 'node' and the 'peer' feature sets def expected_node_features(wumbo_channels=False, extra=[]): """Return the expected node features hexstring for this configuration""" features = [1, 5, 7, 9, 11, 13, 14, 17, 27, 55] if EXPERIMENTAL_FEATURES: # OPT_ONION_MESSAGES features += [39] # option_anchor_outputs features += [21] if wumbo_channels: features += [19] if EXPERIMENTAL_DUAL_FUND: # option_anchor_outputs features += [21] # option_dual_fund features += [29] return hex_bits(features + extra) def expected_channel_features(wumbo_channels=False, extra=[]): """Return the expected channel features hexstring for this configuration""" features = [] return hex_bits(features + extra) def move_matches(exp, mv): if mv['type'] != exp['type']: return False if mv['credit'] != "{}msat".format(exp['credit']): return False if mv['debit'] != "{}msat".format(exp['debit']): return False if mv['tag'] != exp['tag']: return False return True def calc_lease_fee(amt, feerate, rates): fee = rates['lease_fee_base_msat'] fee += amt * rates['lease_fee_basis'] // 10 fee += rates['funding_weight'] * feerate return fee def check_coin_moves(n, account_id, expected_moves, chainparams): moves = n.rpc.call('listcoinmoves_plugin')['coin_moves'] node_id = n.info['id'] acct_moves = [m for m in moves if m['account_id'] == account_id] for mv in acct_moves: print("{{'type': '{}', 'credit': {}, 'debit': {}, 'tag': '{}'}}," .format(mv['type'], Millisatoshi(mv['credit']).millisatoshis, Millisatoshi(mv['debit']).millisatoshis, mv['tag'])) assert mv['version'] == 1 assert mv['node_id'] == node_id assert mv['timestamp'] > 0 assert mv['coin_type'] == chainparams['bip173_prefix'] # chain moves should have blockheights if mv['type'] == 'chain_mvt': assert mv['blockheight'] is not None for num, m in enumerate(expected_moves): # They can group things which are in any order. if isinstance(m, list): number_moves = len(m) for acct_move in acct_moves[:number_moves]: found = None for i in range(len(m)): if move_matches(m[i], acct_move): found = i break if found is None: raise ValueError("Unexpected move {} amongst {}".format(acct_move, m)) del m[i] acct_moves = acct_moves[number_moves:] else: if not move_matches(m, acct_moves[0]): raise ValueError("Unexpected move {}: {} != {}".format(num, acct_moves[0], m)) acct_moves = acct_moves[1:] assert acct_moves == [] def check_coin_moves_idx(n): """ Just check that the counter increments smoothly""" moves = n.rpc.call('listcoinmoves_plugin')['coin_moves'] idx = 0 for m in moves: c_idx = m['movement_idx'] # verify that the index count increments smoothly here, also if c_idx == 0 and idx == 0: continue assert c_idx == idx + 1 idx = c_idx def account_balance(n, account_id): moves = n.rpc.call('listcoinmoves_plugin')['coin_moves'] chan_moves = [m for m in moves if m['account_id'] == account_id] assert len(chan_moves) > 0 m_sum = 0 for m in chan_moves: m_sum += int(m['credit'][:-4]) m_sum -= int(m['debit'][:-4]) return m_sum def first_channel_id(n1, n2): return only_one(only_one(n1.rpc.listpeers(n2.info['id'])['peers'])['channels'])['channel_id'] def basic_fee(feerate): if EXPERIMENTAL_FEATURES or EXPERIMENTAL_DUAL_FUND: # option_anchor_outputs weight = 1124 else: weight = 724 return (weight * feerate) // 1000 def closing_fee(feerate, num_outputs): assert num_outputs == 1 or num_outputs == 2 weight = 424 + 124 * num_outputs return (weight * feerate) // 1000 def scriptpubkey_addr(scriptpubkey): if 'addresses' in scriptpubkey: return scriptpubkey['addresses'][0] elif 'address' in scriptpubkey: # Modern bitcoin (at least, git master) return scriptpubkey['address'] return None