midefos/basicswap
1
0
Fork 0
mirror of https://github.com/basicswap/basicswap.git synced 2025-11-05 10:28:10 +01:00
Code Issues Packages Projects Releases Wiki Activity

Add non-segwit Firo support.

Browse Source 
Rework tests to combine atomic and xmr test cases.
Modify btc interface to support P2WSH_nested_in_BIP16_P2SH
Add coin feature tests to test_btc_xmr.py
This commit is contained in:
tecnovert
2022-11-07 22:31:10 +02:00
parent c0c2c8b9bb
commit ca264db0d0
21 changed files with 1400 additions and 411 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
tests/basicswap/common.py
View File

@@ -308,6 +308,10 @@ def delay_for(delay_event, delay_for=60):
delay_event.wait(delay_for)
def make_boolean(s):
return s.lower() in ['1', 'true']
def make_rpc_func(node_id, base_rpc_port=BASE_RPC_PORT):
node_id = node_id
auth = 'test{0}:test_pass{0}'.format(node_id)

8
tests/basicswap/extended/test_dash.py
View File

@@ -528,13 +528,13 @@ class Test(unittest.TestCase):
def test_08_withdrawal(self):
logging.info('---------- Test DASH withdrawals')
pivx_addr = dashRpc('getnewaddress \"Withdrawal test\"')
wallets0 = read_json_api(TEST_HTTP_PORT + 0, 'wallets')
assert (float(wallets0['DASH']['balance']) > 100)
addr = dashRpc('getnewaddress \"Withdrawal test\"')
wallets = read_json_api(TEST_HTTP_PORT + 0, 'wallets')
assert (float(wallets['DASH']['balance']) > 100)
post_json = {
'value': 100,
'address': pivx_addr,
'address': addr,
'subfee': False,
}
json_rv = json.loads(post_json_req('http://127.0.0.1:{}/json/wallets/dash/withdraw'.format(TEST_HTTP_PORT + 0), post_json))

465
tests/basicswap/extended/test_firo.py Normal file
View File

@@ -0,0 +1,465 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (c) 2022 tecnovert
# Distributed under the MIT software license, see the accompanying
# file LICENSE or http://www.opensource.org/licenses/mit-license.php.
import os
import json
import random
import logging
import unittest
import basicswap.config as cfg
from basicswap.basicswap import (
Coins,
TxStates,
SwapTypes,
BidStates,
)
from basicswap.basicswap_util import (
TxLockTypes,
)
from basicswap.util import (
COIN,
make_int,
format_amount,
)
from basicswap.rpc import (
callrpc_cli,
waitForRPC,
)
from tests.basicswap.common import (
stopDaemons,
wait_for_bid,
make_rpc_func,
read_json_api,
post_json_req,
TEST_HTTP_PORT,
wait_for_offer,
wait_for_in_progress,
wait_for_bid_tx_state,
)
from basicswap.contrib.test_framework.messages import (
FromHex,
CTransaction,
)
from bin.basicswap_run import startDaemon
from basicswap.contrib.rpcauth import generate_salt, password_to_hmac
from tests.basicswap.test_xmr import BaseTest, test_delay_event, callnoderpc
logger = logging.getLogger()
FIRO_BASE_PORT = 34832
FIRO_BASE_RPC_PORT = 35832
FIRO_BASE_ZMQ_PORT = 36832
def firoCli(cmd, node_id=0):
return callrpc_cli(cfg.FIRO_BINDIR, os.path.join(cfg.TEST_DATADIRS, 'firo_' + str(node_id)), 'regtest', cmd, cfg.FIRO_CLI)
def prepareDataDir(datadir, node_id, conf_file, dir_prefix, base_p2p_port, base_rpc_port, num_nodes=3):
node_dir = os.path.join(datadir, dir_prefix + str(node_id))
if not os.path.exists(node_dir):
os.makedirs(node_dir)
cfg_file_path = os.path.join(node_dir, conf_file)
if os.path.exists(cfg_file_path):
return
with open(cfg_file_path, 'w+') as fp:
fp.write('regtest=1\n')
fp.write('port=' + str(base_p2p_port + node_id) + '\n')
fp.write('rpcport=' + str(base_rpc_port + node_id) + '\n')
salt = generate_salt(16)
fp.write('rpcauth={}:{}${}\n'.format('test' + str(node_id), salt, password_to_hmac(salt, 'test_pass' + str(node_id))))
fp.write('daemon=0\n')
fp.write('dandelion=0\n')
fp.write('printtoconsole=0\n')
fp.write('server=1\n')
fp.write('discover=0\n')
fp.write('listenonion=0\n')
fp.write('bind=127.0.0.1\n')
fp.write('findpeers=0\n')
fp.write('debug=1\n')
fp.write('debugexclude=libevent\n')
fp.write('fallbackfee=0.01\n')
fp.write('acceptnonstdtxn=0\n')
# qa/rpc-tests/segwit.py
fp.write('prematurewitness=1\n')
fp.write('walletprematurewitness=1\n')
fp.write('blockversion=4\n')
fp.write('promiscuousmempoolflags=517\n')
for i in range(0, num_nodes):
if node_id == i:
continue
fp.write('addnode=127.0.0.1:{}\n'.format(base_p2p_port + i))
return node_dir
class Test(BaseTest):
__test__ = True
firo_daemons = []
firo_addr = None
test_coin_from = Coins.FIRO
test_atomic = True
test_xmr = False
@classmethod
def setUpClass(cls):
cls.start_ltc_nodes = False
cls.start_xmr_nodes = False
super(Test, cls).setUpClass()
@classmethod
def prepareExtraDataDir(cls, i):
if not cls.restore_instance:
data_dir = prepareDataDir(cfg.TEST_DATADIRS, i, 'firo.conf', 'firo_', base_p2p_port=FIRO_BASE_PORT, base_rpc_port=FIRO_BASE_RPC_PORT)
if os.path.exists(os.path.join(cfg.FIRO_BINDIR, 'firo-wallet')):
callrpc_cli(cfg.FIRO_BINDIR, data_dir, 'regtest', '-wallet=wallet.dat create', 'firo-wallet')
cls.firo_daemons.append(startDaemon(os.path.join(cfg.TEST_DATADIRS, 'firo_' + str(i)), cfg.FIRO_BINDIR, cfg.FIROD))
logging.info('Started %s %d', cfg.FIROD, cls.part_daemons[-1].pid)
waitForRPC(make_rpc_func(i, base_rpc_port=FIRO_BASE_RPC_PORT))
@classmethod
def addPIDInfo(cls, sc, i):
sc.setDaemonPID(Coins.FIRO, cls.firo_daemons[i].pid)
@classmethod
def prepareExtraCoins(cls):
if cls.restore_instance:
void_block_rewards_pubkey = cls.getRandomPubkey()
cls.firo_addr = cls.swap_clients[0].ci(Coins.FIRO).pubkey_to_address(void_block_rewards_pubkey)
else:
num_blocks = 400
cls.firo_addr = callnoderpc(0, 'getnewaddress', ['mining_addr'], base_rpc_port=FIRO_BASE_RPC_PORT)
# cls.firo_addr = callnoderpc(0, 'addwitnessaddress', [cls.firo_addr], base_rpc_port=FIRO_BASE_RPC_PORT)
logging.info('Mining %d Firo blocks to %s', num_blocks, cls.firo_addr)
callnoderpc(0, 'generatetoaddress', [num_blocks, cls.firo_addr], base_rpc_port=FIRO_BASE_RPC_PORT)
firo_addr1 = callnoderpc(1, 'getnewaddress', ['initial addr'], base_rpc_port=FIRO_BASE_RPC_PORT)
# firo_addr1 = callnoderpc(1, 'addwitnessaddress', [firo_addr1], base_rpc_port=FIRO_BASE_RPC_PORT)
for i in range(5):
callnoderpc(0, 'sendtoaddress', [firo_addr1, 1000], base_rpc_port=FIRO_BASE_RPC_PORT)
# Set future block rewards to nowhere (a random address), so wallet amounts stay constant
void_block_rewards_pubkey = cls.getRandomPubkey()
cls.firo_addr = cls.swap_clients[0].ci(Coins.FIRO).pubkey_to_address(void_block_rewards_pubkey)
num_blocks = 100
logging.info('Mining %d Firo blocks to %s', num_blocks, cls.firo_addr)
callnoderpc(0, 'generatetoaddress', [num_blocks, cls.firo_addr], base_rpc_port=FIRO_BASE_RPC_PORT)
@classmethod
def tearDownClass(cls):
logging.info('Finalising FIRO Test')
super(Test, cls).tearDownClass()
stopDaemons(cls.firo_daemons)
@classmethod
def addCoinSettings(cls, settings, datadir, node_id):
settings['chainclients']['firo'] = {
'connection_type': 'rpc',
'manage_daemon': False,
'rpcport': FIRO_BASE_RPC_PORT + node_id,
'rpcuser': 'test' + str(node_id),
'rpcpassword': 'test_pass' + str(node_id),
'datadir': os.path.join(datadir, 'firo_' + str(node_id)),
'bindir': cfg.FIRO_BINDIR,
'use_csv': True,
'use_segwit': False,
}
@classmethod
def coins_loop(cls):
super(Test, cls).coins_loop()
callnoderpc(0, 'generatetoaddress', [1, cls.firo_addr], base_rpc_port=FIRO_BASE_RPC_PORT)
def getBalance(self, js_wallets):
return float(js_wallets[self.test_coin_from.name]['balance']) + float(js_wallets[self.test_coin_from.name]['unconfirmed'])
def getXmrBalance(self, js_wallets):
return float(js_wallets[Coins.XMR.name]['unconfirmed']) + float(js_wallets[Coins.XMR.name]['balance'])
def callnoderpc(self, method, params=[], wallet=None, node_id=0):
return callnoderpc(node_id, method, params, wallet, base_rpc_port=FIRO_BASE_RPC_PORT)
def test_01_firo(self):
logging.info('---------- Test {} segwit'.format(self.test_coin_from.name))
'''
Segwit is not currently enabled:
https://github.com/firoorg/firo/blob/master/src/validation.cpp#L4425
Txns spending segwit utxos don't get mined.
'''
swap_clients = self.swap_clients
addr_plain = firoCli('getnewaddress \"segwit test\"')
addr_witness = firoCli(f'addwitnessaddress {addr_plain}')
addr_witness_info = firoCli(f'validateaddress {addr_witness}')
txid = firoCli(f'sendtoaddress {addr_witness} 1.0')
assert len(txid) == 64
self.callnoderpc('generatetoaddress', [1, self.firo_addr])
'''
TODO: Add back when segwit is active
ro = self.callnoderpc('scantxoutset', ['start', ['addr({})'.format(addr_witness)]])
assert (len(ro['unspents']) == 1)
assert (ro['unspents'][0]['txid'] == txid)
'''
tx_wallet = firoCli(f'gettransaction {txid}')
tx_hex = tx_wallet['hex']
tx = firoCli(f'decoderawtransaction {tx_hex}')
prevout_n = -1
for txo in tx['vout']:
if addr_witness in txo['scriptPubKey']['addresses']:
prevout_n = txo['n']
break
assert prevout_n > -1
tx_funded = firoCli(f'createrawtransaction [{{\\"txid\\":\\"{txid}\\",\\"vout\\":{prevout_n}}}] {{\\"{addr_plain}\\":0.99}}')
tx_signed = firoCli(f'signrawtransaction {tx_funded}')['hex']
# Add scriptsig for txids to match
decoded_tx = CTransaction()
decoded_tx = FromHex(decoded_tx, tx_funded)
decoded_tx.vin[0].scriptSig = bytes.fromhex('16' + addr_witness_info['hex'])
txid_with_scriptsig = decoded_tx.rehash()
tx_funded_decoded = firoCli(f'decoderawtransaction {tx_funded}')
tx_signed_decoded = firoCli(f'decoderawtransaction {tx_signed}')
assert tx_funded_decoded['txid'] != tx_signed_decoded['txid']
assert txid_with_scriptsig == tx_signed_decoded['txid']
def test_02_part_coin(self):
logging.info('---------- Test PART to {}'.format(self.test_coin_from.name))
if not self.test_atomic:
logging.warning('Skipping test')
return
swap_clients = self.swap_clients
offer_id = swap_clients[0].postOffer(Coins.PART, self.test_coin_from, 100 * COIN, 0.1 * COIN, 100 * COIN, SwapTypes.SELLER_FIRST)
wait_for_offer(test_delay_event, swap_clients[1], offer_id)
offers = swap_clients[1].listOffers()
assert (len(offers) == 1)
for offer in offers:
if offer.offer_id == offer_id:
bid_id = swap_clients[1].postBid(offer_id, offer.amount_from)
wait_for_bid(test_delay_event, swap_clients[0], bid_id)
swap_clients[0].acceptBid(bid_id)
wait_for_in_progress(test_delay_event, swap_clients[1], bid_id, sent=True)
wait_for_bid(test_delay_event, swap_clients[0], bid_id, BidStates.SWAP_COMPLETED, wait_for=60)
wait_for_bid(test_delay_event, swap_clients[1], bid_id, BidStates.SWAP_COMPLETED, sent=True, wait_for=60)
js_0 = read_json_api(1800)
js_1 = read_json_api(1801)
assert (js_0['num_swapping'] == 0 and js_0['num_watched_outputs'] == 0)
assert (js_1['num_swapping'] == 0 and js_1['num_watched_outputs'] == 0)
def test_03_coin_part(self):
logging.info('---------- Test {} to PART'.format(self.test_coin_from.name))
swap_clients = self.swap_clients
offer_id = swap_clients[1].postOffer(self.test_coin_from, Coins.PART, 10 * COIN, 9.0 * COIN, 10 * COIN, SwapTypes.SELLER_FIRST)
wait_for_offer(test_delay_event, swap_clients[0], offer_id)
offers = swap_clients[0].listOffers()
for offer in offers:
if offer.offer_id == offer_id:
bid_id = swap_clients[0].postBid(offer_id, offer.amount_from)
wait_for_bid(test_delay_event, swap_clients[1], bid_id)
swap_clients[1].acceptBid(bid_id)
wait_for_in_progress(test_delay_event, swap_clients[0], bid_id, sent=True)
wait_for_bid(test_delay_event, swap_clients[0], bid_id, BidStates.SWAP_COMPLETED, sent=True, wait_for=60)
wait_for_bid(test_delay_event, swap_clients[1], bid_id, BidStates.SWAP_COMPLETED, wait_for=60)
js_0 = read_json_api(1800)
js_1 = read_json_api(1801)
assert (js_0['num_swapping'] == 0 and js_0['num_watched_outputs'] == 0)
assert (js_1['num_swapping'] == 0 and js_1['num_watched_outputs'] == 0)
def test_04_coin_btc(self):
logging.info('---------- Test {} to BTC'.format(self.test_coin_from.name))
swap_clients = self.swap_clients
offer_id = swap_clients[0].postOffer(self.test_coin_from, Coins.BTC, 10 * COIN, 0.1 * COIN, 10 * COIN, SwapTypes.SELLER_FIRST)
wait_for_offer(test_delay_event, swap_clients[1], offer_id)
offers = swap_clients[1].listOffers()
for offer in offers:
if offer.offer_id == offer_id:
bid_id = swap_clients[1].postBid(offer_id, offer.amount_from)
wait_for_bid(test_delay_event, swap_clients[0], bid_id)
swap_clients[0].acceptBid(bid_id)
wait_for_in_progress(test_delay_event, swap_clients[1], bid_id, sent=True)
wait_for_bid(test_delay_event, swap_clients[0], bid_id, BidStates.SWAP_COMPLETED, wait_for=60)
wait_for_bid(test_delay_event, swap_clients[1], bid_id, BidStates.SWAP_COMPLETED, sent=True, wait_for=60)
js_0bid = read_json_api(1800, 'bids/{}'.format(bid_id.hex()))
js_0 = read_json_api(1800)
js_1 = read_json_api(1801)
assert (js_0['num_swapping'] == 0 and js_0['num_watched_outputs'] == 0)
assert (js_1['num_swapping'] == 0 and js_1['num_watched_outputs'] == 0)
def test_05_refund(self):
# Seller submits initiate txn, buyer doesn't respond
logging.info('---------- Test refund, {} to BTC'.format(self.test_coin_from.name))
swap_clients = self.swap_clients
offer_id = swap_clients[0].postOffer(self.test_coin_from, Coins.BTC, 10 * COIN, 0.1 * COIN, 10 * COIN, SwapTypes.SELLER_FIRST,
TxLockTypes.SEQUENCE_LOCK_BLOCKS, 10)
wait_for_offer(test_delay_event, swap_clients[1], offer_id)
offers = swap_clients[1].listOffers()
for offer in offers:
if offer.offer_id == offer_id:
bid_id = swap_clients[1].postBid(offer_id, offer.amount_from)
wait_for_bid(test_delay_event, swap_clients[0], bid_id)
swap_clients[1].abandonBid(bid_id)
swap_clients[0].acceptBid(bid_id)
wait_for_bid(test_delay_event, swap_clients[0], bid_id, BidStates.SWAP_COMPLETED, wait_for=60)
wait_for_bid(test_delay_event, swap_clients[1], bid_id, BidStates.BID_ABANDONED, sent=True, wait_for=60)
js_0 = read_json_api(1800)
js_1 = read_json_api(1801)
assert (js_0['num_swapping'] == 0 and js_0['num_watched_outputs'] == 0)
assert (js_1['num_swapping'] == 0 and js_1['num_watched_outputs'] == 0)
def test_06_self_bid(self):
logging.info('---------- Test same client, BTC to {}'.format(self.test_coin_from.name))
swap_clients = self.swap_clients
js_0_before = read_json_api(1800)
offer_id = swap_clients[0].postOffer(self.test_coin_from, Coins.BTC, 10 * COIN, 10 * COIN, 10 * COIN, SwapTypes.SELLER_FIRST)
wait_for_offer(test_delay_event, swap_clients[0], offer_id)
offers = swap_clients[0].listOffers()
for offer in offers:
if offer.offer_id == offer_id:
bid_id = swap_clients[0].postBid(offer_id, offer.amount_from)
wait_for_bid(test_delay_event, swap_clients[0], bid_id)
swap_clients[0].acceptBid(bid_id)
wait_for_bid_tx_state(test_delay_event, swap_clients[0], bid_id, TxStates.TX_REDEEMED, TxStates.TX_REDEEMED, wait_for=60)
wait_for_bid(test_delay_event, swap_clients[0], bid_id, BidStates.SWAP_COMPLETED, wait_for=60)
js_0 = read_json_api(1800)
assert (js_0['num_swapping'] == 0 and js_0['num_watched_outputs'] == 0)
assert (js_0['num_recv_bids'] == js_0_before['num_recv_bids'] + 1 and js_0['num_sent_bids'] == js_0_before['num_sent_bids'] + 1)
def test_07_error(self):
logging.info('---------- Test error, BTC to {}, set fee above bid value'.format(self.test_coin_from.name))
swap_clients = self.swap_clients
js_0_before = read_json_api(1800)
offer_id = swap_clients[0].postOffer(self.test_coin_from, Coins.BTC, 0.001 * COIN, 1.0 * COIN, 0.001 * COIN, SwapTypes.SELLER_FIRST)
wait_for_offer(test_delay_event, swap_clients[0], offer_id)
offers = swap_clients[0].listOffers()
for offer in offers:
if offer.offer_id == offer_id:
bid_id = swap_clients[0].postBid(offer_id, offer.amount_from)
wait_for_bid(test_delay_event, swap_clients[0], bid_id)
swap_clients[0].acceptBid(bid_id)
swap_clients[0].getChainClientSettings(Coins.BTC)['override_feerate'] = 10.0
swap_clients[0].getChainClientSettings(Coins.FIRO)['override_feerate'] = 10.0
wait_for_bid(test_delay_event, swap_clients[0], bid_id, BidStates.BID_ERROR, wait_for=60)
def test_08_withdrawal(self):
logging.info('---------- Test {} withdrawals'.format(self.test_coin_from.name))
addr = self.callnoderpc('getnewaddress', ['Withdrawal test', ])
wallets = read_json_api(TEST_HTTP_PORT + 0, 'wallets')
assert (float(wallets[self.test_coin_from.name]['balance']) > 100)
post_json = {
'value': 100,
'address': addr,
'subfee': False,
}
json_rv = json.loads(post_json_req('http://127.0.0.1:{}/json/wallets/{}/withdraw'.format(TEST_HTTP_PORT + 0, self.test_coin_from.name.lower()), post_json))
assert (len(json_rv['txid']) == 64)
def test_101_full_swap(self):
logging.info('---------- Test {} to XMR'.format(self.test_coin_from.name))
if not self.test_xmr:
logging.warning('Skipping test')
return
swap_clients = self.swap_clients
js_0 = read_json_api(1800, 'wallets')
node0_from_before = self.getBalance(js_0)
js_1 = read_json_api(1801, 'wallets')
node1_from_before = self.getBalance(js_1)
js_0_xmr = read_json_api(1800, 'wallets/xmr')
js_1_xmr = read_json_api(1801, 'wallets/xmr')
amt_swap = make_int(random.uniform(0.1, 2.0), scale=8, r=1)
rate_swap = make_int(random.uniform(0.2, 20.0), scale=12, r=1)
offer_id = swap_clients[0].postOffer(self.test_coin_from, Coins.XMR, amt_swap, rate_swap, amt_swap, SwapTypes.XMR_SWAP)
wait_for_offer(test_delay_event, swap_clients[1], offer_id)
offers = swap_clients[0].listOffers(filters={'offer_id': offer_id})
offer = offers[0]
bid_id = swap_clients[1].postXmrBid(offer_id, offer.amount_from)
wait_for_bid(test_delay_event, swap_clients[0], bid_id, BidStates.BID_RECEIVED)
swap_clients[0].acceptXmrBid(bid_id)
wait_for_bid(test_delay_event, swap_clients[0], bid_id, BidStates.SWAP_COMPLETED, wait_for=180)
wait_for_bid(test_delay_event, swap_clients[1], bid_id, BidStates.SWAP_COMPLETED, sent=True)
amount_from = float(format_amount(amt_swap, 8))
js_1 = read_json_api(1801, 'wallets')
node1_from_after = self.getBalance(js_1)
assert (node1_from_after > node1_from_before + (amount_from - 0.05))
js_0 = read_json_api(1800, 'wallets')
node0_from_after = self.getBalance(js_0)
# TODO: Discard block rewards
# assert (node0_from_after < node0_from_before - amount_from)
js_0_xmr_after = read_json_api(1800, 'wallets/xmr')
js_1_xmr_after = read_json_api(1801, 'wallets/xmr')
scale_from = 8
amount_to = int((amt_swap * rate_swap) // (10 ** scale_from))
amount_to_float = float(format_amount(amount_to, 12))
node1_xmr_after = float(js_1_xmr_after['unconfirmed']) + float(js_1_xmr_after['balance'])
node1_xmr_before = float(js_1_xmr['unconfirmed']) + float(js_1_xmr['balance'])
assert (node1_xmr_after > node1_xmr_before + (amount_to_float - 0.02))
if __name__ == '__main__':
unittest.main()

8
tests/basicswap/extended/test_pivx.py
View File

@@ -538,13 +538,13 @@ class Test(unittest.TestCase):
def test_08_withdrawal(self):
logging.info('---------- Test PIVX withdrawals')
pivx_addr = pivxRpc('getnewaddress \"Withdrawal test\"')
wallets0 = read_json_api(TEST_HTTP_PORT + 0, 'wallets')
assert (float(wallets0['PIVX']['balance']) > 100)
addr = pivxRpc('getnewaddress \"Withdrawal test\"')
wallets = read_json_api(TEST_HTTP_PORT + 0, 'wallets')
assert (float(wallets['PIVX']['balance']) > 100)
post_json = {
'value': 100,
'address': pivx_addr,
'address': addr,
'subfee': False,
}
json_rv = json.loads(post_json_req('http://127.0.0.1:{}/json/wallets/pivx/withdraw'.format(TEST_HTTP_PORT + 0), post_json))

5
tests/basicswap/extended/test_xmr_persistent.py
View File

@@ -34,6 +34,7 @@ from basicswap.rpc import (
callrpc,
)
from tests.basicswap.common import (
make_boolean,
read_json_api,
waitForServer,
BASE_RPC_PORT,
@@ -46,10 +47,6 @@ from tests.basicswap.common_xmr import (
import bin.basicswap_run as runSystem
def make_boolean(s):
return s.lower() in ['1', 'true']
test_path = os.path.expanduser(os.getenv('TEST_PATH', '/tmp/test_persistent'))
RESET_TEST = make_boolean(os.getenv('RESET_TEST', 'false'))

224
tests/basicswap/test_btc_xmr.py
View File

@@ -27,39 +27,217 @@ from tests.basicswap.common import (
read_json_api,
wait_for_offer,
wait_for_none_active,
BTC_BASE_RPC_PORT,
)
from .test_xmr import BaseTest, test_delay_event
from basicswap.contrib.test_framework.messages import (
ToHex,
FromHex,
CTxIn,
COutPoint,
CTransaction,
CTxInWitness,
)
from basicswap.contrib.test_framework.script import (
CScript,
OP_CHECKLOCKTIMEVERIFY,
OP_CHECKSEQUENCEVERIFY,
)
from .test_xmr import BaseTest, test_delay_event, callnoderpc
logger = logging.getLogger()
class Test(BaseTest):
__test__ = True
@classmethod
def setUpClass(cls):
if not hasattr(cls, 'test_coin_from'):
cls.test_coin_from = Coins.BTC
if not hasattr(cls, 'start_ltc_nodes'):
cls.start_ltc_nodes = False
if not hasattr(cls, 'start_pivx_nodes'):
cls.start_pivx_nodes = False
super(Test, cls).setUpClass()
@classmethod
def tearDownClass(cls):
logging.info('Finalising BTC Test')
super(Test, cls).tearDownClass()
test_coin_from = Coins.BTC
start_ltc_nodes = False
def getBalance(self, js_wallets):
return float(js_wallets[self.test_coin_from.name]['balance']) + float(js_wallets[self.test_coin_from.name]['unconfirmed'])
def getXmrBalance(self, js_wallets):
return float(js_wallets[Coins.XMR.name]['unconfirmed']) + float(js_wallets[Coins.XMR.name]['balance'])
def callnoderpc(self, method, params=[], wallet=None, node_id=0):
return callnoderpc(node_id, method, params, wallet, base_rpc_port=BTC_BASE_RPC_PORT)
def test_001_nested_segwit(self):
logging.info('---------- Test {} p2sh nested segwit'.format(self.test_coin_from.name))
addr_p2sh_segwit = self.callnoderpc('getnewaddress', ['segwit test', 'p2sh-segwit'])
addr_info = self.callnoderpc('getaddressinfo', [addr_p2sh_segwit, ])
assert addr_info['script'] == 'witness_v0_keyhash'
txid = self.callnoderpc('sendtoaddress', [addr_p2sh_segwit, 1.0])
assert len(txid) == 64
self.callnoderpc('generatetoaddress', [1, self.btc_addr])
ro = self.callnoderpc('scantxoutset', ['start', ['addr({})'.format(addr_p2sh_segwit)]])
assert (len(ro['unspents']) == 1)
assert (ro['unspents'][0]['txid'] == txid)
tx_wallet = self.callnoderpc('gettransaction', [txid, ])['hex']
tx = self.callnoderpc('decoderawtransaction', [tx_wallet, ])
prevout_n = -1
for txo in tx['vout']:
if addr_p2sh_segwit == txo['scriptPubKey']['address']:
prevout_n = txo['n']
break
assert prevout_n > -1
tx_funded = self.callnoderpc('createrawtransaction', [[{'txid': txid, 'vout': prevout_n}], {addr_p2sh_segwit: 0.99}])
tx_signed = self.callnoderpc('signrawtransactionwithwallet', [tx_funded, ])['hex']
tx_funded_decoded = self.callnoderpc('decoderawtransaction', [tx_funded, ])
tx_signed_decoded = self.callnoderpc('decoderawtransaction', [tx_signed, ])
assert tx_funded_decoded['txid'] != tx_signed_decoded['txid']
# Add scriptsig for txids to match
addr_p2sh_segwit_info = self.callnoderpc('getaddressinfo', [addr_p2sh_segwit, ])
decoded_tx = FromHex(CTransaction(), tx_funded)
decoded_tx.vin[0].scriptSig = bytes.fromhex('16' + addr_p2sh_segwit_info['hex'])
txid_with_scriptsig = decoded_tx.rehash()
assert txid_with_scriptsig == tx_signed_decoded['txid']
def test_002_native_segwit(self):
logging.info('---------- Test {} p2sh native segwit'.format(self.test_coin_from.name))
addr_segwit = self.callnoderpc('getnewaddress', ['segwit test', 'bech32'])
addr_info = self.callnoderpc('getaddressinfo', [addr_segwit, ])
assert addr_info['iswitness'] is True
txid = self.callnoderpc('sendtoaddress', [addr_segwit, 1.0])
assert len(txid) == 64
tx_wallet = self.callnoderpc('gettransaction', [txid, ])['hex']
tx = self.callnoderpc('decoderawtransaction', [tx_wallet, ])
self.callnoderpc('generatetoaddress', [1, self.btc_addr])
ro = self.callnoderpc('scantxoutset', ['start', ['addr({})'.format(addr_segwit)]])
assert (len(ro['unspents']) == 1)
assert (ro['unspents'][0]['txid'] == txid)
prevout_n = -1
for txo in tx['vout']:
if addr_segwit == txo['scriptPubKey']['address']:
prevout_n = txo['n']
break
assert prevout_n > -1
tx_funded = self.callnoderpc('createrawtransaction', [[{'txid': txid, 'vout': prevout_n}], {addr_segwit: 0.99}])
tx_signed = self.callnoderpc('signrawtransactionwithwallet', [tx_funded, ])['hex']
tx_funded_decoded = self.callnoderpc('decoderawtransaction', [tx_funded, ])
tx_signed_decoded = self.callnoderpc('decoderawtransaction', [tx_signed, ])
assert tx_funded_decoded['txid'] == tx_signed_decoded['txid']
def test_003_cltv(self):
logging.info('---------- Test {} cltv'.format(self.test_coin_from.name))
ci = self.swap_clients[0].ci(self.test_coin_from)
chain_height = self.callnoderpc('getblockcount')
script = CScript([chain_height + 3, OP_CHECKLOCKTIMEVERIFY, ])
script_dest = ci.getScriptDest(script)
tx = CTransaction()
tx.nVersion = ci.txVersion()
tx.vout.append(ci.txoType()(ci.make_int(1.1), script_dest))
tx_hex = ToHex(tx)
tx_funded = self.callnoderpc('fundrawtransaction', [tx_hex])
utxo_pos = 0 if tx_funded['changepos'] == 1 else 1
tx_signed = self.callnoderpc('signrawtransactionwithwallet', [tx_funded['hex'], ])['hex']
txid = self.callnoderpc('sendrawtransaction', [tx_signed, ])
addr_out = self.callnoderpc('getnewaddress', ['csv test', 'bech32'])
pkh = ci.decodeSegwitAddress(addr_out)
script_out = ci.getScriptForPubkeyHash(pkh)
tx_spend = CTransaction()
tx_spend.nVersion = ci.txVersion()
tx_spend.nLockTime = chain_height + 3
tx_spend.vin.append(CTxIn(COutPoint(int(txid, 16), utxo_pos)))
tx_spend.vout.append(ci.txoType()(ci.make_int(1.0999), script_out))
tx_spend.wit.vtxinwit.append(CTxInWitness())
tx_spend.wit.vtxinwit[0].scriptWitness.stack = [script, ]
tx_spend_hex = ToHex(tx_spend)
try:
txid = self.callnoderpc('sendrawtransaction', [tx_spend_hex, ])
assert False, 'Should fail'
except Exception as e:
assert ('non-final' in str(e))
self.callnoderpc('generatetoaddress', [49, self.btc_addr])
txid = self.callnoderpc('sendrawtransaction', [tx_spend_hex, ])
self.callnoderpc('generatetoaddress', [1, self.btc_addr])
ro = self.callnoderpc('listreceivedbyaddress', [0, ])
sum_addr = 0
for entry in ro:
if entry['address'] == addr_out:
sum_addr += entry['amount']
assert (sum_addr == 1.0999)
def test_004_csv(self):
logging.info('---------- Test {} csv'.format(self.test_coin_from.name))
swap_clients = self.swap_clients
ci = self.swap_clients[0].ci(self.test_coin_from)
script = CScript([3, OP_CHECKSEQUENCEVERIFY, ])
script_dest = ci.getScriptDest(script)
tx = CTransaction()
tx.nVersion = ci.txVersion()
tx.vout.append(ci.txoType()(ci.make_int(1.1), script_dest))
tx_hex = ToHex(tx)
tx_funded = self.callnoderpc('fundrawtransaction', [tx_hex])
utxo_pos = 0 if tx_funded['changepos'] == 1 else 1
tx_signed = self.callnoderpc('signrawtransactionwithwallet', [tx_funded['hex'], ])['hex']
txid = self.callnoderpc('sendrawtransaction', [tx_signed, ])
addr_out = self.callnoderpc('getnewaddress', ['csv test', 'bech32'])
pkh = ci.decodeSegwitAddress(addr_out)
script_out = ci.getScriptForPubkeyHash(pkh)
tx_spend = CTransaction()
tx_spend.nVersion = ci.txVersion()
tx_spend.vin.append(CTxIn(COutPoint(int(txid, 16), utxo_pos),
nSequence=3))
tx_spend.vout.append(ci.txoType()(ci.make_int(1.0999), script_out))
tx_spend.wit.vtxinwit.append(CTxInWitness())
tx_spend.wit.vtxinwit[0].scriptWitness.stack = [script, ]
tx_spend_hex = ToHex(tx_spend)
try:
txid = self.callnoderpc('sendrawtransaction', [tx_spend_hex, ])
assert False, 'Should fail'
except Exception as e:
assert ('non-BIP68-final' in str(e))
self.callnoderpc('generatetoaddress', [3, self.btc_addr])
txid = self.callnoderpc('sendrawtransaction', [tx_spend_hex, ])
self.callnoderpc('generatetoaddress', [1, self.btc_addr])
ro = self.callnoderpc('listreceivedbyaddress', [0, ])
sum_addr = 0
for entry in ro:
if entry['address'] == addr_out:
sum_addr += entry['amount']
assert (sum_addr == 1.0999)
def test_005_watchonly(self):
logging.info('---------- Test {} watchonly'.format(self.test_coin_from.name))
addr = self.callnoderpc('getnewaddress', ['watchonly test', 'bech32'])
ro = self.callnoderpc('importaddress', [addr, '', False], node_id=1)
txid = self.callnoderpc('sendtoaddress', [addr, 1.0])
tx_hex = self.callnoderpc('getrawtransaction', [txid, ])
self.callnoderpc('sendrawtransaction', [tx_hex, ], node_id=1)
ro = self.callnoderpc('gettransaction', [txid, ], node_id=1)
assert (ro['txid'] == txid)
balances = self.callnoderpc('getbalances', node_id=1)
assert (balances['watchonly']['trusted'] + balances['watchonly']['untrusted_pending'] >= 1.0)
def test_006_getblock_verbosity(self):
logging.info('---------- Test {} getblock verbosity'.format(self.test_coin_from.name))
best_hash = self.callnoderpc('getbestblockhash')
block = self.callnoderpc('getblock', [best_hash, 2])
assert ('vin' in block['tx'][0])
def test_01_full_swap(self):
logging.info('---------- Test {} to XMR'.format(str(self.test_coin_from)))
logging.info('---------- Test {} to XMR'.format(self.test_coin_from.name))
swap_clients = self.swap_clients
js_0 = read_json_api(1800, 'wallets')
@@ -108,7 +286,7 @@ class Test(BaseTest):
assert (node1_xmr_after > node1_xmr_before + (amount_to_float - 0.02))
def test_02_leader_recover_a_lock_tx(self):
logging.info('---------- Test {} to XMR leader recovers coin a lock tx'.format(str(self.test_coin_from)))
logging.info('---------- Test {} to XMR leader recovers coin a lock tx'.format(self.test_coin_from.name))
swap_clients = self.swap_clients
js_w0_before = read_json_api(1800, 'wallets')
@@ -143,7 +321,7 @@ class Test(BaseTest):
# assert (node0_from_before - node0_from_after < 0.02)
def test_03_follower_recover_a_lock_tx(self):
logging.info('---------- Test {} to XMR follower recovers coin a lock tx'.format(str(self.test_coin_from)))
logging.info('---------- Test {} to XMR follower recovers coin a lock tx'.format(self.test_coin_from.name))
swap_clients = self.swap_clients
js_w0_before = read_json_api(1800, 'wallets')
@@ -186,7 +364,7 @@ class Test(BaseTest):
wait_for_none_active(test_delay_event, 1801)
def test_04_follower_recover_b_lock_tx(self):
logging.info('---------- Test {} to XMR follower recovers coin b lock tx'.format(str(self.test_coin_from)))
logging.info('---------- Test {} to XMR follower recovers coin b lock tx'.format(self.test_coin_from.name))
swap_clients = self.swap_clients

2
tests/basicswap/test_other.py
View File

@@ -212,7 +212,7 @@ class Test(unittest.TestCase):
pk = ci.getPubkey(vk)
sig = ci.signCompact(vk, 'test signing message')
assert (len(sig) == 64)
ci.verifyCompact(pk, 'test signing message', sig)
ci.verifyCompactSig(pk, 'test signing message', sig)
def test_pubkey_to_address(self):
coin_settings = {'rpcport': 0, 'rpcauth': 'none'}

359
tests/basicswap/test_xmr.py
View File

@@ -59,6 +59,7 @@ from basicswap.http_server import (
)
from tests.basicswap.common import (
prepareDataDir,
make_boolean,
make_rpc_func,
checkForks,
stopDaemons,
@@ -80,8 +81,6 @@ from tests.basicswap.common import (
BTC_BASE_RPC_PORT,
LTC_BASE_PORT,
LTC_BASE_RPC_PORT,
PIVX_BASE_PORT,
PIVX_BASE_RPC_PORT,
PREFIX_SECRET_KEY_REGTEST,
)
from bin.basicswap_run import startDaemon, startXmrDaemon
@@ -93,7 +92,6 @@ NUM_NODES = 3
NUM_XMR_NODES = 3
NUM_BTC_NODES = 3
NUM_LTC_NODES = 3
NUM_PIVX_NODES = 3
TEST_DIR = cfg.TEST_DATADIRS
XMR_BASE_P2P_PORT = 17792
@@ -102,6 +100,7 @@ XMR_BASE_ZMQ_PORT = 22792
XMR_BASE_WALLET_RPC_PORT = 23792
test_delay_event = threading.Event()
RESET_TEST = make_boolean(os.getenv('RESET_TEST', 'true'))
def prepareXmrDataDir(datadir, node_id, conf_file):
@@ -153,7 +152,7 @@ def startXmrWalletRPC(node_dir, bin_dir, wallet_bin, node_id, opts=[]):
return subprocess.Popen(args, stdin=subprocess.PIPE, stdout=wallet_stdout, stderr=wallet_stderr, cwd=data_dir)
def prepare_swapclient_dir(datadir, node_id, network_key, network_pubkey, with_coins=set()):
def prepare_swapclient_dir(datadir, node_id, network_key, network_pubkey, with_coins=set(), cls=None):
basicswap_dir = os.path.join(datadir, 'basicswap_' + str(node_id))
if not os.path.exists(basicswap_dir):
os.makedirs(basicswap_dir)
@@ -229,17 +228,8 @@ def prepare_swapclient_dir(datadir, node_id, network_key, network_pubkey, with_c
'use_segwit': True,
}
if Coins.PIVX in with_coins:
settings['chainclients']['pivx'] = {
'connection_type': 'rpc',
'manage_daemon': False,
'rpcport': PIVX_BASE_RPC_PORT + node_id,
'rpcuser': 'test' + str(node_id),
'rpcpassword': 'test_pass' + str(node_id),
'datadir': os.path.join(datadir, 'pivx_' + str(node_id)),
'bindir': cfg.PIVX_BINDIR,
'use_segwit': False,
}
if cls:
cls.addCoinSettings(settings, datadir, node_id)
with open(settings_path, 'w') as fp:
json.dump(settings, fp, indent=4)
@@ -253,10 +243,6 @@ def ltcCli(cmd, node_id=0):
return callrpc_cli(cfg.LITECOIN_BINDIR, os.path.join(TEST_DIR, 'ltc_' + str(node_id)), 'regtest', cmd, cfg.LITECOIN_CLI)
def pivxCli(cmd, node_id=0):
return callrpc_cli(cfg.PIVX_BINDIR, os.path.join(TEST_DIR, 'pivx_' + str(node_id)), 'regtest', cmd, cfg.PIVX_CLI)
def signal_handler(sig, frame):
logging.info('signal {} detected.'.format(sig))
test_delay_event.set()
@@ -298,14 +284,7 @@ def run_coins_loop(cls):
while not test_delay_event.is_set():
pause_event.wait()
try:
if cls.btc_addr is not None:
btcCli('generatetoaddress 1 {}'.format(cls.btc_addr))
if cls.ltc_addr is not None:
ltcCli('generatetoaddress 1 {}'.format(cls.ltc_addr))
if cls.pivx_addr is not None:
pivxCli('generatetoaddress 1 {}'.format(cls.pivx_addr))
if cls.xmr_addr is not None:
callrpc_xmr_na(XMR_BASE_RPC_PORT + 1, 'generateblocks', {'wallet_address': cls.xmr_addr, 'amount_of_blocks': 1})
cls.coins_loop()
except Exception as e:
logging.warning('run_coins_loop ' + str(e))
test_delay_event.wait(1.0)
@@ -320,34 +299,34 @@ def run_loop(cls):
class BaseTest(unittest.TestCase):
__test__ = False
update_thread = None
coins_update_thread = None
http_threads = []
swap_clients = []
part_daemons = []
btc_daemons = []
ltc_daemons = []
xmr_daemons = []
xmr_wallet_auth = []
restore_instance = False
start_ltc_nodes = False
start_xmr_nodes = True
xmr_addr = None
btc_addr = None
ltc_addr = None
@classmethod
def getRandomPubkey(cls):
eckey = ECKey()
eckey.generate()
return eckey.get_pubkey().get_bytes()
@classmethod
def setUpClass(cls):
if not hasattr(cls, 'start_ltc_nodes'):
cls.start_ltc_nodes = False
if not hasattr(cls, 'start_pivx_nodes'):
cls.start_pivx_nodes = False
if not hasattr(cls, 'start_xmr_nodes'):
cls.start_xmr_nodes = True
random.seed(time.time())
cls.update_thread = None
cls.coins_update_thread = None
cls.http_threads = []
cls.swap_clients = []
cls.part_daemons = []
cls.btc_daemons = []
cls.ltc_daemons = []
cls.pivx_daemons = []
cls.xmr_daemons = []
cls.xmr_wallet_auth = []
cls.xmr_addr = None
cls.btc_addr = None
cls.ltc_addr = None
cls.pivx_addr = None
logger.propagate = False
logger.handlers = []
logger.setLevel(logging.INFO) # DEBUG shows many messages from requests.post
@@ -356,16 +335,24 @@ class BaseTest(unittest.TestCase):
stream_stdout.setFormatter(formatter)
logger.addHandler(stream_stdout)
diagrams_dir = 'doc/protocols/sequence_diagrams'
cls.states_bidder = extract_states_from_xu_file(os.path.join(diagrams_dir, 'xmr.bidder.alt.xu'), 'B')
cls.states_offerer = extract_states_from_xu_file(os.path.join(diagrams_dir, 'xmr.offerer.alt.xu'), 'O')
if os.path.isdir(TEST_DIR):
logging.info('Removing ' + TEST_DIR)
for name in os.listdir(TEST_DIR):
if name == 'pivx-params':
continue
fullpath = os.path.join(TEST_DIR, name)
if os.path.isdir(fullpath):
shutil.rmtree(fullpath)
else:
os.remove(fullpath)
if RESET_TEST:
logging.info('Removing ' + TEST_DIR)
for name in os.listdir(TEST_DIR):
if name == 'pivx-params':
continue
fullpath = os.path.join(TEST_DIR, name)
if os.path.isdir(fullpath):
shutil.rmtree(fullpath)
else:
os.remove(fullpath)
else:
logging.info('Restoring instance from ' + TEST_DIR)
cls.restore_instance = True
if not os.path.exists(TEST_DIR):
os.makedirs(TEST_DIR)
@@ -373,39 +360,38 @@ class BaseTest(unittest.TestCase):
cls.stream_fp.setFormatter(formatter)
logger.addHandler(cls.stream_fp)
diagrams_dir = 'doc/protocols/sequence_diagrams'
cls.states_bidder = extract_states_from_xu_file(os.path.join(diagrams_dir, 'xmr.bidder.alt.xu'), 'B')
cls.states_offerer = extract_states_from_xu_file(os.path.join(diagrams_dir, 'xmr.offerer.alt.xu'), 'O')
try:
logging.info('Preparing coin nodes.')
for i in range(NUM_NODES):
data_dir = prepareDataDir(TEST_DIR, i, 'particl.conf', 'part_')
if os.path.exists(os.path.join(cfg.PARTICL_BINDIR, 'particl-wallet')):
callrpc_cli(cfg.PARTICL_BINDIR, data_dir, 'regtest', '-wallet=wallet.dat create', 'particl-wallet')
if not cls.restore_instance:
data_dir = prepareDataDir(TEST_DIR, i, 'particl.conf', 'part_')
if os.path.exists(os.path.join(cfg.PARTICL_BINDIR, 'particl-wallet')):
callrpc_cli(cfg.PARTICL_BINDIR, data_dir, 'regtest', '-wallet=wallet.dat create', 'particl-wallet')
cls.part_daemons.append(startDaemon(os.path.join(TEST_DIR, 'part_' + str(i)), cfg.PARTICL_BINDIR, cfg.PARTICLD))
logging.info('Started %s %d', cfg.PARTICLD, cls.part_daemons[-1].pid)
for i in range(NUM_NODES):
# Load mnemonics after all nodes have started to avoid staking getting stuck in TryToSync
rpc = make_rpc_func(i)
waitForRPC(rpc)
if i == 0:
rpc('extkeyimportmaster', ['abandon baby cabbage dad eager fabric gadget habit ice kangaroo lab absorb'])
elif i == 1:
rpc('extkeyimportmaster', ['pact mammal barrel matrix local final lecture chunk wasp survey bid various book strong spread fall ozone daring like topple door fatigue limb olympic', '', 'true'])
rpc('getnewextaddress', ['lblExtTest'])
rpc('rescanblockchain')
else:
rpc('extkeyimportmaster', [rpc('mnemonic', ['new'])['master']])
# Lower output split threshold for more stakeable outputs
rpc('walletsettings', ['stakingoptions', {'stakecombinethreshold': 100, 'stakesplitthreshold': 200}])
if not cls.restore_instance:
for i in range(NUM_NODES):
# Load mnemonics after all nodes have started to avoid staking getting stuck in TryToSync
rpc = make_rpc_func(i)
waitForRPC(rpc)
if i == 0:
rpc('extkeyimportmaster', ['abandon baby cabbage dad eager fabric gadget habit ice kangaroo lab absorb'])
elif i == 1:
rpc('extkeyimportmaster', ['pact mammal barrel matrix local final lecture chunk wasp survey bid various book strong spread fall ozone daring like topple door fatigue limb olympic', '', 'true'])
rpc('getnewextaddress', ['lblExtTest'])
rpc('rescanblockchain')
else:
rpc('extkeyimportmaster', [rpc('mnemonic', ['new'])['master']])
# Lower output split threshold for more stakeable outputs
rpc('walletsettings', ['stakingoptions', {'stakecombinethreshold': 100, 'stakesplitthreshold': 200}])
for i in range(NUM_BTC_NODES):
data_dir = prepareDataDir(TEST_DIR, i, 'bitcoin.conf', 'btc_', base_p2p_port=BTC_BASE_PORT, base_rpc_port=BTC_BASE_RPC_PORT)
if os.path.exists(os.path.join(cfg.BITCOIN_BINDIR, 'bitcoin-wallet')):
callrpc_cli(cfg.BITCOIN_BINDIR, data_dir, 'regtest', '-wallet=wallet.dat create', 'bitcoin-wallet')
if not cls.restore_instance:
data_dir = prepareDataDir(TEST_DIR, i, 'bitcoin.conf', 'btc_', base_p2p_port=BTC_BASE_PORT, base_rpc_port=BTC_BASE_RPC_PORT)
if os.path.exists(os.path.join(cfg.BITCOIN_BINDIR, 'bitcoin-wallet')):
callrpc_cli(cfg.BITCOIN_BINDIR, data_dir, 'regtest', '-wallet=wallet.dat create', 'bitcoin-wallet')
cls.btc_daemons.append(startDaemon(os.path.join(TEST_DIR, 'btc_' + str(i)), cfg.BITCOIN_BINDIR, cfg.BITCOIND))
logging.info('Started %s %d', cfg.BITCOIND, cls.part_daemons[-1].pid)
@@ -414,29 +400,20 @@ class BaseTest(unittest.TestCase):
if cls.start_ltc_nodes:
for i in range(NUM_LTC_NODES):
data_dir = prepareDataDir(TEST_DIR, i, 'litecoin.conf', 'ltc_', base_p2p_port=LTC_BASE_PORT, base_rpc_port=LTC_BASE_RPC_PORT)
if os.path.exists(os.path.join(cfg.LITECOIN_BINDIR, 'litecoin-wallet')):
callrpc_cli(cfg.LITECOIN_BINDIR, data_dir, 'regtest', '-wallet=wallet.dat create', 'litecoin-wallet')
if not cls.restore_instance:
data_dir = prepareDataDir(TEST_DIR, i, 'litecoin.conf', 'ltc_', base_p2p_port=LTC_BASE_PORT, base_rpc_port=LTC_BASE_RPC_PORT)
if os.path.exists(os.path.join(cfg.LITECOIN_BINDIR, 'litecoin-wallet')):
callrpc_cli(cfg.LITECOIN_BINDIR, data_dir, 'regtest', '-wallet=wallet.dat create', 'litecoin-wallet')
cls.ltc_daemons.append(startDaemon(os.path.join(TEST_DIR, 'ltc_' + str(i)), cfg.LITECOIN_BINDIR, cfg.LITECOIND))
logging.info('Started %s %d', cfg.LITECOIND, cls.part_daemons[-1].pid)
waitForRPC(make_rpc_func(i, base_rpc_port=LTC_BASE_RPC_PORT))
if cls.start_pivx_nodes:
for i in range(NUM_PIVX_NODES):
data_dir = prepareDataDir(TEST_DIR, i, 'pivx.conf', 'pivx_', base_p2p_port=PIVX_BASE_PORT, base_rpc_port=PIVX_BASE_RPC_PORT)
if os.path.exists(os.path.join(cfg.PIVX_BINDIR, 'pivx-wallet')):
callrpc_cli(cfg.PIVX_BINDIR, data_dir, 'regtest', '-wallet=wallet.dat create', 'pivx-wallet')
cls.pivx_daemons.append(startDaemon(os.path.join(TEST_DIR, 'pivx_' + str(i)), cfg.PIVX_BINDIR, cfg.PIVXD))
logging.info('Started %s %d', cfg.PIVXD, cls.part_daemons[-1].pid)
waitForRPC(make_rpc_func(i, base_rpc_port=PIVX_BASE_RPC_PORT))
if cls.start_xmr_nodes:
for i in range(NUM_XMR_NODES):
prepareXmrDataDir(TEST_DIR, i, 'monerod.conf')
if not cls.restore_instance:
prepareXmrDataDir(TEST_DIR, i, 'monerod.conf')
cls.xmr_daemons.append(startXmrDaemon(os.path.join(TEST_DIR, 'xmr_' + str(i)), cfg.XMR_BINDIR, cfg.XMRD))
logging.info('Started %s %d', cfg.XMRD, cls.xmr_daemons[-1].pid)
@@ -450,14 +427,20 @@ class BaseTest(unittest.TestCase):
waitForXMRWallet(i, cls.xmr_wallet_auth[i])
cls.callxmrnodewallet(cls, i, 'create_wallet', {'filename': 'testwallet', 'language': 'English'})
if not cls.restore_instance:
cls.callxmrnodewallet(cls, i, 'create_wallet', {'filename': 'testwallet', 'language': 'English'})
cls.callxmrnodewallet(cls, i, 'open_wallet', {'filename': 'testwallet'})
for i in range(NUM_NODES):
# Hook for descendant classes
cls.prepareExtraDataDir(i)
logging.info('Preparing swap clients.')
eckey = ECKey()
eckey.generate()
cls.network_key = toWIF(PREFIX_SECRET_KEY_REGTEST, eckey.get_bytes())
cls.network_pubkey = eckey.get_pubkey().get_bytes().hex()
if not cls.restore_instance:
eckey = ECKey()
eckey.generate()
cls.network_key = toWIF(PREFIX_SECRET_KEY_REGTEST, eckey.get_bytes())
cls.network_pubkey = eckey.get_pubkey().get_bytes().hex()
for i in range(NUM_NODES):
start_nodes = set()
@@ -465,13 +448,15 @@ class BaseTest(unittest.TestCase):
start_nodes.add(Coins.LTC)
if cls.start_xmr_nodes:
start_nodes.add(Coins.XMR)
if cls.start_pivx_nodes:
start_nodes.add(Coins.PIVX)
prepare_swapclient_dir(TEST_DIR, i, cls.network_key, cls.network_pubkey, start_nodes)
if not cls.restore_instance:
prepare_swapclient_dir(TEST_DIR, i, cls.network_key, cls.network_pubkey, start_nodes, cls)
basicswap_dir = os.path.join(os.path.join(TEST_DIR, 'basicswap_' + str(i)))
settings_path = os.path.join(basicswap_dir, cfg.CONFIG_FILENAME)
with open(settings_path) as fs:
settings = json.load(fs)
if cls.restore_instance and i == 1:
cls.network_key = settings['network_key']
cls.network_pubkey = settings['network_pubkey']
fp = open(os.path.join(basicswap_dir, 'basicswap.log'), 'w')
sc = BasicSwap(fp, basicswap_dir, settings, 'regtest', log_name='BasicSwap{}'.format(i))
sc.setDaemonPID(Coins.BTC, cls.btc_daemons[i].pid)
@@ -479,6 +464,7 @@ class BaseTest(unittest.TestCase):
if cls.start_ltc_nodes:
sc.setDaemonPID(Coins.LTC, cls.ltc_daemons[i].pid)
cls.addPIDInfo(sc, i)
sc.start()
if cls.start_xmr_nodes:
@@ -490,74 +476,75 @@ class BaseTest(unittest.TestCase):
t = HttpThread(cls.swap_clients[i].fp, TEST_HTTP_HOST, TEST_HTTP_PORT + i, False, cls.swap_clients[i])
cls.http_threads.append(t)
t.start()
# Set future block rewards to nowhere (a random address), so wallet amounts stay constant
eckey = ECKey()
eckey.generate()
void_block_rewards_pubkey = eckey.get_pubkey().get_bytes()
void_block_rewards_pubkey = cls.getRandomPubkey()
if cls.restore_instance:
cls.btc_addr = cls.swap_clients[0].ci(Coins.BTC).pubkey_to_segwit_address(void_block_rewards_pubkey)
if cls.start_ltc_nodes:
cls.ltc_addr = cls.swap_clients[0].ci(Coins.LTC).pubkey_to_address(void_block_rewards_pubkey)
if cls.start_xmr_nodes:
cls.xmr_addr = cls.callxmrnodewallet(cls, 1, 'get_address')['address']
else:
cls.btc_addr = callnoderpc(0, 'getnewaddress', ['mining_addr', 'bech32'], base_rpc_port=BTC_BASE_RPC_PORT)
num_blocks = 400 # Mine enough to activate segwit
logging.info('Mining %d Bitcoin blocks to %s', num_blocks, cls.btc_addr)
callnoderpc(0, 'generatetoaddress', [num_blocks, cls.btc_addr], base_rpc_port=BTC_BASE_RPC_PORT)
cls.btc_addr = callnoderpc(0, 'getnewaddress', ['mining_addr', 'bech32'], base_rpc_port=BTC_BASE_RPC_PORT)
num_blocks = 400 # Mine enough to activate segwit
logging.info('Mining %d Bitcoin blocks to %s', num_blocks, cls.btc_addr)
callnoderpc(0, 'generatetoaddress', [num_blocks, cls.btc_addr], base_rpc_port=BTC_BASE_RPC_PORT)
# Switch addresses so wallet amounts stay constant
num_blocks = 100
cls.btc_addr = cls.swap_clients[0].ci(Coins.BTC).pubkey_to_segwit_address(void_block_rewards_pubkey)
logging.info('Mining %d Bitcoin blocks to %s', num_blocks, cls.btc_addr)
callnoderpc(0, 'generatetoaddress', [num_blocks, cls.btc_addr], base_rpc_port=BTC_BASE_RPC_PORT)
checkForks(callnoderpc(0, 'getblockchaininfo', base_rpc_port=BTC_BASE_RPC_PORT))
if cls.start_ltc_nodes:
num_blocks = 400
cls.ltc_addr = callnoderpc(0, 'getnewaddress', ['mining_addr', 'bech32'], base_rpc_port=LTC_BASE_RPC_PORT)
logging.info('Mining %d Litecoin blocks to %s', num_blocks, cls.ltc_addr)
callnoderpc(0, 'generatetoaddress', [num_blocks, cls.ltc_addr], base_rpc_port=LTC_BASE_RPC_PORT)
num_blocks = 31
cls.ltc_addr = cls.swap_clients[0].ci(Coins.LTC).pubkey_to_address(void_block_rewards_pubkey)
logging.info('Mining %d Litecoin blocks to %s', num_blocks, cls.ltc_addr)
callnoderpc(0, 'generatetoaddress', [num_blocks, cls.ltc_addr], base_rpc_port=LTC_BASE_RPC_PORT)
# https://github.com/litecoin-project/litecoin/issues/807
# Block 432 is when MWEB activates. It requires a peg-in. You'll need to generate an mweb address and send some coins to it. Then it will allow you to mine the next block.
mweb_addr = callnoderpc(2, 'getnewaddress', ['mweb_addr', 'mweb'], base_rpc_port=LTC_BASE_RPC_PORT)
callnoderpc(0, 'sendtoaddress', [mweb_addr, 1], base_rpc_port=LTC_BASE_RPC_PORT)
num_blocks = 69
cls.ltc_addr = cls.swap_clients[0].ci(Coins.LTC).pubkey_to_address(void_block_rewards_pubkey)
callnoderpc(0, 'generatetoaddress', [num_blocks, cls.ltc_addr], base_rpc_port=LTC_BASE_RPC_PORT)
checkForks(callnoderpc(0, 'getblockchaininfo', base_rpc_port=LTC_BASE_RPC_PORT))
if cls.start_pivx_nodes:
num_blocks = 400
cls.pivx_addr = callnoderpc(0, 'getnewaddress', ['mining_addr'], base_rpc_port=PIVX_BASE_RPC_PORT)
logging.info('Mining %d PIVX blocks to %s', num_blocks, cls.pivx_addr)
callnoderpc(0, 'generatetoaddress', [num_blocks, cls.pivx_addr], base_rpc_port=PIVX_BASE_RPC_PORT)
btc_addr1 = callnoderpc(1, 'getnewaddress', ['initial addr'], base_rpc_port=BTC_BASE_RPC_PORT)
for i in range(5):
callnoderpc(0, 'sendtoaddress', [btc_addr1, 100], base_rpc_port=BTC_BASE_RPC_PORT)
# Switch addresses so wallet amounts stay constant
num_blocks = 100
cls.pivx_addr = cls.swap_clients[0].ci(Coins.PIVX).pubkey_to_address(void_block_rewards_pubkey)
logging.info('Mining %d PIVX blocks to %s', num_blocks, cls.pivx_addr)
callnoderpc(0, 'generatetoaddress', [num_blocks, cls.pivx_addr], base_rpc_port=PIVX_BASE_RPC_PORT)
cls.btc_addr = cls.swap_clients[0].ci(Coins.BTC).pubkey_to_segwit_address(void_block_rewards_pubkey)
logging.info('Mining %d Bitcoin blocks to %s', num_blocks, cls.btc_addr)
callnoderpc(0, 'generatetoaddress', [num_blocks, cls.btc_addr], base_rpc_port=BTC_BASE_RPC_PORT)
num_blocks = 100
if cls.start_xmr_nodes:
cls.xmr_addr = cls.callxmrnodewallet(cls, 1, 'get_address')['address']
if callrpc_xmr_na(XMR_BASE_RPC_PORT + 1, 'get_block_count')['count'] < num_blocks:
logging.info('Mining %d Monero blocks to %s.', num_blocks, cls.xmr_addr)
callrpc_xmr_na(XMR_BASE_RPC_PORT + 1, 'generateblocks', {'wallet_address': cls.xmr_addr, 'amount_of_blocks': num_blocks})
logging.info('XMR blocks: %d', callrpc_xmr_na(XMR_BASE_RPC_PORT + 1, 'get_block_count')['count'])
checkForks(callnoderpc(0, 'getblockchaininfo', base_rpc_port=BTC_BASE_RPC_PORT))
logging.info('Adding anon outputs')
outputs = []
for i in range(8):
sx_addr = callnoderpc(1, 'getnewstealthaddress')
outputs.append({'address': sx_addr, 'amount': 0.5})
for i in range(6):
callnoderpc(0, 'sendtypeto', ['part', 'anon', outputs])
if cls.start_ltc_nodes:
num_blocks = 400
cls.ltc_addr = callnoderpc(0, 'getnewaddress', ['mining_addr', 'bech32'], base_rpc_port=LTC_BASE_RPC_PORT)
logging.info('Mining %d Litecoin blocks to %s', num_blocks, cls.ltc_addr)
callnoderpc(0, 'generatetoaddress', [num_blocks, cls.ltc_addr], base_rpc_port=LTC_BASE_RPC_PORT)
num_blocks = 31
cls.ltc_addr = cls.swap_clients[0].ci(Coins.LTC).pubkey_to_address(void_block_rewards_pubkey)
logging.info('Mining %d Litecoin blocks to %s', num_blocks, cls.ltc_addr)
callnoderpc(0, 'generatetoaddress', [num_blocks, cls.ltc_addr], base_rpc_port=LTC_BASE_RPC_PORT)
# https://github.com/litecoin-project/litecoin/issues/807
# Block 432 is when MWEB activates. It requires a peg-in. You'll need to generate an mweb address and send some coins to it. Then it will allow you to mine the next block.
mweb_addr = callnoderpc(2, 'getnewaddress', ['mweb_addr', 'mweb'], base_rpc_port=LTC_BASE_RPC_PORT)
callnoderpc(0, 'sendtoaddress', [mweb_addr, 1], base_rpc_port=LTC_BASE_RPC_PORT)
num_blocks = 69
cls.ltc_addr = cls.swap_clients[0].ci(Coins.LTC).pubkey_to_address(void_block_rewards_pubkey)
callnoderpc(0, 'generatetoaddress', [num_blocks, cls.ltc_addr], base_rpc_port=LTC_BASE_RPC_PORT)
checkForks(callnoderpc(0, 'getblockchaininfo', base_rpc_port=LTC_BASE_RPC_PORT))
num_blocks = 100
if cls.start_xmr_nodes:
cls.xmr_addr = cls.callxmrnodewallet(cls, 1, 'get_address')['address']
if callrpc_xmr_na(XMR_BASE_RPC_PORT + 1, 'get_block_count')['count'] < num_blocks:
logging.info('Mining %d Monero blocks to %s.', num_blocks, cls.xmr_addr)
callrpc_xmr_na(XMR_BASE_RPC_PORT + 1, 'generateblocks', {'wallet_address': cls.xmr_addr, 'amount_of_blocks': num_blocks})
logging.info('XMR blocks: %d', callrpc_xmr_na(XMR_BASE_RPC_PORT + 1, 'get_block_count')['count'])
logging.info('Adding anon outputs')
outputs = []
for i in range(8):
sx_addr = callnoderpc(1, 'getnewstealthaddress')
outputs.append({'address': sx_addr, 'amount': 0.5})
for i in range(6):
callnoderpc(0, 'sendtypeto', ['part', 'anon', outputs])
part_addr1 = callnoderpc(1, 'getnewaddress', ['initial addr'])
part_addr2 = callnoderpc(1, 'getnewaddress', ['initial addr 2'])
callnoderpc(0, 'sendtypeto', ['part', 'part', [{'address': part_addr1, 'amount': 100}, {'address': part_addr2, 'amount': 100}]])
cls.prepareExtraCoins()
logging.info('Starting update thread.')
signal.signal(signal.SIGINT, signal_handler)
@@ -599,13 +586,40 @@ class BaseTest(unittest.TestCase):
stopDaemons(cls.part_daemons)
stopDaemons(cls.btc_daemons)
stopDaemons(cls.ltc_daemons)
stopDaemons(cls.pivx_daemons)
super(BaseTest, cls).tearDownClass()
@classmethod
def addCoinSettings(cls, settings, datadir, node_id):
pass
@classmethod
def prepareExtraDataDir(cls, i):
pass
@classmethod
def addPIDInfo(cls, sc, i):
pass
@classmethod
def prepareExtraCoins(cls):
pass
@classmethod
def coins_loop(cls):
if cls.btc_addr is not None:
btcCli('generatetoaddress 1 {}'.format(cls.btc_addr))
if cls.ltc_addr is not None:
ltcCli('generatetoaddress 1 {}'.format(cls.ltc_addr))
if cls.xmr_addr is not None:
callrpc_xmr_na(XMR_BASE_RPC_PORT + 1, 'generateblocks', {'wallet_address': cls.xmr_addr, 'amount_of_blocks': 1})
def callxmrnodewallet(self, node_id, method, params=None):
return callrpc_xmr(XMR_BASE_WALLET_RPC_PORT + node_id, self.xmr_wallet_auth[node_id], method, params)
def getXmrBalance(self, js_wallets):
return float(js_wallets[Coins.XMR.name]['unconfirmed']) + float(js_wallets[Coins.XMR.name]['balance'])
class Test(BaseTest):
__test__ = True
@@ -617,9 +631,9 @@ class Test(BaseTest):
logging.info('---------- Test PART to XMR')
swap_clients = self.swap_clients
start_xmr_amount = self.getXmrBalance(read_json_api(1800, 'wallets'))
js_1 = read_json_api(1801, 'wallets')
assert (make_int(js_1[Coins.XMR.name]['balance'], scale=12) > 0)
assert (make_int(js_1[Coins.XMR.name]['unconfirmed'], scale=12) > 0)
assert (self.getXmrBalance(js_1) > 0.0)
offer_id = swap_clients[0].postOffer(Coins.PART, Coins.XMR, 100 * COIN, 0.11 * XMR_COIN, 100 * COIN, SwapTypes.XMR_SWAP)
wait_for_offer(test_delay_event, swap_clients[1], offer_id)
@@ -640,8 +654,9 @@ class Test(BaseTest):
wait_for_bid(test_delay_event, swap_clients[1], bid_id, BidStates.SWAP_COMPLETED, sent=True)
js_0_end = read_json_api(1800, 'wallets')
end_xmr = float(js_0_end['XMR']['balance']) + float(js_0_end['XMR']['unconfirmed'])
assert (end_xmr > 10.9 and end_xmr < 11.0)
end_xmr_amount = self.getXmrBalance(js_0_end)
xmr_amount_diff = end_xmr_amount - start_xmr_amount
assert (xmr_amount_diff > 10.9 and xmr_amount_diff < 11.0)
bid_id_hex = bid_id.hex()
path = f'bids/{bid_id_hex}/states'