#!/usr/bin/env python3 # -*- coding: utf-8 -*- # Copyright (c) 2020 tecnovert # Distributed under the MIT software license, see the accompanying # file LICENSE or http://www.opensource.org/licenses/mit-license.php. import os import sys import unittest import json import logging import shutil import time import signal import threading from urllib.request import urlopen from coincurve.ecdsaotves import ( ecdsaotves_enc_sign, ecdsaotves_enc_verify, ecdsaotves_dec_sig, ecdsaotves_rec_enc_key) from coincurve.dleag import ( dleag_prove, dleag_verify) import basicswap.config as cfg from basicswap.basicswap import ( BasicSwap, Coins, SwapTypes, BidStates, TxStates, SEQUENCE_LOCK_BLOCKS, ) from basicswap.util import ( COIN, toWIF, dumpje, ) from basicswap.rpc import ( callrpc_cli, waitForRPC, ) from basicswap.contrib.key import ( ECKey, ) from basicswap.http_server import ( HttpThread, ) from bin.basicswap_run import startDaemon logger = logging.getLogger() logger.level = logging.DEBUG if not len(logger.handlers): logger.addHandler(logging.StreamHandler(sys.stdout)) NUM_NODES = 3 BASE_PORT = 14792 BASE_RPC_PORT = 19792 BASE_ZMQ_PORT = 20792 PREFIX_SECRET_KEY_REGTEST = 0x2e TEST_HTML_PORT = 1800 stop_test = False def prepareOtherDir(datadir, nodeId, conf_file='litecoin.conf'): node_dir = os.path.join(datadir, str(nodeId)) if not os.path.exists(node_dir): os.makedirs(node_dir) filePath = os.path.join(node_dir, conf_file) with open(filePath, 'w+') as fp: fp.write('regtest=1\n') fp.write('[regtest]\n') fp.write('port=' + str(BASE_PORT + nodeId) + '\n') fp.write('rpcport=' + str(BASE_RPC_PORT + nodeId) + '\n') fp.write('daemon=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.0002\n') fp.write('acceptnonstdtxn=0\n') def prepareDir(datadir, nodeId, network_key, network_pubkey): node_dir = os.path.join(datadir, str(nodeId)) if not os.path.exists(node_dir): os.makedirs(node_dir) filePath = os.path.join(node_dir, 'particl.conf') with open(filePath, 'w+') as fp: fp.write('regtest=1\n') fp.write('[regtest]\n') fp.write('port=' + str(BASE_PORT + nodeId) + '\n') fp.write('rpcport=' + str(BASE_RPC_PORT + nodeId) + '\n') fp.write('daemon=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('zmqpubsmsg=tcp://127.0.0.1:' + str(BASE_ZMQ_PORT + nodeId) + '\n') fp.write('acceptnonstdtxn=0\n') fp.write('minstakeinterval=5\n') for i in range(0, NUM_NODES): if nodeId == i: continue fp.write('addnode=127.0.0.1:%d\n' % (BASE_PORT + i)) if nodeId < 2: fp.write('spentindex=1\n') fp.write('txindex=1\n') basicswap_dir = os.path.join(datadir, str(nodeId), 'basicswap') if not os.path.exists(basicswap_dir): os.makedirs(basicswap_dir) ltcdatadir = os.path.join(datadir, str(LTC_NODE)) btcdatadir = os.path.join(datadir, str(BTC_NODE)) settings_path = os.path.join(basicswap_dir, cfg.CONFIG_FILENAME) settings = { 'zmqhost': 'tcp://127.0.0.1', 'zmqport': BASE_ZMQ_PORT + nodeId, 'htmlhost': 'localhost', 'htmlport': 12700 + nodeId, 'network_key': network_key, 'network_pubkey': network_pubkey, 'chainclients': { 'particl': { 'connection_type': 'rpc', 'manage_daemon': False, 'rpcport': BASE_RPC_PORT + nodeId, 'datadir': node_dir, 'bindir': cfg.PARTICL_BINDIR, 'blocks_confirmed': 2, # Faster testing }, 'litecoin': { 'connection_type': 'rpc', 'manage_daemon': False, 'rpcport': BASE_RPC_PORT + LTC_NODE, 'datadir': ltcdatadir, 'bindir': cfg.LITECOIN_BINDIR, # 'use_segwit': True, }, 'bitcoin': { 'connection_type': 'rpc', 'manage_daemon': False, 'rpcport': BASE_RPC_PORT + BTC_NODE, 'datadir': btcdatadir, 'bindir': cfg.BITCOIN_BINDIR, 'use_segwit': True, } }, 'check_progress_seconds': 2, 'check_watched_seconds': 4, 'check_expired_seconds': 60, 'check_events_seconds': 1, 'min_delay_auto_accept': 1, 'max_delay_auto_accept': 5 } with open(settings_path, 'w') as fp: json.dump(settings, fp, indent=4) def partRpc(cmd, node_id=0): return callrpc_cli(cfg.PARTICL_BINDIR, os.path.join(cfg.TEST_DATADIRS, str(node_id)), 'regtest', cmd, cfg.PARTICL_CLI) def btcRpc(cmd): return callrpc_cli(cfg.BITCOIN_BINDIR, os.path.join(cfg.TEST_DATADIRS, str(BTC_NODE)), 'regtest', cmd, cfg.BITCOIN_CLI) def signal_handler(sig, frame): global stop_test print('signal {} detected.'.format(sig)) stop_test = True def run_loop(self): while not stop_test: time.sleep(1) for c in self.swap_clients: c.update() btcRpc('generatetoaddress 1 {}'.format(self.btc_addr)) def checkForks(ro): if 'bip9_softforks' in ro: assert(ro['bip9_softforks']['csv']['status'] == 'active') assert(ro['bip9_softforks']['segwit']['status'] == 'active') else: assert(ro['softforks']['csv']['active']) assert(ro['softforks']['segwit']['active']) class Test(unittest.TestCase): @classmethod def setUpClass(cls): super(Test, cls).setUpClass() cls.swap_clients = [] cls.xmr_daemons = [] cls.xmr_wallet_auth = [] cls.part_stakelimit = 0 cls.xmr_addr = None signal.signal(signal.SIGINT, signal_handler) cls.update_thread = threading.Thread(target=run_loop, args=(cls,)) cls.update_thread.start() @classmethod def tearDownClass(cls): global stop_test logging.info('Finalising') stop_test = True cls.update_thread.join() super(Test, cls).tearDownClass() def test_01_part_xmr(self): logging.info('---------- Test PART to XMR') #swap_clients = self.swap_clients #offer_id = swap_clients[0].postOffer(Coins.PART, Coins.XMR, 100 * COIN, 0.5 * COIN, 100 * COIN, SwapTypes.SELLER_FIRST) if __name__ == '__main__': unittest.main()